pesco-ncr/backend/tests/test_state_machine.py

"""Workflow state machine: happy paths, invalid transitions, closure locking,
admin reopen, and rich-text sanitization."""
from .util import (
    create_ncr,
    do_initial_disposition,
    hdr,
    to_closed,
    to_costing,
    to_operations,
    to_qc_inspection,
    user_id_by_email,
)


async def test_full_lifecycle_direct_to_operations(client, team):
    ncr = await create_ncr(client, team)
    assert ncr["stage"] == "new_request"
    assert ncr["ncr_number"].startswith("NCR-")

    ncr = await to_operations(client, team, ncr["id"])
    assert ncr["stage"] == "operations"
    assert ncr["secondary_review_needed"] is False
    assert ncr["qc_authority"] == "AS9100 8.7"

    r = await client.post(
        f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["ops"])
    )
    body = r.json()["ncr"]
    assert body["stage"] == "qc_inspection"
    assert body["operations_complete"] is True
    assert body["operations_completed_by"]["email"] == team["ops"]

    # QC can save repeatedly without closing
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/inspection",
        json={"qc_approval": "no", "inspection_notes": "First pass failed."},
        headers=hdr(team["qc"]),
    )
    assert r.json()["ncr"]["stage"] == "qc_inspection"

    r = await client.post(
        f"/api/ncrs/{ncr['id']}/inspection",
        json={"qc_approval": "yes", "inspection_notes": "Rework verified.", "qc_closed": True},
        headers=hdr(team["qc"]),
    )
    body = r.json()["ncr"]
    assert body["stage"] == "costing"
    assert body["qc_closed"] is True

    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing",
        json={
            "labor_cost": "100.00",
            "material_cost": "50.25",
            "service_cost": "0",
            "other_cost": "10",
        },
        headers=hdr(team["cost"]),
    )
    body = r.json()["ncr"]
    assert body["stage"] == "closed"
    assert body["total_cost"] == "160.25"
    assert body["closed_at"] is not None

    # Transition history is complete and ordered
    stages = [t["to_stage"] for t in body["transitions"]]
    assert stages == ["new_request", "operations", "qc_inspection", "costing", "closed"]


async def test_secondary_disposition_flow(client, team):
    ncr = await create_ncr(client, team)
    second_id = await user_id_by_email(
        client, team["dispo"], team["second"], "secondary_disposition_authority"
    )

    # secondary review without assignees is rejected
    r = await do_initial_disposition(client, team, ncr["id"], secondary=True, secondary_ids=[])
    assert r.status_code == 422

    r = await do_initial_disposition(
        client, team, ncr["id"], secondary=True, secondary_ids=[second_id]
    )
    body = r.json()["ncr"]
    assert body["stage"] == "secondary_disposition"
    assert [u["email"] for u in body["secondary_authorities"]] == [team["second"]]

    # assignee saves without releasing
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/secondary-disposition",
        json={"disposition_notes": "<p>Updated by secondary.</p>", "release": False},
        headers=hdr(team["second"]),
    )
    assert r.json()["ncr"]["stage"] == "secondary_disposition"

    # then releases to operations
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/secondary-disposition",
        json={"work_order": "WO-2002", "release": True},
        headers=hdr(team["second"]),
    )
    body = r.json()["ncr"]
    assert body["stage"] == "operations"
    assert body["work_order"] == "WO-2002"
    # earlier saved notes were not wiped by the release payload
    assert "Updated by secondary" in body["disposition_notes"]


async def test_invalid_transitions_rejected(client, team):
    ncr = await create_ncr(client, team)

    # can't skip ahead from new_request
    r = await client.post(f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["ops"]))
    assert r.status_code == 409
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/inspection",
        json={"qc_closed": True},
        headers=hdr(team["qc"]),
    )
    assert r.status_code == 409
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing",
        json={"labor_cost": "1", "material_cost": "1", "service_cost": "1", "other_cost": "1"},
        headers=hdr(team["cost"]),
    )
    assert r.status_code == 409

    # once in operations, initial disposition can't run again
    await to_operations(client, team, ncr["id"])
    r = await do_initial_disposition(client, team, ncr["id"])
    assert r.status_code == 409


