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

"""The engine's read API: status, task board, task detail and recent turns.

What these tests defend:

* any member, including a viewer, reads every route; a foreign member gets 404 on every
  route, and a task id of another organization is a 404 under the caller's own organization
* ``status`` reports the wallet balance (``credit_ok`` only above zero), the owner's cap, the
  headcount and the stage counts of the organization
* the board lists engine and legacy tasks, and ``stage`` narrows it to one stage
* the task detail carries the pipeline stages, the turns, the gate verdicts, the runs, the
  last diff and the merged sha; a turn excerpt is capped at 1,000 characters
* no response names a model, a provider or an ``ai_model_id`` (every seeded model and
  provider name is scanned in every body)
"""

from __future__ import annotations

from dataclasses import dataclass

from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker

from app.application.container_org import build_model_grants
from app.domain.base import utcnow
from app.domain.identity.models import Role
from app.domain.workforce.engine import EmployeeProfile, EngineTurn, GateName, GateResult
from app.domain.workforce.models import Employee, RoleKey, Run, RunStatus
from app.domain.workforce.pipeline import (
    Mode,
    PipelineKey,
    Stage,
    TaskKind,
    TaskPipelineState,
    pipeline_stages,
)
from app.infrastructure.db.models import AiModelRow, AiProviderRow, MembershipRow, TaskRow
from app.infrastructure.repositories.engine import (
    EmployeeProfileRepository,
    EngineTurnRepository,
    GateResultRepository,
    PipelineStateRepository,
)
from app.infrastructure.repositories.workforce import EmployeeRepository, RunRepository
from tests.billing_helpers import fund_org
from tests.conftest import auth_header, signup

DIFF = "diff --git a/app/login.py b/app/login.py\n+def login(): return True\n"
SHA = "9f8e7d6c5b4a39281706f5e4d3c2b1a098765432"
TITLE = "صفحهٔ ورود"
EMPLOYEE = "برنامه‌نویس یک"


def _base(org_id: str) -> str:
    return f"/api/organizations/{org_id}/engine"


def _payload(name: str, slug: str) -> dict:
    return {
        "org_name": name,
        "slug": slug,
        "app_name": f"app-{slug}",
        "app_type": "web_app",
        "language": "node",
        "framework": "next",
        "git_source": "platform",
    }


@dataclass
class _Org:
    id: str
    project_id: str
    owner_id: str
    owner_token: str


@dataclass
class _Data:
    org: _Org
    model_id: str
    done_task_id: str
    testing_task_id: str
    legacy_task_id: str


async def _make_org(client: AsyncClient, slug: str) -> _Org:
    owner = await signup(client, f"{slug}-owner@example.com")
    response = await client.post(
        "/api/setup/complete",
        json=_payload(f"سازمان {slug}", slug),
        headers=auth_header(owner["access_token"]),
    )
    assert response.status_code == 201, response.text
    body = response.json()
    return _Org(
        id=body["organization"]["id"],
        project_id=body["project"]["id"],
        owner_id=owner["user"]["id"],
        owner_token=owner["access_token"],
    )


async def _join(
    client: AsyncClient,
    session_factory: async_sessionmaker,
    organization_id: str,
    role: Role,
    email: str,
) -> str:
    """Add a fresh account to an organization with ``role``; returns its token."""
    user = await signup(client, email)
    async with session_factory() as session:
        session.add(
            MembershipRow(
                organization_id=organization_id, user_id=user["user"]["id"], role=role.value
            )
        )
        await session.commit()
    return user["access_token"]


async def _engine_task(session, org: _Org, *, title: str, stage: Stage, **fields) -> TaskRow:
    employee_id = fields.pop("employee_id", None)
    row = TaskRow(
        organization_id=org.id,
        title=title,
        description="وصف کار",
        project_id=org.project_id,
        created_by_user_id=org.owner_id,
        assigned_employee_id=employee_id,
    )
    session.add(row)
    await session.flush()
    await PipelineStateRepository(session).create(
        TaskPipelineState(
            task_id=row.id,
            organization_id=org.id,
            project_id=org.project_id,
            pipeline=PipelineKey.PLAN_CODE_TEST,
            kind=TaskKind.FEATURE,
            stage=stage,
            **fields,
        )
    )
    return row


