diff --git a/apps/core/management/commands/seed_walkthrough.py b/apps/core/management/commands/seed_walkthrough.py
index 839b2fd..40106a4 100644
--- a/apps/core/management/commands/seed_walkthrough.py
+++ b/apps/core/management/commands/seed_walkthrough.py
@@ -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")
diff --git a/apps/core/tests.py b/apps/core/tests.py
index ce8297f..7175173 100644
--- a/apps/core/tests.py
+++ b/apps/core/tests.py
@@ -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 = 'Fail'
+ self.assertTrue(self.DETECTOR.search(self.TAG.findall(bad)[0]))
+ good = 'Fail'
+ 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",
+ )