agentic-mobile-control/apps/cloud-api/tests/test_llm_provider_management.py

from __future__ import annotations

from fastapi.testclient import TestClient

from cloud.control_config import CloudControlConfig
from cloud_api.app import create_app


def _create_admin(client: TestClient) -> None:
    client.app.state.cloud_services.user_auth_service.create_user(
        username="admin",
        display_name="Administrator",
        role="admin",
        password="correct-horse-battery-staple",
        must_change_password=False,
    )
    response = client.post(
        "/v1/auth/login",
        json={"username": "admin", "password": "correct-horse-battery-staple"},
    )
    assert response.status_code == 200


def _csrf_headers(client: TestClient) -> dict[str, str]:
    token = client.cookies.get("amcp_csrf")
    assert token is not None
    return {"X-CSRF-Token": token}


def _profile_payload(**overrides: object) -> dict[str, object]:
    payload = {
        "name": "OpenAI Compatible",
        "provider_type": "openai-compatible",
        "model": "gpt-compatible",
        "base_url": "https://compat.example/v1",
        "timeout_seconds": 20,
        "api_key": "provider-secret-value",
    }
    payload.update(overrides)
    return payload


def test_provider_profiles_are_encrypted_redacted_and_activated(monkeypatch) -> None:
    from cryptography.fernet import Fernet

    monkeypatch.setenv(
        "CLOUD_LLM_PROVIDER_ENCRYPTION_KEY", Fernet.generate_key().decode()
    )
    app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
    with TestClient(app) as client:
        _create_admin(client)
        headers = _csrf_headers(client)

        missing_csrf = client.post("/v1/planner/providers", json=_profile_payload())
        assert missing_csrf.status_code == 403

        empty_key = client.post(
            "/v1/planner/providers",
            headers=headers,
            json=_profile_payload(api_key=""),
        )
        assert empty_key.status_code == 422

        created = client.post(
            "/v1/planner/providers", headers=headers, json=_profile_payload()
        )
        assert created.status_code == 201, created.text
        profile = created.json()
        assert profile["has_api_key"] is True
        assert "api_key" not in profile
        assert "ciphertext" not in profile

        repository = client.app.state.cloud_services.repository
        stored = repository.get_llm_provider_profile(profile["id"])
        assert stored is not None
        assert stored.api_key_ciphertext != "provider-secret-value"
        assert "provider-secret-value" not in stored.api_key_ciphertext

        listed = client.get("/v1/planner/providers")
        assert listed.status_code == 200
        assert listed.json()["settings"]["active_profile_id"] is None
        assert "provider-secret-value" not in listed.text
        assert stored.api_key_ciphertext not in listed.text

        activated = client.post(
            f"/v1/planner/providers/{profile['id']}/activate",
            headers=headers,
            json={"expected_settings_revision": 0},
        )
        assert activated.status_code == 200, activated.text
        assert activated.json()["active_profile_id"] == profile["id"]

        disable_active = client.patch(
            f"/v1/planner/providers/{profile['id']}",
            headers=headers,
            json={"enabled": False, "expected_revision": profile["revision"]},
        )
        assert disable_active.status_code == 409


def test_provider_profile_requires_admin_scope(monkeypatch) -> None:
    from cryptography.fernet import Fernet

    monkeypatch.setenv(
        "CLOUD_LLM_PROVIDER_ENCRYPTION_KEY", Fernet.generate_key().decode()
    )
    app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
    with TestClient(app) as client:
        service = client.app.state.cloud_services.user_auth_service
        service.create_user(
            username="operator",
            display_name="Operator",
            role="operator",
            password="correct-horse-battery-staple",
            must_change_password=False,
        )
        assert (
            client.post(
                "/v1/auth/login",
                json={
                    "username": "operator",
                    "password": "correct-horse-battery-staple",
                },
            ).status_code
            == 200
        )

        assert client.get("/v1/planner/providers").status_code == 403
        assert (
            client.post(
                "/v1/planner/providers",
                headers=_csrf_headers(client),
                json=_profile_payload(),
            ).status_code
            == 403
        )


def test_anthropic_profile_accepts_custom_base_url(monkeypatch) -> None:
    from cryptography.fernet import Fernet

    monkeypatch.setenv(
        "CLOUD_LLM_PROVIDER_ENCRYPTION_KEY", Fernet.generate_key().decode()
    )
    app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
    with TestClient(app) as client:
        _create_admin(client)
        response = client.post(
            "/v1/planner/providers",
            headers=_csrf_headers(client),
            json=_profile_payload(
                name="Anthropic proxy",
                provider_type="anthropic",
                base_url="https://anthropic-proxy.example/",
            ),
        )

    assert response.status_code == 201, response.text
    assert response.json()["base_url"] == "https://anthropic-proxy.example"


def test_profile_write_requires_encryption_key(monkeypatch) -> None:
    monkeypatch.delenv("CLOUD_LLM_PROVIDER_ENCRYPTION_KEY", raising=False)
    app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
    with TestClient(app) as client:
        _create_admin(client)
        response = client.post(
            "/v1/planner/providers",
            headers=_csrf_headers(client),
            json=_profile_payload(),
        )

    assert response.status_code == 503
    assert "provider-secret-value" not in response.text