pesco-ncr/backend/tests/test_capa.py

"""CAPA section (API Q1 §5.9.1.2 / §6.4.2): field capture with audit, the
closure gate at costing, effectiveness verification, recurring-issue links,
role enforcement, and the report metrics."""
from .util import (
    answer_capa_no,
    create_ncr,
    hdr,
    to_closed,
    to_costing,
    user_id_by_email,
)


async def _capa(client, team, ncr_id: int, body: dict, as_user: str | None = None):
    return await client.post(
        f"/api/ncrs/{ncr_id}/capa", json=body, headers=hdr(as_user or team["qc"])
    )


async def _lookup_root_cause_ids(client, email: str) -> dict[str, int]:
    r = await client.get("/api/lookups", headers=hdr(email))
    assert r.status_code == 200, r.text
    return {c["name"]: c["id"] for c in r.json()["root_cause_categories"]}


async def _full_capa_yes(client, team, ncr_id: int, *, verify: str | None = None):
    """Answer 'CA required = yes' with a complete plan (optionally verified)."""
    rcc = await _lookup_root_cause_ids(client, team["qc"])
    owner_id = await user_id_by_email(client, team["qc"], team["ops"], "operations")
    body = {
        "root_cause": "Fixture PM interval was never defined.",
        "root_cause_category_id": rcc["Method"],
        "corrective_action_required": True,
        "corrective_action_justification": "Systemic cause; will recur without a fix.",
        "corrective_action_plan": "Define quarterly PM for fixture; update WI-204.",
        "corrective_action_owner_id": owner_id,
        "corrective_action_due_date": "2026-09-01",
    }
    if verify is not None:
        body["effectiveness_result"] = verify
    return await _capa(client, team, ncr_id, body)


# ── roles & availability ─────────────────────────────────────────────────────
async def test_capa_requires_qc_or_disposition_role(client, team):
    ncr = await create_ncr(client, team)
    body = {"root_cause": "Some cause."}

    for user in (team["requester"], team["ops"], team["cost"]):
        r = await _capa(client, team, ncr["id"], body, as_user=user)
        assert r.status_code == 403, user

    for user in (team["qc"], team["dispo"], team["admin"]):
        r = await _capa(client, team, ncr["id"], body, as_user=user)
        assert r.status_code == 200, user

    # available_actions advertises capa only to those roles
    r = await client.get(f"/api/ncrs/{ncr['id']}", headers=hdr(team["qc"]))
    assert "capa" in r.json()["available_actions"]
    r = await client.get(f"/api/ncrs/{ncr['id']}", headers=hdr(team["requester"]))
    assert "capa" not in r.json()["available_actions"]


async def test_capa_locked_once_closed(client, team):
    ncr = await to_closed(client, team, (await create_ncr(client, team))["id"])
    r = await _capa(client, team, ncr["id"], {"root_cause": "Too late."})
    assert r.status_code == 409
    assert "closed" in r.json()["detail"].lower()


# ── the CA question ──────────────────────────────────────────────────────────
async def test_ca_question_requires_justification(client, team):
    ncr = await create_ncr(client, team)

    r = await _capa(client, team, ncr["id"], {"corrective_action_required": True})
    assert r.status_code == 422
    assert "justification" in r.json()["detail"].lower()

    r = await _capa(
        client,
        team,
        ncr["id"],
        {
            "corrective_action_required": True,
            "corrective_action_justification": "Repeat risk without process change.",
        },
    )
    assert r.status_code == 200
    body = r.json()["ncr"]
    assert body["corrective_action_required"] is True
    # answering "yes" stamps the CAPA-opened timestamp
    assert body["corrective_action_opened_at"] is not None

    # once a justification is on record, flipping the answer alone is fine
    r = await _capa(client, team, ncr["id"], {"corrective_action_required": False})
    assert r.status_code == 200

    # field-level audit rows were written
    audit = await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["qc"]))
    fields = {a["field_name"] for a in audit.json()["items"] if a["action"] == "capa"}
    assert "corrective_action_required" in fields
    assert "corrective_action_justification" in fields


async def test_capa_field_validation(client, team):
    ncr = await create_ncr(client, team)
    rcc = await _lookup_root_cause_ids(client, team["qc"])

    # inactive/unknown root cause category rejected (lookups only lists active)
    assert "Retired Cause" not in rcc
    r = await _capa(client, team, ncr["id"], {"root_cause_category_id": 99999})
    assert r.status_code == 422

    # unknown owner rejected
    r = await _capa(client, team, ncr["id"], {"corrective_action_owner_id": 99999})
    assert r.status_code == 422

    # effectiveness verification requires CA required = yes
    r = await _capa(client, team, ncr["id"], {"effectiveness_result": "effective"})
    assert r.status_code == 422


