Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
from dis import dis
import requests
import json
import configparser
import redis
import os
import argparse
import re
from termcolor import colored
# Get environment variables
class InvokerToService():
def __demo_to_aef(self, demo_ip, demo_port, method, demo_url, jwt_token, payload_dict, route, log_level):
url = "http://{}:{}{}".format(demo_ip, demo_port, demo_url)
#url = "http://python_aef:8086/hello"
payload = json.dumps(route, indent=2)
files = {}
headers = {
'Content-Type': 'application/json',
'Authorization': 'Bearer '+jwt_token
}
#print(colored(f"Request Body: {payload}", "yellow"))
try:
if log_level == "debug":
print(colored("''''''''''REQUEST'''''''''''''''''","yellow"))
print(colored(f"Request: to {url}","yellow"))
print(colored(f"Request Headers: {headers}", "yellow"))
print(colored(f"Request Body: {payload}", "yellow"))
print(colored("''''''''''REQUEST'''''''''''''''''\n", "yellow"))
response = requests.request(method.upper(), url, headers=headers, data=payload, files=files, cert=('capif_ops/certs/dummy.crt', 'capif_ops/certs/invoker_private_key.key'), verify=False)
response.raise_for_status()
response_payload = json.loads(response.text)
if log_level == "debug":
print(colored("\n''''''''''RESPONSE'''''''''''''''''","green"))
print(colored(f"Response to: {response.url}","green"))
print(colored(f"Response Headers: {response.headers}","green"))
print(colored(f"Response: {json.dumps(response.json(), indent=2)}","green"))
print(colored(f"Response Status code: {response.status_code}","green"))
print(colored("Success to invoke service","green"))
print(colored("''''''''''RESPONSE'''''''''''''''''\n","green"))
return response_payload
except requests.exceptions.HTTPError as err:
if err.response.status_code == 404:
print(colored("\nUrl not found", "red"))
if err.response.status_code == 405:
print(colored("\nMethod Not Allowed", "red"))
message = json.loads(err.response.text)
status = err.response.status_code
raise Exception(message, status)
def execute_invoker_to_service(self, api_name, method, endpoint, json_route, log_level):
# parser = argparse.ArgumentParser()
# parser.add_argument('--name', metavar= "name", type=str, default="Evolve5G", help="Name to send to the aef service")
# args = parser.parse_args()
input_name = "prueba"
with open('capif_ops/config_files/demo_values.json', 'r') as demo_file:
demo_values = json.load(demo_file)
try:
payload = open(json_route, 'rb')
payload_dict = json.load(payload)
if 'netapp_service_token' in demo_values:
print(colored("\nCalling API","yellow"))
count=-1
for clave, valor in demo_values.items():
if valor == api_name:
count=int(re.match(r'api_name_(\d+)', clave).group(1))
break
if count == -1:
print(colored("\nAPI not found\n","red"))
return
jwt_token = demo_values['netapp_service_token']
invokerID = demo_values['invokerID']
demo_ip = demo_values[f"demo_ipv4_addr_{count}"]
demo_port = demo_values[f"demo_port_{count}"]
result = self.__demo_to_aef(demo_ip, demo_port, method, endpoint, jwt_token, input_name, payload_dict, log_level)
if log_level != "debug":
print(colored(f"\nResponse: {json.dumps(result, indent=2)}", "green"))
print(colored("\nSuccess!!\n","green"))
except Exception as e:
status_code = e.args[0]
if status_code == 401:
print(colored("\nAPI Invoker is not authorized\n", "red"))
elif status_code == 403:
print(colored("\nAPI Invoker does not exist. API Invoker id not found\n", "red"))
elif status_code == "Expecting value: line 1 column 1 (char 0)":
print()
else:
print(colored(f"\n{e}\n", "red"))