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.
This commit is contained in:
ixgadmin 2026-09-17 16:03:14 +00:00
parent dae025f600
commit 5b6c4c48e4
3 changed files with 310 additions and 2 deletions

View file

@ -36,6 +36,7 @@ from apps.compliance.models import (
)
from apps.notifications.models import Notification
from apps.qc.models import QCResult
from apps.shipments.models import Shipment
from apps.subscriptions.models import Subscription
from .tiles import browser_tile_url, upstream_url
@ -52,6 +53,7 @@ OPS_ROUTES = [
"/ops/qc/verify/",
"/ops/coa/",
"/ops/shipments/",
"/ops/shipments/new/",
"/ops/capa/",
"/ops/documents/",
"/ops/esg/",
@ -94,6 +96,25 @@ def batch_screens(batch):
]
def make_sweep_shipment(batch):
"""A shipment so the detail and edit screens render in the route sweeps."""
return Shipment.objects.create(
batch=batch,
status=Shipment.Status.BOOKED,
vessel_name="MV Sweep",
container_ref="IXGU0000000",
etd_nigeria="2026-09-01",
eta_uk="2026-10-01",
)
def shipment_screens(shipment):
return [
f"/ops/shipments/{shipment.pk}/",
f"/ops/shipments/{shipment.pk}/edit/",
]
class PortalPagesRenderTests(TestCase):
def make_user(self, email, role):
return User.objects.create_user(
@ -114,7 +135,8 @@ class PortalPagesRenderTests(TestCase):
def test_ops_pages_render(self):
batch = make_sweep_batch()
routes = OPS_ROUTES + batch_screens(batch)
shipment = make_sweep_shipment(batch)
routes = OPS_ROUTES + batch_screens(batch) + shipment_screens(shipment)
self.sweep(self.make_user("ops-sweep@ixg.local", "ops"), routes, "ops.ixgltd.com")
def test_buyer_pages_render(self):
@ -659,12 +681,14 @@ class TemplateAttributeHygieneTests(TestCase):
def test_ops_pages_render_clean_markup(self):
batch = make_sweep_batch()
shipment = make_sweep_shipment(batch)
self.check_pages(
User.objects.create_user(
email="attr-ops@ixg.local", password="pass1234", role="ops",
full_name="Attr Ops", is_staff=True,
),
OPS_ROUTES + batch_screens(batch), "ops.ixgltd.com",
OPS_ROUTES + batch_screens(batch) + shipment_screens(shipment),
"ops.ixgltd.com",
)
def test_portal_pages_render_clean_markup(self):

View file

@ -21,6 +21,7 @@ from apps.compliance.models import (
IntakeBatchLineage,
OriginZone,
)
from apps.shipments.models import Shipment
from .views import suggested_batch_reference, suggested_intake_reference
@ -104,6 +105,12 @@ class IntakeAccessTests(IntakeTestCase):
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):
@ -351,6 +358,12 @@ class BatchAccessTests(BatchTestCase):
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):
@ -659,3 +672,176 @@ class SuggestedBatchReferenceTests(BatchTestCase):
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"')

98
apps/shipments/tests.py Normal file
View file

@ -0,0 +1,98 @@
"""The shipment API.
The portal form and the API must agree: the fields the buyer's page reads are
required, and a departure after the arrival is refused. The permission boundary
itself is swept in ``apps/core/tests.py``; these pin the serializer rules so the
API cannot accept what the ops form refuses.
"""
from django.contrib.auth import get_user_model
from django.test import TestCase
from rest_framework.test import APIClient
from apps.batches.models import ProcessingBatch
from .models import Shipment
User = get_user_model()
class ShipmentAPITests(TestCase):
def setUp(self):
self.ops = User.objects.create_user(
email="ship-api-ops@ixg.local", password="pass1234", role="ops",
full_name="Shipment API Ops", is_staff=True,
)
self.buyer = User.objects.create_user(
email="ship-api-buyer@ixg.local", password="pass1234", role="buyer",
full_name="Shipment API Buyer", is_staff=True,
)
self.batch = ProcessingBatch.objects.create(
batch_reference="IXG-BATCH-API-SHIP-001", status="ready_to_ship",
)
self.shipment = Shipment.objects.create(
batch=self.batch, status=Shipment.Status.BOOKED,
vessel_name="MV API", container_ref="IXGU0000001",
etd_nigeria="2026-09-15", eta_uk="2026-10-15",
)
self.client = APIClient()
def payload(self, **overrides):
data = {
"batch": str(self.batch.pk),
"status": "booked",
"vessel_name": "MV API",
"container_ref": "IXGU0000002",
"etd_nigeria": "2026-09-20",
"eta_uk": "2026-10-20",
}
data.update(overrides)
return data
def test_a_buyer_cannot_read_or_write_shipments(self):
self.client.force_authenticate(user=self.buyer)
self.assertEqual(self.client.get("/api/shipments/").status_code, 403)
self.assertEqual(
self.client.post("/api/shipments/", self.payload(), format="json").status_code,
403,
)
def test_ops_can_create_a_shipment(self):
self.client.force_authenticate(user=self.ops)
response = self.client.post("/api/shipments/", self.payload(), format="json")
self.assertEqual(response.status_code, 201)
self.assertEqual(Shipment.objects.count(), 2)
def test_each_required_field_is_refused_when_blank(self):
self.client.force_authenticate(user=self.ops)
for field in (
"status", "vessel_name", "container_ref", "etd_nigeria", "eta_uk",
):
with self.subTest(field=field):
response = self.client.post(
"/api/shipments/", self.payload(**{field: ""}), format="json"
)
self.assertEqual(response.status_code, 400)
self.assertIn(field, response.json())
self.assertEqual(Shipment.objects.count(), 1)
def test_a_departure_after_the_arrival_is_refused(self):
self.client.force_authenticate(user=self.ops)
response = self.client.post(
"/api/shipments/",
self.payload(etd_nigeria="2026-11-01", eta_uk="2026-10-01"),
format="json",
)
self.assertEqual(response.status_code, 400)
self.assertIn("eta_uk", response.json())
self.assertEqual(Shipment.objects.count(), 1)
def test_a_partial_status_change_does_not_require_the_other_fields(self):
self.client.force_authenticate(user=self.ops)
response = self.client.patch(
f"/api/shipments/{self.shipment.pk}/",
{"status": "in_transit"},
format="json",
)
self.assertEqual(response.status_code, 200)
self.shipment.refresh_from_db()
self.assertEqual(self.shipment.status, Shipment.Status.IN_TRANSIT)