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

"""Pause, resume, cancel and answer on an engine task, through the HTTP routes.

What these tests defend:

* a member with ``tasks.run`` pauses and resumes a funded task; resume is refused with the
  Persian budget reason while the organization has no credit, and the task stays paused
* cancel ends the task and writes one ``task.control`` audit row that names the action
* an answer is accepted only from ``pending_user_decision``; its text never reaches the audit
  trail, and a blank answer is a 422
* a viewer gets 403, and a member of another organization gets 404, on every action
"""

from __future__ import annotations

from sqlalchemy import select

from app.application.engine.scheduler import BUDGET_PAUSED_MESSAGE
from app.domain.identity.models import Role
from app.domain.workforce.pipeline import Stage
from app.infrastructure.db.models import AuditEventRow
from app.infrastructure.repositories.engine import PipelineStateRepository
from tests.billing_helpers import fund_org
from tests.conftest import auth_header
from tests.test_engine_api import _base, _engine_task, _join, _make_org

ANSWER = "نسخهٔ دوم را انجام بده"


async def _post(client, token: str, organization_id: str, task_id: str, action: str, json=None):
    return await client.post(
        f"{_base(organization_id)}/tasks/{task_id}/{action}",
        json=json,
        headers=auth_header(token),
    )


async def _stage_of(session_factory, task_id: str) -> Stage | None:
    async with session_factory() as session:
        state = await PipelineStateRepository(session).get(task_id)
    return state.stage if state is not None else None


async def _task_at(session_factory, org, stage: Stage, title: str = "صفحهٔ ورود") -> str:
    async with session_factory() as session:
        row = await _engine_task(session, org, title=title, stage=stage)
        await session.commit()
        return row.id


async def test_a_member_pauses_then_resumes_a_funded_task(client, session_factory, seeded):
    org = await _make_org(client, "act-pause")
    await fund_org(session_factory, org.id, 40.0)
    member = await _join(client, session_factory, org.id, Role.MEMBER, "act-member@example.com")
    task_id = await _task_at(session_factory, org, Stage.PENDING_IMPLEMENTATION)

    paused = await _post(client, member, org.id, task_id, "pause")

    assert paused.status_code == 200, paused.text
    assert paused.json()["stage"] == "paused"
    assert paused.json()["status"] == "waiting_approval"

    resumed = await _post(client, member, org.id, task_id, "resume")

    assert resumed.status_code == 200, resumed.text
    assert resumed.json()["stage"] == "pending_manager_review"


async def test_resume_is_refused_without_credit_and_the_task_stays_paused(
    client, session_factory, seeded
):
    org = await _make_org(client, "act-zero")
    member = await _join(
        client, session_factory, org.id, Role.MEMBER, "act-zero-member@example.com"
    )
    task_id = await _task_at(session_factory, org, Stage.PAUSED)

    refused = await _post(client, member, org.id, task_id, "resume")

    assert refused.status_code == 409, refused.text
    assert refused.json()["error"]["message"] == BUDGET_PAUSED_MESSAGE
    assert await _stage_of(session_factory, task_id) is Stage.PAUSED


async def test_cancel_ends_the_task_and_writes_one_audit_row(client, session_factory, seeded):
    org = await _make_org(client, "act-cancel")
    member = await _join(
        client, session_factory, org.id, Role.MEMBER, "act-cancel-member@example.com"
    )
    task_id = await _task_at(session_factory, org, Stage.PENDING_IMPLEMENTATION)

    cancelled = await _post(client, member, org.id, task_id, "cancel")

    assert cancelled.status_code == 200, cancelled.text
    assert cancelled.json()["stage"] == "cancelled"
    assert cancelled.json()["status"] == "cancelled"
    async with session_factory() as session:
        events = list(
            (
                await session.execute(
                    select(AuditEventRow).where(
                        AuditEventRow.action == "task.control",
                        AuditEventRow.target_id == task_id,
                    )
                )
            )
            .scalars()
            .all()
        )
    assert len(events) == 1
    assert events[0].organization_id == org.id
    assert events[0].target_type == "task"
    assert events[0].metadata_json == {"action": "cancel"}


async def test_an_answer_is_accepted_only_from_a_decision_and_stays_out_of_the_audit(
    client, session_factory, seeded
):
    org = await _make_org(client, "act-answer")
    member = await _join(
        client, session_factory, org.id, Role.MEMBER, "act-answer-member@example.com"
    )
    working = await _task_at(session_factory, org, Stage.PENDING_IMPLEMENTATION, "کار یک")
    deciding = await _task_at(session_factory, org, Stage.PENDING_USER_DECISION, "کار دو")

    refused = await _post(
        client, member, org.id, working, "answer", json={"content": ANSWER}
    )
    assert refused.status_code == 409, refused.text
    assert await _stage_of(session_factory, working) is Stage.PENDING_IMPLEMENTATION

    blank = await _post(client, member, org.id, deciding, "answer", json={"content": ""})
    assert blank.status_code == 422, blank.text
    assert await _stage_of(session_factory, deciding) is Stage.PENDING_USER_DECISION

    accepted = await _post(
        client, member, org.id, deciding, "answer", json={"content": ANSWER}
    )
    assert accepted.status_code == 200, accepted.text
    assert accepted.json()["stage"] == "pending_manager_review"

    async with session_factory() as session:
        events = list(
            (
                await session.execute(
                    select(AuditEventRow).where(
                        AuditEventRow.action == "task.control",
                        AuditEventRow.target_id == deciding,
                    )
                )
            )
            .scalars()
            .all()
        )
    assert [event.metadata_json for event in events] == [{"action": "answer"}]
    assert ANSWER not in repr(events[0].metadata_json)


async def test_a_viewer_is_refused_and_a_foreign_member_gets_not_found(
    client, session_factory, seeded
):
    org = await _make_org(client, "act-viewer")
    viewer = await _join(
        client, session_factory, org.id, Role.VIEWER, "act-viewer-user@example.com"
    )
    other = await _make_org(client, "act-other")
    task_id = await _task_at(session_factory, org, Stage.PENDING_USER_DECISION)

    for action in ("pause", "resume", "cancel"):
        response = await _post(client, viewer, org.id, task_id, action)
        assert response.status_code == 403, f"{action}: {response.text}"
    answer = await _post(client, viewer, org.id, task_id, "answer", json={"content": ANSWER})
    assert answer.status_code == 403, answer.text
    assert await _stage_of(session_factory, task_id) is Stage.PENDING_USER_DECISION

    for action in ("pause", "resume", "cancel"):
        response = await _post(client, other.owner_token, org.id, task_id, action)
        assert response.status_code == 404, f"{action}: {response.text}"
    answer = await _post(
        client, other.owner_token, org.id, task_id, "answer", json={"content": ANSWER}
    )
    assert answer.status_code == 404, answer.text

    # The other organization's own routes do not reach this task either.
    inside = await _post(client, other.owner_token, other.id, task_id, "pause")
    assert inside.status_code == 404, inside.text
    assert await _stage_of(session_factory, task_id) is Stage.PENDING_USER_DECISION