"""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 rest_framework.test import APIClient from apps.batches.models import ProcessingBatch from apps.capa.models import CAPARecord from apps.compliance.models import ( DueDiligenceCase, EUDREvidence, HarvestIntake, IntakeBatchLineage, ProductSpecification, ) from apps.notifications.models import Notification from apps.qc.models import QCResult from apps.shipments.models import Shipment from apps.subscriptions.models import Subscription 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/shipments/new/", "/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/", ] def make_sweep_shipment(batch): """A shipment so the detail and edit screens render in the route sweeps.""" return Shipment.objects.create( batch=batch, status=Shipment.Status.BOOKED, vessel_name="MV Sweep", container_ref="IXGU0000000", etd_nigeria="2026-09-01", eta_uk="2026-10-01", ) def shipment_screens(shipment): return [ f"/ops/shipments/{shipment.pk}/", f"/ops/shipments/{shipment.pk}/edit/", ] 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() shipment = make_sweep_shipment(batch) routes = OPS_ROUTES + batch_screens(batch) + shipment_screens(shipment) 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") def test_the_public_signup_page_renders_for_visitors(self): response = Client(SERVER_NAME="portal.ixgltd.com").get("/portal/signup/") self.assertEqual(response.status_code, 200) class RoleRoutingTests(TestCase): """A buyer is not staff, so the middleware must not bounce them off /portal/. The old middleware redirected if any role's pattern matched the path without belonging to the user's role, and the admin patterns include /portal/ — so a real (non-staff) buyer was sent to the landing view, which sent them back. These pin the corrected routing: each role reaches its own portal and is turned away from the others via the landing view. """ def make_user(self, email, role): return User.objects.create_user( email=email, password="pass1234", role=role, is_staff=False ) def test_a_buyer_reaches_the_buyer_portal(self): client = Client(SERVER_NAME="portal.ixgltd.com") client.force_login(self.make_user("route-buyer@ixg.local", "buyer")) self.assertEqual(client.get("/portal/").status_code, 200) def test_a_buyer_is_turned_away_from_ops_and_admin(self): client = Client(SERVER_NAME="portal.ixgltd.com") client.force_login(self.make_user("route-buyer2@ixg.local", "buyer")) for path in ("/ops/", "/admin/"): with self.subTest(path=path): response = client.get(path) self.assertEqual(response.status_code, 302) self.assertEqual(response["Location"], "/") def test_ops_reaches_ops(self): client = Client(SERVER_NAME="ops.ixgltd.com") client.force_login(self.make_user("route-ops@ixg.local", "ops")) self.assertEqual(client.get("/ops/").status_code, 200) def test_the_landing_view_sends_each_role_to_its_portal(self): cases = [("buyer", "/portal/"), ("ops", "/ops/"), ("admin", "/admin/")] for role, destination in cases: with self.subTest(role=role): client = Client(SERVER_NAME="app.ixgltd.com") client.force_login(self.make_user(f"landing-{role}@ixg.local", role)) response = client.get("/") self.assertEqual(response.status_code, 302) self.assertEqual(response["Location"], destination) def test_a_visitor_is_sent_to_sign_in(self): response = Client(SERVER_NAME="app.ixgltd.com").get("/") self.assertEqual(response.status_code, 302) self.assertIn("/oidc/authenticate/", response["Location"]) class AdminAreaRoleTests(TestCase): """The administration area is for the admin role, not for the staff flag. A staff flag on an ops or buyer account used to be enough to reach /admin/, whose index embeds the model registry, and where a later group or permission grant would be inherited. Staff status is derived from the role, but the gate must not rest on the flag alone. """ def make_user(self, email, role, is_staff): return User.objects.create_user( email=email, password="pass1234", role=role, is_staff=is_staff ) def test_a_staff_ops_account_is_turned_away_from_admin(self): client = Client(SERVER_NAME="app.ixgltd.com") client.force_login(self.make_user("gate-ops@ixg.local", "ops", True)) response = client.get("/admin/") self.assertEqual(response.status_code, 302) self.assertEqual(response["Location"], "/") def test_a_staff_buyer_account_is_turned_away_from_admin(self): client = Client(SERVER_NAME="app.ixgltd.com") client.force_login(self.make_user("gate-buyer@ixg.local", "buyer", True)) response = client.get("/admin/") self.assertEqual(response.status_code, 302) self.assertEqual(response["Location"], "/") def test_an_admin_role_reaches_admin(self): client = Client(SERVER_NAME="app.ixgltd.com") client.force_login(self.make_user("gate-admin@ixg.local", "admin", True)) self.assertEqual(client.get("/admin/").status_code, 200) def test_a_superuser_reaches_admin(self): client = Client(SERVER_NAME="app.ixgltd.com") client.force_login( User.objects.create_superuser( email="gate-super@ixg.local", password="pass1234" ) ) self.assertEqual(client.get("/admin/").status_code, 200) def test_the_ops_sidebar_hides_admin_from_a_non_admin(self): client = Client(SERVER_NAME="ops.ixgltd.com") client.force_login(self.make_user("gate-ops2@ixg.local", "ops", True)) self.assertNotContains(client.get("/ops/"), 'href="/admin/"') def test_the_ops_sidebar_shows_admin_to_an_admin(self): client = Client(SERVER_NAME="ops.ixgltd.com") client.force_login(self.make_user("gate-admin2@ixg.local", "admin", True)) self.assertContains(client.get("/ops/"), 'href="/admin/"') class ApiPermissionBoundaryTests(TestCase): """The API permission boundary, pinned for every viewset. Batch writes were closed first; this sweeps the rest. A buyer account is the adversary: it is a legitimate, authenticated user, so ``IsAuthenticated`` alone lets it through. Anything that changes a record — a document, a QC result, a certificate, a shipment status — must be ops/admin only. The buyer portal is server-rendered through the ORM, so it needs no API access; the one exception is the AI chat, which the buyer portal calls from the browser. """ # Every ModelViewSet that changes a record. A buyer must get 403 on write. OPS_WRITE_ENDPOINTS = [ "/api/batches/", "/api/qc-results/", "/api/shipments/", "/api/documents/", "/api/certificates-of-analysis/", "/api/regulatory-certs/", "/api/capa-records/", "/api/esg-reports/", "/api/agents/", "/api/product-specs/", "/api/facilities/", "/api/origin-zones/", "/api/collector-groups/", "/api/harvest-intakes/", "/api/intake-batch-lineage/", "/api/eudr-cases/", "/api/eudr-evidence/", "/api/risk-assessments/", "/api/mitigation-actions/", "/api/dds-references/", ] # Reads a buyer may still make: the chat it calls from the browser, and the # user/notification/subscription rows scoped to the buyer itself. BUYER_READABLE_ENDPOINTS = [ "/api/agents/", "/api/ai-interactions/", "/api/users/", "/api/notifications/", "/api/subscriptions/", ] 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 setUp(self): self.buyer = self.make_user("boundary-buyer@ixg.local", "buyer") self.other_buyer = self.make_user("boundary-other@ixg.local", "buyer") self.ops = self.make_user("boundary-ops@ixg.local", "ops") self.client = APIClient() def test_a_buyer_cannot_write_to_any_ops_api(self): self.client.force_authenticate(user=self.buyer) for endpoint in self.OPS_WRITE_ENDPOINTS: with self.subTest(endpoint=endpoint): self.assertEqual( self.client.post(endpoint, {}, format="json").status_code, 403 ) def test_a_buyer_cannot_read_the_ops_apis(self): self.client.force_authenticate(user=self.buyer) for endpoint in self.OPS_WRITE_ENDPOINTS: if endpoint in self.BUYER_READABLE_ENDPOINTS: continue with self.subTest(endpoint=endpoint): self.assertEqual(self.client.get(endpoint).status_code, 403) def test_ops_is_not_forbidden_on_the_ops_apis(self): self.client.force_authenticate(user=self.ops) for endpoint in self.OPS_WRITE_ENDPOINTS: with self.subTest(endpoint=endpoint): self.assertNotEqual(self.client.get(endpoint).status_code, 403) def test_a_buyer_can_still_reach_the_chat_endpoints(self): self.client.force_authenticate(user=self.buyer) for endpoint in self.BUYER_READABLE_ENDPOINTS: with self.subTest(endpoint=endpoint): self.assertNotEqual(self.client.get(endpoint).status_code, 403) response = self.client.post("/api/ai-interactions/chat/", {}, format="json") self.assertNotEqual(response.status_code, 403) def test_a_buyer_cannot_promote_themselves_to_admin(self): self.client.force_authenticate(user=self.buyer) response = self.client.patch( f"/api/users/{self.buyer.id}/", {"role": "admin"}, format="json" ) self.assertEqual(response.status_code, 403) self.buyer.refresh_from_db() self.assertEqual(self.buyer.role, "buyer") def test_the_user_api_shows_a_buyer_only_themselves(self): self.client.force_authenticate(user=self.buyer) response = self.client.get("/api/users/") self.assertEqual(response.status_code, 200) body = response.json() rows = body["results"] if isinstance(body, dict) else body self.assertEqual([row["email"] for row in rows], [self.buyer.email]) def test_a_user_cannot_reach_another_users_rows(self): note = Notification.objects.create( user=self.other_buyer, title="Other", message="Other" ) sub = Subscription.objects.create( user=self.other_buyer, plan=Subscription.Plan.STARTER ) self.client.force_authenticate(user=self.buyer) self.assertEqual( self.client.get(f"/api/notifications/{note.id}/").status_code, 404 ) self.assertEqual( self.client.get(f"/api/subscriptions/{sub.id}/").status_code, 404 ) 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_the_seeded_buyer_is_not_staff(self): """Staff status is derived from the role; the seed must not disagree.""" buyer = User.objects.get(email="buyer@greenfieldfoods.example") self.assertFalse(buyer.is_staff) def test_re_running_the_seed_corrects_a_staff_buyer(self): """The seed used to set is_staff only on creation, so a buyer whose flag had drifted to True stayed inside the admin area until its next sign-in. Re-running the seed must correct it, not reopen the window.""" User.objects.filter(email="buyer@greenfieldfoods.example").update(is_staff=True) call_command("seed_walkthrough", verbosity=0) buyer = User.objects.get(email="buyer@greenfieldfoods.example") self.assertFalse(buyer.is_staff) 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() shipment = make_sweep_shipment(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) + shipment_screens(shipment), "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