diff --git a/apps/ops/forms.py b/apps/ops/forms.py index 8ac1acc..9c66b08 100644 --- a/apps/ops/forms.py +++ b/apps/ops/forms.py @@ -28,9 +28,12 @@ from apps.batches.models import ProcessingBatch from apps.compliance import services from apps.compliance.models import ( CollectorGroup, + DueDiligenceStatementReference, + EUDREvidence, Facility, HarvestIntake, IntakeBatchLineage, + MitigationAction, OriginZone, ) from apps.shipments.models import Shipment @@ -656,3 +659,286 @@ class ShipmentForm(forms.ModelForm): ), ) return cleaned + + +# --------------------------------------------------------------------------- +# Compliance registers +# +# The facility, origin-zone and collector-group registers, and the evidence, +# mitigation and statement records on a due-diligence case, used to be editable +# only in the Django administration area. These forms bring that data entry into +# the ops portal. They mirror the admin's fields; nothing is removed from admin. +# +# The origin-zone form deliberately keeps the map a *view* of the stored +# geometry rather than a check of it — it does not validate the shape and it +# does not verify the zone. The form only insists that the geometry, if given, +# is valid JSON, because the column is JSON and the map cannot draw a broken +# value. Geometry validation and the two-step verification are separate work. + + +class FacilityForm(forms.ModelForm): + class Meta: + model = Facility + fields = ["name", "country", "region", "address", "facility_type", "is_active"] + labels = { + "name": "Facility name", + "country": "Country", + "region": "Region", + "address": "Address", + "facility_type": "Facility type", + "is_active": "Active", + } + help_texts = { + "name": "The site's name, as it should appear on records.", + "country": "The country the site operates in.", + "region": "The state or region, if it helps identify the site.", + "address": "Optional. The site's address.", + "facility_type": "Production sites collect; processing sites handle the shea.", + "is_active": "Uncheck to retire a site without deleting its history.", + } + widgets = { + "name": forms.TextInput(attrs={"class": "form-control"}), + "country": forms.TextInput(attrs={"class": "form-control"}), + "region": forms.TextInput(attrs={"class": "form-control"}), + "address": forms.Textarea(attrs={"class": "form-control", "rows": 3}), + "facility_type": forms.Select(attrs={"class": "form-select"}), + "is_active": forms.CheckboxInput(attrs={"class": "form-check-input"}), + } + + +class OriginZoneForm(forms.ModelForm): + geometry = forms.JSONField( + required=False, + label="Geometry (GeoJSON)", + help_text=( + "The zone's polygon as GeoJSON, for example " + '{"type": "Polygon", "coordinates": [[[7.60, 12.99], ...]]}. ' + "Stored as given: the map below draws it, it does not check the shape." + ), + error_messages={"invalid": "Enter valid GeoJSON, or leave this field blank."}, + widget=forms.Textarea( + attrs={"class": "form-control", "rows": 6, "spellcheck": "false"} + ), + ) + + class Meta: + model = OriginZone + fields = [ + "facility", + "zone_code", + "country", + "region", + "geometry", + "coordinate_source", + "coordinates_collected_by", + "coordinates_collected_at", + "evidence_file", + "verification_status", + ] + labels = { + "facility": "Facility", + "zone_code": "Zone code", + "country": "Country", + "region": "Region", + "coordinate_source": "Coordinate source", + "coordinates_collected_by": "Coordinates collected by", + "coordinates_collected_at": "Coordinates collected on", + "evidence_file": "Supporting file name", + "verification_status": "Verification status", + } + help_texts = { + "zone_code": "Must be unique. The short code the zone is known by.", + "country": "The country the zone lies in.", + "region": "The state or region.", + "coordinate_source": "Where the coordinates came from, e.g. a survey or a map.", + "coordinates_collected_by": "Who collected the coordinates.", + "coordinates_collected_at": "The date the coordinates were collected.", + "evidence_file": "The name of the supporting record. Files are not uploaded by the platform yet, so this is the name it is filed under.", + "verification_status": "Unverified until the zone has been checked. Setting it to Verified or Rejected records who did so and when.", + } + widgets = { + "facility": forms.Select(attrs={"class": "form-select"}), + "zone_code": forms.TextInput(attrs={"class": "form-control"}), + "country": forms.TextInput(attrs={"class": "form-control"}), + "region": forms.TextInput(attrs={"class": "form-control"}), + "coordinate_source": forms.TextInput(attrs={"class": "form-control"}), + "coordinates_collected_by": forms.TextInput( + attrs={"class": "form-control"} + ), + "coordinates_collected_at": forms.DateInput( + format="%Y-%m-%d", attrs={"class": "form-control", "type": "date"} + ), + "evidence_file": forms.TextInput(attrs={"class": "form-control"}), + "verification_status": forms.Select(attrs={"class": "form-select"}), + } + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + for name in ("facility", "zone_code", "country"): + self.fields[name].required = True + + def clean_geometry(self): + # An empty field must become {}, not None: the column is a non-null JSON + # field and a zone may legitimately have no polygon stored yet. + value = self.cleaned_data.get("geometry") + return value if value not in (None, "") else {} + + def clean_zone_code(self): + code = (self.cleaned_data.get("zone_code") or "").strip() + if not code: + raise forms.ValidationError("A zone code is required.") + clashes = OriginZone.objects.filter(zone_code__iexact=code) + if self.instance.pk: + clashes = clashes.exclude(pk=self.instance.pk) + if clashes.exists(): + raise forms.ValidationError( + f"Zone {code} already exists. Zone codes must be unique." + ) + return code + + +class CollectorGroupForm(forms.ModelForm): + origin_zone = ZoneChoiceField( + queryset=OriginZone.objects.order_by("zone_code"), + label="Origin zone", + required=False, + empty_label="No origin zone set", + help_text=( + "The zone this group normally collects from. Optional, and it can be " + "changed if the group draws from more than one area." + ), + ) + + class Meta: + model = CollectorGroup + fields = ["name", "contact_reference", "origin_zone", "notes"] + labels = { + "name": "Group name", + "contact_reference": "Contact reference", + "notes": "Notes", + } + help_texts = { + "name": "The group's name, as it should appear on records.", + "contact_reference": "Optional. A reference for the group's contact.", + "notes": "Anything the next person needs to know about this group.", + } + widgets = { + "name": forms.TextInput(attrs={"class": "form-control"}), + "contact_reference": forms.TextInput(attrs={"class": "form-control"}), + "notes": forms.Textarea(attrs={"class": "form-control", "rows": 3}), + } + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.fields["name"].required = True + + +class MitigationActionForm(forms.ModelForm): + """Add or correct a mitigation action on a due-diligence case.""" + + class Meta: + model = MitigationAction + fields = ["action", "status", "owner", "due_date", "evidence_file"] + labels = { + "action": "Action", + "status": "Status", + "owner": "Owner", + "due_date": "Due date", + "evidence_file": "Supporting file name", + } + help_texts = { + "action": "What will be done to address the risk.", + "status": "Open until the action has been carried out; Verified once it has been checked.", + "owner": "Who is responsible for the action.", + "due_date": "When it is expected to be done.", + "evidence_file": "The name of the supporting record. Files are not uploaded by the platform yet, so this is the name it is filed under.", + } + widgets = { + "action": forms.Textarea(attrs={"class": "form-control", "rows": 3}), + "status": forms.Select(attrs={"class": "form-select"}), + "owner": forms.TextInput(attrs={"class": "form-control"}), + "due_date": forms.DateInput( + format="%Y-%m-%d", attrs={"class": "form-control", "type": "date"} + ), + "evidence_file": forms.TextInput(attrs={"class": "form-control"}), + } + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.fields["action"].required = True + self.fields["status"].required = True + + +class EUDREvidenceForm(forms.ModelForm): + """Attach an item of evidence to a due-diligence case. + + ``file`` is a stored reference (a file name or path), not an uploaded file: + the platform has no upload screen yet, so the evidence record points at the + document the way the admin form always has. A real upload is separate work. + """ + + class Meta: + model = EUDREvidence + fields = ["evidence_type", "file", "source", "collected_at", "notes"] + labels = { + "evidence_type": "Evidence type", + "file": "File reference", + "source": "Source", + "collected_at": "Collected on", + "notes": "Notes", + } + help_texts = { + "evidence_type": "Geolocation, harvest document and supplier identity are the three the completeness check requires.", + "file": "The name of the file the evidence is stored under.", + "source": "Where the evidence came from.", + "collected_at": "The date the evidence was collected.", + "notes": "Anything the next person needs to know about this evidence.", + } + widgets = { + "evidence_type": forms.Select(attrs={"class": "form-select"}), + "file": forms.TextInput(attrs={"class": "form-control"}), + "source": forms.TextInput(attrs={"class": "form-control"}), + "collected_at": forms.DateInput( + format="%Y-%m-%d", attrs={"class": "form-control", "type": "date"} + ), + "notes": forms.Textarea(attrs={"class": "form-control", "rows": 3}), + } + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.fields["evidence_type"].required = True + self.fields["file"].required = True + + +class StatementReferenceForm(forms.ModelForm): + """Correct an external-submission (DDS) reference on a case.""" + + class Meta: + model = DueDiligenceStatementReference + fields = ["dds_reference", "submission_date", "reconciliation_status", "notes"] + labels = { + "dds_reference": "DDS reference", + "submission_date": "Submission date", + "reconciliation_status": "Reconciliation status", + "notes": "Notes", + } + help_texts = { + "dds_reference": "The reference the external submission was given.", + "submission_date": "The date it was submitted.", + "reconciliation_status": "Pending until the submission has been reconciled against the case.", + "notes": "Anything the next person needs to know about this submission.", + } + widgets = { + "dds_reference": forms.TextInput(attrs={"class": "form-control"}), + "submission_date": forms.DateInput( + format="%Y-%m-%d", attrs={"class": "form-control", "type": "date"} + ), + "reconciliation_status": forms.Select(attrs={"class": "form-select"}), + "notes": forms.Textarea(attrs={"class": "form-control", "rows": 3}), + } + + def clean_dds_reference(self): + reference = (self.cleaned_data.get("dds_reference") or "").strip() + if not reference: + raise forms.ValidationError("A DDS reference is required.") + return reference diff --git a/apps/ops/tests.py b/apps/ops/tests.py index e9af0df..23325cf 100644 --- a/apps/ops/tests.py +++ b/apps/ops/tests.py @@ -12,13 +12,18 @@ 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, ) from apps.shipments.models import Shipment @@ -845,3 +850,409 @@ class ShipmentEditTests(ShipmentTestCase): 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")) diff --git a/apps/ops/urls.py b/apps/ops/urls.py index c955b49..9f0d9a1 100644 --- a/apps/ops/urls.py +++ b/apps/ops/urls.py @@ -13,6 +13,15 @@ urlpatterns = [ path('intakes/', views.IntakeListView.as_view(), name='intake_list'), path('intakes/new/', views.IntakeCreateView.as_view(), name='intake_create'), path('intakes//edit/', views.IntakeUpdateView.as_view(), name='intake_update'), + path('facilities/', views.FacilityListView.as_view(), name='facility_list'), + path('facilities/new/', views.FacilityCreateView.as_view(), name='facility_create'), + path('facilities//edit/', views.FacilityUpdateView.as_view(), name='facility_update'), + path('origin-zones/', views.OriginZoneListView.as_view(), name='origin_zone_list'), + path('origin-zones/new/', views.OriginZoneCreateView.as_view(), name='origin_zone_create'), + path('origin-zones//edit/', views.OriginZoneUpdateView.as_view(), name='origin_zone_update'), + path('collector-groups/', views.CollectorGroupListView.as_view(), name='collector_group_list'), + path('collector-groups/new/', views.CollectorGroupCreateView.as_view(), name='collector_group_create'), + path('collector-groups//edit/', views.CollectorGroupUpdateView.as_view(), name='collector_group_update'), path('qc/', views.QCListView.as_view(), name='qc_list'), path('qc//', views.QCDetailView.as_view(), name='qc_detail'), path('shipments/', views.ShipmentListView.as_view(), name='shipment_list'), @@ -31,6 +40,11 @@ urlpatterns = [ path('coa//', views.COAReviewDetailView.as_view(), name='coa_review_detail'), path('eudr/', views.EUDRCaseListView.as_view(), name='eudr_list'), path('eudr//', views.EUDRCaseDetailView.as_view(), name='eudr_detail'), + path('eudr//evidence/new/', views.EUDREvidenceCreateView.as_view(), name='eudr_evidence_create'), + path('eudr//evidence//edit/', views.EUDREvidenceUpdateView.as_view(), name='eudr_evidence_update'), + path('eudr//mitigation/new/', views.MitigationActionCreateView.as_view(), name='eudr_mitigation_create'), + path('eudr//mitigation//edit/', views.MitigationActionUpdateView.as_view(), name='eudr_mitigation_update'), + path('eudr//statement//edit/', views.StatementReferenceUpdateView.as_view(), name='eudr_statement_update'), path('help/', views.GuideIndexView.as_view(), name='guide_index'), path('help//', views.GuideDetailView.as_view(), name='guide_detail'), ] diff --git a/apps/ops/views.py b/apps/ops/views.py index 9938cad..543c613 100644 --- a/apps/ops/views.py +++ b/apps/ops/views.py @@ -11,6 +11,7 @@ from django.views import View from django.shortcuts import get_object_or_404, render from apps.batches.models import ProcessingBatch +from apps.core.tiles import browser_tile_url from apps.qc.models import QCResult from apps.shipments.models import Shipment from apps.capa.models import CAPARecord @@ -21,12 +22,29 @@ from apps.notifications.models import Notification from apps.compliance.models import ( CollectorGroup, DueDiligenceCase, + DueDiligenceStatementReference, + EUDREvidence, + Facility, HarvestIntake, IntakeBatchLineage, + MitigationAction, + OriginZone, ) +from apps.compliance import services from apps.guides.views import GuideDetailView as BaseGuideDetailView from apps.guides.views import GuideIndexView as BaseGuideIndexView -from .forms import BatchForm, BatchLineageFormSet, HarvestIntakeForm, ShipmentForm +from .forms import ( + BatchForm, + BatchLineageFormSet, + CollectorGroupForm, + EUDREvidenceForm, + FacilityForm, + HarvestIntakeForm, + MitigationActionForm, + OriginZoneForm, + ShipmentForm, + StatementReferenceForm, +) def suggested_intake_reference(today=None): @@ -777,3 +795,381 @@ class GuideIndexView(OpsRequiredMixin, BaseGuideIndexView): class GuideDetailView(OpsRequiredMixin, BaseGuideDetailView): template_name = 'guides/detail_ops.html' + + +# --------------------------------------------------------------------------- +# Compliance registers +# +# Facility, origin zone and collector group used to be editable only in the +# Django administration area, which left the platform unable to register a new +# supply site, zone or group from the portal. These screens close that gap. The +# admin registrations stay: the portal and the admin are two write paths to the +# same records, and the forms here mirror the admin's fields. + + +class FacilityListView(OpsRequiredMixin, ListView): + model = Facility + template_name = 'ops/facility_list.html' + context_object_name = 'facilities' + paginate_by = 25 + + def get_queryset(self): + qs = Facility.objects.all().order_by('name') + q = self.request.GET.get('q') + active = self.request.GET.get('active') + if q: + qs = qs.filter(name__icontains=q) + if active == '1': + qs = qs.filter(is_active=True) + elif active == '0': + qs = qs.filter(is_active=False) + return qs + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['current_q'] = self.request.GET.get('q', '') + ctx['current_active'] = self.request.GET.get('active', '') + return ctx + + +class FacilityFormContextMixin: + model = Facility + form_class = FacilityForm + template_name = 'ops/register_form.html' + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['cancel_url'] = reverse('ops:facility_list') + return ctx + + +class FacilityCreateView(FacilityFormContextMixin, OpsRequiredMixin, CreateView): + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = 'New facility' + ctx['submit_label'] = 'Create facility' + return ctx + + def form_valid(self, form): + response = super().form_valid(form) + messages.success(self.request, f"Facility {self.object.name} created.") + return response + + def get_success_url(self): + return reverse('ops:facility_list') + + +class FacilityUpdateView(FacilityFormContextMixin, OpsRequiredMixin, UpdateView): + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = f"Edit facility {self.object.name}" + ctx['submit_label'] = 'Save changes' + return ctx + + def form_valid(self, form): + response = super().form_valid(form) + messages.success(self.request, f"Facility {self.object.name} updated.") + return response + + def get_success_url(self): + return reverse('ops:facility_list') + + +def apply_zone_verification(zone, user, previous_status=None): + """Record who verified or rejected a zone, and when. + + A zone moved back to Unverified has its stamp cleared, so the record and the + audit agree. An unchanged status keeps the original verifier and date — a + later edit of an unrelated field must not silently re-attribute the check. + """ + if zone.verification_status == OriginZone.VerificationStatus.UNVERIFIED: + zone.verified_by = None + zone.verified_at = None + elif previous_status != zone.verification_status or zone.verified_at is None: + zone.verified_by = user + zone.verified_at = timezone.now() + + +class OriginZoneListView(OpsRequiredMixin, ListView): + model = OriginZone + template_name = 'ops/origin_zone_list.html' + context_object_name = 'zones' + paginate_by = 25 + + def get_queryset(self): + qs = OriginZone.objects.select_related('facility').order_by('zone_code') + q = self.request.GET.get('q') + status = self.request.GET.get('status') + if q: + qs = qs.filter(zone_code__icontains=q) + if status: + qs = qs.filter(verification_status=status) + return qs + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['status_choices'] = OriginZone.VerificationStatus.choices + ctx['current_q'] = self.request.GET.get('q', '') + ctx['current_status'] = self.request.GET.get('status', '') + return ctx + + +class OriginZoneFormContextMixin: + model = OriginZone + form_class = OriginZoneForm + template_name = 'ops/register_form.html' + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['cancel_url'] = reverse('ops:origin_zone_list') + ctx['show_zone_map'] = True + ctx['tile_url'] = browser_tile_url() + zone = getattr(self, 'object', None) + geometry = getattr(zone, 'geometry', None) if zone is not None else None + ctx['zone_has_geometry'] = bool(geometry and geometry.get('coordinates')) + ctx['zone_geometry_json'] = json.dumps(geometry) if ctx['zone_has_geometry'] else '' + return ctx + + def form_valid(self, form): + apply_zone_verification( + form.instance, + self.request.user, + previous_status=form.initial.get('verification_status'), + ) + return super().form_valid(form) + + +class OriginZoneCreateView(OriginZoneFormContextMixin, OpsRequiredMixin, CreateView): + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = 'New origin zone' + ctx['submit_label'] = 'Create zone' + return ctx + + def form_valid(self, form): + response = super().form_valid(form) + messages.success( + self.request, + f"Origin zone {self.object.zone_code} created. " + "The map below is a view of the stored polygon, not a check of it.", + ) + return response + + def get_success_url(self): + return reverse('ops:origin_zone_update', kwargs={'pk': self.object.pk}) + + +class OriginZoneUpdateView(OriginZoneFormContextMixin, OpsRequiredMixin, UpdateView): + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = f"Edit origin zone {self.object.zone_code}" + ctx['submit_label'] = 'Save changes' + return ctx + + def form_valid(self, form): + response = super().form_valid(form) + messages.success( + self.request, f"Origin zone {self.object.zone_code} updated." + ) + return response + + def get_success_url(self): + return reverse('ops:origin_zone_update', kwargs={'pk': self.object.pk}) + + +class CollectorGroupListView(OpsRequiredMixin, ListView): + model = CollectorGroup + template_name = 'ops/collector_group_list.html' + context_object_name = 'groups' + paginate_by = 25 + + def get_queryset(self): + qs = CollectorGroup.objects.select_related('origin_zone').order_by('name') + q = self.request.GET.get('q') + if q: + qs = qs.filter(name__icontains=q) + return qs + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['current_q'] = self.request.GET.get('q', '') + return ctx + + +class CollectorGroupFormContextMixin: + model = CollectorGroup + form_class = CollectorGroupForm + template_name = 'ops/register_form.html' + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['cancel_url'] = reverse('ops:collector_group_list') + return ctx + + +class CollectorGroupCreateView(CollectorGroupFormContextMixin, OpsRequiredMixin, CreateView): + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = 'New collector group' + ctx['submit_label'] = 'Create group' + return ctx + + def form_valid(self, form): + response = super().form_valid(form) + messages.success(self.request, f"Collector group {self.object.name} created.") + return response + + def get_success_url(self): + return reverse('ops:collector_group_list') + + +class CollectorGroupUpdateView(CollectorGroupFormContextMixin, OpsRequiredMixin, UpdateView): + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = f"Edit collector group {self.object.name}" + ctx['submit_label'] = 'Save changes' + return ctx + + def form_valid(self, form): + response = super().form_valid(form) + messages.success(self.request, f"Collector group {self.object.name} updated.") + return response + + def get_success_url(self): + return reverse('ops:collector_group_list') + + +# --------------------------------------------------------------------------- +# Records that belong to a due-diligence case +# +# Mitigation actions, EUDR evidence and external-submission references used to +# be creatable/editable only in admin (or, for evidence, not at all from the +# portal). These screens are reached from the case detail and return to it. + + +class CaseChildMixin(OpsRequiredMixin): + template_name = 'ops/register_form.html' + + def get_case(self): + return get_object_or_404(DueDiligenceCase, pk=self.kwargs['case_pk']) + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['case'] = self.get_case() + ctx['cancel_url'] = reverse( + 'ops:eudr_detail', kwargs={'pk': self.kwargs['case_pk']} + ) + return ctx + + def get_success_url(self): + return reverse('ops:eudr_detail', kwargs={'pk': self.kwargs['case_pk']}) + + +class EUDREvidenceCreateView(CaseChildMixin, CreateView): + model = EUDREvidence + form_class = EUDREvidenceForm + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = ( + f"Attach evidence — {self.get_case().batch.batch_reference}" + ) + ctx['submit_label'] = 'Attach evidence' + return ctx + + def form_valid(self, form): + form.instance.case = self.get_case() + response = super().form_valid(form) + # Adding evidence changes the evidence completeness result, so re-run it + # exactly as the API does. + services.run_completeness_check(self.object.case) + messages.success( + self.request, + "Evidence attached. The case's completeness check has been re-run.", + ) + return response + + +class EUDREvidenceUpdateView(CaseChildMixin, UpdateView): + model = EUDREvidence + form_class = EUDREvidenceForm + + def get_queryset(self): + return EUDREvidence.objects.filter(case_id=self.kwargs['case_pk']) + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = ( + f"Edit evidence — {self.get_case().batch.batch_reference}" + ) + ctx['submit_label'] = 'Save changes' + return ctx + + def form_valid(self, form): + response = super().form_valid(form) + services.run_completeness_check(self.object.case) + messages.success(self.request, 'Evidence updated.') + return response + + +class MitigationActionCreateView(CaseChildMixin, CreateView): + model = MitigationAction + form_class = MitigationActionForm + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = ( + f"Add mitigation action — {self.get_case().batch.batch_reference}" + ) + ctx['submit_label'] = 'Add action' + return ctx + + def form_valid(self, form): + form.instance.case = self.get_case() + response = super().form_valid(form) + messages.success(self.request, 'Mitigation action added.') + return response + + +class MitigationActionUpdateView(CaseChildMixin, UpdateView): + model = MitigationAction + form_class = MitigationActionForm + + def get_queryset(self): + return MitigationAction.objects.filter(case_id=self.kwargs['case_pk']) + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = ( + f"Edit mitigation action — {self.get_case().batch.batch_reference}" + ) + ctx['submit_label'] = 'Save changes' + return ctx + + def form_valid(self, form): + response = super().form_valid(form) + messages.success(self.request, 'Mitigation action updated.') + return response + + +class StatementReferenceUpdateView(CaseChildMixin, UpdateView): + model = DueDiligenceStatementReference + form_class = StatementReferenceForm + + def get_queryset(self): + return DueDiligenceStatementReference.objects.filter( + case_id=self.kwargs['case_pk'] + ) + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + ctx['page_title'] = ( + f"Edit submission reference — {self.get_case().batch.batch_reference}" + ) + ctx['submit_label'] = 'Save changes' + return ctx + + def form_valid(self, form): + response = super().form_valid(form) + messages.success(self.request, 'Submission reference updated.') + return response diff --git a/templates/ops/base_ops.html b/templates/ops/base_ops.html index 257d073..1266856 100644 --- a/templates/ops/base_ops.html +++ b/templates/ops/base_ops.html @@ -29,6 +29,7 @@ body { font-size: .875rem; background: #f5f6fa; } .guide-content pre { background: #f8fafc; border: 1px solid #e2e8f0; border-radius: .375rem; padding: .75rem; } .guide-content blockquote { border-left: 4px solid #a7f3d0; padding-left: 1rem; color: #475569; } +{% block extra_head %}{% endblock %} @@ -37,6 +38,9 @@ body { font-size: .875rem; background: #f5f6fa; }