7 class PrhLibrary(object):
13 def find_one_of_log_entryies(searched_entries):
14 print(type(searched_entries))
15 client = docker.from_env()
16 container = client.containers.get('prh')
17 print("Check for log searches for pattern: ", searched_entries)
18 for line in container.logs(stream=True):
19 print("Check for log analysis line: ", line )
20 for searched_entry in searched_entries:
21 if searched_entry in line.strip():
27 def find_log_json(prefix, json_message):
29 print("Prefix: " + str(prefix))
30 print("Json: " + str(json_message))
32 decoded_message = json.loads(json_message)
33 except json.JSONDecodeError:
34 print("Could not decode given message")
36 pattern = re.compile(prefix + "(.*)$")
37 client = docker.from_env()
38 container = client.containers.get('prh')
39 for line in container.logs(stream=True):
40 print("Check for log analysis line: ", line )
41 if PrhLibrary.__same_json_in_log(decoded_message, line, pattern):
47 def create_invalid_notification(json_file):
49 input = json.loads(json_file)
50 output["correlationId"] = PrhLibrary.__extract_correlation_id_value(input)
51 output["oamV4IpAddress"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "oamV4IpAddress")
52 output["oamV6IpAddress"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "oamV6IpAddress")
53 output["serialNumber"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "serialNumber")
54 output["vendorName"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "vendorName")
55 output["modelNumber"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "modelNumber")
56 output["unitType"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "unitType")
57 output['nfNamingCode'] = ''
58 output['softwareVersion'] = ''
60 output["additionalFields"] = PrhLibrary.__get_additional_fields_as_key_value_pairs(input)
62 return json.dumps(output)
65 def create_pnf_ready_notification_as_pnf_ready(json_file):
67 input = json.loads(json_file)[0]
69 output["correlationId"] = PrhLibrary.__extract_correlation_id_value(input)
70 output["serialNumber"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "serialNumber")
71 output["equip-vendor"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "vendorName")
72 output["equip-model"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "modelNumber")
73 output["equip-type"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "unitType")
74 output["nf-role"] = PrhLibrary.__extract_nf_role(input)
75 output["sw-version"] = ""
77 output["additionalFields"] = PrhLibrary.__get_additional_fields_as_key_value_pairs(input)
79 return json.dumps(output)
82 def ensure_container_is_running(name):
83 client = docker.from_env()
85 if not PrhLibrary.is_in_status(client, name, "running"):
86 print ("starting container", name)
87 container = client.containers.get(name)
89 PrhLibrary.wait_for_status(client, name, "running")
91 PrhLibrary.print_status(client)
94 def ensure_container_is_exited(name):
95 client = docker.from_env()
97 if not PrhLibrary.is_in_status(client, name, "exited"):
98 print ("stopping container", name)
99 container = client.containers.get(name)
101 PrhLibrary.wait_for_status(client, name, "exited")
103 PrhLibrary.print_status(client)
106 def print_status(client):
107 print("containers status")
108 for c in client.containers.list(all=True):
109 print(c.name, " ", c.status)
112 def wait_for_status(client, name, status):
113 while not PrhLibrary.is_in_status(client, name, status):
114 print ("waiting for container: ", name, "to be in status: ", status)
118 def is_in_status(client, name, status):
119 return len(client.containers.list(all=True, filters={"name": "^/"+name+"$", "status": status})) == 1
122 def create_pnf_name(json_file):
123 json_to_python = json.loads(json_file)
124 correlation_id = json_to_python.get("event").get("commonEventHeader").get("sourceName") + '",' if "sourceName" in json_to_python["event"]["commonEventHeader"] else '",'
125 return correlation_id
128 def __get_additional_fields_as_key_value_pairs(content):
129 return content.get("event").get("pnfRegistrationFields").get(
130 "additionalFields") if "additionalFields" in content["event"]["pnfRegistrationFields"] else {}
133 def __extract_value_from_pnfRegistrationFields(content, key):
134 return content["event"]["pnfRegistrationFields"][key] if key in content["event"]["pnfRegistrationFields"] else ''
137 def __extract_correlation_id_value(content):
138 return content["event"]["commonEventHeader"]["sourceName"] if "sourceName" in content["event"]["commonEventHeader"] else ''
141 def __extract_nf_role(content):
142 return content["event"]["commonEventHeader"]["nfNamingCode"] if "nfNamingCode" in content["event"]["commonEventHeader"] else ''
145 def __same_json_in_log(decoded_message, line, pattern):
146 extracted_json = PrhLibrary.__extract_json(line, pattern)
147 if extracted_json is not None:
148 print("Found json: " + extracted_json)
150 if json.loads(extracted_json) == decoded_message:
152 except json.JSONDecodeError:
153 print("Could not decode")
157 def __extract_json(line, pattern):
158 full_message = PrhLibrary.__extract_full_message_from_line(line)
159 if full_message is not None:
160 match = pattern.match(full_message)
162 return match.group(1).replace("\\n", "\n").replace("\\t", "\t")
166 def __extract_full_message_from_line(line):
167 split = line.split("|")