"""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. 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. 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 from decimal import Decimal from html.parser import HTMLParser from pathlib import Path 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, SimpleTestCase, TestCase, override_settings from apps.batches.models import ProcessingBatch from apps.capa.models import CAPARecord from apps.compliance.models import ( DueDiligenceCase, EUDREvidence, HarvestIntake, IntakeBatchLineage, ProductSpecification, ) from apps.qc.models import QCResult from .tiles import browser_tile_url, upstream_url User = get_user_model() OPS_ROUTES = [ "/ops/", "/ops/intakes/", "/ops/intakes/new/", "/ops/batches/", "/ops/batches/new/", "/ops/qc/", "/ops/qc/verify/", "/ops/coa/", "/ops/shipments/", "/ops/capa/", "/ops/documents/", "/ops/esg/", "/ops/eudr/", "/ops/help/", ] PORTAL_ROUTES = [ "/portal/", "/portal/batches/", "/portal/documents/", "/portal/compliance/", "/portal/subscription/", "/portal/help/", ] def make_sweep_batch(): """A batch with one lineage row, so the edit and lineage screens render. The route sweeps otherwise run against an empty database and never render ``batch_form.html`` in edit mode or ``batch_lineage.html`` at all — only the source guard covers them, and that is the weaker guard. """ batch = ProcessingBatch.objects.create(batch_reference="IXG-SWEEP-001") intake = HarvestIntake.objects.create( intake_reference="IXG-SWEEP-INTAKE-001", weight_kg=Decimal("100.00") ) IntakeBatchLineage.objects.create( batch=batch, harvest_intake=intake, weight_allocated_kg=Decimal("40.00") ) return batch def batch_screens(batch): return [ f"/ops/batches/{batch.pk}/", f"/ops/batches/{batch.pk}/edit/", f"/ops/batches/{batch.pk}/lineage/", ] class PortalPagesRenderTests(TestCase): def make_user(self, email, role): return User.objects.create_user( email=email, password="pass1234", role=role, full_name=email.split("@")[0], is_staff=True, ) def sweep(self, user, routes, host): client = Client(SERVER_NAME=host) client.force_login(user) for route in routes: with self.subTest(route=route): response = client.get(route) body = response.content self.assertEqual(response.status_code, 200, f"{route} did not render") for marker in (b"Traceback", b"TemplateDoesNotExist", b"Django Version"): self.assertNotIn(marker, body, f"{route} rendered an error page") def test_ops_pages_render(self): batch = make_sweep_batch() routes = OPS_ROUTES + batch_screens(batch) self.sweep(self.make_user("ops-sweep@ixg.local", "ops"), routes, "ops.ixgltd.com") def test_buyer_pages_render(self): self.sweep(self.make_user("buyer-sweep@ixg.local", "buyer"), PORTAL_ROUTES, "portal.ixgltd.com") class TileProxyTests(TestCase): """Tiles come from our origin, so the map does not depend on the viewer's network being able to reach a third-party tile server.""" def setUp(self): self.cache = Path(tempfile.mkdtemp(prefix="ixg-tile-cache-")) self.addCleanup(shutil.rmtree, self.cache, True) override = override_settings(LEAFLET_TILE_CACHE_DIR=str(self.cache)) override.enable() self.addCleanup(override.disable) self.user = User.objects.create_user( email="tile-viewer@ixg.local", password="pass1234", role="admin", full_name="Tile Viewer", is_staff=True, ) self.anon = Client(SERVER_NAME="app.ixgltd.com") self.viewer = Client(SERVER_NAME="app.ixgltd.com") self.viewer.force_login(self.user) @staticmethod def fake_upstream(payload): """Stand in for urllib.request.urlopen returning a tile.""" response = MagicMock() response.read.return_value = payload context = MagicMock() context.__enter__.return_value = response return lambda *a, **k: context def test_a_viewer_must_be_signed_in(self): self.assertEqual(self.anon.get("/tiles/13/4085/3448.png").status_code, 403) def test_tiles_outside_the_zoom_range_are_refused(self): self.assertEqual(self.viewer.get("/tiles/20/1/1.png").status_code, 404) # x beyond the span of that zoom level self.assertEqual(self.viewer.get("/tiles/1/5/1.png").status_code, 404) def test_a_tile_is_fetched_from_the_upstream_and_served(self): payload = b"\x89PNG\r\n\x1a\n fake tile" with patch("apps.core.tiles.urllib.request.urlopen", self.fake_upstream(payload)): r = self.viewer.get("/tiles/13/4085/3448.png") self.assertEqual(r.status_code, 200) self.assertEqual(r["Content-Type"], "image/png") self.assertEqual(r.content, payload) self.assertTrue((self.cache / "13" / "4085" / "3448.png").is_file()) def test_a_second_request_is_served_from_the_cache(self): payload = b"\x89PNG\r\n\x1a\n cached tile" first = MagicMock(side_effect=self.fake_upstream(payload)) with patch("apps.core.tiles.urllib.request.urlopen", first): self.viewer.get("/tiles/13/4085/3448.png") self.assertEqual(first.call_count, 1) with patch("apps.core.tiles.urllib.request.urlopen") as second: r = self.viewer.get("/tiles/13/4085/3448.png") self.assertEqual(second.call_count, 0, "cache should have been used") self.assertEqual(r.status_code, 200) self.assertEqual(r.content, payload) def test_an_unreachable_upstream_answers_an_empty_tile_not_an_error_page(self): with patch( "apps.core.tiles.urllib.request.urlopen", side_effect=urllib.error.HTTPError("u", 403, "Forbidden", None, None), ): r = self.viewer.get("/tiles/13/4085/3448.png") self.assertEqual(r.status_code, 204) self.assertEqual(r.content, b"") def test_browser_tile_url_is_same_origin_when_the_proxy_is_on(self): self.assertEqual(browser_tile_url(), "/tiles/{z}/{x}/{y}.png") def test_browser_tile_url_is_the_upstream_when_the_proxy_is_off(self): with override_settings(LEAFLET_TILE_PROXY_ENABLED=False): self.assertEqual(browser_tile_url(), settings.LEAFLET_TILE_URL) def test_the_upstream_url_gets_the_coordinates_substituted(self): with override_settings( LEAFLET_TILE_URL="https://example.invalid/{s}/{z}/{x}/{y}.png" ): self.assertEqual( 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) def test_every_intake_keeps_within_what_it_received(self): """The rule the ops lineage screen enforces, applied to the training data. It was not applied here, and nobody noticed: both demo batches were traced back to the same three intakes, so the 820 kg intake 0141 was allocated 800 kg twice over. The screen now refuses that, so the data must not need it. """ for intake in HarvestIntake.objects.all(): allocated = sum(row.weight_allocated_kg or 0 for row in intake.batch_lineage.all()) with self.subTest(intake=intake.intake_reference): self.assertLessEqual( allocated, intake.weight_kg, f"{intake.intake_reference} received {intake.weight_kg} kg " f"and is allocated {allocated} kg", ) def test_the_two_passing_batches_have_their_own_intakes(self): completed = set( ProcessingBatch.objects.get(batch_reference="IXG-BATCH-2026-014") .intake_lineage.values_list("harvest_intake__intake_reference", flat=True) ) inflight = set( ProcessingBatch.objects.get(batch_reference="IXG-BATCH-2026-005") .intake_lineage.values_list("harvest_intake__intake_reference", flat=True) ) self.assertTrue(completed, "the completed batch needs a lineage to show") self.assertTrue(inflight, "the in-flight batch needs a lineage to show") self.assertEqual( completed & inflight, set(), "an intake cannot supply two batches' worth of shea", ) def test_re_running_the_seed_removes_records_it_no_longer_makes(self): batch = ProcessingBatch.objects.get(batch_reference="IXG-BATCH-2026-005") stray = HarvestIntake.objects.get(intake_reference="IXG-INTAKE-2026-0141") IntakeBatchLineage.objects.create( batch=batch, harvest_intake=stray, weight_allocated_kg=Decimal("10.00") ) # Evidence is keyed on the filename, so reassigning the in-flight batch's # intakes used to leave the old harvest_document row in place. case = DueDiligenceCase.objects.get(batch=batch) stale = EUDREvidence.objects.create( case=case, evidence_type=EUDREvidence.EvidenceType.HARVEST_DOCUMENT, file="IXG-INTAKE-2026-0141-weighbridge-ticket.pdf", ) call_command("seed_walkthrough", verbosity=0) self.assertFalse( batch.intake_lineage.filter(harvest_intake=stray).exists(), "a link the seed no longer makes must not survive a re-run", ) self.assertEqual(batch.intake_lineage.count(), 3) self.assertFalse( EUDREvidence.objects.filter(pk=stale.pk).exists(), "stale evidence the seed no longer references must not survive a re-run", ) self.assertEqual(case.evidence.count(), 4) class RawTagScanner(HTMLParser): """Collect each start tag's raw source text, quote-aware. A regex such as ``<[^>]*>`` ends the tag at the first ``>``, which a quoted attribute value may contain: ```` then arrives truncated, and a real leak after the quote is never seen. ``html.parser`` knows where the tag actually ends. """ def __init__(self): super().__init__(convert_charrefs=False) self.tags = [] def handle_starttag(self, tag, attrs): self.tags.append((self.getpos()[0], self.get_starttag_text())) def handle_startendtag(self, tag, attrs): self.tags.append((self.getpos()[0], self.get_starttag_text())) def scan_tags(html): """Return ``(line, raw_tag)`` for every start tag in ``html``.""" scanner = RawTagScanner() scanner.feed(html) scanner.close() return scanner.tags 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>])') @classmethod def offenders(cls, html): found = [] for _, tag in scan_tags(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(scan_tags(bad)[0][1])) good = 'Fail' self.assertFalse(self.DETECTOR.search(scan_tags(good)[0][1])) def test_a_quoted_angle_bracket_does_not_hide_a_defect(self): # The old ``<[^>]*>`` tag regex stopped at the ``>`` inside the title, so # the unquoted attribute after it was never examined. unquoted = 'Fail' self.assertTrue(self.offenders(unquoted)) leaked = 'Approved' self.assertEqual(self.leaked_classes(leaked), ["bg-success"]) # Two more defects from the same family, both of which shipped and neither of # which any status code, test or screenshot review caught. # # 1. The *inverse* of the mistake above, made when the unquoted attributes were # fixed in bulk: the quotes went around two attributes instead of around one # value, so `method=post novalidate` became `method="post novalidate"`. A # form's method is then an unrecognised keyword, the browser falls back to # GET, and the form silently saves nothing. # 2. A conditional class list written *outside* the quotes — # `` # — leaves the colour as a bare attribute. Bootstrap's `.badge` on its own # is white text on a transparent background, so the status is invisible. FORM_METHOD = re.compile(r']*\bmethod="([^"]*)"') ATTRIBUTE_NAMES = { "post", "get", "dialog", "novalidate", "async", "defer", "multiple", "required", "checked", "selected", "disabled", "readonly", "autofocus", "hidden", "target", "method", "action", "enctype", "accept", } # Any bare hyphenated token is a class that has escaped its quotes unless it # is an attribute that legitimately carries no value. An allowlist of known # Bootstrap prefixes missed the rest of the library (``form-control``, # ``alert-danger``, ``is-invalid``, ``rounded-*``, ``shadow-*``); the default # has to be "leak". Class lists inside quotes may contain spaces # legitimately, so only value-less tokens are considered. VALUELESS_ATTRIBUTES = ATTRIBUTE_NAMES | { "allowfullscreen", "autoplay", "controls", "default", "inert", "ismap", "itemscope", "loop", "muted", "nomodule", "open", "playsinline", "reversed", "truespeed", # Hyphenated boolean attributes from common front-end libraries. "x-cloak", "x-ignore", "x-transition", } QUOTED_VALUE = re.compile(r'\s([\w-]+)="([^"]*)"') ANY_QUOTED = re.compile(r'"[^"]*"|\'[^\']*\'') @classmethod def is_leaked_class(cls, token): return ( "-" in token and "=" not in token and token.lower() not in cls.VALUELESS_ATTRIBUTES ) @classmethod def folded_attributes(cls, html): """Quoted values that are really two attributes run together.""" found = [] for _, tag in scan_tags(html): for name, value in cls.QUOTED_VALUE.findall(tag): tokens = value.split() if len(tokens) > 1 and all(t in cls.ATTRIBUTE_NAMES for t in tokens): found.append(f"{name}={value!r}") return found @classmethod def leaked_classes(cls, html): """Classes sitting outside their class attribute.""" found = [] for _, tag in scan_tags(html): # Drop the angle brackets first: with them left on, the last token in # the tag arrives as "bg-success>" and matches nothing. without_values = cls.ANY_QUOTED.sub(" ", tag[1:-1]) for token in without_values.split(): if cls.is_leaked_class(token): found.append(token) return found def test_the_folded_attribute_detector_catches_the_real_defect(self): self.assertTrue(self.folded_attributes('
')) for good in ( '', '
', '', '', ): with self.subTest(good=good): self.assertEqual(self.folded_attributes(good), []) def test_the_leaked_class_detector_catches_the_real_defect(self): # The first is the defect that shipped; the rest are from the same family # and were missed by the old Bootstrap-prefix allowlist. cases = { 'Approved': "bg-success", '
': "form-control", '
': "alert-danger", '': "is-invalid", '
': "rounded-3", '
': "shadow-sm", } for bad, expected in cases.items(): with self.subTest(bad=bad): self.assertIn(expected, self.leaked_classes(bad)) def test_the_leaked_class_detector_leaves_legitimate_markup_alone(self): for good in ( 'Approved', '
', '', '', '', '', '
hidden until ready
', ): with self.subTest(good=good): self.assertEqual(self.leaked_classes(good), []) def check_pages(self, user, routes, host): """Every rendered page, held to all four markup rules at once.""" client = Client(SERVER_NAME=host) client.force_login(user) for route in routes: html = client.get(route).content.decode() with self.subTest(route=route, rule="unquoted multi-token attribute"): self.assertEqual( self.offenders(html), [], f"{route} renders unquoted multi-token attributes", ) with self.subTest(route=route, rule="form method"): for value in self.FORM_METHOD.findall(html): self.assertIn( value, ("post", "get"), f"{route} renders a form whose method is {value!r}", ) with self.subTest(route=route, rule="folded attribute"): self.assertEqual( self.folded_attributes(html), [], f"{route} folds two attributes into one quoted value", ) with self.subTest(route=route, rule="leaked class"): self.assertEqual( self.leaked_classes(html), [], f"{route} renders a class outside its class attribute", ) def test_ops_pages_render_clean_markup(self): batch = make_sweep_batch() 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 + batch_screens(batch), "ops.ixgltd.com", ) def test_portal_pages_render_clean_markup(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", ) class TemplateSourceHygieneTests(SimpleTestCase): """The same class-leak check, on the template source instead of a response. The route sweep above does not render every *detail* page — and the detail pages are exactly where the status badges are. Seven templates kept their badge colour outside the class attribute, which showed up only as a blank Status column in a screenshot. Scanning the source is sound for this particular question, and only this one: once `{% %}` tags are stripped, a class list split across `{% if %}` branches is unambiguous. The *unquoted attribute* question stays on rendered pages, because there a template can be ambiguous and a response cannot. """ QUOTED = re.compile(r'"[^"]*"|\'[^\']*\'') TEMPLATE_TAG = re.compile(r"\{%[^%]*%\}", re.S) def test_no_template_puts_a_class_outside_its_class_attribute(self): root = Path(settings.BASE_DIR) / "templates" offenders = [] for path in sorted(root.rglob("*.html")): text = path.read_text() for line, tag in scan_tags(text): stripped = self.QUOTED.sub(" ", self.TEMPLATE_TAG.sub(" ", tag[1:-1])) for token in stripped.split(): if TemplateAttributeHygieneTests.is_leaked_class(token): offenders.append(f"{path.relative_to(root)}:{line} {token}") self.assertEqual( offenders, [], "a class list is sitting outside its class attribute" )
total