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. Model identity and account material stay inside the console — a customer
   never sees a raw model, and nothing that could act as a credential (a Google
   refresh token, an API key) ever comes back out.
"""

from __future__ import annotations

from httpx import AsyncClient

from tests.conftest import auth_header, signup


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


async def test_normal_user_cannot_create_model(client: AsyncClient):
    token = (await signup(client, "peon2@example.com"))["access_token"]
    response = await client.post(
        "/api/admin/models",
        json={"internal_name": "evil-model"},
        headers=auth_header(token),
    )
    assert response.status_code == 403


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


async def test_model_creation_resolves_the_agy_provider_itself(
    client: AsyncClient, platform_owner_token: str
):
    """The provider is not a client choice: agy is the only one."""
    created = await client.post(
        "/api/admin/models",
        json={
            "internal_name": "gemini-3-pro-high",
            "capabilities": ["text", "reasoning"],
            "input_price_per_1m": 0.3,
            "output_price_per_1m": 1.2,
            "reasoning_effort": "high",
        },
        headers=auth_header(platform_owner_token),
    )
    assert created.status_code == 201, created.text
    body = created.json()
    assert body["internal_name"] == "gemini-3-pro-high"
    # The display name falls back to the model id; nothing is typed twice.
    assert body["display_name_for_admin"] == "gemini-3-pro-high"
    assert body["reasoning_effort"] == "high"
    assert body["provider_id"], "the agy provider must be resolved server-side"

    listing = await client.get("/api/admin/models", headers=auth_header(platform_owner_token))
    assert listing.status_code == 200
    assert [m["internal_name"] for m in listing.json()] == ["gemini-3-pro-high"]


async def test_duplicate_model_name_is_rejected(client: AsyncClient, platform_owner_token: str):
    payload = {"internal_name": "claude-sonnet-4-5-thinking", "reasoning_effort": "high"}
    first = await client.post(
        "/api/admin/models", json=payload, headers=auth_header(platform_owner_token)
    )
    assert first.status_code == 201
    second = await client.post(
        "/api/admin/models", json=payload, headers=auth_header(platform_owner_token)
    )
    assert second.status_code == 409


async def test_unknown_reasoning_effort_is_rejected(
    client: AsyncClient, platform_owner_token: str
):
    response = await client.post(
        "/api/admin/models",
        json={"internal_name": "gemini-3-flash", "reasoning_effort": "ultra"},
        headers=auth_header(platform_owner_token),
    )
    assert response.status_code == 422


async def test_agy_account_and_model_endpoints_are_owner_only(
    client: AsyncClient, platform_owner_token: str
):
    """The account catalog is owner-only and never carries token material."""
    token = (await signup(client, "peon3@example.com"))["access_token"]
    response = await client.get(
        "/api/admin/agy/accounts/whatever/models", headers=auth_header(token)
    )
    assert response.status_code == 403

    unauthenticated = await client.get("/api/admin/agy/accounts/whatever/models")
    assert unauthenticated.status_code == 401


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