- def check_for_log(search_for):
- client = docker.from_env()
- container = client.containers.get('prh')
- for line in container.logs(stream=True):
- if search_for in line.strip():
+ def get_current_utc_datetime():
+ return datetime.datetime.utcnow()
+
+ def get_docker_logs_since_test_start(self, container_id):
+ return self.get_docker_logs(container_id, self.test_start_time)
+
+ @staticmethod
+ def get_docker_logs(container_id, since=None):
+ container = PrhLibrary.__get_docker_container(container_id)
+ return container.logs(stream=False, since=since)
+
+ def wait_for_one_of_docker_log_entries(self, container_id, searched_entries):
+ print("Looking for: %s" % searched_entries)
+ container = PrhLibrary.__get_docker_container(container_id)
+ print("Log lines:")
+ for line in container.logs(stream=True, since=self.test_start_time):
+ print(line)
+ for searched_entry in searched_entries:
+ if searched_entry in line.decode(encoding='utf-8').strip():
+ return True
+ else:
+ return False
+
+ def wait_for_log_entry_with_json_message(self, prefix, json_message):
+ print("Looking for:")
+ print("Prefix: " + str(prefix))
+ print("Json: " + str(json_message))
+ try:
+ decoded_message = json.loads(json_message)
+ except json.JSONDecodeError:
+ print("Could not decode given message")
+ return False
+ pattern = re.compile(prefix + "(.*)$")
+ container = PrhLibrary.__get_docker_container('prh')
+ print("Log lines:")
+ for line in container.logs(stream=True, since=self.test_start_time):
+ print(line)
+ if PrhLibrary.__same_json_in_log(decoded_message, line, pattern):