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

"""§18.3 — the minimal end-to-end smoke path.

One test walks the whole product in order, through the real HTTP API and the
real application services, against the agy-shaped catalogue (the seeded fixture
installs the agy provider and its models) with the fake model provider from
conftest standing in for the Google round-trip:

    signup -> organization -> project -> repo -> preview domain -> sandbox
           -> employee -> manager message -> task -> run -> usage -> audit

The point is not coverage of edge cases (the focused suites do that) but to
prove the seams line up: that an object created in one step is actually
accepted by the next one. A green unit suite with a broken seam between two
sections is exactly the failure this test exists to catch.

It also re-asserts the two product invariants that must hold at every step:
a customer never sees a raw provider/model, and another tenant's objects are
invisible (404, never 403).
"""

from __future__ import annotations

import pytest_asyncio
from httpx import AsyncClient

from app.domain.identity.models import PlatformRole
from app.domain.workforce.pipeline import Stage
from app.infrastructure.repositories.engine import PipelineStateRepository
from tests.billing_helpers import fund_org
from tests.conftest import auth_header, signup
from tests.legacy_tasks import legacy_task


@pytest_asyncio.fixture(autouse=True)
async def _platform_seed(seeded):
    """The platform rows a fresh install has: templates, policies, models, rates."""
    return seeded


