Assert rendered pages carry no unquoted multi-token attributes

The ops templates shipped with 346 of them: an unquoted attribute value ends at
the first whitespace, so the styles were never applied and status badges rendered
invisible. Assert it on the rendered HTML rather than the template source, because
a template may hold {{ }} inside an unquoted attribute that renders harmlessly —
the source is ambiguous, the response is not.
This commit is contained in:
ixgadmin 2026-09-17 14:00:10 +00:00
parent db05d6052a
commit e2a6784505
2 changed files with 233 additions and 26 deletions

View file

@ -42,6 +42,7 @@ from reportlab.pdfgen import canvas
from apps.batches.models import ProcessingBatch
from apps.capa.models import CAPARecord
from apps.compliance import services
from apps.compliance.models import (
CollectorGroup,
DueDiligenceCase,
@ -166,7 +167,7 @@ class Command(BaseCommand):
)
def _reset(self) -> None:
for ref in (COMPLETED_BATCH, INFLIGHT_BATCH):
for ref in (COMPLETED_BATCH, INFLIGHT_BATCH, FAILED_BATCH):
ProcessingBatch.objects.filter(batch_reference=ref).delete()
self.stdout.write(" reset: removed seeded batches")
@ -282,6 +283,50 @@ class Command(BaseCommand):
qc.save()
return qc
def _failed_qc(self, batch: ProcessingBatch, enterer: User, spec):
"""The exception-branch result: a real failure, judged by the real evaluator.
The values fail on **moisture, 0.31 % against the 0.100 % limit** — the
story the CAPA and the walkthrough tell. ``overall_pass`` and
``result_status`` are computed by ``services.evaluate_qc_data`` rather than
asserted here, so a stored status can never contradict the stored values
(this record used to carry ``within_spec`` beside ``overall_pass = False``).
Left awaiting verification on purpose: that is what puts it in the QC
Approvals queue, so the four-eyes control can be walked in training.
"""
qc = batch.qc_results.order_by("-tested_at").first() or QCResult(batch=batch)
qc.ffa_percentage = Decimal("0.42")
qc.moisture_percentage = Decimal("0.31")
qc.peroxide_value = Decimal("1.20")
qc.visual_grade = QCResult.VisualGrade.A
qc.tested_by = LAB_NAME
qc.lab_reference = "NAS-2026-0921"
qc.entered_by = enterer
qc.verified_by = None
qc.approved_at = None
qc.verification_status = QCResult.VerificationStatus.DRAFT
qc.method_ffa = "ISO 660"
qc.method_moisture = "ISO 662"
qc.method_peroxide = "ISO 3960"
values = {
"ffa": str(qc.ffa_percentage),
"moisture": str(qc.moisture_percentage),
"peroxide": str(qc.peroxide_value),
"visual": qc.visual_grade,
}
overall_pass, result_status, details, spec_version = services.evaluate_qc_data(
values, spec=spec
)
qc.overall_pass = overall_pass
qc.result_status = result_status
qc.spec_version = spec_version
# Mirrors QCResultCreateSerializer: a failing result quarantines the batch.
qc.is_quarantine = not overall_pass
qc.save()
return qc, details
def _documents(self, batch: ProcessingBatch, qc: QCResult, enterer: User,
verifier: User, coa_issue: date, is_completed: bool,
coa_published: bool = True) -> CertificateOfAnalysis:
@ -606,25 +651,47 @@ class Command(BaseCommand):
self._esg(inflight, [(date(2026, 9, 6), 27, 16, "Synthetic report: intake day observations.")])
self.stdout.write(f" {INFLIGHT_BATCH}: in transit, CoA awaiting release, EUDR complete for review")
# 3. exception branch
failed = ProcessingBatch.objects.filter(batch_reference=FAILED_BATCH).first()
if failed:
# exactly one CAPA per demo batch, however often this command is re-run
CAPARecord.objects.filter(batch=failed).delete()
CAPARecord.objects.create(
batch=failed,
issue_description=(
"QC result outside specification: moisture 0.31% against a limit of 0.20% "
"(synthetic training record)."
),
root_cause="Drying time shortened during a heavy-rain week at the processing site.",
corrective_action="Batch placed on hold and re-dried; re-sampled and re-tested by the laboratory.",
preventive_action="Minimum drying time added to the processing checklist; daily moisture spot checks.",
status=CAPARecord.Status.IN_PROGRESS,
assigned_to="IXG Quality (dummy owner)",
due_date=date(2026, 10, 15),
)
self.stdout.write(f" {FAILED_BATCH}: CAPA record replaced (exception branch example)")
# 3. exception branch. The batch is created here as well, so this branch is
# reproducible on an empty database rather than depending on a record
# that happened to already exist. No intake lineage on purpose: its risk
# assessment is the "untraceable origin" example.
failed = self._batch(
FAILED_BATCH, [], enterer,
harvest_date=date(2026, 7, 15), weight_kg=Decimal("900.00"),
grade=ProcessingBatch.Grade.A, status=ProcessingBatch.Status.QC_FAIL,
notes="Synthetic batch used to demonstrate the exception path (QC failure).",
)
qc_failed, details = self._failed_qc(failed, enterer, spec)
# The batch follows its result, as it does through the API.
failed.status = ProcessingBatch.Status.QC_FAIL
failed.save(update_fields=["status", "updated_at"])
# The CAPA is raised with the description the platform itself writes
# (QCResultCreateSerializer), then worked — the root cause and the
# actions are what an operator fills in, and the status moves on.
# exactly one CAPA per demo batch, however often this command is re-run
CAPARecord.objects.filter(batch=failed).delete()
CAPARecord.objects.create(
batch=failed,
issue_description=f"QC failure on batch {failed.batch_reference}",
root_cause="Drying time shortened during a heavy-rain week at the processing site.",
corrective_action="Batch placed on hold and re-dried; re-sampled and re-tested by the laboratory.",
preventive_action="Minimum drying time added to the processing checklist; daily moisture spot checks.",
status=CAPARecord.Status.IN_PROGRESS,
assigned_to="IXG Quality (dummy owner)",
due_date=date(2026, 10, 15),
)
failing = ", ".join(
f"{name} {detail['detail']}"
for name, detail in details.items()
if detail["status"] == "outside"
)
self.stdout.write(
f" {FAILED_BATCH}: QC {qc_failed.get_result_status_display()} "
f"({failing or 'no parameter detail'}), quarantined, awaiting "
f"verification; CAPA raised and worked"
)
buyer = self._buyer()
self.stdout.write(f" buyer: {buyer.email} ({buyer.role}) with an active Professional subscription")

