smart-org/apps/api/tests/test_admin_control_plane.py

"""AI Control Plane: access control and the no-leak guarantee.

Two properties matter more than any feature here:

1. A normal customer can never reach these endpoints.
2. A provider API key can never come back out of the API.
"""

from __future__ import annotations

from httpx import AsyncClient

from tests.conftest import auth_header, signup

SECRET_KEY = "sk-super-secret-value-do-not-leak-1234567890"


async def test_normal_user_cannot_list_providers(client: AsyncClient):
    token = (await signup(client, "peon@example.com"))["access_token"]
    response = await client.get("/api/admin/providers", headers=auth_header(token))
    assert response.status_code == 403
    assert response.json()["error"]["code"] == "permission_denied"


async def test_normal_user_cannot_create_provider(client: AsyncClient):
    token = (await signup(client, "peon2@example.com"))["access_token"]
    response = await client.post(
        "/api/admin/providers",
        json={"name": "evil", "type": "mock", "base_url": ""},
        headers=auth_header(token),
    )
    assert response.status_code == 403


async def test_admin_endpoints_require_authentication(client: AsyncClient):
    response = await client.get("/api/admin/providers")
    assert response.status_code == 401


async def test_platform_owner_can_create_and_list_provider(
    client: AsyncClient, platform_owner_token: str
):
    created = await client.post(
        "/api/admin/providers",
        json={"name": "openrouter", "type": "openrouter", "base_url": "https://openrouter.ai/api/v1"},
        headers=auth_header(platform_owner_token),
    )
    assert created.status_code == 201, created.text
    assert created.json()["name"] == "openrouter"

    listing = await client.get("/api/admin/providers", headers=auth_header(platform_owner_token))
    assert listing.status_code == 200
    assert [p["name"] for p in listing.json()] == ["openrouter"]


async def test_duplicate_provider_name_is_rejected(client: AsyncClient, platform_owner_token: str):
    payload = {"name": "dupe", "type": "mock", "base_url": ""}
    first = await client.post(
        "/api/admin/providers", json=payload, headers=auth_header(platform_owner_token)
    )
    assert first.status_code == 201
    second = await client.post(
        "/api/admin/providers", json=payload, headers=auth_header(platform_owner_token)
    )
    assert second.status_code == 409


async def test_credential_creation_returns_only_a_masked_key(
    client: AsyncClient, platform_owner_token: str
):
    provider = (
        await client.post(
            "/api/admin/providers",
            json={"name": "p1", "type": "openai_compatible", "base_url": "https://x.test/v1"},
            headers=auth_header(platform_owner_token),
        )
    ).json()

    response = await client.post(
        f"/api/admin/providers/{provider['id']}/credentials",
        json={"name": "primary", "api_key": SECRET_KEY},
        headers=auth_header(platform_owner_token),
    )
    assert response.status_code == 201, response.text
    body = response.json()
    assert SECRET_KEY not in str(body)
    assert body["masked_key"].startswith("•")
    assert body["masked_key"].endswith(SECRET_KEY[-4:])


async def test_listing_credentials_never_exposes_the_key(
    client: AsyncClient, platform_owner_token: str
):
    provider = (
        await client.post(
            "/api/admin/providers",
            json={"name": "p2", "type": "openai_compatible", "base_url": "https://y.test/v1"},
            headers=auth_header(platform_owner_token),
        )
    ).json()
    await client.post(
        f"/api/admin/providers/{provider['id']}/credentials",
        json={"name": "primary", "api_key": SECRET_KEY},
        headers=auth_header(platform_owner_token),
    )

    response = await client.get(
        f"/api/admin/providers/{provider['id']}/credentials",
        headers=auth_header(platform_owner_token),
    )
    assert response.status_code == 200
    assert SECRET_KEY not in response.text
    assert response.json()[0]["masked_key"] == "••••••"


async def test_model_creation_and_listing(client: AsyncClient, platform_owner_token: str):
    provider = (
        await client.post(
            "/api/admin/providers",
            json={"name": "p3", "type": "mock", "base_url": ""},
            headers=auth_header(platform_owner_token),
        )
    ).json()

    created = await client.post(
        "/api/admin/models",
        json={
            "provider_id": provider["id"],
            "internal_name": "internal-coder-v1",
            "display_name_for_admin": "Coder V1",
            "capabilities": ["text", "code"],
            "input_price_per_1m": 0.3,
            "output_price_per_1m": 1.2,
        },
        headers=auth_header(platform_owner_token),
    )
    assert created.status_code == 201, created.text

    listing = await client.get("/api/admin/models", headers=auth_header(platform_owner_token))
    assert listing.status_code == 200
    assert listing.json()[0]["internal_name"] == "internal-coder-v1"


async def test_model_requires_existing_provider(client: AsyncClient, platform_owner_token: str):
    response = await client.post(
        "/api/admin/models",
        json={
            "provider_id": "does-not-exist",
            "internal_name": "x",
            "display_name_for_admin": "X",
        },
        headers=auth_header(platform_owner_token),
    )
    assert response.status_code == 404


async def test_routing_policy_creation_and_default_swap(
    client: AsyncClient, platform_owner_token: str
):
    first = await client.post(
        "/api/admin/routing-policies",
        json={"name": "cheap", "mode": "cheap", "is_default": True},
        headers=auth_header(platform_owner_token),
    )
    assert first.status_code == 201

    second = await client.post(
        "/api/admin/routing-policies",
        json={"name": "coding", "mode": "coding", "is_default": True},
        headers=auth_header(platform_owner_token),
    )
    assert second.status_code == 201

    listing = (
        await client.get("/api/admin/routing-policies", headers=auth_header(platform_owner_token))
    ).json()
    defaults = [p for p in listing if p["is_default"]]
    assert len(defaults) == 1
    assert defaults[0]["name"] == "coding"


async def test_rate_card_upsert_updates_in_place(client: AsyncClient, platform_owner_token: str):
    payload = {
        "unit": "model_input_tokens",
        "name": "input",
        "customer_price": 0.0000005,
        "internal_cost_estimate": 0.0000002,
    }
    first = await client.put(
        "/api/admin/rate-cards", json=payload, headers=auth_header(platform_owner_token)
    )
    assert first.status_code == 200, first.text

    payload["customer_price"] = 0.0000009
    second = await client.put(
        "/api/admin/rate-cards", json=payload, headers=auth_header(platform_owner_token)
    )
    assert second.json()["customer_price"] == 0.0000009

    listing = (
        await client.get("/api/admin/rate-cards", headers=auth_header(platform_owner_token))
    ).json()
    assert len([c for c in listing if c["unit"] == "model_input_tokens"]) == 1


async def test_platform_owner_sees_all_organizations(
    client: AsyncClient, platform_owner_token: str
):
    customer = (await signup(client, "tenant-x@example.com"))["access_token"]
    await client.post(
        "/api/organizations", json={"name": "شرکت مشتری"}, headers=auth_header(customer)
    )

    response = await client.get(
        "/api/admin/organizations", headers=auth_header(platform_owner_token)
    )
    assert response.status_code == 200
    assert [o["name"] for o in response.json()] == ["شرکت مشتری"]
    assert response.json()[0]["member_count"] == 1


async def test_customer_cannot_see_other_organizations_through_admin(
    client: AsyncClient,
):
    customer = (await signup(client, "tenant-y@example.com"))["access_token"]
    response = await client.get("/api/admin/organizations", headers=auth_header(customer))
    assert response.status_code == 403