async def test_end_to_end_smoke_path(client: AsyncClient, session_factory):
    """Decision D-D: the task made over HTTP enters the engine and refuses the legacy run, so
    the run, its metering and its audit are walked on a pre-v2 task (see step 10)."""
    # -- 1. signup -------------------------------------------------------- #
    # The platform owner, not a plain manager: step 6 drives the sandbox
    # directly, and v2 makes container/command access owner-only
    # (PLAN-v2 §7.7). The org-scoped prefix still scopes the tenant and the
    # audit trail, so this is the same path a customer walks minus the
    # infrastructure step.
    signup_body = await signup(
        client, "smoke@example.com", name="کاربر دود", platform_role=PlatformRole.OWNER
    )
    token = signup_body["access_token"]
    headers = auth_header(token)

    me = await client.get("/api/auth/me", headers=headers)
    assert me.status_code == 200, me.text
    assert me.json()["user"]["email"] == "smoke@example.com"

    # -- 2. create organization ------------------------------------------- #
    org = await client.post(
        "/api/organizations", json={"name": "سازمان دود"}, headers=headers
    )
    assert org.status_code == 201, org.text
    org_id = org.json()["id"]
    assert org.json()["role"] == "owner"

    # -- 3. create project ------------------------------------------------ #
    project = await client.post(
        f"/api/organizations/{org_id}/projects",
        json={"name": "پروژهٔ دود", "description": "مسیر دود انتها به انتها"},
        headers=headers,
    )
    assert project.status_code == 201, project.text
    project_id = project.json()["id"]
    assert project.json()["slug"], "a project must get a slug for its preview host"

    # -- 4. ensure repo --------------------------------------------------- #
    repo = await client.post(
        f"/api/organizations/{org_id}/projects/{project_id}/repo/ensure",
        headers=headers,
    )
    assert repo.status_code in (200, 201), repo.text
    assert repo.json()["private"] is True, "a customer repo must be private"

    status_body = await client.get(
        f"/api/organizations/{org_id}/projects/{project_id}/repo/status",
        headers=headers,
    )
    assert status_body.status_code == 200, status_body.text
    assert status_body.json()["exists"] is True

    # -- 5. create preview domain ----------------------------------------- #
    domain = await client.post(
        f"/api/organizations/{org_id}/projects/{project_id}/preview/domains",
        json={"subdomain": "smoke-app"},
        headers=headers,
    )
    assert domain.status_code == 201, domain.text
    domain_body = domain.json()
    assert domain_body["host"] == "smoke-app.msa.artacloud.ir", domain_body
    assert 8080 <= domain_body["port"] <= 8999, "port must come from the managed range"
    assert domain_body["protected"] is False

    # -- 6. sandbox workspace + one metered command ------------------------ #
    templates = await client.get(
        f"/api/organizations/{org_id}/sandbox/templates", headers=headers
    )
    assert templates.status_code == 200, templates.text
    assert templates.json(), "a seeded install must expose sandbox templates"
    template_id = templates.json()[0]["id"]

    workspace = await client.post(
        f"/api/organizations/{org_id}/sandbox/workspaces",
        json={"project_id": project_id, "template_id": template_id},
        headers=headers,
    )
    assert workspace.status_code in (200, 201), workspace.text
    assert workspace.json()["status"] in {"ready", "starting", "created"}

    session = await client.post(
        f"/api/organizations/{org_id}/sandbox/sessions",
        json={"workspace_id": workspace.json()["id"]},
        headers=headers,
    )
    assert session.status_code in (200, 201), session.text
    session_id = session.json()["id"]

    command = await client.post(
        f"/api/organizations/{org_id}/sandbox/sessions/{session_id}/commands",
        # A caller that tries to raise its own limits must not be able to: the
        # request schema accepts only `command`, so the extra keys are ignored
        # and the platform's policy values apply.
        json={
            "command": "python -m pytest -q",
            "timeout_seconds": 999_999,
            "max_output_bytes": 10**9,
        },
        headers=headers,
    )
    assert command.status_code in (200, 201), command.text
    command_body = command.json()
    assert command_body["command"] == "python -m pytest -q"
    for platform_owned in ("timeout_seconds", "max_output_bytes"):
        assert platform_owned not in command_body, (
            f"the caller must not be able to set {platform_owned}"
        )

    # -- 7. create employee ------------------------------------------------ #
    agent_templates = await client.get(
        f"/api/organizations/{org_id}/agent-templates", headers=headers
    )
    assert agent_templates.status_code == 200, agent_templates.text
    agent_template = agent_templates.json()[0]

    employee = await client.post(
        f"/api/organizations/{org_id}/employees",
        json={
            "name": "کارمند دود",
            "role_key": agent_template["role_key"],
            "template_id": agent_template["id"],
        },
        headers=headers,
    )
    assert employee.status_code == 201, employee.text
    employee_id = employee.json()["id"]

    # The employee record is customer-facing: it must not name a model.
    assert "model" not in str(employee.json()).lower()

    # -- 8. send a manager message ----------------------------------------- #
    # PLAN-v2 D3: the engine may only run on models the owner granted to this
    # organization, so the manager's chat needs a grant before it can answer.
    # Before the engine existed any active model would do; a chat without a
    # grant now correctly answers nothing but the "waiting for a model" note.
    models = await client.get("/api/admin/models", headers=headers)
    assert models.status_code == 200, models.text
    ai_model_id = models.json()[0]["id"]
    grant = await client.put(
        f"/api/admin/model-grants/{org_id}/models/{ai_model_id}",
        json={"daily_credit_cap": None},
        headers=headers,
    )
    assert grant.status_code in (200, 201), grant.text

    thread = await client.post(
        f"/api/organizations/{org_id}/threads",
        json={"title": "گفت‌وگو با مدیر", "project_id": project_id},
        headers=headers,
    )
    assert thread.status_code == 201, thread.text
    thread_id = thread.json()["id"]

    # prepaid (D-B): the HTTP org starts at zero credit, and the chat costs model calls.
    await fund_org(session_factory, org_id)

    message = await client.post(
        f"/api/organizations/{org_id}/threads/{thread_id}/messages",
        json={"content": "برای ماژول پرداخت تست بنویس.", "employee_id": employee_id},
        headers=headers,
    )
    assert message.status_code in (200, 201), message.text
    message_body = message.json()
    assert message_body["user_message"]["role"] == "user"
    assistant = message_body["assistant_message"]
    assert assistant["content"], "the manager must actually answer"
    # The reply is customer-facing text: it must not name a model or provider.
    assert "provider" not in assistant["content"].lower()
    assert "model" not in assistant["content"].lower()

    # -- 9. create task ---------------------------------------------------- #
    task = await client.post(
        f"/api/organizations/{org_id}/tasks",
        json={
            "title": "افزودن تست برای ماژول پرداخت",
            "description": "پوشش تست واحد برای محاسبهٔ تخفیف",
            "project_id": project_id,
            "thread_id": thread_id,
            "assigned_employee_id": employee_id,
        },
        headers=headers,
    )
    assert task.status_code == 201, task.text
    task_id = task.json()["id"]
    # Decision D-D: a task made over HTTP is the engine's from manager review, so it has a
    # pipeline state and the legacy run button refuses it.
    async with session_factory() as session:
        state = await PipelineStateRepository(session).get(task_id)
    assert state is not None and state.stage is Stage.PENDING_MANAGER_REVIEW
    refused = await client.post(
        f"/api/organizations/{org_id}/tasks/{task_id}/runs",
        json={"employee_id": employee_id},
        headers=headers,
    )
    assert refused.status_code == 409, refused.text

    # -- 10. run task ------------------------------------------------------ #
    # The legacy run path serves pre-v2 tasks, which have no pipeline state, so the run, its
    # metering and its audit are checked on one of those (decision D-D).
    run_task_id = await legacy_task(
        session_factory,
        org_id,
        signup_body["user"]["id"],
        title="افزودن تست برای ماژول پرداخت",
        project_id=project_id,
        assigned_employee_id=employee_id,
    )
    run = await client.post(
        f"/api/organizations/{org_id}/tasks/{run_task_id}/runs",
        json={"employee_id": employee_id},
        headers=headers,
    )
    assert run.status_code in (200, 201), run.text
    run_body = run.json()
    assert run_body["status"] == "completed", run_body
    assert run_body["steps"], "a run must leave a timeline"
    assert run_body["artifacts"], "a completed run must produce an artifact"
    assert [s["index"] for s in run_body["steps"]] == sorted(
        s["index"] for s in run_body["steps"]
    ), "timeline steps must be ordered"

    # The run payload is customer-facing: no raw model identity anywhere.
    lowered = str(run_body).lower()
    for forbidden in ("provider", "api_key", "internal_name", "model"):
        assert forbidden not in lowered, f"{forbidden} leaked into the run payload"

    # -- 11. see usage summary --------------------------------------------- #
    summary = await client.get(
        f"/api/organizations/{org_id}/billing/summary", headers=headers
    )
    assert summary.status_code == 200, summary.text
    usage = summary.json()
    assert usage["total_spend"] > 0, "a completed run must be metered"
    assert usage["by_unit"], "usage must be attributable to a billing unit"
    assert usage["balance"] is not None

    events = await client.get(
        f"/api/organizations/{org_id}/billing/usage-events", headers=headers
    )
    assert events.status_code == 200, events.text
    assert any(e["run_id"] == run_body["id"] for e in events.json()), (
        "the run's usage must be traceable back to the run"
    )

    # -- 12. audit trail --------------------------------------------------- #
    audit = await client.get(f"/api/organizations/{org_id}/audit", headers=headers)
    assert audit.status_code == 200, audit.text
    actions = {e["action"] for e in audit.json()}
    for expected in ("organization.create", "project.create", "domain.create"):
        assert expected in actions, f"{expected} was not audited"


