Source code for nomnomdata.engine.testing

import json
import logging
import re
from os import environ
from pprint import pformat

from httmock import HTTMock, urlmatch

from .globals import current_engine
from .nominode import NominodeContext


class EngineContextMock:
    def __init__(self, engine=None, task_parameters=None):
        from .engine import Engine

        self.engine = engine or Engine(uuid="TEST-ENGINE", alias="TEST", mock=True)
        task_parameters = task_parameters.copy() if task_parameters else {}
        self.engine._current_parameters = task_parameters
        self.nominode_ctx = NominodeContextMock(task_parameters=task_parameters)

    def __enter__(self):
        current_engine._set(self.engine)
        self.nominode_ctx.__enter__()
        return self

    def __exit__(self, *args):
        current_engine._set(None)
        self.nominode_ctx.__exit__(*args)


[docs]class ExecutionContextMock(EngineContextMock): def __enter__(self): super().__enter__() self.nominode_context = NominodeContext.from_env() self.nominode_context.__enter__() return self def __exit__(self, *args): self.nominode_context.__exit__(*args) super().__exit__(*args)
class NominodeContextMock(HTTMock): def __init__(self, task_parameters=None): super().__init__(self.api_match) self.logger = logging.getLogger("nomigen.nominode-mock") task_parameters = task_parameters or {} self.secrets = { k: {**secret, "alias": f"Connection Alias {i}"} if secret else {"alias": f"Connection {i}"} for i, (k, secret) in enumerate(task_parameters.pop("config", {}).items()) } self.secrets = self.secrets if self.secrets else {"result": None} self.params = {**task_parameters, "alias": "Testing Task Alias"} self.calls = [] self.environ = None def __enter__(self): self.environ = environ.copy() environ["execution_uuid"] = "TEST_UUID" environ["task_uuid"] = "TASK_UUID" environ["project_uuid"] = "TEST_PROJECT" environ["nomnom_api"] = "http://127.0.0.1:9090" environ["token"] = "token" environ["NND_LOG_LEVEL"] = logging.getLevelName(logging.getLogger().level) super().__enter__() return self def __exit__(self, *args): super().__exit__(*args) environ.pop("execution_uuid") environ.pop("task_uuid") environ.pop("project_uuid") environ.pop("nomnom_api") environ.pop("token") @urlmatch(netloc=r"(.*\.)?127.0.0.1:9090$") def api_match(self, url, request): self.calls.append((url.path, json.loads(request.body))) match = re.match(r"/connection/(?P<uuid>.+)/update", url.path) if match: json_data = request.body loaded = json.loads(json_data) uuid = match.groupdict()["uuid"] self.params["config"][uuid] = json.loads(loaded["parameters"]) self.logger.debug("Caught connections update. Test creds updated") elif url.path == "/execution/log/TEST_UUID": json_data = request.body loaded = json.loads(json_data) elif url.path == "/task/TASK_UUID/update": json_data = request.body loaded = json.loads(json_data) self.logger.debug("Caught task update {}".format(pformat(loaded))) elif url.path == "/execution/update/TEST_UUID": json_data = request.body loaded = json.loads(json_data) self.logger.debug( "Caught execution progress update {}".format(pformat(loaded)) ) elif url.path == "/execution/decode/TEST_UUID": return json.dumps(self.secrets) elif url.path == "/task/TASK_UUID/parameters": json_data = request.body loaded = json.loads(json_data) self.logger.debug("Caught task parameter update {}".format(pformat(loaded))) return json.dumps({"result": "success"}) elif url.path == "/execution/checkout/TEST_UUID": return json.dumps({"parameters": self.params, "task_uuid": "TASK_UUID"}) else: self.logger.info( f"Unknown api endpoint called {url.path}, \n Body {request.body}" ) return '{"you_logged":"test"}'