import os import tempfile from decimal import Decimal from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase, override_settings from django.urls import reverse from django.utils import timezone from rest_framework.test import APIClient from apps.batches.models import ProcessingBatch from apps.capa.models import CAPARecord from apps.documents.models import CertificateOfAnalysis, Document from apps.notifications.models import Notification from apps.qc.models import QCResult from .models import ( CollectorGroup, DueDiligenceCase, Facility, HarvestIntake, IntakeBatchLineage, OriginZone, ProductSpecification, SpecParameter, ) from . import services User = get_user_model() def make_user(email, role): # Mirrors the Keycloak OIDC backend, which creates all users as is_staff=True. return User.objects.create_user( email=email, password="pass1234", role=role, full_name=email.split("@")[0], is_staff=True, ) def make_batch(user): return ProcessingBatch.objects.create( batch_reference=f"IXG-BATCH-TEST-{ProcessingBatch.objects.count() + 1}", grade="A", weight_kg=Decimal("1000.00"), origin_facility="ASI - Katsina State, Nigeria", created_by=user, ) class SpecEvaluationTests(TestCase): def setUp(self): self.spec = ProductSpecification.objects.create( product="Unrefined Shea Butter", version="9.9", status="approved", is_current=True ) SpecParameter.objects.create(spec=self.spec, parameter="ffa", unit="%", max_value=Decimal("3.00"), sort_order=1) SpecParameter.objects.create(spec=self.spec, parameter="moisture", unit="%", max_value=Decimal("0.10"), sort_order=2) SpecParameter.objects.create(spec=self.spec, parameter="peroxide", unit="meq O2/kg", max_value=Decimal("10.00"), sort_order=3) SpecParameter.objects.create(spec=self.spec, parameter="visual", unit="grade", allowed_values="A,B,C", sort_order=4) def test_pass_within_spec(self): values = {"ffa": "2.0", "moisture": "0.05", "peroxide": "5.0", "visual_grade": "A"} overall_pass, result_status, _, version = services.evaluate_qc_data(values, spec=self.spec) self.assertTrue(overall_pass) self.assertEqual(result_status, QCResult.ResultStatus.WITHIN_SPEC) self.assertEqual(version, "9.9") def test_fail_outside_spec(self): values = {"ffa": "5.0", "moisture": "0.05", "peroxide": "5.0", "visual_grade": "A"} overall_pass, result_status, _, _ = services.evaluate_qc_data(values, spec=self.spec) self.assertFalse(overall_pass) self.assertEqual(result_status, QCResult.ResultStatus.OUTSIDE_SPEC) def test_fail_visual_grade(self): values = {"ffa": "2.0", "moisture": "0.05", "peroxide": "5.0", "visual_grade": "Fail"} overall_pass, result_status, _, _ = services.evaluate_qc_data(values, spec=self.spec) self.assertFalse(overall_pass) def test_fallback_legacy_limits_when_no_spec(self): values = {"ffa": "2.0", "moisture": "0.05", "peroxide": "5.0", "visual_grade": "A"} overall_pass, result_status, _, version = services.evaluate_qc_data(values, spec=None) self.assertTrue(overall_pass) self.assertEqual(version, "") class QCWorkflowTests(TestCase): def setUp(self): self.ops_a = make_user("ops.a@ixg.local", "ops") self.ops_b = make_user("ops.b@ixg.local", "ops") self.buyer = make_user("buyer@ixg.local", "buyer") self.batch = make_batch(self.ops_a) self.client = APIClient() def create_qc(self, values, user=None): self.client.force_authenticate(user=user or self.ops_a) payload = { "batch": str(self.batch.id), "ffa_percentage": values[0], "moisture_percentage": values[1], "peroxide_value": values[2], "visual_grade": values[3], } return self.client.post("/api/qc-results/", payload, format="json") def test_qc_fail_creates_capa_and_quarantine(self): resp = self.create_qc(["9.0", "0.9", "20", "Fail"]) self.assertEqual(resp.status_code, 201) qc = QCResult.objects.get(batch=self.batch) self.assertFalse(qc.overall_pass) self.assertTrue(qc.is_quarantine) self.assertEqual(qc.entered_by, self.ops_a) self.assertEqual(qc.result_status, QCResult.ResultStatus.OUTSIDE_SPEC) self.batch.refresh_from_db() self.assertEqual(self.batch.status, ProcessingBatch.Status.QC_FAIL) self.assertTrue(CAPARecord.objects.filter(batch=self.batch).exists()) def test_qc_pass_sets_batch_qc_pass(self): resp = self.create_qc(["2.0", "0.05", "5.0", "A"]) self.assertEqual(resp.status_code, 201) qc = QCResult.objects.get(batch=self.batch) self.assertTrue(qc.overall_pass) self.batch.refresh_from_db() self.assertEqual(self.batch.status, ProcessingBatch.Status.QC_PASS) def test_verify_four_eyes_blocks_submitter(self): qc = QCResult.objects.create( batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, entered_by=self.ops_a, ) self.client.force_authenticate(user=self.ops_a) resp = self.client.post(f"/api/qc-results/{qc.id}/verify/", {"approved": "true"}, format="json") self.assertEqual(resp.status_code, 400) qc.refresh_from_db() self.assertEqual(qc.verification_status, QCResult.VerificationStatus.DRAFT) def test_verify_approval_generates_coa(self): qc = QCResult.objects.create( batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, entered_by=self.ops_a, spec_version="1.0", ) self.client.force_authenticate(user=self.ops_b) resp = self.client.post(f"/api/qc-results/{qc.id}/verify/", {"approved": "true"}, format="json") self.assertEqual(resp.status_code, 200) qc.refresh_from_db() self.assertEqual(qc.verification_status, QCResult.VerificationStatus.APPROVED) self.assertEqual(qc.verified_by, self.ops_b) coa = CertificateOfAnalysis.objects.get(qc_result=qc) self.assertEqual(coa.status, CertificateOfAnalysis.Status.PENDING_APPROVAL) self.assertTrue(coa.sha256_hash) self.assertTrue(os.path.exists(coa.file_url.replace("/media/", "media/"))) @override_settings(MEDIA_ROOT=tempfile.mkdtemp(prefix="ixg_test_media_")) def test_publish_coa_creates_buyer_document_and_notification(self): qc = QCResult.objects.create( batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, entered_by=self.ops_a, spec_version="1.0", ) coa = services.create_coa_for_qc(qc, approved_by=self.ops_b) services.publish_coa(coa, released_by=self.ops_b) coa.refresh_from_db() self.assertEqual(coa.status, CertificateOfAnalysis.Status.PUBLISHED) buyer_doc = Document.objects.filter(batch=self.batch, doc_type="coa", visibility="buyer") self.assertEqual(buyer_doc.count(), 1) self.assertEqual(buyer_doc.first().file_url, coa.file_url) self.assertTrue(Notification.objects.filter(user=self.buyer, type="coa_published").exists()) def test_buyer_verification_forbidden(self): qc = QCResult.objects.create( batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, entered_by=self.ops_a, ) self.client.force_authenticate(user=self.buyer) resp = self.client.post(f"/api/qc-results/{qc.id}/verify/", {"approved": "true"}, format="json") self.assertEqual(resp.status_code, 403) class EUDRWorkflowTests(TestCase): def setUp(self): self.ops = make_user("ops.eudr@ixg.local", "ops") self.buyer = make_user("buyer.eudr@ixg.local", "buyer") self.batch = make_batch(self.ops) def test_completeness_missing_lineage_blocks(self): case = DueDiligenceCase.objects.create(batch=self.batch) services.run_completeness_check(case) case.refresh_from_db() self.assertEqual(case.status, DueDiligenceCase.Status.INCOMPLETE) self.assertFalse(case.completeness_results["geolocation"]["pass"]) self.assertFalse(case.completeness_results["lineage"]["pass"]) def test_completeness_unverified_zone_blocks(self): facility = Facility.objects.create(name="ASI Katsina", country="Nigeria", region="Katsina") zone = OriginZone.objects.create( facility=facility, zone_code="KAT-01", country="Nigeria", region="Katsina", verification_status=OriginZone.VerificationStatus.UNVERIFIED, ) group = CollectorGroup.objects.create(name="Group A", origin_zone=zone) intake = HarvestIntake.objects.create( intake_reference="INT-001", collector_group=group, origin_zone=zone, intake_date="2026-01-10", weight_kg=Decimal("500.00"), ) IntakeBatchLineage.objects.create(batch=self.batch, harvest_intake=intake, weight_allocated_kg=Decimal("500.00")) case = services.ensure_due_diligence_case(self.batch) services.run_completeness_check(case) case.refresh_from_db() self.assertEqual(case.status, DueDiligenceCase.Status.INCOMPLETE) self.assertFalse(case.completeness_results["geolocation"]["pass"]) def test_verified_zone_with_lineage_passes_geolocation(self): facility = Facility.objects.create(name="ASI Katsina", country="Nigeria", region="Katsina") zone = OriginZone.objects.create( facility=facility, zone_code="KAT-02", country="Nigeria", region="Katsina", verification_status=OriginZone.VerificationStatus.VERIFIED, ) group = CollectorGroup.objects.create(name="Group B", origin_zone=zone) intake = HarvestIntake.objects.create( intake_reference="INT-002", collector_group=group, origin_zone=zone, intake_date="2026-01-10", weight_kg=Decimal("500.00"), ) IntakeBatchLineage.objects.create(batch=self.batch, harvest_intake=intake, weight_allocated_kg=Decimal("500.00")) case = services.ensure_due_diligence_case(self.batch) services.run_completeness_check(case) case.refresh_from_db() self.assertTrue(case.completeness_results["geolocation"]["pass"]) self.assertTrue(case.completeness_results["lineage"]["pass"]) def test_risk_high_without_lineage(self): case = DueDiligenceCase.objects.create(batch=self.batch) risk = services.evaluate_risk(case, user=self.ops) case.refresh_from_db() self.assertEqual(risk.risk_level, "high") self.assertEqual(case.risk_level, DueDiligenceCase.RiskLevel.HIGH) self.assertEqual(case.status, DueDiligenceCase.Status.MITIGATION_REQUIRED) def test_approve_case_requires_complete_state(self): case = DueDiligenceCase.objects.create(batch=self.batch) with self.assertRaises(ValueError): services.approve_case(case, self.ops) def test_record_submission(self): case = DueDiligenceCase.objects.create(batch=self.batch) ref = services.record_submission(case, self.ops, "DD-2026-0001", "2026-08-05") self.assertEqual(ref.dds_reference, "DD-2026-0001") self.assertEqual(case.statement_references.count(), 1) class BuyerIsolationTests(TestCase): def setUp(self): self.buyer = make_user("iso.buyer@ixg.local", "buyer") self.ops = make_user("iso.ops@ixg.local", "ops") self.client = APIClient() def test_buyer_cannot_access_compliance_api(self): self.client.force_authenticate(user=self.buyer) resp = self.client.get("/api/product-specs/") self.assertEqual(resp.status_code, 403) resp = self.client.get("/api/eudr-cases/") self.assertEqual(resp.status_code, 403) def test_ops_can_access_compliance_api(self): self.client.force_authenticate(user=self.ops) resp = self.client.get("/api/product-specs/") self.assertEqual(resp.status_code, 200) resp = self.client.get("/api/eudr-cases/") self.assertEqual(resp.status_code, 200) def test_buyer_forbidden_from_ops(self): self.client.force_login(self.buyer) resp = self.client.get("/ops/") self.assertEqual(resp.status_code, 403) class PageRenderTests(TestCase): def setUp(self): self.ops = make_user("page.ops@ixg.local", "ops") self.buyer = make_user("page.buyer@ixg.local", "buyer") self.batch = make_batch(self.ops) self.qc = QCResult.objects.create( batch=self.batch, ffa_percentage=Decimal("2.0"), moisture_percentage=Decimal("0.05"), peroxide_value=Decimal("5.0"), visual_grade="A", overall_pass=True, entered_by=self.ops, spec_version="1.0", ) self.coa = services.create_coa_for_qc(self.qc, approved_by=self.ops) self.case = DueDiligenceCase.objects.create(batch=self.batch) def test_ops_pages_render(self): self.client.force_login(self.ops) for url in [ "/ops/qc/verify/", f"/ops/qc/verify/{self.qc.id}/", "/ops/coa/", f"/ops/coa/{self.coa.id}/", "/ops/eudr/", f"/ops/eudr/{self.case.id}/", ]: resp = self.client.get(url) self.assertEqual(resp.status_code, 200, msg=f"GET {url}") def test_buyer_compliance_page_renders(self): self.client.force_login(self.buyer) resp = self.client.get("/portal/compliance/") self.assertEqual(resp.status_code, 200) def test_ops_post_publish_coa(self): self.client.force_login(self.ops) resp = self.client.post(f"/ops/coa/{self.coa.id}/", {"action": "approve"}) self.assertEqual(resp.status_code, 302) self.coa.refresh_from_db() self.assertEqual(self.coa.status, CertificateOfAnalysis.Status.PUBLISHED)