async def test_smoke_path_is_invisible_to_another_tenant(client: AsyncClient):
    """The same path, viewed by an outsider: every object must 404, not 403."""
    owner_token = (await signup(client, "smoke-owner@example.com"))["access_token"]
    owner_headers = auth_header(owner_token)
    org_id = (
        await client.post(
            "/api/organizations", json={"name": "سازمان مالک"}, headers=owner_headers
        )
    ).json()["id"]
    project_id = (
        await client.post(
            f"/api/organizations/{org_id}/projects",
            json={"name": "پروژهٔ مالک"},
            headers=owner_headers,
        )
    ).json()["id"]

    outsider_token = (await signup(client, "smoke-outsider@example.com"))["access_token"]
    outsider_headers = auth_header(outsider_token)

    # The outsider is authenticated and knows both ids.
    probes = [
        f"/api/organizations/{org_id}",
        f"/api/organizations/{org_id}/projects/{project_id}",
        f"/api/organizations/{org_id}/employees",
        f"/api/organizations/{org_id}/billing/summary",
        f"/api/organizations/{org_id}/audit",
        f"/api/organizations/{org_id}/sandbox/templates",
    ]
    for path in probes:
        response = await client.get(path, headers=outsider_headers)
        assert response.status_code == 404, f"{path} returned {response.status_code}"