async def _seed(session_factory: async_sessionmaker, org: _Org) -> _Data:
    """One finished engine task (turn, gate, run, diff, sha), one at a test stage, one legacy."""
    async with session_factory() as session:
        model = (await session.execute(select(AiModelRow).limit(1))).scalars().first()
        assert model is not None
        developer = await EmployeeRepository(session).create(
            Employee(organization_id=org.id, name=EMPLOYEE, role_key=RoleKey.DEVELOPER)
        )
        await EmployeeProfileRepository(session).upsert(
            EmployeeProfile(
                employee_id=developer.id,
                ai_model_id=model.id,
                roles=[RoleKey.DEVELOPER.value],
            )
        )
        done = await _engine_task(
            session,
            org,
            title=TITLE,
            stage=Stage.COMPLETED,
            employee_id=developer.id,
            last_diff=DIFF,
            merged_sha=SHA,
        )
        testing = await _engine_task(session, org, title="جدول کاربران", stage=Stage.PENDING_TEST)
        legacy = TaskRow(
            organization_id=org.id,
            title="کار قدیمی",
            project_id=org.project_id,
            created_by_user_id=org.owner_id,
        )
        session.add(legacy)
        await session.flush()
        run = await RunRepository(session).create(
            Run(task_id=done.id, employee_id=developer.id, status=RunStatus.COMPLETED)
        )
        await EngineTurnRepository(session).create(
            EngineTurn(
                task_id=done.id,
                run_id=run.id,
                employee_id=developer.id,
                role=RoleKey.DEVELOPER,
                mode=Mode.WORK,
                stage=Stage.PENDING_IMPLEMENTATION,
                reply_excerpt="پاسخ " * 400,
                ok=True,
                finished_at=utcnow(),
            )
        )
        await GateResultRepository(session).create(
            GateResult(task_id=done.id, gate=GateName.GREEN, ok=True, detail="آزمون‌ها سبز شد.")
        )
        await session.commit()
        return _Data(
            org=org,
            model_id=model.id,
            done_task_id=done.id,
            testing_task_id=testing.id,
            legacy_task_id=legacy.id,
        )


async def _get(client: AsyncClient, token: str, url: str):
    return await client.get(url, headers=auth_header(token))


# ---------------------------------------------------------------------------
# status
# ---------------------------------------------------------------------------


async def test_status_reports_credit_policy_team_and_stage_counts(
    client, session_factory, seeded
):
    org = await _make_org(client, "eng-status")
    await fund_org(session_factory, org.id, 40.0)
    await _seed(session_factory, org)
    async with session_factory() as session:
        cap = (await build_model_grants(session).get_org_policy(org.id)).max_employees
        headcount = await EmployeeRepository(session).count_for_org(org.id)

    response = await _get(client, org.owner_token, f"{_base(org.id)}/status")

    assert response.status_code == 200, response.text
    body = response.json()
    assert body["credit_ok"] is True
    assert body["balance"] == 40.0
    assert body["currency"] == "USD"
    assert body["max_employees"] == cap
    assert body["team_size"] == headcount
    assert body["stage_counts"] == {"completed": 1, "pending_test": 1}


async def test_status_is_not_credit_ok_without_credit(client, session_factory, seeded):
    org = await _make_org(client, "eng-zero")

    response = await _get(client, org.owner_token, f"{_base(org.id)}/status")

    assert response.status_code == 200, response.text
    assert response.json()["credit_ok"] is False
    assert response.json()["balance"] == 0.0


# ---------------------------------------------------------------------------
# board
# ---------------------------------------------------------------------------


async def test_the_board_lists_engine_and_legacy_tasks_and_filters_by_stage(
    client, session_factory, seeded
):
    org = await _make_org(client, "eng-board")
    data = await _seed(session_factory, org)
    board = f"{_base(org.id)}/tasks"

    everything = await _get(client, org.owner_token, board)

    assert everything.status_code == 200, everything.text
    rows = {row["id"]: row for row in everything.json()}
    assert set(rows) == {data.done_task_id, data.testing_task_id, data.legacy_task_id}
    assert rows[data.legacy_task_id]["stage"] is None
    assert rows[data.legacy_task_id]["pipeline"] is None
    assert rows[data.legacy_task_id]["cycle"] == 0
    assert rows[data.testing_task_id]["stage"] == "pending_test"
    assert rows[data.done_task_id]["assignee_name"] == EMPLOYEE

    testing = await client.get(
        board, params={"stage": "pending_test"}, headers=auth_header(org.owner_token)
    )
    assert [row["id"] for row in testing.json()] == [data.testing_task_id]

    limited = await client.get(board, params={"limit": 1}, headers=auth_header(org.owner_token))
    assert len(limited.json()) == 1

    unknown = await client.get(
        board, params={"stage": "nope"}, headers=auth_header(org.owner_token)
    )
    assert unknown.status_code == 422


