661 lines
28 KiB
Python
661 lines
28 KiB
Python
"""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 .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")
|
|
|
|
|
|
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())
|
|
|
|
|
|
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")
|