ixgadmin/ixg_platform
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
ixg_platform/apps/ops/tests.py
ixgadmin 5b6c4c48e4 Test the shipment screen, its API rules and the date inputs 
Cover access on GET and POST for a buyer, each required field when blank, an ETD
after the ETA, a batch that already has a shipment (queried, not blocked), a
status change, and an edit updating in place. Add the shipment routes to both
OPS_ROUTES sweeps so the detail and edit screens render clean markup, and assert
the native date inputs carry ISO values — the defect that left the intake and
batch edit dates blank.
2026-09-17 16:03:14 +00:00

847 lines
36 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 apps.batches.models import ProcessingBatch
from apps.compliance.models import (
CollectorGroup,
Facility,
HarvestIntake,
IntakeBatchLineage,
OriginZone,
)
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"')
Reference in a new issue View git blame Copy permalink