8 class PrhLibrary(object):
9 ROBOT_LIBRARY_SCOPE = 'TEST SUITE'
10 ROBOT_LISTENER_API_VERSION = 2
13 self.ROBOT_LIBRARY_LISTENER = self
15 def _start_test(self, name, attrs):
16 # http://robotframework.org/robotframework/latest/RobotFrameworkUserGuide.html#test-libraries-as-listeners
17 self.test_start_time = self.get_current_utc_datetime()
20 def get_current_utc_datetime():
21 return datetime.datetime.utcnow()
23 def get_docker_logs_since_test_start(self, container_id):
24 return self.get_docker_logs(container_id, self.test_start_time)
27 def get_docker_logs(container_id, since=None):
28 container = PrhLibrary.__get_docker_container(container_id)
29 return container.logs(stream=False, since=since)
31 def wait_for_one_of_docker_log_entries(self, container_id, searched_entries):
32 print("Looking for: %s" % searched_entries)
33 container = PrhLibrary.__get_docker_container(container_id)
35 for line in container.logs(stream=True, since=self.test_start_time):
37 for searched_entry in searched_entries:
38 if searched_entry in line.decode(encoding='utf-8').strip():
43 def wait_for_log_entry_with_json_message(self, prefix, json_message):
45 print("Prefix: " + str(prefix))
46 print("Json: " + str(json_message))
48 decoded_message = json.loads(json_message)
49 except json.JSONDecodeError:
50 print("Could not decode given message")
52 pattern = re.compile(prefix + "(.*)$")
53 container = PrhLibrary.__get_docker_container('prh')
55 for line in container.logs(stream=True, since=self.test_start_time):
57 if PrhLibrary.__same_json_in_log(decoded_message, line, pattern):
63 def create_invalid_notification(json_file):
65 input = json.loads(json_file)
66 output["correlationId"] = PrhLibrary.__extract_correlation_id_value(input)
67 output["oamV4IpAddress"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "oamV4IpAddress")
68 output["oamV6IpAddress"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "oamV6IpAddress")
69 output["serialNumber"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "serialNumber")
70 output["vendorName"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "vendorName")
71 output["modelNumber"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "modelNumber")
72 output["unitType"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "unitType")
73 output['nfNamingCode'] = ''
74 output['softwareVersion'] = ''
76 output["additionalFields"] = PrhLibrary.__get_additional_fields_as_key_value_pairs(input)
78 return json.dumps(output)
81 def create_pnf_ready_notification_as_pnf_ready(json_file):
83 input = json.loads(json_file)[0]
85 output["correlationId"] = PrhLibrary.__extract_correlation_id_value(input)
86 output["serialNumber"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "serialNumber")
87 output["equip-vendor"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "vendorName")
88 output["equip-model"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "modelNumber")
89 output["equip-type"] = PrhLibrary.__extract_value_from_pnfRegistrationFields(input, "unitType")
90 output["nf-role"] = PrhLibrary.__extract_nf_role(input)
91 output["sw-version"] = ""
93 output["additionalFields"] = PrhLibrary.__get_additional_fields_as_key_value_pairs(input)
95 return json.dumps(output)
98 def ensure_container_is_running(name):
99 client = docker.from_env()
101 if not PrhLibrary.is_in_status(client, name, "running"):
102 print ("starting container", name)
103 container = client.containers.get(name)
105 PrhLibrary.wait_for_status(client, name, "running")
107 PrhLibrary.print_status(client)
110 def ensure_container_is_exited(name):
111 client = docker.from_env()
113 if not PrhLibrary.is_in_status(client, name, "exited"):
114 print ("stopping container", name)
115 container = client.containers.get(name)
117 PrhLibrary.wait_for_status(client, name, "exited")
119 PrhLibrary.print_status(client)
122 def print_status(client):
123 print("containers status")
124 for c in client.containers.list(all=True):
125 print(c.name, " ", c.status)
128 def wait_for_status(client, name, status):
129 while not PrhLibrary.is_in_status(client, name, status):
130 print ("waiting for container: ", name, "to be in status: ", status)
134 def is_in_status(client, name, status):
135 return len(client.containers.list(all=True, filters={"name": "^/"+name+"$", "status": status})) == 1
138 def create_pnf_name(json_file):
139 json_to_python = json.loads(json_file)
140 correlation_id = json_to_python.get("event").get("commonEventHeader").get("sourceName") + '",' if "sourceName" in json_to_python["event"]["commonEventHeader"] else '",'
141 return correlation_id
144 def __get_additional_fields_as_key_value_pairs(content):
145 return content.get("event").get("pnfRegistrationFields").get(
146 "additionalFields") if "additionalFields" in content["event"]["pnfRegistrationFields"] else {}
149 def __extract_value_from_pnfRegistrationFields(content, key):
150 return content["event"]["pnfRegistrationFields"][key] if key in content["event"]["pnfRegistrationFields"] else ''
153 def __extract_correlation_id_value(content):
154 return content["event"]["commonEventHeader"]["sourceName"] if "sourceName" in content["event"]["commonEventHeader"] else ''
157 def __extract_nf_role(content):
158 return content["event"]["commonEventHeader"]["nfNamingCode"] if "nfNamingCode" in content["event"]["commonEventHeader"] else ''
161 def __same_json_in_log(decoded_message, line, pattern):
162 extracted_json = PrhLibrary.__extract_json(line, pattern)
163 if extracted_json is not None:
164 print("Found json: " + extracted_json)
166 if json.loads(extracted_json) == decoded_message:
168 except json.JSONDecodeError:
169 print("Could not decode")
173 def __extract_json(line, pattern):
174 full_message = PrhLibrary.__extract_full_message_from_line(line)
175 if full_message is not None:
176 match = pattern.match(full_message.decode(encoding='utf-8'))
178 return match.group(1).replace("\\n", "\n").replace("\\t", "\t")
182 def __extract_full_message_from_line(line):
183 split = line.split(bytes('|', 'utf-8'))
189 def __get_docker_container(container_id):
190 docker_client = docker.from_env()
191 return docker_client.containers.get(container_id)