cmvr_edge_ai/tests/unit/test_server_config.py

from __future__ import annotations

import pytest
from pydantic import ValidationError

from cmvr_edge_ai.config import load_config_data


def _pipeline(*, enabled: bool = True) -> dict[str, object]:
    return {
        "enabled": enabled,
        "nodes": {"source": {"uses": "test.source@1"}},
    }


def test_server_config_maps_namespaced_categories_to_enabled_pipelines() -> None:
    config = load_config_data(
        {
            "api_version": "cmvr.edge.ai/v1",
            "pipelines": {"mobile_phone_remote": _pipeline()},
            "server": {
                "enabled": True,
                "http": {
                    "bind": "0.0.0.0",
                    "port": 18080,
                    "bearer_token": "deployment-secret",
                    "allow_insecure_remote": True,
                    "max_request_bytes": 2_000_000,
                    "max_image_bytes": 1_000_000,
                },
                "routes": {
                    "detect.mobile_phone": {
                        "pipeline": "mobile_phone_remote",
                        "model_id": "yolov8n-mobile-phone@1",
                        "queue_capacity": 3,
                    }
                },
            },
        }
    )

    assert config.server is not None
    assert config.server.http.bind == "0.0.0.0"
    assert config.server.http.port == 18080
    assert config.server.http.bearer_token is not None
    assert config.server.http.bearer_token.get_secret_value() == "deployment-secret"
    assert config.server.http.model_dump(mode="json")["bearer_token"] == "**********"
    assert config.server.http.allow_insecure_remote is True
    route = config.server.routes["detect.mobile_phone"]
    assert route.pipeline == "mobile_phone_remote"
    assert route.model_id == "yolov8n-mobile-phone@1"
    assert route.queue_capacity == 3


@pytest.mark.parametrize(
    ("server", "message"),
    [
        (
            {"enabled": True, "routes": {}},
            "enabled server must configure at least one route",
        ),
        (
            {
                "enabled": True,
                "routes": {
                    "detect": {
                        "pipeline": "remote",
                        "model_id": "model@1",
                    }
                },
            },
            "invalid invocation category",
        ),
        (
            {
                "enabled": True,
                "http": {
                    "max_request_bytes": 1024,
                    "max_image_bytes": 2048,
                },
                "routes": {
                    "detect.remote": {
                        "pipeline": "remote",
                        "model_id": "model@1",
                    }
                },
            },
            "max_image_bytes must not exceed max_request_bytes",
        ),
    ],
)
def test_server_config_rejects_ambiguous_or_unsafe_values(
    server: dict[str, object], message: str
) -> None:
    with pytest.raises(ValidationError, match=message):
        load_config_data(
            {
                "api_version": "cmvr.edge.ai/v1",
                "pipelines": {"remote": _pipeline()},
                "server": server,
            }
        )


def test_server_routes_must_reference_one_unique_enabled_pipeline() -> None:
    with pytest.raises(ValidationError, match="only one public category"):
        load_config_data(
            {
                "api_version": "cmvr.edge.ai/v1",
                "pipelines": {"remote": _pipeline()},
                "server": {
                    "enabled": True,
                    "routes": {
                        "detect.first": {
                            "pipeline": "remote",
                            "model_id": "first@1",
                        },
                        "detect.second": {
                            "pipeline": "remote",
                            "model_id": "second@1",
                        },
                    },
                },
            }
        )

    with pytest.raises(ValidationError, match="disabled pipeline"):
        load_config_data(
            {
                "api_version": "cmvr.edge.ai/v1",
                "pipelines": {"remote": _pipeline(enabled=False)},
                "server": {
                    "enabled": True,
                    "routes": {
                        "detect.remote": {
                            "pipeline": "remote",
                            "model_id": "model@1",
                        }
                    },
                },
            }
        )

    with pytest.raises(ValidationError, match="unknown pipeline"):
        load_config_data(
            {
                "api_version": "cmvr.edge.ai/v1",
                "pipelines": {"active": _pipeline()},
                "server": {
                    "enabled": True,
                    "routes": {
                        "detect.remote": {
                            "pipeline": "missing",
                            "model_id": "model@1",
                        }
                    },
                },
            }
        )


@pytest.mark.parametrize(
    ("http", "message"),
    [
        (
            {"bind": "0.0.0.0"},
            "non-loopback server.http.bind requires bearer_token",
        ),
        (
            {"bind": "edge-host", "bearer_token": "secret"},
            "requires TLS or allow_insecure_remote=true",
        ),
        (
            {"tls_certfile": "/certs/server.crt"},
            "tls_certfile and tls_keyfile must be configured together",
        ),
        (
            {"bearer_token": "contains whitespace"},
            "bearer_token must be a non-empty token without whitespace",
        ),
    ],
)
def test_server_http_rejects_incomplete_remote_security(
    http: dict[str, object], message: str
) -> None:
    with pytest.raises(ValidationError, match=message):
        load_config_data(
            {
                "api_version": "cmvr.edge.ai/v1",
                "pipelines": {"remote": _pipeline()},
                "server": {
                    "enabled": True,
                    "http": http,
                    "routes": {
                        "detect.remote": {
                            "pipeline": "remote",
                            "model_id": "model@1",
                        }
                    },
                },
            }
        )


def test_server_http_accepts_loopback_without_auth_and_remote_with_tls() -> None:
    loopback = load_config_data(
        {
            "api_version": "cmvr.edge.ai/v1",
            "pipelines": {"remote": _pipeline()},
            "server": {
                "enabled": True,
                "http": {"bind": "::1"},
                "routes": {
                    "detect.remote": {
                        "pipeline": "remote",
                        "model_id": "model@1",
                    }
                },
            },
        }
    )
    assert loopback.server is not None
    assert loopback.server.http.bearer_token is None

    secured = load_config_data(
        {
            "api_version": "cmvr.edge.ai/v1",
            "pipelines": {"remote": _pipeline()},
            "server": {
                "enabled": True,
                "http": {
                    "bind": "192.0.2.10",
                    "bearer_token": "deployment-secret",
                    "tls_certfile": "/certs/server.crt",
                    "tls_keyfile": "/certs/server.key",
                },
                "routes": {
                    "detect.remote": {
                        "pipeline": "remote",
                        "model_id": "model@1",
                    }
                },
            },
        }
    )
    assert secured.server is not None
    assert str(secured.server.http.tls_certfile) == "/certs/server.crt"
    assert str(secured.server.http.tls_keyfile) == "/certs/server.key"