ixgadmin/ixg_platform
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
ixg_platform/apps/ops/tests.py
ixgadmin d74f0814da Finish the wiki root move, zone map view and spec approval 
- Follow home/user-guides/ through one GUIDE_PATH_PREFIX constant.
- Add a read-only Origin Zone detail view with the map as the main panel.
- Name due diligence explicitly and add its section headings.
- Add admin-only, four-eyes specification approval (portal and API).
- Empty the New-zone geometry placeholder instead of showing null.
2026-10-27 17:41:08 +00:00

1617 lines
68 KiB
Python
Raw Blame History

"""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}/",
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)
def test_the_new_form_geometry_is_empty_not_null(self):
r = self.client_ops.get("/ops/origin-zones/new/")
self.assertEqual(r.context["form"]["geometry"].value(), "")
def test_a_zone_has_a_read_only_detail_view_with_the_map(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}/")
self.assertEqual(r.status_code, 200)
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())
self.assertIn(f"/ops/origin-zones/{self.zone.pk}/edit/", html)
def test_the_detail_view_says_when_there_is_no_geometry(self):
r = self.client_ops.get(f"/ops/origin-zones/{self.zone.pk}/")
html = r.content.decode()
self.assertIn("No polygon coordinates are stored", html)
self.assertNotIn('id="zone-map"', html)
def test_the_list_links_each_zone_to_its_detail_view(self):
html = self.client_ops.get("/ops/origin-zones/").content.decode()
self.assertIn(f'href="/ops/origin-zones/{self.zone.pk}/"', 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 DueDiligenceNamingTests(EUDRCaseChildBase):
def test_the_sidebar_and_case_screen_name_due_diligence(self):
list_html = self.client_ops.get("/ops/eudr/").content.decode()
self.assertIn("Due Diligence (EUDR)", list_html)
detail = self.client_ops.get(self.case_url()).content.decode()
self.assertIn("Due Diligence Case:", detail)
self.assertIn("due diligence case", detail)
for heading in (
"Due diligence completeness check",
"Intake lineage (origin)",
"Evidence",
"Risk assessments",
"Mitigation actions",
"Statement references (DDS)",
):
with self.subTest(heading=heading):
self.assertIn(heading, detail)
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.created_by, self.ops)
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)
class SpecificationApprovalTests(SpecificationBase):
"""Approval is an administrator action and never a single-person act."""
def setUp(self):
super().setUp()
self.admin = User.objects.create_user(
email="spec-admin@ixg.local", password="pass1234", role="admin",
full_name="Spec Admin", is_staff=True,
)
self.other_admin = User.objects.create_user(
email="spec-admin2@ixg.local", password="pass1234", role="admin",
full_name="Other Spec Admin", is_staff=True,
)
self.client_admin = Client(SERVER_NAME="ops.ixgltd.com")
self.client_admin.force_login(self.admin)
self.client_other_admin = Client(SERVER_NAME="ops.ixgltd.com")
self.client_other_admin.force_login(self.other_admin)
def approve(self, client, spec):
return client.post(f"/ops/specifications/{spec.pk}/approve/", follow=True)
def test_an_ops_officer_cannot_approve_a_specification(self):
self.client_ops.post("/ops/specifications/new/", self.create_payload())
spec = ProductSpecification.objects.get(version="2.0")
r = self.client_ops.post(f"/ops/specifications/{spec.pk}/approve/")
self.assertEqual(r.status_code, 403)
spec.refresh_from_db()
self.assertEqual(spec.status, ProductSpecification.Status.DRAFT)
self.assertFalse(spec.is_current)
def test_an_admin_approves_an_ops_draft_and_supersedes_the_old_version(self):
self.client_ops.post("/ops/specifications/new/", self.create_payload())
spec = ProductSpecification.objects.get(version="2.0")
r = self.approve(self.client_admin, spec)
self.assertEqual(r.status_code, 200)
spec.refresh_from_db()
self.spec.refresh_from_db()
self.assertEqual(spec.status, ProductSpecification.Status.APPROVED)
self.assertTrue(spec.is_current)
self.assertEqual(spec.approved_by, self.admin)
self.assertIsNotNone(spec.approved_at)
self.assertEqual(self.spec.status, ProductSpecification.Status.SUPERSEDED)
self.assertFalse(self.spec.is_current)
self.assertContains(r, "approved and made current")
def test_an_admin_cannot_approve_a_specification_they_created(self):
self.client_admin.post("/ops/specifications/new/", self.create_payload())
spec = ProductSpecification.objects.get(version="2.0")
self.assertEqual(spec.created_by, self.admin)
r = self.approve(self.client_admin, spec)
self.assertContains(r, "another administrator must approve it")
spec.refresh_from_db()
self.assertEqual(spec.status, ProductSpecification.Status.DRAFT)
self.assertFalse(spec.is_current)
self.assertIsNone(spec.approved_by)
def test_a_different_admin_can_approve_it(self):
self.client_admin.post("/ops/specifications/new/", self.create_payload())
spec = ProductSpecification.objects.get(version="2.0")
self.approve(self.client_other_admin, spec)
spec.refresh_from_db()
self.assertEqual(spec.status, ProductSpecification.Status.APPROVED)
self.assertEqual(spec.approved_by, self.other_admin)
def test_a_specification_without_parameters_cannot_be_approved(self):
spec = ProductSpecification.objects.create(
product="Unrefined Shea Butter", version="9.0"
)
r = self.approve(self.client_admin, spec)
self.assertContains(r, "at least one parameter")
spec.refresh_from_db()
self.assertEqual(spec.status, ProductSpecification.Status.DRAFT)
def test_an_already_approved_specification_cannot_be_approved_again(self):
r = self.approve(self.client_admin, self.spec)
self.assertContains(r, "Only a draft specification")
self.spec.refresh_from_db()
self.assertEqual(self.spec.status, ProductSpecification.Status.APPROVED)
def test_the_approve_control_only_appears_for_admins_who_may_use_it(self):
self.client_admin.post("/ops/specifications/new/", self.create_payload())
spec = ProductSpecification.objects.get(version="2.0")
approve_url = f"/ops/specifications/{spec.pk}/approve/"
self.assertNotContains(self.client_ops.get("/ops/specifications/"), approve_url)
self.assertNotContains(self.client_admin.get("/ops/specifications/"), approve_url)
self.assertContains(self.client_other_admin.get("/ops/specifications/"), approve_url)
Reference in a new issue View git blame Copy permalink