# ── the closure gate ─────────────────────────────────────────────────────────
async def test_close_blocked_until_ca_question_answered(client, team):
    ncr = await create_ncr(client, team)
    await to_costing(client, team, ncr["id"])

    costing_body = {
        "labor_cost": "10", "material_cost": "0", "service_cost": "0", "other_cost": "0",
    }
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing", json=costing_body, headers=hdr(team["cost"])
    )
    assert r.status_code == 409
    assert "Corrective Action Required" in r.json()["detail"]

    await answer_capa_no(client, team, ncr["id"])
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing", json=costing_body, headers=hdr(team["cost"])
    )
    assert r.status_code == 200
    assert r.json()["ncr"]["stage"] == "closed"


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

    r = await _full_capa_yes(client, team, ncr["id"])  # complete plan, unverified
    assert r.status_code == 200, r.text

    costing_body = {
        "labor_cost": "10", "material_cost": "0", "service_cost": "0", "other_cost": "0",
    }
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing", json=costing_body, headers=hdr(team["cost"])
    )
    assert r.status_code == 409
    assert "effectiveness" in r.json()["detail"].lower()

    # a failed verification does not satisfy the gate
    r = await _capa(client, team, ncr["id"], {"effectiveness_result": "not_effective"})
    assert r.status_code == 200
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing", json=costing_body, headers=hdr(team["cost"])
    )
    assert r.status_code == 409

    # verified effective → server-stamps who/when, and the NCR can close
    r = await _capa(client, team, ncr["id"], {"effectiveness_result": "effective"})
    body = r.json()["ncr"]
    assert body["effectiveness_verified_by"]["email"] == team["qc"]
    assert body["effectiveness_verified_at"] is not None

    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing", json=costing_body, headers=hdr(team["cost"])
    )
    assert r.status_code == 200
    assert r.json()["ncr"]["stage"] == "closed"


async def test_close_blocked_when_plan_incomplete(client, team):
    ncr = await create_ncr(client, team)
    await to_costing(client, team, ncr["id"])
    # CA = yes but no plan/owner/due date/root cause
    r = await _capa(
        client,
        team,
        ncr["id"],
        {
            "corrective_action_required": True,
            "corrective_action_justification": "Needs a systemic fix.",
        },
    )
    assert r.status_code == 200
    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
    detail = r.json()["detail"]
    for fragment in ("root cause", "plan", "owner", "due date"):
        assert fragment in detail, detail


# ── recurring-issue links ────────────────────────────────────────────────────
async def test_recurring_issue_links(client, team):
    prior = await create_ncr(client, team)
    ncr = await create_ncr(client, team)

    # self-link and unknown ids rejected
    r = await _capa(
        client, team, ncr["id"], {"is_recurring": True, "related_ncr_ids": [ncr["id"]]}
    )
    assert r.status_code == 422
    r = await _capa(client, team, ncr["id"], {"related_ncr_ids": [999999]})
    assert r.status_code == 422

    r = await _capa(
        client,
        team,
        ncr["id"],
        {"is_recurring": True, "related_ncr_ids": [prior["id"]]},
    )
    assert r.status_code == 200
    body = r.json()["ncr"]
    assert body["is_recurring"] is True
    assert [l["ncr_number"] for l in body["related_ncrs"]] == [prior["ncr_number"]]

    # the prior NCR shows the reverse reference
    r = await client.get(f"/api/ncrs/{prior['id']}", headers=hdr(team["requester"]))
    assert [l["ncr_number"] for l in r.json()["referenced_by"]] == [ncr["ncr_number"]]

    # clearing the links removes them
    r = await _capa(client, team, ncr["id"], {"related_ncr_ids": []})
    assert r.json()["ncr"]["related_ncrs"] == []


# ── metrics ──────────────────────────────────────────────────────────────────
async def test_capa_metrics_in_reports_summary(client, team):
    ncr = await create_ncr(client, team)
    rcc = await _lookup_root_cause_ids(client, team["qc"])
    owner_id = await user_id_by_email(client, team["qc"], team["ops"], "operations")

    # an overdue CAPA: required, past due, not verified effective
    r = await _capa(
        client,
        team,
        ncr["id"],
        {
            "root_cause": "Gauge past calibration due date.",
            "root_cause_category_id": rcc["Method"],
            "corrective_action_required": True,
            "corrective_action_justification": "Calibration program gap.",
            "corrective_action_plan": "Add gauge to the calibration recall system.",
            "corrective_action_owner_id": owner_id,
            "corrective_action_due_date": "2020-01-01",
        },
    )
    assert r.status_code == 200, r.text

    r = await client.get("/api/reports/summary", headers=hdr(team["qc"]))
    assert r.status_code == 200
    data = r.json()
    assert data["overdue_capa_count"] >= 1
    assert data["root_cause_pct"] is not None and data["root_cause_pct"] > 0
    # this CA-required NCR is not verified, so the pct must be < 100 when present
    if data["effectiveness_verified_pct"] is not None:
        assert data["effectiveness_verified_pct"] < 100
    assert any(c["name"] == "Method" for c in data["by_root_cause_category"])

    # close a verified-effective CAPA and the avg close time appears
    ncr2 = await create_ncr(client, team)
    await to_costing(client, team, ncr2["id"])
    r = await _full_capa_yes(client, team, ncr2["id"], verify="effective")
    assert r.status_code == 200, r.text
    r = await client.get("/api/reports/summary", headers=hdr(team["qc"]))
    assert r.json()["avg_capa_close_days"] is not None


async def test_csv_export_includes_capa_columns(client, team):
    await create_ncr(client, team)
    r = await client.get("/api/ncrs/export.csv", headers=hdr(team["qc"]))
    assert r.status_code == 200
    header = r.text.splitlines()[0]
    for col in (
        "root_cause_category",
        "corrective_action_required",
        "corrective_action_owner",
        "corrective_action_due_date",
        "effectiveness_result",
        "is_recurring",
    ):
        assert col in header