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>
99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
"""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 ( # noqa: E402
|
|
Base,
|
|
Department,
|
|
DeviationCategory,
|
|
RootCauseCategory,
|
|
User,
|
|
UserRole,
|
|
)
|
|
|
|
|
|
@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))
|
|
db.add(RootCauseCategory(name="Method", is_active=True))
|
|
db.add(RootCauseCategory(name="Retired Cause", is_active=False))
|
|
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"]),
|
|
}
|