async def test_closed_ncr_is_fully_locked(client, team):
    ncr = await to_closed(client, team, (await create_ncr(client, team))["id"])
    assert ncr["stage"] == "closed"

    for path, payload, user in [
        ("initial-disposition", {"secondary_review_needed": False}, team["dispo"]),
        ("secondary-disposition", {"release": True}, team["second"]),
        ("operations-complete", None, team["ops"]),
        ("inspection", {"qc_closed": True}, team["qc"]),
        ("costing", {"labor_cost": "9", "material_cost": "9", "service_cost": "9", "other_cost": "9"}, team["cost"]),
    ]:
        r = await client.post(
            f"/api/ncrs/{ncr['id']}/{path}",
            json=payload,
            headers=hdr(user),
        )
        assert r.status_code == 409, f"{path}: {r.status_code} {r.text}"
        assert "closed" in r.json()["detail"].lower()


async def test_admin_reopen_with_reason(client, team):
    ncr = await to_closed(client, team, (await create_ncr(client, team))["id"])

    # non-admin cannot reopen
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/reopen",
        json={"to_stage": "costing", "reason": "Costs were entered incorrectly."},
        headers=hdr(team["cost"]),
    )
    assert r.status_code == 403

    # reason is required (min length)
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/reopen",
        json={"to_stage": "costing", "reason": ""},
        headers=hdr(team["admin"]),
    )
    assert r.status_code == 422

    # reopening to 'closed' is not a valid target
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/reopen",
        json={"to_stage": "closed", "reason": "does not make sense"},
        headers=hdr(team["admin"]),
    )
    assert r.status_code == 422

    r = await client.post(
        f"/api/ncrs/{ncr['id']}/reopen",
        json={"to_stage": "costing", "reason": "Costs were entered incorrectly."},
        headers=hdr(team["admin"]),
    )
    body = r.json()["ncr"]
    assert body["stage"] == "costing"
    assert body["closed_at"] is None
    # costs preserved for correction
    assert body["labor_cost"] == "125.50"

    # reopen is recorded with its reason in the transition history + audit trail
    reopen_t = [t for t in body["transitions"] if t["action"] == "reopen"]
    assert len(reopen_t) == 1
    assert "Costs were entered incorrectly." in reopen_t[0]["note"]

    audit = await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["admin"]))
    actions = [a["action"] for a in audit.json()["items"]]
    assert "reopen" in actions

    # workflow resumes: costing can close it again
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing",
        json={"labor_cost": "200", "material_cost": "0", "service_cost": "0", "other_cost": "0"},
        headers=hdr(team["cost"]),
    )
    assert r.json()["ncr"]["stage"] == "closed"


async def test_reopen_only_from_closed(client, team):
    ncr = await create_ncr(client, team)
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/reopen",
        json={"to_stage": "new_request", "reason": "not closed yet"},
        headers=hdr(team["admin"]),
    )
    assert r.status_code == 409


async def test_rich_text_is_sanitized(client, team):
    ncr = await create_ncr(client, team)
    r = await do_initial_disposition(
        client,
        team,
        ncr["id"],
        notes='<p onclick="evil()">Keep</p><script>alert("xss")</script><a href="javascript:x()">link</a>',
    )
    notes = r.json()["ncr"]["disposition_notes"]
    assert "<script" not in notes
    assert "onclick" not in notes
    assert "javascript:" not in notes
    assert "Keep" in notes


async def test_audit_trail_field_level(client, team):
    ncr = await create_ncr(client, team)
    await to_operations(client, team, ncr["id"])

    r = await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["qc"]))
    items = r.json()["items"]
    by_field = {i["field_name"]: i for i in items if i["field_name"]}
    assert by_field["stage"]["old_value"] == "new_request"
    assert by_field["stage"]["new_value"] == "operations"
    assert by_field["work_order"]["new_value"] == "WO-1001"
    assert any(i["action"] == "create" for i in items)