spencerm/pesco-ncr
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: PESCO NCR system

Browse Source 
Complete Non-Conformance Report system replacing the PowerApps/SharePoint
prototype: FastAPI + SQLAlchemy 2 (async) + Alembic + MySQL 8 backend,
React 18 + Vite + TypeScript + MUI frontend, Entra ID auth (MSAL / JWKS,
group-gated), Microsoft Graph delegated Mail.Send notifications (OBO),
six-stage workflow state machine with server-side enforcement, atomic
NCR-YYYY-NNNN numbering, attachments with camera capture, immutable
field-level audit trail, admin reopen, reports + CSV export, WeasyPrint
PDF traveler, Power BI reporting views + read-only DB user, documented
VISUAL ERP job-lookup stub, pytest suite (26 tests), docker-compose
deployment.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
ang3l12
2026-07-13 11:41:22 -06:00
commit dea316b113
111 changed files with 13817 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
backend/tests/__init__.py Normal file
View File

89
backend/tests/conftest.py Normal file
View File

@@ -0,0 +1,89 @@
"""Test configuration.
The environment MUST be set before any `app.*` import (settings are cached):
tests run against a file-backed SQLite database with AUTH_MODE=dev, which
exercises the same SQLAlchemy models, state machine, numbering, and
permission code paths as MySQL. To run the suite against a real MySQL
instance instead:
DATABASE_URL="mysql+aiomysql://user:pass@host/db_test?charset=utf8mb4" pytest
"""
import asyncio
import os
import tempfile
import uuid
_TMPDIR = tempfile.mkdtemp(prefix="pesco-ncr-tests-")
os.environ.setdefault("DATABASE_URL", f"sqlite+aiosqlite:///{_TMPDIR}/test.db")
os.environ["AUTH_MODE"] = "dev"
os.environ["ATTACHMENTS_DIR"] = os.path.join(_TMPDIR, "attachments")
os.environ["INITIAL_ADMIN_EMAILS"] = ""
os.environ["JOB_LOOKUP_PROVIDER"] = "null"
# Disabled by default so mutation responses have empty `warnings`;
# notification-specific tests flip the AppSetting row explicitly.
os.environ["NOTIFICATIONS_ENABLED_DEFAULT"] = "false"
import pytest # noqa: E402
from httpx import ASGITransport, AsyncClient # noqa: E402
from app.database import get_engine, get_session_factory # noqa: E402
from app.main import app # noqa: E402
from app.models import Base, Department, DeviationCategory, User, UserRole # noqa: E402
@pytest.fixture(scope="session", autouse=True)
def _create_schema():
async def _run():
engine = get_engine()
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with get_session_factory()() as db:
db.add(Department(name="Machining", is_active=True))
db.add(Department(name="Inactive Dept", is_active=False))
db.add(DeviationCategory(name="Dimensional", is_active=True))
await db.commit()
asyncio.run(_run())
yield
@pytest.fixture
async def client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
@pytest.fixture
def make_user():
async def _make(roles: list[str], name: str | None = None) -> str:
email = f"user-{uuid.uuid4().hex[:10]}@pescoinc.biz"
async with get_session_factory()() as db:
user = User(
email=email,
display_name=name or f"Test {email.split('@')[0]}",
is_active=True,
)
db.add(user)
await db.flush()
for role in roles:
db.add(UserRole(user_id=user.id, role=role))
await db.commit()
return email
return _make
@pytest.fixture
async def team(make_user) -> dict[str, str]:
"""One user per workflow role, fresh for each test."""
return {
"requester": await make_user(["requester"]),
"dispo": await make_user(["requester", "disposition_authority"]),
"second": await make_user(["requester", "secondary_disposition_authority"]),
"second2": await make_user(["requester", "secondary_disposition_authority"]),
"ops": await make_user(["requester", "operations"]),
"qc": await make_user(["requester", "qc_inspector"]),
"cost": await make_user(["requester", "costing"]),
"admin": await make_user(["admin"]),
}

62
backend/tests/test_attachments.py Normal file
View File

