ixg_platform/apps/compliance/tests.py

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 LineageAllocationAPITests(TestCase):
    """The API refuses the same over-allocation the ops lineage screen refuses."""

    def setUp(self):
        self.ops = make_user("lineage.ops@ixg.local", "ops")
        self.batch = make_batch(self.ops)
        facility = Facility.objects.create(
            name="Lineage Facility", country="Nigeria", region="Katsina"
        )
        zone = OriginZone.objects.create(
            facility=facility, zone_code="LIN-01", country="Nigeria",
            region="Katsina", verification_status=OriginZone.VerificationStatus.VERIFIED,
        )
        group = CollectorGroup.objects.create(name="Lineage Group", origin_zone=zone)
        self.intake = HarvestIntake.objects.create(
            intake_reference="INT-LIN-001", collector_group=group,
            origin_zone=zone, intake_date="2026-01-10",
            weight_kg=Decimal("100.00"),
        )
        self.client = APIClient()
        self.client.force_authenticate(user=self.ops)

    def post(self, weight):
        return self.client.post(
            "/api/intake-batch-lineage/",
            {
                "batch": str(self.batch.id),
                "harvest_intake": str(self.intake.id),
                "weight_allocated_kg": weight,
            },
            format="json",
        )

    def test_an_allocation_larger_than_the_intake_is_refused(self):
        response = self.post("9999.00")
        self.assertEqual(response.status_code, 400)
        self.assertFalse(IntakeBatchLineage.objects.exists())

    def test_an_allocation_that_overdraws_with_another_batch_is_refused(self):
        other = make_batch(self.ops)
        IntakeBatchLineage.objects.create(
            batch=other, harvest_intake=self.intake,
            weight_allocated_kg=Decimal("80.00"),
        )
        response = self.post("30.00")
        self.assertEqual(response.status_code, 400)
        self.assertEqual(IntakeBatchLineage.objects.count(), 1)

    def test_what_is_left_of_the_intake_can_be_allocated(self):
        other = make_batch(self.ops)
        IntakeBatchLineage.objects.create(
            batch=other, harvest_intake=self.intake,
            weight_allocated_kg=Decimal("80.00"),
        )
        response = self.post("20.00")
        self.assertEqual(response.status_code, 201)
        self.assertEqual(IntakeBatchLineage.objects.count(), 2)


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)


class OriginZoneMapAdminTests(TestCase):
    """The admin map that closes the "no map view" gap, and its empty states.

    The map is display only: these tests pin the fact that it draws what is
    stored, and that a zone without coordinates explains itself instead of
    rendering an empty box or raising.
    """

    def setUp(self):
        self.admin = User.objects.create_user(
            email="zoneadmin@ixg.local",
            password="pass1234",
            role="admin",
            is_staff=True,
            is_superuser=True,
        )
        self.facility = Facility.objects.create(
            name="IXG Processing Facility - Katsina", country="Nigeria",
            region="Katsina State", facility_type=Facility.FacilityType.PROCESSING,
        )
        self.zone = OriginZone.objects.create(
            facility=self.facility, zone_code="NG-KT-Z99", country="Nigeria",
            region="Katsina State",
            geometry={
                "type": "Polygon",
                "coordinates": [[[7.6012, 12.9905], [7.618, 12.9905],
                                 [7.618, 13.0042], [7.6012, 13.0042],
                                 [7.6012, 12.9905]]],
            },
        )

    def test_zone_map_draws_the_stored_polygon(self):
        self.client.force_login(self.admin)
        resp = self.client.get(f"/admin/compliance/originzone/{self.zone.id}/change/")
        self.assertEqual(resp.status_code, 200)
        html = resp.content.decode()
        self.assertIn('id="zone-map"', html)
        self.assertIn("7.6012", html)                     # the stored coordinates
        self.assertIn("vendor/leaflet/leaflet.js", html)  # the library is loaded
        self.assertIn("js/ixg_zone_map.js", html)

    def test_zone_without_coordinates_says_so(self):
        bare = OriginZone.objects.create(
            facility=self.facility, zone_code="NG-KT-Z98", country="Nigeria",
            geometry={},
        )
        self.client.force_login(self.admin)
        resp = self.client.get(f"/admin/compliance/originzone/{bare.id}/change/")
        self.assertEqual(resp.status_code, 200)
        html = resp.content.decode()
        self.assertIn("No polygon coordinates are stored", html)
        self.assertNotIn('id="zone-map"', html)

    def test_zone_map_on_the_add_page_is_not_an_error(self):
        self.client.force_login(self.admin)
        resp = self.client.get("/admin/compliance/originzone/add/")
        self.assertEqual(resp.status_code, 200)
        self.assertNotIn('id="zone-map"', resp.content.decode())