dea316b113
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>
52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
"""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))
|