thePlug/tests/api/test_proxy.py

import importlib

from core.security import get_auth_context, require_authenticated_user
from models.proxy import ProxyDocument
from models.service import ServiceEndpointDocument
from services.mitmproxy.validator import ProxyCodeValidationError

proxy_router_module = importlib.import_module("api.proxy.router")


def make_proxy_document(
    *,
    proxy_port: int,
    name: str = "svc",
    code: str = "proxy-code",
    port: int = 9000,
) -> ProxyDocument:
    return ProxyDocument(
        proxy_port=proxy_port,
        name=name,
        code=code,
        service=ServiceEndpointDocument(port=port),
    )


def test_proxy_requires_auth_without_dependency_overrides(client):
    client.app.dependency_overrides.pop(require_authenticated_user, None)
    client.app.dependency_overrides.pop(get_auth_context, None)

    response = client.get("/proxy")

    assert response.status_code == 401
    assert response.json()["detail"] == "Could not validate credentials"


def test_list_proxies_returns_documents(client, monkeypatch: object):
    docs = [
        make_proxy_document(proxy_port=8080, name="api"),
        make_proxy_document(proxy_port=8081, name="worker"),
    ]

    async def fake_list_proxies() -> list[ProxyDocument]:
        return docs

    monkeypatch.setattr(proxy_router_module, "list_proxy_documents", fake_list_proxies)

    response = client.get("/proxy")

    assert response.status_code == 200
    body = response.json()
    assert [item["proxy_port"] for item in body] == [8080, 8081]


def test_get_proxy_returns_404_when_not_found(client, monkeypatch: object):
    async def fake_get_proxy_by_port(proxy_port: int) -> None:
        return None

    monkeypatch.setattr(
        proxy_router_module,
        "get_proxy_by_port",
        fake_get_proxy_by_port,
    )

    response = client.get("/proxy/9999")

    assert response.status_code == 404
    assert response.json()["detail"] == "Proxy not found"


def test_create_proxy_returns_409_on_port_conflict(client, monkeypatch: object):
    existing = make_proxy_document(proxy_port=3671)

    async def fake_get_proxy_by_port(proxy_port: int) -> ProxyDocument | None:
        assert proxy_port == 3671
        return existing

    monkeypatch.setattr(
        proxy_router_module,
        "get_proxy_by_port",
        fake_get_proxy_by_port,
    )

    response = client.post(
        "/proxy",
        json={
            "proxy_port": 3671,
            "name": "new",
            "code": "abc",
            "service": {"port": 8080},
        },
    )

    assert response.status_code == 409
    assert response.json()["detail"] == "Proxy already exists on port 3671"


def test_create_proxy_returns_created_document(client, monkeypatch: object):
    captured: dict[str, object] = {}

    async def fake_get_proxy_by_port(proxy_port: int) -> ProxyDocument | None:
        assert proxy_port == 4040
        return None

    async def fake_create_proxy(
        *,
        proxy_port: int,
        name: str,
        service: dict[str, object],
        code: str,
    ) -> ProxyDocument:
        captured["proxy_port"] = proxy_port
        captured["name"] = name
        captured["service"] = service
        captured["code"] = code
        return make_proxy_document(
            proxy_port=proxy_port,
            name=name,
            code=code,
            port=int(service["port"]),
        )

    async def fake_validate_proxy_code(code: str) -> None:
        assert code == "def proxy_handler():\n    return 'ok'\n"

    monkeypatch.setattr(
        proxy_router_module,
        "get_proxy_by_port",
        fake_get_proxy_by_port,
    )
    monkeypatch.setattr(
        proxy_router_module,
        "validate_proxy_code",
        fake_validate_proxy_code,
    )
    monkeypatch.setattr(proxy_router_module, "create_proxy", fake_create_proxy)

    response = client.post(
        "/proxy",
        json={
            "proxy_port": 4040,
            "name": "edge",
            "code": "def proxy_handler():\n    return 'ok'\n",
            "service": {"port": 8181},
        },
    )

    assert response.status_code == 201
    assert response.json()["proxy_port"] == 4040
    assert captured == {
        "proxy_port": 4040,
        "name": "edge",
        "service": {"port": 8181},
        "code": "def proxy_handler():\n    return 'ok'\n",
    }


def test_create_proxy_returns_422_when_code_validation_fails(
    client,
    monkeypatch: object,
):
    create_called = False

    async def fake_get_proxy_by_port(proxy_port: int) -> ProxyDocument | None:
        assert proxy_port == 4000
        return None

    async def fake_validate_proxy_code(code: str) -> None:
        assert code == "invalid-code"
        raise ProxyCodeValidationError(
            "Proxy code contains ruff violations",
            errors=[
                {
                    "code": "F821",
                    "message": "Undefined name `missing_symbol`",
                    "location": {"row": 1, "column": 1},
                }
            ],
        )

    async def fake_create_proxy(
        *,
        proxy_port: int,
        name: str,
        service: dict[str, object],
        code: str,
    ) -> ProxyDocument:
        nonlocal create_called
        create_called = True
        return make_proxy_document(proxy_port=proxy_port, name=name, code=code)

    monkeypatch.setattr(
        proxy_router_module,
        "get_proxy_by_port",
        fake_get_proxy_by_port,
    )
    monkeypatch.setattr(
        proxy_router_module,
        "validate_proxy_code",
        fake_validate_proxy_code,
    )
    monkeypatch.setattr(proxy_router_module, "create_proxy", fake_create_proxy)

    response = client.post(
        "/proxy",
        json={
            "proxy_port": 4000,
            "name": "broken",
            "code": "invalid-code",
            "service": {"port": 9000},
        },
    )

    assert response.status_code == 422
    body = response.json()["detail"]
    assert body["message"] == "Proxy code contains ruff violations"
    assert body["errors"][0]["code"] == "F821"
    assert create_called is False


def test_update_proxy_returns_422_when_code_validation_fails(
    client,
    monkeypatch: object,
):
    update_called = False

    async def fake_validate_proxy_code(code: str) -> None:
        assert code == "invalid-code"
        raise ProxyCodeValidationError(
            "Proxy code has invalid Python syntax",
            errors=[
                {
                    "code": "COMPILE",
                    "message": "invalid syntax",
                    "location": {"row": 1, "column": 1},
                }
            ],
        )

    async def fake_update_proxy(
        proxy_port: int,
        payload: object,
    ) -> ProxyDocument | None:
        nonlocal update_called
        update_called = True
        return make_proxy_document(proxy_port=proxy_port)

    monkeypatch.setattr(
        proxy_router_module,
        "validate_proxy_code",
        fake_validate_proxy_code,
    )
    monkeypatch.setattr(proxy_router_module, "update_proxy", fake_update_proxy)

    response = client.put(
        "/proxy/8123",
        json={"code": "invalid-code"},
    )

    assert response.status_code == 422
    body = response.json()["detail"]
    assert body["message"] == "Proxy code has invalid Python syntax"
    assert body["errors"][0]["code"] == "COMPILE"
    assert update_called is False