c1cd1bc9df
Implements tester feedback against API Q1 §5.9.1.2 / §6.4.2: - Root Cause field + 6M Root Cause Category lookup (Man/Machine/Method/ Material/Measurement/Environment), separate from Deviation Detail/Category - "Corrective Action Required?" Yes/No gate on every NCR with a required justification - Corrective action plan with owner + due date; owner is notified by email - Effectiveness verification (result, notes, server-stamped verifier/date) required before an NCR can close when corrective action is required — costing returns 409 listing the missing pieces - Recurring-issue flag with bidirectional NCR-to-NCR links; prior NCRs show a warning when later NCRs reference them - Dashboard metrics: % root cause completed, % CAPA verified effective, avg CAPA close time, overdue CAPA count, NCRs by root cause category - CAPA section in the NCR detail UI, printable PDF, CSV export, and the vw_ncr_full Power BI view; admin list manager for root cause categories - Migrations 0003 (schema + seeded 6M lookup) and 0004 (view refresh); demo seed data exercises every metric Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
233 lines
8.1 KiB
Python
233 lines
8.1 KiB
Python
"""Server-side role enforcement per stage and admin-area authorization."""
|
|
from .util import (
|
|
create_ncr,
|
|
do_initial_disposition,
|
|
hdr,
|
|
to_costing,
|
|
to_operations,
|
|
to_qc_inspection,
|
|
user_id_by_email,
|
|
)
|
|
|
|
|
|
async def test_stage_actions_require_stage_role(client, team):
|
|
ncr = await create_ncr(client, team)
|
|
|
|
# a plain requester can't perform initial disposition
|
|
r = await do_initial_disposition(client, team, ncr["id"], as_user=team["requester"])
|
|
assert r.status_code == 403
|
|
# nor can operations/qc/costing roles
|
|
r = await do_initial_disposition(client, team, ncr["id"], as_user=team["ops"])
|
|
assert r.status_code == 403
|
|
|
|
await to_operations(client, team, ncr["id"])
|
|
# only operations can mark complete
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["requester"])
|
|
)
|
|
assert r.status_code == 403
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["qc"])
|
|
)
|
|
assert r.status_code == 403
|
|
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["ops"])
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
# only QC can edit inspection fields
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/inspection",
|
|
json={"qc_approval": "yes"},
|
|
headers=hdr(team["ops"]),
|
|
)
|
|
assert r.status_code == 403
|
|
|
|
# only costing can cost
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/inspection",
|
|
json={"qc_approval": "yes", "qc_closed": True},
|
|
headers=hdr(team["qc"]),
|
|
)
|
|
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["qc"]),
|
|
)
|
|
assert r.status_code == 403
|
|
|
|
|
|
async def test_admin_can_act_at_every_stage(client, team):
|
|
ncr = await create_ncr(client, team)
|
|
r = await do_initial_disposition(client, team, ncr["id"], as_user=team["admin"])
|
|
assert r.status_code == 200
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["admin"])
|
|
)
|
|
assert r.status_code == 200
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/inspection",
|
|
json={"qc_approval": "yes", "qc_closed": True},
|
|
headers=hdr(team["admin"]),
|
|
)
|
|
assert r.status_code == 200
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/capa",
|
|
json={
|
|
"corrective_action_required": False,
|
|
"corrective_action_justification": "Isolated incident; no systemic cause.",
|
|
},
|
|
headers=hdr(team["admin"]),
|
|
)
|
|
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["admin"]),
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["ncr"]["stage"] == "closed"
|
|
|
|
|
|
async def test_secondary_restricted_to_assignees(client, team):
|
|
ncr = await create_ncr(client, team)
|
|
second_id = await user_id_by_email(
|
|
client, team["dispo"], team["second"], "secondary_disposition_authority"
|
|
)
|
|
await do_initial_disposition(
|
|
client, team, ncr["id"], secondary=True, secondary_ids=[second_id]
|
|
)
|
|
|
|
# another user holding the secondary role but NOT assigned is rejected
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/secondary-disposition",
|
|
json={"release": True},
|
|
headers=hdr(team["second2"]),
|
|
)
|
|
assert r.status_code == 403
|
|
|
|
# the assignee is allowed
|
|
r = await client.post(
|
|
f"/api/ncrs/{ncr['id']}/secondary-disposition",
|
|
json={"release": True},
|
|
headers=hdr(team["second"]),
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["ncr"]["stage"] == "operations"
|
|
|
|
|
|
async def test_secondary_queue_filtered_by_identity(client, team):
|
|
ncr = await create_ncr(client, team)
|
|
second_id = await user_id_by_email(
|
|
client, team["dispo"], team["second"], "secondary_disposition_authority"
|
|
)
|
|
await do_initial_disposition(
|
|
client, team, ncr["id"], secondary=True, secondary_ids=[second_id]
|
|
)
|
|
|
|
r = await client.get("/api/ncrs?queue=secondary", headers=hdr(team["second"]))
|
|
assert any(item["id"] == ncr["id"] for item in r.json()["items"])
|
|
|
|
r = await client.get("/api/ncrs?queue=secondary", headers=hdr(team["second2"]))
|
|
assert not any(item["id"] == ncr["id"] for item in r.json()["items"])
|
|
|
|
|
|
async def test_create_requires_valid_disposition_authority(client, team):
|
|
from .util import lookup_ids
|
|
|
|
dept_id, cat_id = await lookup_ids(client, team["requester"])
|
|
ops_id = await user_id_by_email(client, team["requester"], team["ops"], "operations")
|
|
r = await client.post(
|
|
"/api/ncrs",
|
|
json={
|
|
"job_number": "J1",
|
|
"department_id": dept_id,
|
|
"deviation_category_id": cat_id,
|
|
"disposition_authority_id": ops_id, # lacks the role
|
|
"deviation_detail": "Detail long enough.",
|
|
},
|
|
headers=hdr(team["requester"]),
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
|
|
async def test_secondary_assignees_must_hold_role(client, team):
|
|
ncr = await create_ncr(client, team)
|
|
ops_id = await user_id_by_email(client, team["dispo"], team["ops"], "operations")
|
|
r = await do_initial_disposition(
|
|
client, team, ncr["id"], secondary=True, secondary_ids=[ops_id]
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
|
|
async def test_audit_endpoint_restricted(client, team):
|
|
ncr = await create_ncr(client, team)
|
|
r = await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["requester"]))
|
|
assert r.status_code == 403
|
|
r = await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["qc"]))
|
|
assert r.status_code == 200
|
|
r = await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["admin"]))
|
|
assert r.status_code == 200
|
|
|
|
|
|
async def test_admin_area_requires_admin(client, team):
|
|
for path in ("/api/admin/users", "/api/admin/departments", "/api/admin/settings",
|
|
"/api/admin/audit"):
|
|
r = await client.get(path, headers=hdr(team["requester"]))
|
|
assert r.status_code == 403, path
|
|
r = await client.get(path, headers=hdr(team["admin"]))
|
|
assert r.status_code == 200, path
|
|
|
|
|
|
async def test_role_assignment_and_lockout_protection(client, team, make_user):
|
|
target = await make_user(["requester"])
|
|
r = await client.get("/api/admin/users", headers=hdr(team["admin"]))
|
|
target_id = next(u["id"] for u in r.json() if u["email"] == target)
|
|
admin_id = next(u["id"] for u in r.json() if u["email"] == team["admin"])
|
|
|
|
r = await client.put(
|
|
f"/api/admin/users/{target_id}/roles",
|
|
json={"roles": ["requester", "qc_inspector"]},
|
|
headers=hdr(team["admin"]),
|
|
)
|
|
assert r.status_code == 200
|
|
assert set(r.json()["roles"]) == {"requester", "qc_inspector"}
|
|
|
|
# unknown role rejected
|
|
r = await client.put(
|
|
f"/api/admin/users/{target_id}/roles",
|
|
json={"roles": ["superuser"]},
|
|
headers=hdr(team["admin"]),
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
# admin cannot remove their own admin role
|
|
r = await client.put(
|
|
f"/api/admin/users/{admin_id}/roles",
|
|
json={"roles": ["requester"]},
|
|
headers=hdr(team["admin"]),
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
|
|
async def test_everyone_can_view_and_search(client, team):
|
|
ncr = await create_ncr(client, team)
|
|
r = await client.get(f"/api/ncrs/{ncr['id']}", headers=hdr(team["ops"]))
|
|
assert r.status_code == 200
|
|
# available_actions reflect the viewer's role
|
|
assert "initial_disposition" not in r.json()["available_actions"]
|
|
r = await client.get(f"/api/ncrs/{ncr['id']}", headers=hdr(team["dispo"]))
|
|
assert "initial_disposition" in r.json()["available_actions"]
|
|
|
|
r = await client.get(
|
|
f"/api/ncrs?q={ncr['ncr_number']}", headers=hdr(team["requester"])
|
|
)
|
|
assert r.json()["total"] >= 1
|
|
|
|
|
|
async def test_unknown_dev_user_rejected(client):
|
|
r = await client.get("/api/me", headers=hdr("ghost@pescoinc.biz"))
|
|
assert r.status_code == 401
|