# ---------------------------------------------------------------------------
# detail and turns
# ---------------------------------------------------------------------------


async def test_task_detail_returns_stages_turns_gates_runs_diff_and_sha(
    client, session_factory, seeded
):
    org = await _make_org(client, "eng-detail")
    data = await _seed(session_factory, org)

    response = await _get(client, org.owner_token, f"{_base(org.id)}/tasks/{data.done_task_id}")

    assert response.status_code == 200, response.text
    body = response.json()
    assert body["stage"] == "completed"
    assert body["description"] == "وصف کار"
    assert body["stages"] == [str(s) for s in pipeline_stages(PipelineKey.PLAN_CODE_TEST)]
    assert body["merged_sha"] == SHA
    assert body["diff"] == DIFF
    assert body["assignee_name"] == EMPLOYEE
    [turn] = body["turns"]
    assert turn["employee_name"] == EMPLOYEE
    assert turn["role"] == "developer"
    assert turn["mode"] == str(Mode.WORK)
    assert len(turn["reply_excerpt"]) == 1000
    [gate] = body["gates"]
    assert gate["gate"] == "green"
    assert gate["ok"] is True
    [run] = body["runs"]
    assert run["status"] == "completed"


async def test_recent_turns_carry_the_task_title_and_the_employee_name(
    client, session_factory, seeded
):
    org = await _make_org(client, "eng-turns")
    data = await _seed(session_factory, org)

    response = await _get(client, org.owner_token, f"{_base(org.id)}/turns")

    assert response.status_code == 200, response.text
    [turn] = response.json()
    assert turn["task_id"] == data.done_task_id
    assert turn["task_title"] == TITLE
    assert turn["employee_name"] == EMPLOYEE

    too_many = await _get(client, org.owner_token, f"{_base(org.id)}/turns?limit=0")
    assert too_many.status_code == 422


# ---------------------------------------------------------------------------
# who may read
# ---------------------------------------------------------------------------


async def test_a_viewer_reads_every_route(client, session_factory, seeded):
    org = await _make_org(client, "eng-viewer")
    data = await _seed(session_factory, org)
    token = await _join(
        client, session_factory, org.id, Role.VIEWER, "eng-viewer-user@example.com"
    )

    for path in ("/status", "/tasks", f"/tasks/{data.done_task_id}", "/turns"):
        response = await _get(client, token, f"{_base(org.id)}{path}")
        assert response.status_code == 200, f"{path}: {response.text}"


async def test_a_foreign_member_gets_404_on_every_route(client, session_factory, seeded):
    org_a = await _make_org(client, "eng-a")
    data = await _seed(session_factory, org_a)
    org_b = await _make_org(client, "eng-b")

    for path in ("/status", "/tasks", f"/tasks/{data.done_task_id}", "/turns"):
        response = await _get(client, org_b.owner_token, f"{_base(org_a.id)}{path}")
        assert response.status_code == 404, f"{path}: {response.text}"

    # A's task id under B's own organization is not found, and B's board stays empty.
    inside_b = await _get(client, org_b.owner_token, f"{_base(org_b.id)}/tasks/{data.done_task_id}")
    assert inside_b.status_code == 404
    board_b = await _get(client, org_b.owner_token, f"{_base(org_b.id)}/tasks")
    assert board_b.json() == []
    turns_b = await _get(client, org_b.owner_token, f"{_base(org_b.id)}/turns")
    assert turns_b.json() == []


# ---------------------------------------------------------------------------
# no model identity
# ---------------------------------------------------------------------------


async def test_no_response_names_a_model_a_provider_or_an_ai_model_id(
    client, session_factory, seeded
):
    org = await _make_org(client, "eng-names")
    data = await _seed(session_factory, org)
    async with session_factory() as session:
        models = list((await session.execute(select(AiModelRow))).scalars().all())
        providers = list((await session.execute(select(AiProviderRow))).scalars().all())
    forbidden = {data.model_id}
    for model in models:
        forbidden.update({model.internal_name, model.display_name_for_admin})
    forbidden.update(provider.name for provider in providers)
    # Names shorter than three characters would match ordinary text; the seed has none.
    forbidden = {value for value in forbidden if len(value) >= 3}
    assert forbidden

    base = _base(org.id)
    paths = (
        "/status",
        "/tasks",
        f"/tasks/{data.done_task_id}",
        f"/tasks/{data.testing_task_id}",
        "/turns",
    )
    for path in paths:
        response = await _get(client, org.owner_token, f"{base}{path}")
        assert response.status_code == 200, f"{path}: {response.text}"
        for value in forbidden:
            assert value not in response.text, f"{path} names {value!r}"