cmvr_edge_ai/tests/integration/test_application_cli.py

from __future__ import annotations

import asyncio
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

from cmvr_edge_ai.application import (
    ApplicationState,
    EdgeAIApplication,
    create_default_registry,
    validate_application,
)
from cmvr_edge_ai.cli import main
from cmvr_edge_ai.config import load_config


PROJECT_ROOT = Path(__file__).resolve().parents[2]
MINIMAL_CONFIG = PROJECT_ROOT / "tests" / "fixtures" / "minimal_pipeline.yaml"
ACTIVE_DETECTION_CONFIG = PROJECT_ROOT / "configs" / "active_detection.yaml"


def test_application_runs_the_finite_minimal_pipeline() -> None:
    async def exercise():  # type: ignore[no-untyped-def]
        loop = asyncio.get_running_loop()
        host_executor = ThreadPoolExecutor(max_workers=1)
        loop.set_default_executor(host_executor)
        application = EdgeAIApplication(
            load_config(MINIMAL_CONFIG),
            create_default_registry(discover_entry_points=False),
        )
        try:
            await application.start()
            running_health = await application.health()
            await application.wait()
            await application.stop()
            # The application owns only its bounded executor. Stopping it must
            # not poison an embedding event loop's pre-existing executor.
            host_executor_result = await asyncio.to_thread(lambda: "still-usable")
            return application, running_health, host_executor_result
        finally:
            # Python 3.12 Runner shuts its default executor down from another
            # helper thread. Explicit cleanup keeps this regression deterministic
            # in restricted containers where that helper cannot wake the loop.
            host_executor.shutdown(wait=True, cancel_futures=True)
            loop._default_executor = None  # type: ignore[attr-defined]

    application, running_health, host_executor_result = asyncio.run(exercise())

    assert running_health["state"] == "running"
    assert set(running_health["pipelines"]) == {"minimal"}
    assert application.pipelines == ("minimal",)
    assert application.state is ApplicationState.STOPPED
    assert host_executor_result == "still-usable"


def test_cli_validates_and_runs_minimal_config(
    capsys, monkeypatch
) -> None:  # type: ignore[no-untyped-def]
    # ``configure_logging(force=True)`` intentionally owns process logging in
    # production. Avoid leaking that global CLI side effect into later tests.
    monkeypatch.setattr("cmvr_edge_ai.cli.configure_logging", lambda *_: None)
    assert main(["validate", "--config", str(MINIMAL_CONFIG)]) == 0
    validation_output = capsys.readouterr()
    assert "configuration is valid; pipelines: minimal" in validation_output.out
    assert validation_output.err == ""

    assert (
        main(
            [
                "run",
                "--config",
                str(MINIMAL_CONFIG),
                "--log-level",
                "WARNING",
            ]
        )
        == 0
    )
    run_output = capsys.readouterr()
    assert run_output.err == ""


def test_cli_lists_versioned_builtin_plugins(capsys) -> None:  # type: ignore[no-untyped-def]
    assert main(["plugins"]) == 0

    output = capsys.readouterr().out
    assert "core.sequence_source@1\tsource" in output
    assert "detection.model@1\toperator" in output
    assert "media.video_decoder.pyav@1\toperator" in output
    assert "safety.robot_command_gate@1\toperator" in output


def test_cli_lists_detection_model_metadata(capsys) -> None:  # type: ignore[no-untyped-def]
    assert main(["models"]) == 0

    output = capsys.readouterr().out
    assert (
        "construction-ppe-yolov8@1\tConstruction PPE YOLOv8s\t" "ultralytics-yolo"
    ) in output
    assert "No-Helmet" in output
    assert "No-Vest" in output
    assert (
        "ppe-6classes-yolov8n@1\tPPE Detection YOLOv8n (6 Classes)\t"
        "ultralytics-yolo"
    ) in output
    assert "Gloves,Vest,goggles,helmet,mask,safety_shoe" in output
    assert (
        "people-talking-yolov8x@1\tPeople Talking YOLOv8x\t"
        "ultralytics-yolo\tlabel,talking on phone"
    ) in output
    assert (
        "yolov8n-mobile-phone@1\tYOLOv8n Mobile Phone\t"
        "ultralytics-yolo\tmobile_phone"
    ) in output
    assert (
        "construction-ppe-yolov8@2\tConstruction PPE YOLOv8s ONNX\t"
        "onnxruntime-yolov8"
    ) in output
    assert (
        "ppe-6classes-yolov8n@2\tPPE Detection YOLOv8n (6 Classes) ONNX\t"
        "onnxruntime-yolov8"
    ) in output
    assert (
        "people-talking-yolov8x@2\tPeople Talking YOLOv8x ONNX\t"
        "onnxruntime-yolov8\tlabel,talking on phone"
    ) in output
    assert (
        "yolov8n-mobile-phone@2\tYOLOv8n Mobile Phone ONNX\t"
        "onnxruntime-yolov8\tmobile_phone"
    ) in output


def test_active_detection_config_compiles_without_loading_optional_runtimes() -> None:
    config = load_config(ACTIVE_DETECTION_CONFIG)
    registry = create_default_registry(discover_entry_points=False)
    compiled = validate_application(
        config,
        registry,
        ("detection",),
    )

    assert set(config.pipelines) == {"detection"}
    assert len(compiled) == 1
    assert compiled[0].pipeline_id == "detection"
    assert "ppe_alert_platform" in config.endpoints
    assert (
        config.pipelines["detection"]
        .nodes["alert_platform"]
        .params["endpoint"]
        == "ppe_alert_platform"
    )
    assert (
        config.pipelines["detection"]
        .nodes["alert_platform"]
        .params["failure_mode"]
        == "log_and_drop"
    )
    assert (
        config.pipelines["detection"]
        .nodes["alert_platform"]
        .params["grpc_endpoint"]
        == "cmvr_es"
    )
    phone_detector = config.pipelines["detection"].nodes["phone_detector"]
    assert phone_detector.params["model"] == "people-talking-yolov8x@2"
    assert phone_detector.params["detect_labels"] == ["talking on phone"]
    phone_gate = config.pipelines["detection"].nodes["phone_repeat_gate"]
    assert phone_gate.params["rules"][0]["labels"] == ["talking on phone"]
    assert set(compiled[0].plugin_specs) == {
        "camera",
        "decoder",
        "detector",
        "repeat_gate",
        "phone_detector",
        "phone_repeat_gate",
        "alert_platform",
    }