Underground-Digital/Workflow-Engine

1import os2from typing import Literal3 4import pytest5from _pytest.monkeypatch import MonkeyPatch6from jinja2 import Template7 8from core.helper.code_executor.code_executor import CodeExecutor, CodeLanguage9 10MOCK = os.getenv("MOCK_SWITCH", "false") == "true"11 12 13class MockedCodeExecutor:14    @classmethod15    def invoke(cls, language: Literal["python3", "javascript", "jinja2"], code: str, inputs: dict) -> dict:16        # invoke directly17        match language:18            case CodeLanguage.PYTHON3:19                return {"result": 3}20            case CodeLanguage.JINJA2:21                return {"result": Template(code).render(inputs)}22            case _:23                raise Exception("Language not supported")24 25 26@pytest.fixture27def setup_code_executor_mock(request, monkeypatch: MonkeyPatch):28    if not MOCK:29        yield30        return31 32    monkeypatch.setattr(CodeExecutor, "execute_workflow_code_template", MockedCodeExecutor.invoke)33    yield34    monkeypatch.undo()35