ixgadmin/ixg_platform
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
ixg_platform/apps/ops/tests.py
ixgadmin df2a90bc71 Add the product specification screen to the ops portal 
Operations can now create and edit a specification and its parameters
together, over the existing model, mirroring the admin's inline.

The approval half is deliberately held: the form does not expose status,
is_current, approved_by or approved_at, so a portal-created specification
is a draft and cannot be approved here. The edit screen warns when the
record is not a draft, because its parameters are the limits QC results
are judged against. Nothing is removed from the admin.

The parameter formset refuses an empty specification (with no parameters
every QC result would pass), a numeric parameter with no limits, and a
visual parameter with no allowed values.
2026-09-25 16:35:45 +00:00

1476 lines
61 KiB
Python
Raw Permalink 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}/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)
Reference in a new issue View git blame Copy permalink