View file

@ -1,12 +1,16 @@
"""Cross-cutting checks: every portal page renders, and map tiles are served
from our own origin.
"""Cross-cutting checks: every portal page renders, map tiles are served
from our own origin, and the training data the walkthrough depends on is
coherent.
Neither was covered before. The page sweep exists because a view can lose a
line — a `template_name`, say — and 500 in front of a user while the rest of the
None of these were covered before. The page sweep exists because a view can lose
a line — a `template_name`, say — and 500 in front of a user while the rest of the
suite stays green; it fetches each read-only page with an empty database, so it
covers the empty-state branches too. The tile tests cover the map's only
external dependency, which used to be fetched by the viewer's browser.
covers the empty-state branches too. The tile tests cover the map's only external
dependency, which used to be fetched by the viewer's browser. The seed tests cover
the exception branch, which used to depend on a record that happened to exist
already, with a stored status that contradicted its own values.
"""
import re
import shutil
import tempfile
import urllib.error
@ -15,8 +19,14 @@ from unittest.mock import MagicMock, patch
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.management import call_command
from django.test import Client, TestCase, override_settings
from apps.batches.models import ProcessingBatch
from apps.capa.models import CAPARecord
from apps.compliance.models import ProductSpecification
from apps.qc.models import QCResult
from .tiles import browser_tile_url, upstream_url
User = get_user_model()
@ -155,3 +165,133 @@ class TileProxyTests(TestCase):
upstream_url(13, 4085, 3448),
"https://example.invalid/a/13/4085/3448.png",
)
class SeedWalkthroughTests(TestCase):
"""The walkthrough's exception branch is training data, and it has to hold
together: a failure the platform's own evaluator would recognise, a batch that
follows it, and one CAPA — reproducible on an empty database rather than
depending on a record that happened to exist already."""
def setUp(self):
self.media = Path(tempfile.mkdtemp(prefix="ixg-seed-media-"))
self.addCleanup(shutil.rmtree, self.media, True)
override = override_settings(MEDIA_ROOT=self.media)
override.enable()
self.addCleanup(override.disable)
# The seed uses the accounts that already exist rather than inventing staff.
User.objects.create_user(
email="test@ixg.local", password="pass1234", role="ops",
full_name="Test User", is_staff=True,
)
User.objects.create_user(
email="dbg@ixg.local", password="pass1234", role="ops",
full_name="Debug User", is_staff=True,
)
call_command("seed_walkthrough", verbosity=0)
def test_the_exception_branch_is_reproducible(self):
batch = ProcessingBatch.objects.get(batch_reference="IXG-TEST-001")
self.assertEqual(batch.status, ProcessingBatch.Status.QC_FAIL)
self.assertFalse(batch.intake_lineage.exists(), "the example has no lineage")
qc = QCResult.objects.get(batch=batch)
self.assertFalse(qc.overall_pass)
self.assertEqual(qc.result_status, QCResult.ResultStatus.OUTSIDE_SPEC)
self.assertTrue(qc.is_quarantine, "a failing result quarantines the batch")
self.assertEqual(qc.verification_status, QCResult.VerificationStatus.DRAFT)
self.assertIsNone(qc.verified_by, "it must still need a second pair of eyes")
def test_the_stored_status_cannot_contradict_the_stored_values(self):
qc = QCResult.objects.get(batch__batch_reference="IXG-TEST-001")
limit = ProductSpecification.current("Unrefined Shea Butter").parameters.get(
parameter="moisture"
).max_value
self.assertGreater(
qc.moisture_percentage, limit,
"the failure must be a genuine one, judged by the real evaluator",
)
def test_the_passing_batches_stay_within_specification(self):
for reference in ("IXG-BATCH-2026-014", "IXG-BATCH-2026-005"):
with self.subTest(batch=reference):
qc = QCResult.objects.get(batch__batch_reference=reference)
self.assertTrue(qc.overall_pass)
self.assertEqual(qc.result_status, QCResult.ResultStatus.WITHIN_SPEC)
self.assertFalse(qc.is_quarantine)
def test_one_capa_per_demo_batch(self):
capas = CAPARecord.objects.filter(batch__batch_reference="IXG-TEST-001")
self.assertEqual(capas.count(), 1)
self.assertEqual(
capas.first().issue_description, "QC failure on batch IXG-TEST-001"
)
def test_re_running_the_seed_does_not_duplicate_records(self):
call_command("seed_walkthrough", verbosity=0)
self.assertEqual(
QCResult.objects.filter(batch__batch_reference="IXG-TEST-001").count(), 1
)
self.assertEqual(CAPARecord.objects.count(), 1)
self.assertEqual(ProcessingBatch.objects.filter(batch_reference="IXG-TEST-001").count(), 1)
class TemplateAttributeHygieneTests(TestCase):
"""No ops or buyer page may render an unquoted attribute that holds more than
one token.
An unquoted attribute value ends at the first whitespace, so
``class=badge bg-danger`` reaches the browser as ``class="badge"`` plus two
meaningless attributes: the badge renders white-on-white and the failure is
invisible. Every ops template shipped that way for months, so this asserts it
on the *rendered* pages — the only place the question has an unambiguous
answer, since a template may contain ``{{ }}`` in an unquoted attribute that
renders harmlessly.
"""
DETECTOR = re.compile(r'\s([\w-]+)=([^\s"\'=<>`]+(?:\s+[^\s"\'=<>`]+)+)(?=[\s>])')
TAG = re.compile(r"<[a-zA-Z/][^>]*>")
@classmethod
def offenders(cls, html):
found = []
for tag in cls.TAG.findall(html):
for match in cls.DETECTOR.finditer(tag):
found.append(match.group(0).strip())
return found
def test_the_detector_catches_the_real_defect(self):
bad = '<span class=badge bg-danger>Fail</span>'
self.assertTrue(self.DETECTOR.search(self.TAG.findall(bad)[0]))
good = '<span class="badge bg-danger">Fail</span>'
self.assertFalse(self.DETECTOR.search(self.TAG.findall(good)[0]))
def check_pages(self, user, routes, host):
client = Client(SERVER_NAME=host)
client.force_login(user)
for route in routes:
with self.subTest(route=route):
html = client.get(route).content.decode()
self.assertEqual(
self.offenders(html), [],
f"{route} renders unquoted multi-token attributes",
)
def test_ops_pages_have_no_unquoted_multi_token_attributes(self):
self.check_pages(
User.objects.create_user(
email="attr-ops@ixg.local", password="pass1234", role="ops",
full_name="Attr Ops", is_staff=True,
),
OPS_ROUTES, "ops.ixgltd.com",
)
def test_portal_pages_have_no_unquoted_multi_token_attributes(self):
self.check_pages(
User.objects.create_user(
email="attr-buyer@ixg.local", password="pass1234", role="buyer",
full_name="Attr Buyer", is_staff=True,
),
PORTAL_ROUTES, "portal.ixgltd.com",
)