import re from flask import current_app, Flask from typing import List class OnlineHost: def __init__(self, id: str, url: str): self.id = id self.url = url # I decided to just use dict everywhere instead because I have to use dict to read it from json # class SSHHostKey: # def __init__(self, key_type=None, content=None, sha256=None): # self.key_type = key_type # self.content = content # self.sha256 = sha256 class VirtualMachine: def __init__(self, id, host, ipv4=None, ipv6=None, state="unknown", ssh_host_keys: List[dict] = list()): self.id = id self.host = host self.ipv4 = ipv4 self.ipv6 = ipv6 self.state = state self.ssh_host_keys = ssh_host_keys class VirtualizationInterface: def capacity_avaliable(self, additional_ram_bytes: int) -> bool: pass def get(self, id: str) -> VirtualMachine: pass def list_ids(self) -> list: pass def create(self, email: str, id: str, template_image_file_name: str, vcpus: int, memory: int, ssh_authorized_keys: list): pass def destroy(self, email: str, id: str): pass def vm_state_command(self, email: str, id: str, command: str): pass def validate_capsul_id(id): if not re.match(r"^(cvm|capsul)-[a-z0-9]{10}$", id): raise ValueError(f"vm id \"{id}\" must match \"^capsul-[a-z0-9]{{10}}$\"") def authorized_as_hub(headers): if headers.get('Authorization'): auth_header_value = headers.get('Authorization').replace("Bearer ", "") return auth_header_value == current_app.config["HUB_TOKEN"] return False def my_exec_info_message(exec_info): return "{}: {}".format(".".join([exec_info[0].__module__, exec_info[0].__name__]), exec_info[1]) mylog_current_test_id_container = { 'value': '', } def set_mylog_test_id(test_id): mylog_current_test_id_container['value'] = ".".join(test_id.split(".")[-2:]) def log_output_for_tests(app: Flask, message: str): if app.config['TESTING'] and mylog_current_test_id_container['value'] != "": file_object = open('unittest-log-output.log', 'a') file_object.write(f"{mylog_current_test_id_container['value']}: {message}\n") file_object.close() def mylog_debug(app: Flask, message: str): log_output_for_tests(app, f"DEBUG: {message}") app.logger.debug(message) def mylog_info(app: Flask, message: str): log_output_for_tests(app, f"INFO: {message}") app.logger.info(message) def mylog_warning(app: Flask, message: str): log_output_for_tests(app, f"WARNING: {message}") app.logger.warning(message) def mylog_error(app: Flask, message: str): log_output_for_tests(app, f"ERROR: {message}") app.logger.error(message) def mylog_critical(app: Flask, message: str): log_output_for_tests(app, f"CRITICAL: {message}") app.logger.critical(message)