@@ -0,0 +1,62 @@
"""Attachment upload validation, metadata, download, and closure locking."""
from .util import create_ncr, hdr, to_closed
TINY_PNG = (
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01"
b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
)
async def test_upload_download_and_metadata(client, team):
ncr = await create_ncr(client, team)
r = await client.post(
f"/api/ncrs/{ncr['id']}/attachments",
files=[
("files", ("photo one.png", TINY_PNG, "image/png")),
("files", ("notes.txt", b"observed at station 4", "text/plain")),
],
headers=hdr(team["requester"]),
)
assert r.status_code == 201, r.text
items = r.json()
assert len(items) == 2
png = next(i for i in items if i["is_image"])
assert png["original_filename"] == "photo one.png"
assert png["uploaded_by"]["email"] == team["requester"]
assert png["size_bytes"] == len(TINY_PNG)
r = await client.get(
f"/api/attachments/{png['id']}/download", headers=hdr(team["ops"])
)
assert r.status_code == 200
assert r.content == TINY_PNG
# attachment add shows in detail + audit
detail = (await client.get(f"/api/ncrs/{ncr['id']}", headers=hdr(team["qc"]))).json()
assert len(detail["attachments"]) == 2
audit = (
await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["qc"]))
).json()
assert sum(1 for a in audit["items"] if a["action"] == "attachment_add") == 2
async def test_disallowed_type_rejected(client, team):
ncr = await create_ncr(client, team)
r = await client.post(
f"/api/ncrs/{ncr['id']}/attachments",
files=[("files", ("malware.exe", b"MZ...", "application/octet-stream"))],
headers=hdr(team["requester"]),
)
assert r.status_code == 422
assert "not allowed" in r.json()["detail"]
async def test_attachments_locked_when_closed(client, team):
ncr = await to_closed(client, team, (await create_ncr(client, team))["id"])
r = await client.post(
f"/api/ncrs/{ncr['id']}/attachments",
files=[("files", ("late.png", TINY_PNG, "image/png"))],
headers=hdr(team["requester"]),
)
assert r.status_code == 409

51
backend/tests/test_numbering.py Normal file
View File

@@ -0,0 +1,51 @@
"""NCR numbering: format, per-year sequence + rollover, and concurrency."""
import asyncio
import re
from datetime import datetime
from app.database import get_session_factory
from app.services.numbering import allocate_ncr_number
from .util import create_ncr
NCR_RE = re.compile(r"^NCR-(\d{4})-(\d{4})$")
async def test_number_format_and_sequence(client, team):
first = await create_ncr(client, team)
second = await create_ncr(client, team)
m1, m2 = NCR_RE.match(first["ncr_number"]), NCR_RE.match(second["ncr_number"])
assert m1 and m2, (first["ncr_number"], second["ncr_number"])
assert int(m1.group(1)) == datetime.now().year
assert int(m2.group(2)) == int(m1.group(2)) + 1
async def test_year_rollover_resets_sequence():
async with get_session_factory()() as db:
n1, year1, seq1 = await allocate_ncr_number(db, now=datetime(2098, 12, 31))
n2, year2, seq2 = await allocate_ncr_number(db, now=datetime(2099, 1, 1))
n3, _, seq3 = await allocate_ncr_number(db, now=datetime(2099, 6, 15))
await db.rollback()
assert (year1, seq1) == (2098, 1) and n1 == "NCR-2098-0001"
assert (year2, seq2) == (2099, 1) and n2 == "NCR-2099-0001"
assert seq3 == 2 and n3 == "NCR-2099-0002"
async def test_zero_padding():
async with get_session_factory()() as db:
number, _, _ = await allocate_ncr_number(db, now=datetime(2097, 3, 1))
await db.rollback()
assert number == "NCR-2097-0001"
async def test_concurrent_submissions_never_collide(client, team):
"""Twelve simultaneous submissions must all succeed with distinct numbers."""
results = await asyncio.gather(
*[create_ncr(client, team, job_number=f"J-CONC-{i}") for i in range(12)]
)
numbers = [r["ncr_number"] for r in results]
assert len(set(numbers)) == 12, numbers
seqs = sorted(int(NCR_RE.match(n).group(2)) for n in numbers)
assert seqs == list(range(seqs[0], seqs[0] + 12))

