test_engine_runtime_registry.py 3.44 KB
from __future__ import annotations

from apps.web_api.runtime.engine_registry import EngineRuntimeRegistry, build_default_engine_catalog


def test_default_engine_catalog_covers_mainline_engines_and_launch_metadata():
    registry = EngineRuntimeRegistry(build_default_engine_catalog())

    assert registry.snapshot_all() == {
        "insight": {
            "name": "insight",
            "available": True,
            "launch_mode": "streamlit",
            "script_path": "apps/engine_console/insight_engine_streamlit_app.py",
            "local_runner": "insight",
        },
        "media": {
            "name": "media",
            "available": True,
            "launch_mode": "streamlit",
            "script_path": "apps/engine_console/media_engine_streamlit_app.py",
            "local_runner": "media",
        },
        "query": {
            "name": "query",
            "available": True,
            "launch_mode": "streamlit",
            "script_path": "apps/engine_console/query_engine_streamlit_app.py",
            "local_runner": "query",
        },
        "forum": {
            "name": "forum",
            "available": True,
            "launch_mode": "callable",
            "script_path": None,
            "local_runner": None,
        },
    }


def test_snapshot_helpers_clone_entries_and_filter_streamlit_scripts():
    registry = EngineRuntimeRegistry(build_default_engine_catalog())

    snapshot = registry.snapshot("query")
    snapshot["script_path"] = "mutated.py"

    assert registry.snapshot("query") == {
        "name": "query",
        "available": True,
        "launch_mode": "streamlit",
        "script_path": "apps/engine_console/query_engine_streamlit_app.py",
        "local_runner": "query",
    }
    assert registry.engines_for_launch_mode("streamlit") == ["insight", "media", "query"]
    assert registry.streamlit_script_snapshot() == {
        "insight": "apps/engine_console/insight_engine_streamlit_app.py",
        "media": "apps/engine_console/media_engine_streamlit_app.py",
        "query": "apps/engine_console/query_engine_streamlit_app.py",
    }
    assert registry.streamlit_scripts() == registry.streamlit_script_snapshot()


def test_local_runner_snapshot_omits_missing_engines_and_non_local_entries():
    registry = EngineRuntimeRegistry(build_default_engine_catalog())

    assert registry.local_runner_snapshot() == {
        "insight": "insight",
        "media": "media",
        "query": "query",
    }
    assert registry.local_runner_snapshot(["forum", "query", "missing"]) == {
        "query": "query",
    }
    assert registry.local_runners(["forum", "query", "missing"]) == {
        "query": "query",
    }


def test_availability_updates_and_reset_restore_default_catalog():
    registry = EngineRuntimeRegistry(build_default_engine_catalog())

    assert registry.is_available("media") is True
    assert registry.get_local_runner("media") == "media"
    assert registry.get_script_path("media") == "apps/engine_console/media_engine_streamlit_app.py"

    registry.set_available("media", False)

    assert registry.is_available("media") is False
    assert registry.snapshot("media")["available"] is False

    registry.reset()

    assert registry.is_available("media") is True
    assert registry.snapshot("media") == {
        "name": "media",
        "available": True,
        "launch_mode": "streamlit",
        "script_path": "apps/engine_console/media_engine_streamlit_app.py",
        "local_runner": "media",
    }