"""The ops intake and batch screens. Two things matter here and neither is cosmetic: that only operations can create records, and that the forms refuse what the platform's own checks exist to catch. The model tolerates a missing intake date, weight or zone because records also arrive by API — the portal form does not. And the model has no opinion at all about a batch's status or about how much of an intake can be allocated, which is where the real rules live: a QC failure must not be typable, and an intake cannot give away more shea than it received. """ from decimal import Decimal from django.contrib.auth import get_user_model from django.test import Client, TestCase from django.utils import timezone from apps.batches.models import ProcessingBatch from apps.compliance.models import ( CollectorGroup, DueDiligenceCase, DueDiligenceStatementReference, EUDREvidence, Facility, HarvestIntake, IntakeBatchLineage, MitigationAction, OriginZone, ProductSpecification, SpecParameter, ) from apps.shipments.models import Shipment from .views import suggested_batch_reference, suggested_intake_reference User = get_user_model() class IntakeTestCase(TestCase): def setUp(self): self.ops = User.objects.create_user( email="intake-ops@ixg.local", password="pass1234", role="ops", full_name="Intake Officer", is_staff=True, ) self.buyer = User.objects.create_user( email="intake-buyer@ixg.local", password="pass1234", role="buyer", full_name="Intake Buyer", is_staff=True, ) self.facility = Facility.objects.create(name="IXG Test Facility") self.zone = OriginZone.objects.create( facility=self.facility, zone_code="NG-KT-ZTEST", country="Nigeria", region="Katsina State", verification_status="verified", ) self.other_zone = OriginZone.objects.create( facility=self.facility, zone_code="NG-KT-ZTEST2", country="Nigeria", region="Katsina State", verification_status="unverified", ) self.group = CollectorGroup.objects.create( name="Test Collective", contact_reference="CG-TEST-001", origin_zone=self.zone, ) self.client_ops = Client(SERVER_NAME="ops.ixgltd.com") self.client_ops.force_login(self.ops) self.client_buyer = Client(SERVER_NAME="ops.ixgltd.com") self.client_buyer.force_login(self.buyer) self.intake = HarvestIntake.objects.create( intake_reference="IXG-INTAKE-2026-9001", collector_group=self.group, origin_zone=self.zone, intake_date="2026-09-01", weight_kg=Decimal("500.00"), grade="A", status="received", ) def payload(self, **overrides): data = { "intake_reference": "IXG-INTAKE-2026-9002", "collector_group": str(self.group.pk), "origin_zone": str(self.zone.pk), "intake_date": "2026-09-10", "weight_kg": "640.00", "grade": "A", "receiving_officer": "Intake Officer", "status": "received", "evidence_file": "weighbridge-ticket-9002.pdf", } data.update(overrides) return data def errors_for(self, response, field=None): form = response.context["form"] if field: return " ".join(form.errors.get(field, [])) return " ".join(str(v) for v in form.errors.values()) class IntakeAccessTests(IntakeTestCase): def test_ops_can_open_the_list(self): r = self.client_ops.get("/ops/intakes/") self.assertEqual(r.status_code, 200) self.assertContains(r, "IXG-INTAKE-2026-9001") def test_ops_can_open_the_create_form(self): self.assertEqual(self.client_ops.get("/ops/intakes/new/").status_code, 200) def test_ops_can_open_the_edit_form(self): r = self.client_ops.get(f"/ops/intakes/{self.intake.pk}/edit/") self.assertEqual(r.status_code, 200) self.assertContains(r, "IXG-INTAKE-2026-9001") def test_a_buyer_cannot_open_the_intake_screens(self): for path in ("/ops/intakes/", "/ops/intakes/new/", f"/ops/intakes/{self.intake.pk}/edit/"): with self.subTest(path=path): self.assertEqual(self.client_buyer.get(path).status_code, 403) def test_the_intake_list_appears_in_the_sidebar(self): self.assertContains(self.client_ops.get("/ops/intakes/"), "Harvest Intakes") def test_the_edit_form_carries_an_iso_intake_date(self): """A native date input discards a d/m/Y value, so the field would be empty on an edit — and saving would then demand it again.""" r = self.client_ops.get(f"/ops/intakes/{self.intake.pk}/edit/") self.assertContains(r, 'value="2026-09-01"') class IntakeCreateTests(IntakeTestCase): def test_a_complete_intake_is_recorded(self): r = self.client_ops.post("/ops/intakes/new/", self.payload(), follow=True) self.assertEqual(r.status_code, 200) intake = HarvestIntake.objects.get(intake_reference="IXG-INTAKE-2026-9002") self.assertEqual(intake.weight_kg, Decimal("640.00")) self.assertEqual(intake.origin_zone, self.zone) self.assertEqual(intake.collector_group, self.group) self.assertContains(r, "recorded") def test_the_form_suggests_the_next_reference_and_today(self): r = self.client_ops.get("/ops/intakes/new/") self.assertContains(r, 'value="IXG-INTAKE-2026-9002"') self.assertContains(r, "Intake Officer") def test_a_duplicate_reference_is_refused(self): r = self.client_ops.post( "/ops/intakes/new/", self.payload(intake_reference="IXG-INTAKE-2026-9001") ) self.assertEqual(r.status_code, 200) self.assertIn("already exists", self.errors_for(r, "intake_reference")) self.assertEqual(HarvestIntake.objects.count(), 1) def test_a_duplicate_differing_only_in_case_is_refused(self): r = self.client_ops.post( "/ops/intakes/new/", self.payload(intake_reference="ixg-intake-2026-9001") ) self.assertIn("already exists", self.errors_for(r, "intake_reference")) def test_a_zero_or_negative_weight_is_refused(self): for weight in ("0", "-5"): with self.subTest(weight=weight): r = self.client_ops.post("/ops/intakes/new/", self.payload(weight_kg=weight)) self.assertIn("greater than zero", self.errors_for(r, "weight_kg")) def test_the_fields_the_checks_read_are_required(self): cases = { "intake_reference": "", "collector_group": "", "origin_zone": "", "intake_date": "", "weight_kg": "", "grade": "", } for field, blank in cases.items(): with self.subTest(field=field): r = self.client_ops.post("/ops/intakes/new/", self.payload(**{field: blank})) self.assertEqual(r.status_code, 200) self.assertTrue( self.errors_for(r, field), f"{field} was accepted blank" ) def test_an_unknown_grade_is_refused(self): r = self.client_ops.post("/ops/intakes/new/", self.payload(grade="Z")) self.assertTrue(self.errors_for(r, "grade")) def test_a_zone_that_does_not_match_the_group_is_queried(self): r = self.client_ops.post( "/ops/intakes/new/", self.payload(origin_zone=str(self.other_zone.pk)) ) self.assertIn("registered to", self.errors_for(r, "origin_zone")) def test_an_unverified_zone_is_allowed_but_flagged(self): """Recording against a zone still awaiting verification must stay possible.""" group_without_zone = CollectorGroup.objects.create(name="Unzoned Collective") r = self.client_ops.post( "/ops/intakes/new/", self.payload(collector_group=str(group_without_zone.pk), origin_zone=str(self.other_zone.pk)), ) self.assertEqual(r.status_code, 302) self.assertTrue( HarvestIntake.objects.filter(intake_reference="IXG-INTAKE-2026-9002").exists() ) def test_the_form_labels_an_unverified_zone(self): r = self.client_ops.get("/ops/intakes/new/") self.assertContains(r, "unverified") class IntakeEditTests(IntakeTestCase): def test_an_intake_can_be_corrected(self): data = self.payload(intake_reference="IXG-INTAKE-2026-9001", weight_kg="510.00") r = self.client_ops.post(f"/ops/intakes/{self.intake.pk}/edit/", data, follow=True) self.assertEqual(r.status_code, 200) self.intake.refresh_from_db() self.assertEqual(self.intake.weight_kg, Decimal("510.00")) self.assertContains(r, "updated") def test_saving_an_intake_under_its_own_reference_is_not_a_duplicate(self): data = self.payload(intake_reference="IXG-INTAKE-2026-9001") r = self.client_ops.post(f"/ops/intakes/{self.intake.pk}/edit/", data) self.assertEqual(r.status_code, 302) def test_correcting_does_not_create_a_second_record(self): data = self.payload(intake_reference="IXG-INTAKE-2026-9001", grade="B") self.client_ops.post(f"/ops/intakes/{self.intake.pk}/edit/", data) self.assertEqual(HarvestIntake.objects.count(), 1) class SuggestedReferenceTests(IntakeTestCase): def test_the_suggestion_continues_the_year_sequence(self): HarvestIntake.objects.create( intake_reference="IXG-INTAKE-2026-9042", intake_date="2026-09-02", weight_kg=Decimal("10.00"), grade="A", ) self.assertEqual(suggested_intake_reference(), "IXG-INTAKE-2026-9043") def test_the_suggestion_ignores_other_years_and_odd_references(self): HarvestIntake.objects.create( intake_reference="IXG-INTAKE-2025-9999", intake_date="2025-09-02", weight_kg=Decimal("10.00"), grade="A", ) HarvestIntake.objects.create( intake_reference="IXG-INTAKE-2026-ABC", intake_date="2026-09-02", weight_kg=Decimal("10.00"), grade="A", ) self.assertEqual(suggested_intake_reference(), "IXG-INTAKE-2026-9002") class BatchTestCase(TestCase): """A zone, a group, a facility and three intakes — the world a batch lives in.""" def setUp(self): self.ops = User.objects.create_user( email="batch-ops@ixg.local", password="pass1234", role="ops", full_name="Batch Officer", is_staff=True, ) self.buyer = User.objects.create_user( email="batch-buyer@ixg.local", password="pass1234", role="buyer", full_name="Batch Buyer", is_staff=True, ) self.facility = Facility.objects.create( name="IXG Test Processing Facility", country="Nigeria", region="Katsina State", ) self.zone = OriginZone.objects.create( facility=self.facility, zone_code="NG-KT-ZBATCH", country="Nigeria", region="Katsina State", verification_status="verified", ) self.group = CollectorGroup.objects.create( name="Batch Test Collective", contact_reference="CG-BATCH-001", origin_zone=self.zone, ) self.intakes = [ HarvestIntake.objects.create( intake_reference=f"IXG-INTAKE-2026-91{n:02d}", collector_group=self.group, origin_zone=self.zone, intake_date=f"2026-09-0{n}", weight_kg=Decimal(weight), grade="A", status="verified", ) for n, weight in ((1, "820.00"), (2, "640.00"), (3, "600.00")) ] self.client_ops = Client(SERVER_NAME="ops.ixgltd.com") self.client_ops.force_login(self.ops) self.client_buyer = Client(SERVER_NAME="ops.ixgltd.com") self.client_buyer.force_login(self.buyer) self.batch = ProcessingBatch.objects.create( batch_reference="IXG-BATCH-2026-901", harvest_date="2026-09-04", weight_kg=Decimal("1500.00"), grade="A", status="harvested", origin_facility=self.facility.name, created_by=self.ops, ) def payload(self, **overrides): data = { "batch_reference": "IXG-BATCH-2026-902", "grade": "A", "weight_kg": "1200.00", "harvest_date": "2026-09-10", "origin_facility": self.facility.name, "notes": "Synthetic batch for tests.", "status": "harvested", } data.update(overrides) return data def errors_for(self, response, field=None): form = response.context["form"] if field: return " ".join(form.errors.get(field, [])) return " ".join(str(v) for v in form.errors.values()) def lineage_payload(self, rows, initial=0): """A formset POST body. Each row: {intake, weight, id?, delete?}.""" data = { "lineage-TOTAL_FORMS": str(len(rows) + 1), "lineage-INITIAL_FORMS": str(initial), "lineage-MIN_NUM_FORMS": "0", "lineage-MAX_NUM_FORMS": "1000", } for i, row in enumerate(rows): data[f"lineage-{i}-harvest_intake"] = str(row.get("intake") or "") data[f"lineage-{i}-weight_allocated_kg"] = str(row.get("weight") or "") if row.get("id"): data[f"lineage-{i}-id"] = str(row["id"]) if row.get("delete"): data[f"lineage-{i}-DELETE"] = "on" data[f"lineage-{len(rows)}-harvest_intake"] = "" data[f"lineage-{len(rows)}-weight_allocated_kg"] = "" return data def lineage_errors(self, response): formset = response.context["formset"] return " ".join( str(e) for form in formset.forms for e in form.errors.values() ) class BatchAccessTests(BatchTestCase): def test_ops_can_open_every_batch_screen(self): for path in ( "/ops/batches/", "/ops/batches/new/", f"/ops/batches/{self.batch.pk}/edit/", f"/ops/batches/{self.batch.pk}/lineage/", ): with self.subTest(path=path): self.assertEqual(self.client_ops.get(path).status_code, 200) def test_a_buyer_cannot_open_them(self): for path in ( "/ops/batches/new/", f"/ops/batches/{self.batch.pk}/edit/", f"/ops/batches/{self.batch.pk}/lineage/", ): with self.subTest(path=path): self.assertEqual(self.client_buyer.get(path).status_code, 403) def test_a_buyer_cannot_post_a_new_batch(self): r = self.client_buyer.post("/ops/batches/new/", self.payload()) self.assertEqual(r.status_code, 403) self.assertFalse( ProcessingBatch.objects.filter(batch_reference="IXG-BATCH-2026-902").exists() ) def test_a_buyer_cannot_post_an_edit(self): r = self.client_buyer.post( f"/ops/batches/{self.batch.pk}/edit/", self.payload(batch_reference="IXG-BATCH-2026-901", status="delivered"), ) self.assertEqual(r.status_code, 403) self.batch.refresh_from_db() self.assertEqual(self.batch.status, "harvested") def test_the_list_offers_a_new_batch_control(self): r = self.client_ops.get("/ops/batches/") self.assertContains(r, "New batch") self.assertIn("/ops/batches/new/", r.content.decode()) def test_the_edit_form_carries_an_iso_harvest_date(self): """A native date input discards a d/m/Y value, so the field would be empty on an edit — and saving would then demand it again.""" r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/edit/") self.assertContains(r, 'value="2026-09-04"') class BatchCreateTests(BatchTestCase): def test_a_complete_batch_is_created_and_sent_to_its_lineage(self): r = self.client_ops.post("/ops/batches/new/", self.payload()) self.assertEqual(r.status_code, 302) batch = ProcessingBatch.objects.get(batch_reference="IXG-BATCH-2026-902") self.assertEqual(batch.weight_kg, Decimal("1200.00")) self.assertEqual(batch.origin_facility, self.facility.name) # Created by the signed-in user, and pointed straight at the lineage, # because a batch with no intake behind it is not traceable. self.assertEqual(batch.created_by, self.ops) self.assertEqual(r["Location"], f"/ops/batches/{batch.pk}/lineage/") def test_the_form_suggests_the_next_reference(self): r = self.client_ops.get("/ops/batches/new/") self.assertContains(r, f'value="{suggested_batch_reference()}"') def test_a_duplicate_reference_is_refused(self): r = self.client_ops.post( "/ops/batches/new/", self.payload(batch_reference="IXG-BATCH-2026-901") ) self.assertEqual(r.status_code, 200) self.assertIn("already exists", self.errors_for(r, "batch_reference")) def test_a_duplicate_differing_only_in_case_is_refused(self): r = self.client_ops.post( "/ops/batches/new/", self.payload(batch_reference="ixg-batch-2026-901") ) self.assertIn("already exists", self.errors_for(r, "batch_reference")) def test_a_zero_or_negative_weight_is_refused(self): for weight in ("0", "-5"): with self.subTest(weight=weight): r = self.client_ops.post("/ops/batches/new/", self.payload(weight_kg=weight)) self.assertIn("greater than zero", self.errors_for(r, "weight_kg")) def test_the_fields_a_batch_needs_are_required(self): for field in ("batch_reference", "grade", "weight_kg", "harvest_date", "origin_facility", "status"): with self.subTest(field=field): r = self.client_ops.post("/ops/batches/new/", self.payload(**{field: ""})) self.assertEqual(r.status_code, 200) self.assertTrue( self.errors_for(r, field), f"{field} was accepted blank" ) def test_a_new_batch_cannot_be_born_mid_journey(self): for status in ("in_transit", "delivered", "ready_to_ship"): with self.subTest(status=status): r = self.client_ops.post("/ops/batches/new/", self.payload(status=status)) self.assertTrue(self.errors_for(r, "status")) def test_a_qc_status_cannot_be_typed_by_hand(self): for status in ("qc_pass", "qc_fail"): with self.subTest(status=status): r = self.client_ops.post("/ops/batches/new/", self.payload(status=status)) self.assertTrue(self.errors_for(r, "status")) self.assertFalse( ProcessingBatch.objects.filter(batch_reference="IXG-BATCH-2026-902").exists() ) def test_the_origin_facility_comes_from_the_register(self): r = self.client_ops.post( "/ops/batches/new/", self.payload(origin_facility="A site nobody registered") ) self.assertTrue(self.errors_for(r, "origin_facility")) def test_the_origin_facility_help_text_matches_what_the_form_does(self): r = self.client_ops.get("/ops/batches/new/") # An existing unregistered value is kept, so the help must not claim the # batch and the register cannot disagree. self.assertContains(r, "keeps its own value") self.assertNotContains(r, "cannot disagree") def test_the_status_card_says_which_qc_statuses_are_offered(self): r = self.client_ops.get("/ops/batches/new/") self.assertContains(r, "never offered here") self.assertContains(r, "appears only as the current value") class BatchStatusRuleTests(BatchTestCase): """The rule that makes the screen worth having.""" def test_a_journey_status_is_moved_by_hand(self): r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/edit/", self.payload(batch_reference="IXG-BATCH-2026-901", status="ready_to_ship"), ) self.assertEqual(r.status_code, 302) self.batch.refresh_from_db() self.assertEqual(self.batch.status, "ready_to_ship") def test_a_passed_batch_can_go_on_to_ship(self): self.batch.status = "qc_pass" self.batch.save() r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/edit/", self.payload(batch_reference="IXG-BATCH-2026-901", status="ready_to_ship"), ) self.assertEqual(r.status_code, 302) self.batch.refresh_from_db() self.assertEqual(self.batch.status, "ready_to_ship") def test_a_failed_batch_keeps_its_failed_status(self): """The screen must not be a way to make a quality failure disappear.""" self.batch.status = "qc_fail" self.batch.save() r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/edit/") html = r.content.decode() self.assertIn("Set by the QC result", html) # Only the current status is on offer, so it cannot be typed over. self.assertNotIn('value="ready_to_ship"', html) self.assertNotIn('value="delivered"', html) r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/edit/", self.payload(batch_reference="IXG-BATCH-2026-901", status="ready_to_ship"), ) self.batch.refresh_from_db() self.assertEqual(self.batch.status, "qc_fail") self.assertEqual(r.status_code, 302) def test_a_qc_status_cannot_be_hand_set_on_a_journey_batch(self): r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/edit/", self.payload(batch_reference="IXG-BATCH-2026-901", status="qc_fail"), ) self.assertTrue(self.errors_for(r, "status")) self.batch.refresh_from_db() self.assertEqual(self.batch.status, "harvested") def test_editing_updates_in_place(self): r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/edit/", self.payload(batch_reference="IXG-BATCH-2026-901", weight_kg="1400.00"), ) self.assertEqual(r.status_code, 302) self.batch.refresh_from_db() self.assertEqual(self.batch.weight_kg, Decimal("1400.00")) self.assertEqual(ProcessingBatch.objects.count(), 1) def test_an_existing_facility_that_is_not_registered_is_kept_and_flagged(self): """History is not silently rewritten, but the drift is made visible.""" self.batch.origin_facility = "ASI - Katsina State, Nigeria" self.batch.save() r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/edit/") self.assertContains(r, "not in the facility register") def test_saving_under_its_own_reference_is_not_a_duplicate(self): r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/edit/", self.payload(batch_reference="IXG-BATCH-2026-901"), ) self.assertEqual(r.status_code, 302) class BatchLineageTests(BatchTestCase): def test_the_screen_shows_what_is_allocated_and_the_batch_weight(self): IntakeBatchLineage.objects.create( batch=self.batch, harvest_intake=self.intakes[0], weight_allocated_kg=Decimal("500.00"), ) r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/lineage/") self.assertContains(r, "500.00") self.assertContains(r, "1500.00") def test_an_allocation_is_created_against_the_batch(self): r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/lineage/", self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "800.00"}]), ) self.assertEqual(r.status_code, 302) row = IntakeBatchLineage.objects.get(batch=self.batch) self.assertEqual(row.harvest_intake, self.intakes[0]) self.assertEqual(row.weight_allocated_kg, Decimal("800.00")) def test_the_batch_detail_page_shows_the_lineage(self): IntakeBatchLineage.objects.create( batch=self.batch, harvest_intake=self.intakes[2], weight_allocated_kg=Decimal("600.00"), ) r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/") self.assertContains(r, "Intake Lineage") self.assertContains(r, self.intakes[2].intake_reference) self.assertContains(r, "600.00") def test_a_batch_with_no_lineage_is_told_so(self): r = self.client_ops.get(f"/ops/batches/{self.batch.pk}/") self.assertContains(r, "No harvest intakes are linked to this batch") def test_more_than_the_intake_received_is_refused(self): r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/lineage/", self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "900.00"}]), ) self.assertEqual(r.status_code, 200) self.assertIn("cannot be allocated", self.lineage_errors(r)) self.assertFalse(IntakeBatchLineage.objects.exists()) def test_an_intake_already_taken_by_another_batch_is_refused(self): """The rule that counts every batch — and the defect in the old seed data.""" other = ProcessingBatch.objects.create( batch_reference="IXG-BATCH-2026-903", weight_kg=Decimal("800.00"), grade="A", status="processing", origin_facility=self.facility.name, ) IntakeBatchLineage.objects.create( batch=other, harvest_intake=self.intakes[0], weight_allocated_kg=Decimal("800.00"), ) r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/lineage/", self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "800.00"}]), ) self.assertEqual(r.status_code, 200) message = self.lineage_errors(r) self.assertIn("already allocated", message) self.assertIn("IXG-BATCH-2026-903", message) self.assertEqual(IntakeBatchLineage.objects.filter(batch=self.batch).count(), 0) def test_what_is_left_of_an_intake_can_still_be_allocated(self): other = ProcessingBatch.objects.create( batch_reference="IXG-BATCH-2026-904", weight_kg=Decimal("500.00"), grade="A", status="processing", origin_facility=self.facility.name, ) IntakeBatchLineage.objects.create( batch=other, harvest_intake=self.intakes[0], weight_allocated_kg=Decimal("500.00"), ) r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/lineage/", self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "300.00"}]), ) self.assertEqual(r.status_code, 302) self.assertEqual( IntakeBatchLineage.objects.get(batch=self.batch).weight_allocated_kg, Decimal("300.00"), ) def test_the_same_intake_twice_in_one_batch_is_refused(self): r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/lineage/", self.lineage_payload([ {"intake": self.intakes[0].pk, "weight": "200.00"}, {"intake": self.intakes[0].pk, "weight": "300.00"}, ]), ) self.assertEqual(r.status_code, 200) self.assertIn("already linked to this batch", self.lineage_errors(r)) def test_a_zero_allocation_is_refused(self): r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/lineage/", self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "0"}]), ) self.assertTrue(self.lineage_errors(r)) def test_an_allocation_can_be_corrected_and_removed(self): row = IntakeBatchLineage.objects.create( batch=self.batch, harvest_intake=self.intakes[0], weight_allocated_kg=Decimal("500.00"), ) r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/lineage/", self.lineage_payload( [{"id": row.pk, "intake": self.intakes[0].pk, "weight": "520.00"}], initial=1, ), ) self.assertEqual(r.status_code, 302) row.refresh_from_db() self.assertEqual(row.weight_allocated_kg, Decimal("520.00")) r = self.client_ops.post( f"/ops/batches/{self.batch.pk}/lineage/", self.lineage_payload( [{"id": row.pk, "intake": self.intakes[0].pk, "weight": "520.00", "delete": True}], initial=1, ), ) self.assertEqual(r.status_code, 302) self.assertFalse(IntakeBatchLineage.objects.filter(batch=self.batch).exists()) # The intake itself is untouched: removing a link is not deleting material. self.assertTrue(HarvestIntake.objects.filter(pk=self.intakes[0].pk).exists()) def test_a_buyer_cannot_post_an_allocation(self): r = self.client_buyer.post( f"/ops/batches/{self.batch.pk}/lineage/", self.lineage_payload([{"intake": self.intakes[0].pk, "weight": "500.00"}]), ) self.assertEqual(r.status_code, 403) self.assertFalse(IntakeBatchLineage.objects.exists()) class SuggestedBatchReferenceTests(BatchTestCase): def test_the_suggestion_continues_the_year_sequence(self): self.assertEqual(suggested_batch_reference(), "IXG-BATCH-2026-902") def test_the_suggestion_ignores_other_years_and_odd_references(self): ProcessingBatch.objects.create( batch_reference="IXG-BATCH-2025-9999", weight_kg=Decimal("10.00"), grade="A", status="harvested", ) ProcessingBatch.objects.create( batch_reference="IXG-TEST-001", weight_kg=Decimal("10.00"), grade="A", status="harvested", ) self.assertEqual(suggested_batch_reference(), "IXG-BATCH-2026-902") class ShipmentTestCase(TestCase): def setUp(self): self.ops = User.objects.create_user( email="ship-ops@ixg.local", password="pass1234", role="ops", full_name="Shipment Officer", is_staff=True, ) self.buyer = User.objects.create_user( email="ship-buyer@ixg.local", password="pass1234", role="buyer", full_name="Shipment Buyer", is_staff=True, ) self.batch = ProcessingBatch.objects.create( batch_reference="IXG-BATCH-2026-903", harvest_date="2026-09-04", weight_kg=Decimal("1500.00"), grade="A", status="ready_to_ship", ) self.other_batch = ProcessingBatch.objects.create( batch_reference="IXG-BATCH-2026-904", status="ready_to_ship", ) self.shipment = Shipment.objects.create( batch=self.batch, status=Shipment.Status.BOOKED, vessel_name="MV Existing", container_ref="IXGU7654321", etd_nigeria="2026-09-15", eta_uk="2026-10-15", ) self.client_ops = Client(SERVER_NAME="ops.ixgltd.com") self.client_ops.force_login(self.ops) self.client_buyer = Client(SERVER_NAME="ops.ixgltd.com") self.client_buyer.force_login(self.buyer) def payload(self, **overrides): data = { "batch": str(self.other_batch.pk), "status": "booked", "origin_port": "Apapa Port, Lagos", "destination_port": "Port of Felixstowe", "vessel_name": "MV Test", "container_ref": "IXGU1234567", "etd_nigeria": "2026-09-20", "eta_uk": "2026-10-20", "bl_number": "", "freight_forwarder": "", } data.update(overrides) return data def errors_for(self, response, field=None): form = response.context["form"] if field: return " ".join(form.errors.get(field, [])) return " ".join(str(v) for v in form.errors.values()) class ShipmentAccessTests(ShipmentTestCase): def test_ops_can_open_every_shipment_screen(self): for path in ( "/ops/shipments/", "/ops/shipments/new/", f"/ops/shipments/{self.shipment.pk}/", f"/ops/shipments/{self.shipment.pk}/edit/", ): with self.subTest(path=path): self.assertEqual(self.client_ops.get(path).status_code, 200) def test_a_buyer_cannot_open_them(self): for path in ( "/ops/shipments/new/", f"/ops/shipments/{self.shipment.pk}/", f"/ops/shipments/{self.shipment.pk}/edit/", ): with self.subTest(path=path): self.assertEqual(self.client_buyer.get(path).status_code, 403) def test_a_buyer_cannot_post_a_new_shipment(self): r = self.client_buyer.post("/ops/shipments/new/", self.payload()) self.assertEqual(r.status_code, 403) self.assertEqual(Shipment.objects.count(), 1) def test_a_buyer_cannot_post_an_edit(self): r = self.client_buyer.post( f"/ops/shipments/{self.shipment.pk}/edit/", self.payload(batch=str(self.batch.pk), status="delivered"), ) self.assertEqual(r.status_code, 403) self.shipment.refresh_from_db() self.assertEqual(self.shipment.status, Shipment.Status.BOOKED) def test_the_list_offers_a_new_shipment_control(self): r = self.client_ops.get("/ops/shipments/") self.assertContains(r, "New shipment") self.assertIn("/ops/shipments/new/", r.content.decode()) def test_the_detail_offers_an_edit_control(self): r = self.client_ops.get(f"/ops/shipments/{self.shipment.pk}/") self.assertIn(f"/ops/shipments/{self.shipment.pk}/edit/", r.content.decode()) class ShipmentCreateTests(ShipmentTestCase): def test_a_complete_shipment_is_created(self): r = self.client_ops.post("/ops/shipments/new/", self.payload()) self.assertEqual(r.status_code, 302) shipment = Shipment.objects.get(batch=self.other_batch) self.assertEqual(shipment.vessel_name, "MV Test") self.assertEqual(shipment.container_ref, "IXGU1234567") self.assertEqual(str(shipment.etd_nigeria), "2026-09-20") self.assertEqual(str(shipment.eta_uk), "2026-10-20") self.assertEqual(r["Location"], f"/ops/shipments/{shipment.pk}/") def test_each_required_field_is_refused_when_blank(self): for field in ( "batch", "status", "vessel_name", "container_ref", "etd_nigeria", "eta_uk", ): with self.subTest(field=field): r = self.client_ops.post("/ops/shipments/new/", self.payload(**{field: ""})) self.assertEqual(r.status_code, 200) self.assertTrue(self.errors_for(r, field)) self.assertEqual(Shipment.objects.count(), 1) def test_a_departure_after_the_arrival_is_refused(self): r = self.client_ops.post( "/ops/shipments/new/", self.payload(etd_nigeria="2026-11-01", eta_uk="2026-10-01"), ) self.assertEqual(r.status_code, 200) self.assertIn("before the ETD", self.errors_for(r, "eta_uk")) self.assertEqual(Shipment.objects.count(), 1) def test_a_batch_that_already_has_a_shipment_is_queried_not_blocked(self): r = self.client_ops.post( "/ops/shipments/new/", self.payload(batch=str(self.batch.pk)), follow=True, ) self.assertEqual(r.status_code, 200) self.assertEqual(Shipment.objects.filter(batch=self.batch).count(), 2) self.assertContains(r, "check it is not a duplicate") def test_the_batch_dropdown_names_the_status_and_existing_shipments(self): r = self.client_ops.get("/ops/shipments/new/") self.assertContains(r, "IXG-BATCH-2026-903 — Ready to Ship (1 other shipment)") self.assertContains(r, "IXG-BATCH-2026-904 — Ready to Ship") class ShipmentEditTests(ShipmentTestCase): def test_an_edit_updates_in_place(self): r = self.client_ops.post( f"/ops/shipments/{self.shipment.pk}/edit/", self.payload( batch=str(self.batch.pk), vessel_name="MV Corrected", status="in_transit", ), ) self.assertEqual(r.status_code, 302) self.assertEqual(Shipment.objects.count(), 1) self.shipment.refresh_from_db() self.assertEqual(self.shipment.vessel_name, "MV Corrected") self.assertEqual(self.shipment.status, Shipment.Status.IN_TRANSIT) def test_the_form_lists_the_batches_other_shipments(self): other = Shipment.objects.create( batch=self.batch, status=Shipment.Status.LOADING, vessel_name="MV Second", container_ref="IXGU9999999", etd_nigeria="2026-09-18", eta_uk="2026-10-18", ) r = self.client_ops.get(f"/ops/shipments/{self.shipment.pk}/edit/") self.assertContains(r, "Other shipments on this batch") self.assertContains(r, other.container_ref) def test_the_edit_form_carries_iso_dates(self): """A native date input discards a d/m/Y value, so the dates would be empty on an edit and saving a status change would demand them again.""" r = self.client_ops.get(f"/ops/shipments/{self.shipment.pk}/edit/") self.assertContains(r, 'value="2026-09-15"') self.assertContains(r, 'value="2026-10-15"') # --------------------------------------------------------------------------- # Compliance registers # # Facility, origin zone and collector group used to be admin-only. These tests # pin that operations can now register them from the portal, that only # operations may, and that the origin-zone map is a view of the stored geometry # rather than a check of it. class ComplianceRegisterBase(TestCase): def setUp(self): self.ops = User.objects.create_user( email="reg-ops@ixg.local", password="pass1234", role="ops", full_name="Register Officer", is_staff=True, ) self.buyer = User.objects.create_user( email="reg-buyer@ixg.local", password="pass1234", role="buyer", full_name="Register Buyer", is_staff=True, ) self.facility = Facility.objects.create( name="IXG Register Facility", country="Nigeria", region="Katsina State", facility_type=Facility.FacilityType.PROCESSING, ) self.zone = OriginZone.objects.create( facility=self.facility, zone_code="NG-KT-RZ01", country="Nigeria", region="Katsina State", verification_status="verified", verified_by=self.ops, verified_at=timezone.now(), ) self.group = CollectorGroup.objects.create( name="Register Collective", contact_reference="CG-R-001", origin_zone=self.zone, ) self.client_ops = Client(SERVER_NAME="ops.ixgltd.com") self.client_ops.force_login(self.ops) self.client_buyer = Client(SERVER_NAME="ops.ixgltd.com") self.client_buyer.force_login(self.buyer) class RegisterAccessTests(ComplianceRegisterBase): def test_ops_can_open_the_register_lists(self): for path in ("/ops/facilities/", "/ops/origin-zones/", "/ops/collector-groups/"): with self.subTest(path=path): self.assertEqual(self.client_ops.get(path).status_code, 200) def test_ops_can_open_the_create_forms(self): for path in ( "/ops/facilities/new/", "/ops/origin-zones/new/", "/ops/collector-groups/new/", ): with self.subTest(path=path): self.assertEqual(self.client_ops.get(path).status_code, 200) def test_a_buyer_cannot_open_the_registers(self): for path in ( "/ops/facilities/", "/ops/facilities/new/", f"/ops/facilities/{self.facility.pk}/edit/", "/ops/origin-zones/", "/ops/origin-zones/new/", f"/ops/origin-zones/{self.zone.pk}/edit/", "/ops/collector-groups/", "/ops/collector-groups/new/", f"/ops/collector-groups/{self.group.pk}/edit/", ): with self.subTest(path=path): self.assertEqual(self.client_buyer.get(path).status_code, 403) def test_the_sidebar_links_to_the_registers(self): html = self.client_ops.get("/ops/facilities/").content.decode() self.assertIn("/ops/facilities/", html) self.assertIn("/ops/origin-zones/", html) self.assertIn("/ops/collector-groups/", html) class FacilityRegisterTests(ComplianceRegisterBase): def payload(self, **overrides): data = { "name": "New Facility", "country": "Nigeria", "region": "Kaduna State", "address": "", "facility_type": "production", "is_active": "on", } data.update(overrides) return data def test_a_facility_is_created_and_listed(self): r = self.client_ops.post("/ops/facilities/new/", self.payload(), follow=True) self.assertEqual(r.status_code, 200) self.assertTrue(Facility.objects.filter(name="New Facility").exists()) self.assertContains(r, "New Facility") def test_a_facility_can_be_edited(self): r = self.client_ops.post( f"/ops/facilities/{self.facility.pk}/edit/", self.payload(name="Renamed Facility"), ) self.assertEqual(r.status_code, 302) self.facility.refresh_from_db() self.assertEqual(self.facility.name, "Renamed Facility") def test_name_and_country_are_required(self): for field in ("name", "country"): with self.subTest(field=field): r = self.client_ops.post( "/ops/facilities/new/", self.payload(**{field: ""}) ) self.assertEqual(r.status_code, 200) self.assertTrue(r.context["form"].errors.get(field)) def test_retiring_a_facility_keeps_the_record(self): r = self.client_ops.post( f"/ops/facilities/{self.facility.pk}/edit/", self.payload(name=self.facility.name, is_active=""), ) self.assertEqual(r.status_code, 302) self.facility.refresh_from_db() self.assertFalse(self.facility.is_active) self.assertTrue(Facility.objects.filter(pk=self.facility.pk).exists()) class OriginZoneRegisterTests(ComplianceRegisterBase): POLYGON = ( '{"type": "Polygon", "coordinates": [[[7.6012, 12.9905], ' "[7.618, 12.9905], [7.618, 13.0042], [7.6012, 13.0042], " "[7.6012, 12.9905]]]}" ) def payload(self, **overrides): data = { "facility": str(self.facility.pk), "zone_code": "NG-KT-RZ02", "country": "Nigeria", "region": "Katsina State", "geometry": self.POLYGON, "coordinate_source": "Field survey", "coordinates_collected_by": "Field Officer", "coordinates_collected_at": "2026-09-01", "evidence_file": "survey-2026-09-01.pdf", "verification_status": "unverified", } data.update(overrides) return data def test_a_zone_is_created_with_its_geometry(self): r = self.client_ops.post("/ops/origin-zones/new/", self.payload(), follow=True) self.assertEqual(r.status_code, 200) zone = OriginZone.objects.get(zone_code="NG-KT-RZ02") self.assertEqual(zone.geometry["type"], "Polygon") self.assertContains(r, "NG-KT-RZ02") def test_creating_a_verified_zone_records_who_and_when(self): self.client_ops.post( "/ops/origin-zones/new/", self.payload(verification_status="verified") ) zone = OriginZone.objects.get(zone_code="NG-KT-RZ02") self.assertEqual(zone.verified_by, self.ops) self.assertIsNotNone(zone.verified_at) def test_moving_a_zone_back_to_unverified_clears_the_stamp(self): self.client_ops.post( f"/ops/origin-zones/{self.zone.pk}/edit/", self.payload(zone_code=self.zone.zone_code, verification_status="unverified"), ) self.zone.refresh_from_db() self.assertIsNone(self.zone.verified_by) self.assertIsNone(self.zone.verified_at) def test_an_unchanged_verified_zone_keeps_its_verifier(self): other = User.objects.create_user( email="other-verifier@ixg.local", password="pass1234", role="ops", full_name="Other Verifier", is_staff=True, ) self.zone.verified_by = other self.zone.save(update_fields=["verified_by"]) self.client_ops.post( f"/ops/origin-zones/{self.zone.pk}/edit/", self.payload( zone_code=self.zone.zone_code, verification_status="verified" ), ) self.zone.refresh_from_db() self.assertEqual(self.zone.verified_by, other) def test_invalid_geometry_json_is_refused(self): r = self.client_ops.post( "/ops/origin-zones/new/", self.payload(geometry="{not json") ) self.assertEqual(r.status_code, 200) self.assertTrue(r.context["form"].errors.get("geometry")) def test_a_duplicate_zone_code_is_refused_case_insensitively(self): r = self.client_ops.post( "/ops/origin-zones/new/", self.payload(zone_code=self.zone.zone_code.lower()), ) self.assertEqual(r.status_code, 200) self.assertTrue(r.context["form"].errors.get("zone_code")) def test_the_edit_form_draws_the_stored_polygon(self): self.zone.geometry = { "type": "Polygon", "coordinates": [[[7.6012, 12.9905], [7.618, 12.9905], [7.618, 13.0042], [7.6012, 13.0042], [7.6012, 12.9905]]], } self.zone.save() r = self.client_ops.get(f"/ops/origin-zones/{self.zone.pk}/edit/") html = r.content.decode() self.assertIn('id="zone-map"', html) self.assertIn("7.6012", html) self.assertIn("vendor/leaflet/leaflet.js", html) self.assertIn("js/ixg_zone_map.js", html) self.assertIn("a view, not a check", html.lower()) def test_a_zone_without_geometry_says_so(self): r = self.client_ops.get(f"/ops/origin-zones/{self.zone.pk}/edit/") html = r.content.decode() self.assertIn("No polygon coordinates are stored", html) self.assertNotIn('id="zone-map"', html) class CollectorGroupRegisterTests(ComplianceRegisterBase): def payload(self, **overrides): data = { "name": "New Collective", "contact_reference": "CG-NEW-001", "origin_zone": str(self.zone.pk), "notes": "", } data.update(overrides) return data def test_a_group_is_created_and_listed(self): r = self.client_ops.post( "/ops/collector-groups/new/", self.payload(), follow=True ) self.assertEqual(r.status_code, 200) group = CollectorGroup.objects.get(name="New Collective") self.assertEqual(group.origin_zone, self.zone) self.assertContains(r, "New Collective") def test_a_group_can_be_edited(self): r = self.client_ops.post( f"/ops/collector-groups/{self.group.pk}/edit/", self.payload(name="Renamed Collective"), ) self.assertEqual(r.status_code, 302) self.group.refresh_from_db() self.assertEqual(self.group.name, "Renamed Collective") def test_the_origin_zone_is_optional(self): self.client_ops.post( "/ops/collector-groups/new/", self.payload(origin_zone="") ) group = CollectorGroup.objects.get(name="New Collective") self.assertIsNone(group.origin_zone) class EUDRCaseChildBase(ComplianceRegisterBase): def setUp(self): super().setUp() self.batch = ProcessingBatch.objects.create( batch_reference="IXG-BATCH-2026-910", harvest_date="2026-09-04", weight_kg=Decimal("500.00"), grade="A", status="harvested", origin_facility=self.facility.name, created_by=self.ops, ) self.case = DueDiligenceCase.objects.create(batch=self.batch) def case_url(self, path=""): return f"/ops/eudr/{self.case.pk}/{path}" def another_case(self): batch = ProcessingBatch.objects.create( batch_reference="IXG-BATCH-2026-911", harvest_date="2026-09-05", weight_kg=Decimal("400.00"), grade="A", status="harvested", origin_facility=self.facility.name, created_by=self.ops, ) return DueDiligenceCase.objects.create(batch=batch) class EUDREvidenceTests(EUDRCaseChildBase): def payload(self, **overrides): data = { "evidence_type": "geolocation", "file": "geolocation.geojson", "source": "Field survey", "collected_at": "2026-09-01", "notes": "", } data.update(overrides) return data def test_evidence_is_attached_and_shown_on_the_case(self): r = self.client_ops.post(self.case_url("evidence/new/"), self.payload(), follow=True) self.assertEqual(r.status_code, 200) self.assertEqual(self.case.evidence.count(), 1) self.assertContains(r, "geolocation.geojson") def test_attaching_evidence_reruns_the_completeness_check(self): self.client_ops.post( self.case_url("evidence/new/"), self.payload(evidence_type="geolocation") ) self.case.refresh_from_db() detail = self.case.completeness_results["evidence"]["detail"] self.assertNotIn("geolocation", detail) def test_evidence_can_be_edited(self): evidence = EUDREvidence.objects.create( case=self.case, evidence_type="other", file="old.pdf" ) r = self.client_ops.post( self.case_url(f"evidence/{evidence.pk}/edit/"), self.payload(file="new.pdf") ) self.assertEqual(r.status_code, 302) evidence.refresh_from_db() self.assertEqual(evidence.file, "new.pdf") def test_evidence_cannot_be_reached_through_another_case(self): other = self.another_case() evidence = EUDREvidence.objects.create( case=self.case, evidence_type="other", file="x.pdf" ) r = self.client_ops.get(f"/ops/eudr/{other.pk}/evidence/{evidence.pk}/edit/") self.assertEqual(r.status_code, 404) def test_a_buyer_cannot_attach_evidence(self): r = self.client_buyer.post(self.case_url("evidence/new/"), self.payload()) self.assertEqual(r.status_code, 403) class MitigationActionTests(EUDRCaseChildBase): def payload(self, **overrides): data = { "action": "Re-survey the origin zone", "status": "open", "owner": "Ops Officer", "due_date": "2026-10-01", "evidence_file": "", } data.update(overrides) return data def test_an_action_is_added_and_shown_on_the_case(self): r = self.client_ops.post( self.case_url("mitigation/new/"), self.payload(), follow=True ) self.assertEqual(r.status_code, 200) self.assertEqual(self.case.mitigation_actions.count(), 1) self.assertContains(r, "Re-survey the origin zone") def test_an_action_can_be_edited(self): action = MitigationAction.objects.create(case=self.case, action="Old action") r = self.client_ops.post( self.case_url(f"mitigation/{action.pk}/edit/"), self.payload(action="Corrected action", status="completed"), ) self.assertEqual(r.status_code, 302) action.refresh_from_db() self.assertEqual(action.action, "Corrected action") self.assertEqual(action.status, MitigationAction.Status.COMPLETED) def test_a_buyer_cannot_add_an_action(self): r = self.client_buyer.post(self.case_url("mitigation/new/"), self.payload()) self.assertEqual(r.status_code, 403) class StatementReferenceEditTests(EUDRCaseChildBase): def setUp(self): super().setUp() self.reference = DueDiligenceStatementReference.objects.create( case=self.case, dds_reference="DD-2026-00001", submission_date="2026-09-01", submitted_by=self.ops, ) def test_a_reference_can_be_reconciled(self): r = self.client_ops.post( self.case_url(f"statement/{self.reference.pk}/edit/"), { "dds_reference": "DD-2026-00001", "submission_date": "2026-09-01", "reconciliation_status": "reconciled", "notes": "", }, ) self.assertEqual(r.status_code, 302) self.reference.refresh_from_db() self.assertEqual( self.reference.reconciliation_status, DueDiligenceStatementReference.ReconciliationStatus.RECONCILED, ) def test_a_reference_is_required(self): r = self.client_ops.post( self.case_url(f"statement/{self.reference.pk}/edit/"), { "dds_reference": "", "submission_date": "2026-09-01", "reconciliation_status": "pending", "notes": "", }, ) self.assertEqual(r.status_code, 200) self.assertTrue(r.context["form"].errors.get("dds_reference")) # --------------------------------------------------------------------------- # Product specifications # # The portal can create and edit a specification and its parameters, but the # approval half is held: a portal-created specification is a draft and the form # cannot set status or is_current. These tests pin both halves. class SpecificationBase(TestCase): def setUp(self): self.ops = User.objects.create_user( email="spec-ops@ixg.local", password="pass1234", role="ops", full_name="Spec Officer", is_staff=True, ) self.buyer = User.objects.create_user( email="spec-buyer@ixg.local", password="pass1234", role="buyer", full_name="Spec Buyer", is_staff=True, ) self.client_ops = Client(SERVER_NAME="ops.ixgltd.com") self.client_ops.force_login(self.ops) self.client_buyer = Client(SERVER_NAME="ops.ixgltd.com") self.client_buyer.force_login(self.buyer) self.spec = ProductSpecification.objects.create( product="Unrefined Shea Butter", version="7.1", status=ProductSpecification.Status.APPROVED, is_current=True, ) self.ffa = SpecParameter.objects.create( spec=self.spec, parameter=SpecParameter.Parameter.FFA, unit="%", max_value=Decimal("3.000"), sort_order=1, ) def management(self, total, initial=0): return { "params-TOTAL_FORMS": str(total), "params-INITIAL_FORMS": str(initial), "params-MIN_NUM_FORMS": "0", "params-MAX_NUM_FORMS": "1000", } def create_payload(self, **overrides): data = { "product": "Unrefined Shea Butter", "version": "2.0", "effective_from": "", "effective_to": "", **self.management(1), "params-0-id": "", "params-0-parameter": "moisture", "params-0-unit": "%", "params-0-min_value": "", "params-0-max_value": "0.100", "params-0-allowed_values": "", "params-0-required": "on", "params-0-sort_order": "1", "params-0-DELETE": "", } data.update(overrides) return data def edit_payload(self, params, **overrides): data = { "product": self.spec.product, "version": self.spec.version, "effective_from": "", "effective_to": "", **self.management(len(params), initial=len(params)), } for i, p in enumerate(params): data[f"params-{i}-id"] = str(p.pk) data[f"params-{i}-parameter"] = p.parameter data[f"params-{i}-unit"] = p.unit data[f"params-{i}-min_value"] = str(p.min_value) if p.min_value is not None else "" data[f"params-{i}-max_value"] = str(p.max_value) if p.max_value is not None else "" data[f"params-{i}-allowed_values"] = p.allowed_values data[f"params-{i}-required"] = "on" if p.required else "" data[f"params-{i}-sort_order"] = str(p.sort_order) data[f"params-{i}-DELETE"] = "" data.update(overrides) return data class SpecificationAccessTests(SpecificationBase): def test_ops_can_open_the_specification_screens(self): for path in ( "/ops/specifications/", "/ops/specifications/new/", f"/ops/specifications/{self.spec.pk}/edit/", ): with self.subTest(path=path): self.assertEqual(self.client_ops.get(path).status_code, 200) def test_a_buyer_cannot_open_the_specification_screens(self): for path in ( "/ops/specifications/", "/ops/specifications/new/", f"/ops/specifications/{self.spec.pk}/edit/", ): with self.subTest(path=path): self.assertEqual(self.client_buyer.get(path).status_code, 403) def test_the_sidebar_links_to_specifications(self): self.assertContains(self.client_ops.get("/ops/specifications/"), "Specifications") class SpecificationCreateTests(SpecificationBase): def test_a_specification_is_created_as_a_draft_with_its_parameters(self): r = self.client_ops.post( "/ops/specifications/new/", self.create_payload(), follow=True ) self.assertEqual(r.status_code, 200) spec = ProductSpecification.objects.get(version="2.0") self.assertEqual(spec.status, ProductSpecification.Status.DRAFT) self.assertFalse(spec.is_current) self.assertEqual(spec.parameters.count(), 1) self.assertEqual(spec.parameters.first().parameter, "moisture") def test_the_approval_fields_cannot_be_set_from_the_portal(self): r = self.client_ops.post( "/ops/specifications/new/", self.create_payload(status="approved", is_current="on"), ) self.assertEqual(r.status_code, 302) spec = ProductSpecification.objects.get(version="2.0") self.assertEqual(spec.status, ProductSpecification.Status.DRAFT) self.assertFalse(spec.is_current) def test_a_specification_with_no_parameters_is_refused(self): payload = self.create_payload(**self.management(0)) for key in list(payload): if key.startswith("params-0-"): del payload[key] r = self.client_ops.post("/ops/specifications/new/", payload) self.assertEqual(r.status_code, 200) errors = " ".join(str(e) for e in r.context["formset"].non_form_errors()) self.assertIn("at least one parameter", errors) def test_a_duplicate_product_and_version_is_refused(self): r = self.client_ops.post( "/ops/specifications/new/", self.create_payload(product=self.spec.product, version=self.spec.version), ) self.assertEqual(r.status_code, 200) self.assertTrue(r.context["form"].errors.get("version")) def test_a_numeric_parameter_with_no_limits_is_refused(self): r = self.client_ops.post( "/ops/specifications/new/", self.create_payload(**{"params-0-max_value": ""}), ) self.assertEqual(r.status_code, 200) form = r.context["formset"].forms[0] self.assertTrue(form.errors.get("min_value")) def test_a_visual_parameter_without_allowed_values_is_refused(self): r = self.client_ops.post( "/ops/specifications/new/", self.create_payload(**{ "params-0-parameter": "visual", "params-0-max_value": "", "params-0-allowed_values": "", }), ) self.assertEqual(r.status_code, 200) form = r.context["formset"].forms[0] self.assertTrue(form.errors.get("allowed_values")) def test_an_effective_to_before_effective_from_is_refused(self): r = self.client_ops.post( "/ops/specifications/new/", self.create_payload(effective_from="2026-10-01", effective_to="2026-09-01"), ) self.assertEqual(r.status_code, 200) self.assertTrue(r.context["form"].errors.get("effective_to")) class SpecificationEditTests(SpecificationBase): def test_editing_updates_a_parameter_in_place(self): before = ProductSpecification.objects.count() self.ffa.max_value = Decimal("2.500") r = self.client_ops.post( f"/ops/specifications/{self.spec.pk}/edit/", self.edit_payload([self.ffa]), follow=True, ) self.assertEqual(r.status_code, 200) self.ffa.refresh_from_db() self.assertEqual(self.ffa.max_value, Decimal("2.500")) self.assertEqual(ProductSpecification.objects.count(), before) def test_a_parameter_can_be_removed(self): visual = SpecParameter.objects.create( spec=self.spec, parameter=SpecParameter.Parameter.VISUAL, allowed_values="A,B,C", sort_order=2, ) payload = self.edit_payload([self.ffa, visual]) payload["params-1-DELETE"] = "on" r = self.client_ops.post( f"/ops/specifications/{self.spec.pk}/edit/", payload ) self.assertEqual(r.status_code, 302) self.assertEqual(self.spec.parameters.count(), 1) def test_editing_an_approved_specification_warns_and_keeps_its_status(self): r = self.client_ops.get(f"/ops/specifications/{self.spec.pk}/edit/") self.assertContains(r, "This specification is not a draft") self.client_ops.post( f"/ops/specifications/{self.spec.pk}/edit/", self.edit_payload([self.ffa]) ) self.spec.refresh_from_db() self.assertEqual(self.spec.status, ProductSpecification.Status.APPROVED) def test_a_buyer_cannot_post_a_specification(self): r = self.client_buyer.post("/ops/specifications/new/", self.create_payload()) self.assertEqual(r.status_code, 403)