223
backend/tests/test_permissions.py Normal file
View File

@@ -0,0 +1,223 @@
"""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']}/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

244
backend/tests/test_state_machine.py Normal file
View File

@@ -0,0 +1,244 @@
"""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)

108
backend/tests/util.py Normal file
View File

@@ -0,0 +1,108 @@
"""Shared helpers: drive the API exactly the way the frontend does."""
from httpx import AsyncClient
def hdr(email: str) -> dict[str, str]:
return {"X-Dev-User": email}
async def lookup_ids(client: AsyncClient, email: str) -> tuple[int, int]:
r = await client.get("/api/lookups", headers=hdr(email))
assert r.status_code == 200, r.text
body = r.json()
return body["departments"][0]["id"], body["deviation_categories"][0]["id"]
async def user_id_by_email(client: AsyncClient, as_email: str, email: str, role: str) -> int:
r = await client.get(f"/api/users?role={role}", headers=hdr(as_email))
assert r.status_code == 200, r.text
for u in r.json():
if u["email"] == email:
return u["id"]
raise AssertionError(f"user {email} with role {role} not found")
async def create_ncr(client: AsyncClient, team: dict, **overrides) -> dict:
dept_id, cat_id = await lookup_ids(client, team["requester"])
dispo_id = await user_id_by_email(
client, team["requester"], team["dispo"], "disposition_authority"
)
payload = {
"job_number": "J12345",
"department_id": dept_id,
"deviation_category_id": cat_id,
"disposition_authority_id": dispo_id,
"deviation_detail": "Bore diameter out of tolerance on 3 pieces.",
}
payload.update(overrides)
r = await client.post("/api/ncrs", json=payload, headers=hdr(team["requester"]))
assert r.status_code == 201, r.text
return r.json()["ncr"]
async def do_initial_disposition(
client: AsyncClient,
team: dict,
ncr_id: int,
*,
secondary: bool = False,
secondary_ids: list[int] | None = None,
as_user: str | None = None,
notes: str = "<p>Rework per instructions.</p>",
):
body = {
"qc_authority": "AS9100 8.7",
"work_order": "WO-1001",
"disposition_notes": notes,
"secondary_review_needed": secondary,
}
if secondary_ids is not None:
body["secondary_authority_ids"] = secondary_ids
return await client.post(
f"/api/ncrs/{ncr_id}/initial-disposition",
json=body,
headers=hdr(as_user or team["dispo"]),
)
async def to_operations(client: AsyncClient, team: dict, ncr_id: int) -> dict:
"""Walk a fresh NCR straight to the Operations stage."""
r = await do_initial_disposition(client, team, ncr_id, secondary=False)
assert r.status_code == 200, r.text
return r.json()["ncr"]
async def to_qc_inspection(client: AsyncClient, team: dict, ncr_id: int) -> dict:
await to_operations(client, team, ncr_id)
r = await client.post(
f"/api/ncrs/{ncr_id}/operations-complete", headers=hdr(team["ops"])
)
assert r.status_code == 200, r.text
return r.json()["ncr"]
async def to_costing(client: AsyncClient, team: dict, ncr_id: int) -> dict:
await to_qc_inspection(client, team, ncr_id)
r = await client.post(
f"/api/ncrs/{ncr_id}/inspection",
json={"qc_approval": "yes", "inspection_notes": "All good.", "qc_closed": True},
headers=hdr(team["qc"]),
)
assert r.status_code == 200, r.text
return r.json()["ncr"]
async def to_closed(client: AsyncClient, team: dict, ncr_id: int) -> dict:
await to_costing(client, team, ncr_id)
r = await client.post(
f"/api/ncrs/{ncr_id}/costing",
json={
"labor_cost": "125.50",
"material_cost": "60.00",
"service_cost": "0",
"other_cost": "14.50",
},
headers=hdr(team["cost"]),
)
assert r.status_code == 200, r.text
return r.json()["ncr"]