from decimal import Decimal from datetime import datetime from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from rest_framework import status from rest_framework.test import APITestCase from checkpoint.models import CheckPoint, FacilityType from record.models import ( CollectionCategory, CollectionType, CommodityCategory, CommodityType, DestructionSubmission, Document, ExportSubmission, HealthInspectionSubmission, ImportSubmission, MetadataDeletionRequest, RevenueFineSubmission, Submission, SubmissionStatus, SubmissionType, UnitOfMeasurement, UnitConversion, ) User = get_user_model() class PartialOnlyViewSetUpdateTests(APITestCase): def setUp(self): self.admin_user = User.objects.create_user( email="partial-admin@example.com", password="password123", name="Partial Admin", role=User.Role.ADMIN, ) self.facility_type = FacilityType.objects.create(name="Border Point") self.checkpoint = CheckPoint.objects.create( name="Assigned CP", location="Kakarbhitta", facility_type=self.facility_type, ) self.submission_type = SubmissionType.objects.create( name="Import", code_prefix="IMP", rank=1 ) self.commodity_category = CommodityCategory.objects.create(name="Agriculture") self.commodity_type = CommodityType.objects.create( category=self.commodity_category, name="Seeds", ) self.from_unit = UnitOfMeasurement.objects.create( name="Kilogram", code="KG", dimension=UnitOfMeasurement.DimensionChoices.WEIGHT, ) self.to_unit = UnitOfMeasurement.objects.create( name="Gram", code="G", dimension=UnitOfMeasurement.DimensionChoices.WEIGHT, ) self.unit_conversion = UnitConversion.objects.create( from_unit=self.from_unit, to_unit=self.to_unit, factor=Decimal("1000"), ) SubmissionStatus.objects.bulk_create( [SubmissionStatus(name="Pending", code="PENDING")] ) self.submission_status = SubmissionStatus.objects.get(code="PENDING") self.collection_category = CollectionCategory.objects.create(name="Penalties") self.collection_type = CollectionType.objects.create( category=self.collection_category, name="Late Fee", ) self.submission = Submission.objects.create( submission_type=self.submission_type, checkpoint=self.checkpoint, created_by=self.admin_user, updated_by=self.admin_user, status=self.submission_status, remarks="Original submission", ) self.document = Document.objects.create( submission=self.submission, uploaded_by=self.admin_user, file="uploads/documents/test.txt", remarks="Original document", ) self.client.force_authenticate(user=self.admin_user) def test_admin_can_patch_record_partial_only_viewsets(self): cases = ( ( "/api/v1/record/submission-types/", self.submission_type, {"category_description": "Updated submission type"}, "category_description", "Updated submission type", ), ( "/api/v1/record/commodity-categories/", self.commodity_category, {"category_description": "Updated commodity category"}, "category_description", "Updated commodity category", ), ( "/api/v1/record/commodity-types/", self.commodity_type, {"description": "Updated commodity type"}, "description", "Updated commodity type", ), ( "/api/v1/record/units-of-measurement/", self.from_unit, {"name": "Kilograms"}, "name", "Kilograms", ), ( "/api/v1/record/unit-conversions/", self.unit_conversion, {"factor": "1000.5000000000"}, "factor", Decimal("1000.5000000000"), ), ( "/api/v1/record/submission-status/", self.submission_status, {"name": "Awaiting"}, "name", "Awaiting", ), ( "/api/v1/record/collection-categories/", self.collection_category, {"description": "Updated collection category"}, "description", "Updated collection category", ), ( "/api/v1/record/collection-types/", self.collection_type, {"description": "Updated collection type"}, "description", "Updated collection type", ), ( "/api/v1/record/documents/", self.document, {"remarks": "Updated document"}, "remarks", "Updated document", ), ) for endpoint, obj, payload, field_name, expected_value in cases: with self.subTest(endpoint=endpoint): response = self.client.patch( f"{endpoint}{obj.id}/", payload, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) obj.refresh_from_db() self.assertEqual(getattr(obj, field_name), expected_value) class ExportSubmissionCheckpointAssignmentTests(APITestCase): endpoint = "/api/v1/record/export-submissions/" def setUp(self): self.facility_type = FacilityType.objects.create(name="Border Point") self.assigned_checkpoint = CheckPoint.objects.create( name="Assigned CP", location="Kakarbhitta", facility_type=self.facility_type, ) self.other_checkpoint = CheckPoint.objects.create( name="Other CP", location="Birgunj", facility_type=self.facility_type, ) self.user = User.objects.create_user( email="staff@example.com", password="password123", name="Staff User", checkpoint=self.assigned_checkpoint, role=User.Role.STAFF, ) self.submission_type = SubmissionType.objects.create( name="Export", code_prefix="EXP", rank=1 ) self.commodity_category = CommodityCategory.objects.create(name="Livestock") self.commodity_type = CommodityType.objects.create( category=self.commodity_category, name="Goat", ) self.unit_of_measurement = UnitOfMeasurement.objects.create( name="Kilogram", code="KG", ) SubmissionStatus.objects.bulk_create( [SubmissionStatus(name="Pending", code="PENDING")] ) self.client.force_authenticate(user=self.user) def _payload(self): return { "submission_type": str(self.submission_type.id), "commodity_type": str(self.commodity_type.id), "exporter_name": "Everest Exports", "destination_country": "India", "quantity": "12.50", "goods_description": "Goat products", "base_amount": "1000.00", "currency": "NPR", "unit_of_measurement": str(self.unit_of_measurement.id), "remarks": "Initial export submission", } def test_create_submission_uses_authenticated_users_assigned_checkpoint(self): response = self.client.post(self.endpoint, self._payload(), format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) created = ExportSubmission.objects.get(id=response.data["id"]) self.assertEqual(created.checkpoint_id, self.assigned_checkpoint.id) def test_payload_checkpoint_is_ignored_and_user_checkpoint_is_used(self): payload = self._payload() payload["checkpoint"] = str(self.other_checkpoint.id) response = self.client.post(self.endpoint, payload, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) created = ExportSubmission.objects.get(id=response.data["id"]) self.assertEqual(created.checkpoint_id, self.assigned_checkpoint.id) def test_create_submission_fails_if_user_has_no_assigned_checkpoint(self): self.user.checkpoint = None self.user.save(update_fields=["checkpoint"]) response = self.client.post(self.endpoint, self._payload(), format="json") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("checkpoint", [detail.get("field") for detail in details]) class ImportSubmissionInlineDocumentUploadTests(APITestCase): endpoint = "/api/v1/record/import-submissions/" def setUp(self): self.facility_type = FacilityType.objects.create(name="Border Point") self.assigned_checkpoint = CheckPoint.objects.create( name="Assigned CP", location="Kakarbhitta", facility_type=self.facility_type, ) self.user = User.objects.create_user( email="importstaff@example.com", password="password123", name="Import Staff", checkpoint=self.assigned_checkpoint, role=User.Role.STAFF, ) self.submission_type = SubmissionType.objects.create( name="Import", code_prefix="IMP", rank=11 ) self.commodity_category = CommodityCategory.objects.create(name="Food") self.commodity_type = CommodityType.objects.create( category=self.commodity_category, name="Dairy", ) self.unit_of_measurement = UnitOfMeasurement.objects.create( name="Carton", code="CTN", ) SubmissionStatus.objects.bulk_create( [SubmissionStatus(name="Pending", code="PENDING")] ) self.pending_status = SubmissionStatus.objects.get(code="PENDING") self.client.force_authenticate(user=self.user) def _payload(self): return { "submission_type": str(self.submission_type.id), "commodity_type": str(self.commodity_type.id), "detailed_description": "Milk products", "approval_no": "APP-IMP-001", "approval_time": "2026-04-09T10:30:00", "manufacturing_date": "2026-03-01", "batch_no": "BATCH-001", "base_amount": "1200.50", "currency": "NPR", "health_certificate_no": "HC-7788", "health_certificate_issue_date": "2026-03-15", "health_certificate_issue_agency": "Food Dept", "invoice_no": "INV-9090", "invoice_date": "2026-03-20", "origin_of_consignment": "Nepal", "unit_of_measurement": str(self.unit_of_measurement.id), "remarks": "Import with docs", } def _create_import_submission(self): return ImportSubmission.objects.create( submission_type=self.submission_type, checkpoint=self.assigned_checkpoint, created_by=self.user, updated_by=self.user, status=self.pending_status, commodity_type=self.commodity_type, detailed_description="Created directly for patch test", approval_no="APP-IMP-002", approval_time=datetime(2026, 4, 9, 10, 30, 0), manufacturing_date="2026-03-01", batch_no="BATCH-XYZ", base_amount="100.00", currency="NPR", health_certificate_no="HC-ABC", health_certificate_issue_date="2026-03-15", health_certificate_issue_agency="Food Dept", invoice_no="INV-XYZ", invoice_date="2026-03-20", origin_of_consignment="Nepal", unit_of_measurement=self.unit_of_measurement, ) def test_create_import_submission_accepts_documents_in_same_request(self): payload = self._payload() payload["documents"] = [ SimpleUploadedFile( "invoice.pdf", b"%PDF-1.4 fake content", content_type="application/pdf" ), SimpleUploadedFile( "certificate.jpg", b"\xff\xd8\xff\xe0jpeg", content_type="image/jpeg", ), ] payload["document_remarks"] = "Attached during create" response = self.client.post(self.endpoint, data=payload, format="multipart") self.assertEqual(response.status_code, status.HTTP_201_CREATED) submission_id = response.data["id"] docs = Document.objects.filter(submission_id=submission_id).order_by("created_at") self.assertEqual(docs.count(), 2) self.assertEqual(docs[0].uploaded_by_id, self.user.id) self.assertEqual(docs[0].remarks, "Attached during create") self.assertTrue(docs[0].file.name.startswith(f"uploads/documents/{submission_id}/")) def test_patch_import_submission_can_append_documents(self): submission = self._create_import_submission() response = self.client.patch( f"{self.endpoint}{submission.id}/", data={ "documents": [ SimpleUploadedFile( "extra-note.txt", b"extra attachment", content_type="text/plain", ) ], "document_remarks": "Patched attachment", }, format="multipart", ) self.assertEqual(response.status_code, status.HTTP_200_OK) docs = Document.objects.filter(submission=submission) self.assertEqual(docs.count(), 1) self.assertEqual(docs.first().remarks, "Patched attachment") class SubmissionCheckpointScopeTests(APITestCase): endpoint = "/api/v1/record/submissions/" dashboard_summary_endpoint = "/api/v1/record/dashboard/summary/" dashboard_monthly_endpoint = "/api/v1/record/dashboard/monthly/" dashboard_quarterly_endpoint = "/api/v1/record/dashboard/quarterly/" def setUp(self): self.facility_type = FacilityType.objects.create(name="Border Point") self.checkpoint_one = CheckPoint.objects.create( name="Checkpoint One", location="Kakarbhitta", facility_type=self.facility_type, ) self.checkpoint_two = CheckPoint.objects.create( name="Checkpoint Two", location="Birgunj", facility_type=self.facility_type, ) self.staff_one = User.objects.create_user( email="staff1@example.com", password="password123", name="Staff One", checkpoint=self.checkpoint_one, role=User.Role.STAFF, ) self.staff_one_peer = User.objects.create_user( email="staff1b@example.com", password="password123", name="Staff One Peer", checkpoint=self.checkpoint_one, role=User.Role.STAFF, ) self.staff_two = User.objects.create_user( email="staff2@example.com", password="password123", name="Staff Two", checkpoint=self.checkpoint_two, role=User.Role.STAFF, ) self.admin_user = User.objects.create_user( email="admin@example.com", password="password123", name="Admin User", role=User.Role.ADMIN, ) self.submission_type = SubmissionType.objects.create( name="Export Scoped", code_prefix="EXS", rank=2 ) self.commodity_category = CommodityCategory.objects.create(name="Medicines") self.commodity_type = CommodityType.objects.create( category=self.commodity_category, name="Vaccine", ) self.unit_of_measurement = UnitOfMeasurement.objects.create( name="Box", code="BOX", ) self.collection_category = CollectionCategory.objects.create(name="Penalties") self.collection_type = CollectionType.objects.create( category=self.collection_category, name="Late Filing Fee", ) SubmissionStatus.objects.bulk_create( [SubmissionStatus(name="Pending", code="PENDING")] ) self.pending_status = SubmissionStatus.objects.get(code="PENDING") self.submission_staff_one = self._create_export_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="100.00" ) self.submission_staff_one_peer = self._create_export_submission( created_by=self.staff_one_peer, checkpoint=self.checkpoint_one, base_amount="200.00", ) self.submission_staff_two = self._create_export_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="300.00" ) def _create_export_submission(self, created_by, checkpoint, base_amount): return ExportSubmission.objects.create( submission_type=self.submission_type, checkpoint=checkpoint, created_by=created_by, updated_by=created_by, status=self.pending_status, commodity_type=self.commodity_type, exporter_name="Scoped Exporter", destination_country="India", export_date="2081-05-10", quantity="10.00", goods_description="Scoped goods", base_amount=base_amount, currency="NPR", unit_of_measurement=self.unit_of_measurement, remarks="Scoped test submission", ) def _create_import_summary_submission(self, created_by, checkpoint, base_amount): amount_suffix = str(base_amount).replace(".", "") return ImportSubmission.objects.create( submission_type=self.submission_type, checkpoint=checkpoint, created_by=created_by, updated_by=created_by, status=self.pending_status, commodity_type=self.commodity_type, detailed_description="Summary import goods", approval_no=f"APP-{checkpoint.id.hex[:8]}-{amount_suffix}", approval_time=datetime(2026, 4, 9, 10, 30, 0), import_date="2081-05-12", manufacturing_date="2026-03-01", batch_no=f"BATCH-{amount_suffix}", base_amount=base_amount, currency="NPR", health_certificate_no=f"HC-{amount_suffix}", health_certificate_issue_date="2026-03-15", health_certificate_issue_agency="Food Dept", invoice_no=f"INV-{amount_suffix}", invoice_date="2026-03-20", origin_of_consignment="Nepal", unit_of_measurement=self.unit_of_measurement, remarks="Scoped import summary submission", ) def _create_revenue_summary_submission(self, created_by, checkpoint, base_amount): amount_suffix = str(base_amount).replace(".", "") return RevenueFineSubmission.objects.create( submission_type=self.submission_type, checkpoint=checkpoint, created_by=created_by, updated_by=created_by, status=self.pending_status, collection_type=self.collection_type, revenue_date="2081-05-15", payment_refrence=f"PAY-{checkpoint.id.hex[:8]}-{amount_suffix}", reason="Summary revenue collection", base_amount=base_amount, currency="NPR", remarks="Scoped revenue summary submission", ) @staticmethod def _result_ids(response): results = response.data.get("results", response.data) return {item["id"] for item in results} def test_staff_list_only_sees_submissions_from_assigned_checkpoint(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get(self.endpoint) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertSetEqual( self._result_ids(response), {str(self.submission_staff_one.id), str(self.submission_staff_one_peer.id)}, ) def test_staff_can_filter_by_user_within_same_checkpoint(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get(self.endpoint, {"user": str(self.staff_one_peer.id)}) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertSetEqual(self._result_ids(response), {str(self.submission_staff_one_peer.id)}) def test_staff_cannot_filter_by_user_from_other_checkpoint(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get(self.endpoint, {"user": str(self.staff_two.id)}) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("user", [detail.get("field") for detail in details]) def test_staff_cannot_filter_by_other_checkpoint(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get( self.endpoint, {"checkpoint": str(self.checkpoint_two.id)} ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("checkpoint", [detail.get("field") for detail in details]) def test_admin_can_filter_by_any_checkpoint(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.endpoint, {"checkpoint": str(self.checkpoint_two.id)} ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertSetEqual(self._result_ids(response), {str(self.submission_staff_two.id)}) def test_staff_can_delete_submission_from_same_checkpoint_group(self): self.client.force_authenticate(user=self.staff_one) response = self.client.delete( f"{self.endpoint}{self.submission_staff_one_peer.id}/" ) self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) self.assertFalse( ExportSubmission.objects.filter(id=self.submission_staff_one_peer.id).exists() ) def test_staff_cannot_delete_submission_from_other_checkpoint(self): self.client.force_authenticate(user=self.staff_one) response = self.client.delete(f"{self.endpoint}{self.submission_staff_two.id}/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_staff_dashboard_summary_is_scoped_to_assigned_checkpoint(self): self._create_import_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="150.00", ) self._create_import_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="250.00", ) self._create_revenue_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="80.00", ) self._create_revenue_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="120.00", ) self.client.force_authenticate(user=self.staff_one) response = self.client.get(self.dashboard_summary_endpoint, {"year": 2081}) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(Decimal(str(response.data["total_import_value"])), Decimal("150")) self.assertEqual(Decimal(str(response.data["total_export_value"])), Decimal("300")) self.assertEqual(Decimal(str(response.data["total_revenue"])), Decimal("80")) import_rows = response.data["reports_by_checkpoint"]["imports"] export_rows = response.data["reports_by_checkpoint"]["exports"] revenue_rows = response.data["reports_by_checkpoint"]["revenue"] checkpoint_totals = response.data["checkpoint_totals"] self.assertSetEqual( {row["checkpoint__name"] for row in import_rows}, {self.checkpoint_one.name}, ) self.assertSetEqual( {row["checkpoint__name"] for row in export_rows}, {self.checkpoint_one.name}, ) self.assertSetEqual( {row["checkpoint__name"] for row in revenue_rows}, {self.checkpoint_one.name}, ) self.assertEqual(len(checkpoint_totals), 1) self.assertEqual(checkpoint_totals[0]["checkpoint_name"], self.checkpoint_one.name) self.assertEqual(Decimal(str(checkpoint_totals[0]["import_value"])), Decimal("150")) self.assertEqual(Decimal(str(checkpoint_totals[0]["export_value"])), Decimal("300")) self.assertEqual(Decimal(str(checkpoint_totals[0]["revenue_value"])), Decimal("80")) def test_admin_dashboard_summary_includes_all_checkpoints(self): self._create_import_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="150.00", ) self._create_import_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="250.00", ) self._create_revenue_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="80.00", ) self._create_revenue_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="120.00", ) self.client.force_authenticate(user=self.admin_user) response = self.client.get(self.dashboard_summary_endpoint, {"year": 2081}) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(Decimal(str(response.data["total_import_value"])), Decimal("400")) self.assertEqual(Decimal(str(response.data["total_export_value"])), Decimal("600")) self.assertEqual(Decimal(str(response.data["total_revenue"])), Decimal("200")) revenue_rows = response.data["reports_by_checkpoint"]["revenue"] checkpoint_totals = response.data["checkpoint_totals"] self.assertSetEqual( {row["checkpoint__name"] for row in revenue_rows}, {self.checkpoint_one.name, self.checkpoint_two.name}, ) self.assertSetEqual( {row["checkpoint_name"] for row in checkpoint_totals}, {self.checkpoint_one.name, self.checkpoint_two.name}, ) def test_admin_dashboard_summary_can_filter_by_checkpoint(self): self._create_import_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="150.00", ) self._create_import_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="250.00", ) self._create_revenue_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="80.00", ) self._create_revenue_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="120.00", ) self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.dashboard_summary_endpoint, {"year": 2081, "checkpoint": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(Decimal(str(response.data["total_import_value"])), Decimal("250")) self.assertEqual(Decimal(str(response.data["total_export_value"])), Decimal("300")) self.assertEqual(Decimal(str(response.data["total_revenue"])), Decimal("120")) import_rows = response.data["reports_by_checkpoint"]["imports"] export_rows = response.data["reports_by_checkpoint"]["exports"] revenue_rows = response.data["reports_by_checkpoint"]["revenue"] checkpoint_totals = response.data["checkpoint_totals"] self.assertSetEqual( {row["checkpoint__name"] for row in import_rows}, {self.checkpoint_two.name}, ) self.assertSetEqual( {row["checkpoint__name"] for row in export_rows}, {self.checkpoint_two.name}, ) self.assertSetEqual( {row["checkpoint__name"] for row in revenue_rows}, {self.checkpoint_two.name}, ) self.assertEqual(len(checkpoint_totals), 1) self.assertEqual(checkpoint_totals[0]["checkpoint_name"], self.checkpoint_two.name) self.assertEqual(Decimal(str(checkpoint_totals[0]["import_value"])), Decimal("250")) self.assertEqual(Decimal(str(checkpoint_totals[0]["export_value"])), Decimal("300")) self.assertEqual(Decimal(str(checkpoint_totals[0]["revenue_value"])), Decimal("120")) def test_admin_dashboard_summary_can_filter_by_multiple_checkpoints(self): self._create_import_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="150.00", ) self._create_import_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="250.00", ) self._create_revenue_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="80.00", ) self._create_revenue_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="120.00", ) self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.dashboard_summary_endpoint, { "year": 2081, "checkpoints": f"{self.checkpoint_one.id},{self.checkpoint_two.id}", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(Decimal(str(response.data["total_import_value"])), Decimal("400")) self.assertEqual(Decimal(str(response.data["total_export_value"])), Decimal("600")) self.assertEqual(Decimal(str(response.data["total_revenue"])), Decimal("200")) def test_staff_dashboard_summary_cannot_filter_other_checkpoint(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get( self.dashboard_summary_endpoint, {"year": 2081, "checkpoint": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("checkpoint", [detail.get("field") for detail in details]) def test_staff_dashboard_summary_cannot_filter_other_checkpoint_with_checkpoints_param(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get( self.dashboard_summary_endpoint, {"year": 2081, "checkpoints": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("checkpoint", [detail.get("field") for detail in details]) def test_dashboard_summary_accepts_fiscal_year_label(self): self._create_import_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="250.00", ) self._create_revenue_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="120.00", ) self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.dashboard_summary_endpoint, {"fiscal_year": "2081/82", "checkpoint": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["fiscal_year"], "2081/82") self.assertEqual(Decimal(str(response.data["total_import_value"])), Decimal("250")) self.assertEqual(Decimal(str(response.data["total_export_value"])), Decimal("300")) self.assertEqual(Decimal(str(response.data["total_revenue"])), Decimal("120")) def test_dashboard_summary_rejects_invalid_fiscal_year_label(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.dashboard_summary_endpoint, {"fiscal_year": "2081/90"}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("fiscal_year", [detail.get("field") for detail in details]) def test_admin_dashboard_monthly_can_filter_by_checkpoint(self): self._create_import_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="150.00", ) self._create_import_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="250.00", ) self._create_revenue_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="80.00", ) self._create_revenue_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="120.00", ) self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.dashboard_monthly_endpoint, {"year": 2081, "checkpoint": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) month_five = next( item for item in response.data["monthly_trends"] if item["nepali_month"] == 5 ) self.assertEqual(Decimal(str(month_five["total_import"])), Decimal("250")) self.assertEqual(Decimal(str(month_five["total_export"])), Decimal("300")) self.assertEqual(Decimal(str(month_five["total_revenue"])), Decimal("120")) def test_admin_dashboard_monthly_accepts_fiscal_year_label(self): self._create_import_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="250.00", ) self._create_revenue_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="120.00", ) self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.dashboard_monthly_endpoint, {"fiscal_year": "2081/82", "checkpoint": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["fiscal_year"], "2081/82") month_five = next( item for item in response.data["monthly_trends"] if item["nepali_month"] == 5 ) self.assertEqual(Decimal(str(month_five["total_import"])), Decimal("250")) self.assertEqual(Decimal(str(month_five["total_export"])), Decimal("300")) self.assertEqual(Decimal(str(month_five["total_revenue"])), Decimal("120")) def test_staff_dashboard_monthly_cannot_filter_other_checkpoint(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get( self.dashboard_monthly_endpoint, {"year": 2081, "checkpoint": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("checkpoint", [detail.get("field") for detail in details]) def test_admin_dashboard_quarterly_can_filter_by_checkpoint(self): self._create_import_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="150.00", ) self._create_import_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="250.00", ) self._create_revenue_summary_submission( created_by=self.staff_one, checkpoint=self.checkpoint_one, base_amount="80.00", ) self._create_revenue_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="120.00", ) self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.dashboard_quarterly_endpoint, {"year": 2081, "checkpoint": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) quarter_one = next( item for item in response.data["quarterly_trends"] if item["nepali_quarter"] == 1 ) self.assertEqual(Decimal(str(quarter_one["total_import"])), Decimal("250")) self.assertEqual(Decimal(str(quarter_one["total_export"])), Decimal("300")) self.assertEqual(Decimal(str(quarter_one["total_revenue"])), Decimal("120")) def test_admin_dashboard_quarterly_accepts_fiscal_year_label(self): self._create_import_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="250.00", ) self._create_revenue_summary_submission( created_by=self.staff_two, checkpoint=self.checkpoint_two, base_amount="120.00", ) self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.dashboard_quarterly_endpoint, {"fiscal_year": "2081/82", "checkpoint": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["fiscal_year"], "2081/82") quarter_one = next( item for item in response.data["quarterly_trends"] if item["nepali_quarter"] == 1 ) self.assertEqual(Decimal(str(quarter_one["total_import"])), Decimal("250")) self.assertEqual(Decimal(str(quarter_one["total_export"])), Decimal("300")) self.assertEqual(Decimal(str(quarter_one["total_revenue"])), Decimal("120")) def test_staff_dashboard_quarterly_cannot_filter_other_checkpoint(self): self.client.force_authenticate(user=self.staff_one) response = self.client.get( self.dashboard_quarterly_endpoint, {"year": 2081, "checkpoint": str(self.checkpoint_two.id)}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) details = response.data["error"]["details"] self.assertIn("checkpoint", [detail.get("field") for detail in details]) class DashboardMonthlyFilterAndExportTests(APITestCase): export_monthly_endpoint = "/api/v1/record/dashboard/export-monthly/" export_file_endpoint = "/api/v1/record/dashboard/monthly-export/" def setUp(self): self.facility_type = FacilityType.objects.create(name="Dry Port") self.checkpoint = CheckPoint.objects.create( name="Birgunj CP", location="Birgunj", facility_type=self.facility_type, ) self.staff_user = User.objects.create_user( email="dashboard-staff@example.com", password="password123", name="Dashboard Staff", checkpoint=self.checkpoint, role=User.Role.STAFF, ) self.client.force_authenticate(user=self.staff_user) self.submission_type = SubmissionType.objects.create( name="Export Dashboard", code_prefix="EXD", rank=10 ) SubmissionStatus.objects.bulk_create( [SubmissionStatus(name="Pending", code="PENDING")] ) self.status_pending = SubmissionStatus.objects.get(code="PENDING") self.unit = UnitOfMeasurement.objects.create(name="Kg", code="KG-DASH") self.category_animal = CommodityCategory.objects.create(name="Animal Products") self.category_fish = CommodityCategory.objects.create(name="Fish Products") self.commodity_goat = CommodityType.objects.create( category=self.category_animal, name="Goat Meat", ) self.commodity_fish = CommodityType.objects.create( category=self.category_fish, name="Contaminated Fish", ) self._create_export_submission( commodity_type=self.commodity_goat, quantity="10.00", nepali_year=2081, nepali_month=4, nepali_day=10, ) self._create_export_submission( commodity_type=self.commodity_fish, quantity="20.00", nepali_year=2081, nepali_month=5, nepali_day=15, ) self._create_export_submission( commodity_type=self.commodity_goat, quantity="5.00", nepali_year=2081, nepali_month=5, nepali_day=28, ) def _create_export_submission( self, commodity_type, quantity, nepali_year, nepali_month, nepali_day ): submission = ExportSubmission.objects.create( submission_type=self.submission_type, checkpoint=self.checkpoint, created_by=self.staff_user, updated_by=self.staff_user, status=self.status_pending, commodity_type=commodity_type, exporter_name="Dashboard Exporter", destination_country="India", quantity=quantity, goods_description="Dashboard export goods", base_amount="1000.00", currency="NPR", unit_of_measurement=self.unit, remarks="Dashboard testing", ) ExportSubmission.objects.filter(id=submission.id).update( nepali_year=nepali_year, nepali_month=nepali_month, nepali_day=nepali_day, nepali_quarter=1 if nepali_month in (4, 5, 6) else 2, export_nepali_year=nepali_year, export_nepali_month=nepali_month, export_nepali_day=nepali_day, export_nepali_quarter=1 if nepali_month in (4, 5, 6) else 2, ) @staticmethod def _month_items(response, month): for row in response.data["monthly_data"]: if row["nepali_month"] == month: return row["items"] return [] def test_export_monthly_supports_bs_date_range_filter(self): response = self.client.get( self.export_monthly_endpoint, { "year": 2081, "from_bs_date": "2081-05-01", "to_bs_date": "2081-05-30", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) month4_items = self._month_items(response, 4) month5_items = self._month_items(response, 5) self.assertEqual(month4_items, []) self.assertEqual(len(month5_items), 2) def test_export_monthly_supports_commodity_filters(self): response = self.client.get( self.export_monthly_endpoint, {"year": 2081, "commodity_type": str(self.commodity_goat.id)}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) month4_items = self._month_items(response, 4) month5_items = self._month_items(response, 5) all_names = {item["commodity"] for item in month4_items + month5_items} self.assertSetEqual(all_names, {self.commodity_goat.name}) def test_export_monthly_supports_commodity_category_filter(self): response = self.client.get( self.export_monthly_endpoint, {"year": 2081, "commodity_category": str(self.category_fish.id)}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) month5_items = self._month_items(response, 5) all_names = {item["commodity"] for item in month5_items} self.assertSetEqual(all_names, {self.commodity_fish.name}) def test_monthly_export_endpoint_supports_excel_pdf_and_print(self): csv_response = self.client.get( self.export_file_endpoint, { "year": 2081, "report_category": "export", "export_format": "excel", "commodity_type": str(self.commodity_goat.id), }, ) self.assertEqual(csv_response.status_code, status.HTTP_200_OK) self.assertIn("text/csv", csv_response["Content-Type"]) self.assertIn(self.commodity_goat.name, csv_response.content.decode("utf-8")) self.assertNotIn(self.commodity_fish.name, csv_response.content.decode("utf-8")) pdf_response = self.client.get( self.export_file_endpoint, { "year": 2081, "report_category": "export", "export_format": "pdf", }, ) self.assertEqual(pdf_response.status_code, status.HTTP_200_OK) self.assertIn("application/pdf", pdf_response["Content-Type"]) self.assertTrue(pdf_response.content.startswith(b"%PDF-1.4")) print_response = self.client.get( self.export_file_endpoint, { "year": 2081, "report_category": "export", "export_format": "print", }, ) self.assertEqual(print_response.status_code, status.HTTP_200_OK) self.assertIn("text/html", print_response["Content-Type"]) self.assertIn("", print_response.content.decode("utf-8")) class DashboardMonthlyConversionAggregationTests(APITestCase): export_monthly_endpoint = "/api/v1/record/dashboard/export-monthly/" def setUp(self): self.facility_type = FacilityType.objects.create(name="Dry Port") self.checkpoint = CheckPoint.objects.create( name="Birgunj CP", location="Birgunj", facility_type=self.facility_type, ) self.staff_user = User.objects.create_user( email="dashboard-conversion@example.com", password="password123", name="Dashboard Conversion User", checkpoint=self.checkpoint, role=User.Role.STAFF, ) self.client.force_authenticate(user=self.staff_user) self.submission_type = SubmissionType.objects.create( name="Export Conversion Dashboard", code_prefix="EXC", rank=12 ) SubmissionStatus.objects.bulk_create( [SubmissionStatus(name="Pending", code="PENDING")] ) self.status_pending = SubmissionStatus.objects.get(code="PENDING") self.unit_kg = UnitOfMeasurement.objects.create( name="Kilogram", code="KG-MONTHLY", dimension=UnitOfMeasurement.DimensionChoices.WEIGHT, ) self.unit_gm = UnitOfMeasurement.objects.create( name="Gram", code="GM-MONTHLY", dimension=UnitOfMeasurement.DimensionChoices.WEIGHT, ) UnitConversion.objects.create( from_unit=self.unit_kg, to_unit=self.unit_gm, factor=Decimal("1000"), ) self.category = CommodityCategory.objects.create(name="Animal Products") self.commodity = CommodityType.objects.create( category=self.category, name="Goat Meat", ) self._create_export_submission( quantity="2.00", unit=self.unit_kg, nepali_month=4, ) self._create_export_submission( quantity="500.00", unit=self.unit_gm, nepali_month=4, ) def _create_export_submission(self, quantity, unit, nepali_month): submission = ExportSubmission.objects.create( submission_type=self.submission_type, checkpoint=self.checkpoint, created_by=self.staff_user, updated_by=self.staff_user, status=self.status_pending, commodity_type=self.commodity, exporter_name="Dashboard Exporter", destination_country="India", quantity=quantity, goods_description="Dashboard export goods", base_amount="1000.00", currency="NPR", unit_of_measurement=unit, remarks="Dashboard conversion testing", ) ExportSubmission.objects.filter(id=submission.id).update( nepali_year=2081, nepali_month=nepali_month, nepali_day=10, nepali_quarter=1, export_nepali_year=2081, export_nepali_month=nepali_month, export_nepali_day=10, export_nepali_quarter=1, ) def test_export_monthly_merges_convertible_units_without_response_shape_change(self): response = self.client.get(self.export_monthly_endpoint, {"year": 2081}) self.assertEqual(response.status_code, status.HTTP_200_OK) month4 = next( row for row in response.data["monthly_data"] if row["nepali_month"] == 4 ) self.assertEqual(len(month4["items"]), 1) item = month4["items"][0] self.assertSetEqual(set(item.keys()), {"commodity", "unit", "total_quantity"}) self.assertEqual(item["commodity"], "Goat Meat") self.assertEqual(item["unit"], "Kilogram") self.assertEqual(Decimal(str(item["total_quantity"])), Decimal("2.5")) class CommodityMetadataPermissionTests(APITestCase): category_endpoint = "/api/v1/record/commodity-categories/" type_endpoint = "/api/v1/record/commodity-types/" def setUp(self): self.staff_user = User.objects.create_user( email="commodity-staff@example.com", password="password123", name="Commodity Staff", role=User.Role.STAFF, ) self.category = CommodityCategory.objects.create(name="Agriculture") self.commodity_type = CommodityType.objects.create( category=self.category, name="Seeds", ) def test_staff_can_create_commodity_category(self): self.client.force_authenticate(user=self.staff_user) response = self.client.post( self.category_endpoint, {"name": "Livestock", "category_description": "Animal-related goods"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertTrue(CommodityCategory.objects.filter(name="Livestock").exists()) def test_staff_can_create_commodity_type(self): self.client.force_authenticate(user=self.staff_user) response = self.client.post( self.type_endpoint, { "category": str(self.category.id), "name": "Animal Feed", "description": "Processed feed product", }, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertTrue( CommodityType.objects.filter( category=self.category, name="Animal Feed", ).exists() ) def test_staff_cannot_patch_commodity_category(self): self.client.force_authenticate(user=self.staff_user) response = self.client.patch( f"{self.category_endpoint}{self.category.id}/", {"name": "Updated Agriculture"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_staff_cannot_delete_commodity_type(self): self.client.force_authenticate(user=self.staff_user) response = self.client.delete(f"{self.type_endpoint}{self.commodity_type.id}/") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) class CollectionMetadataPermissionTests(APITestCase): category_endpoint = "/api/v1/record/collection-categories/" type_endpoint = "/api/v1/record/collection-types/" def setUp(self): self.staff_user = User.objects.create_user( email="collection-staff@example.com", password="password123", name="Collection Staff", role=User.Role.STAFF, ) self.category = CollectionCategory.objects.create(name="Penalties") self.collection_type = CollectionType.objects.create( category=self.category, name="Late Fee", ) def test_staff_can_create_collection_category(self): self.client.force_authenticate(user=self.staff_user) response = self.client.post( self.category_endpoint, {"name": "Service Charges", "description": "Service-related charges"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertTrue(CollectionCategory.objects.filter(name="Service Charges").exists()) def test_staff_can_create_collection_type(self): self.client.force_authenticate(user=self.staff_user) response = self.client.post( self.type_endpoint, { "category": str(self.category.id), "name": "Penalty Charge", "description": "Charge for policy violation", }, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertTrue( CollectionType.objects.filter( category=self.category, name="Penalty Charge", ).exists() ) def test_staff_cannot_patch_collection_category(self): self.client.force_authenticate(user=self.staff_user) response = self.client.patch( f"{self.category_endpoint}{self.category.id}/", {"name": "Updated Penalties"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_staff_cannot_delete_collection_type(self): self.client.force_authenticate(user=self.staff_user) response = self.client.delete(f"{self.type_endpoint}{self.collection_type.id}/") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) class MasterDataDeletionRequestWorkflowTests(APITestCase): endpoint = "/api/v1/record/deletion-requests/" def setUp(self): self.admin_user = User.objects.create_user( email="metadata-admin@example.com", password="password123", name="Metadata Admin", role=User.Role.ADMIN, ) self.staff_user = User.objects.create_user( email="metadata-staff@example.com", password="password123", name="Metadata Staff", role=User.Role.STAFF, ) self.other_staff_user = User.objects.create_user( email="metadata-other-staff@example.com", password="password123", name="Metadata Other Staff", role=User.Role.STAFF, ) self.commodity_category = CommodityCategory.objects.create(name="Livestock") self.commodity_type = CommodityType.objects.create( category=self.commodity_category, name="Goat Meat", ) self.collection_category = CollectionCategory.objects.create(name="Penalties") self.collection_type = CollectionType.objects.create( category=self.collection_category, name="Late Fee", ) @staticmethod def _extract_results(data): if isinstance(data, dict) and "results" in data: return data["results"] return data def test_staff_can_create_master_data_deletion_request(self): self.client.force_authenticate(user=self.staff_user) response = self.client.post( self.endpoint, { "target_type": MetadataDeletionRequest.TargetType.COMMODITY_TYPE, "target_id": str(self.commodity_type.id), "reason": "Duplicate commodity entry", }, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) created_request = MetadataDeletionRequest.objects.get(id=response.data["id"]) self.assertEqual( created_request.status, MetadataDeletionRequest.StatusChoices.PENDING ) self.assertEqual(created_request.requested_by_id, self.staff_user.id) self.assertEqual(created_request.target_name, str(self.commodity_type)) def test_staff_cannot_create_duplicate_pending_delete_request(self): MetadataDeletionRequest.objects.create( target_type=MetadataDeletionRequest.TargetType.COMMODITY_TYPE, target_id=self.commodity_type.id, target_name=str(self.commodity_type), requested_by=self.staff_user, ) self.client.force_authenticate(user=self.staff_user) response = self.client.post( self.endpoint, { "target_type": MetadataDeletionRequest.TargetType.COMMODITY_TYPE, "target_id": str(self.commodity_type.id), }, format="json", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_patch_is_not_allowed_for_requests(self): delete_request = MetadataDeletionRequest.objects.create( target_type=MetadataDeletionRequest.TargetType.COMMODITY_TYPE, target_id=self.commodity_type.id, target_name=str(self.commodity_type), requested_by=self.staff_user, ) self.client.force_authenticate(user=self.admin_user) response = self.client.patch( f"{self.endpoint}{delete_request.id}/", {"reason": "Updated reason"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED) def test_delete_is_not_allowed_for_requests(self): delete_request = MetadataDeletionRequest.objects.create( target_type=MetadataDeletionRequest.TargetType.COMMODITY_TYPE, target_id=self.commodity_type.id, target_name=str(self.commodity_type), requested_by=self.staff_user, ) self.client.force_authenticate(user=self.admin_user) response = self.client.delete(f"{self.endpoint}{delete_request.id}/") self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED) def test_legacy_metadata_route_is_removed(self): self.client.force_authenticate(user=self.staff_user) response = self.client.get("/api/v1/record/metadata-delete-requests/") self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_non_admin_only_sees_own_delete_requests(self): own_request = MetadataDeletionRequest.objects.create( target_type=MetadataDeletionRequest.TargetType.COMMODITY_TYPE, target_id=self.commodity_type.id, target_name=str(self.commodity_type), requested_by=self.staff_user, ) MetadataDeletionRequest.objects.create( target_type=MetadataDeletionRequest.TargetType.COLLECTION_TYPE, target_id=self.collection_type.id, target_name=str(self.collection_type), requested_by=self.other_staff_user, ) self.client.force_authenticate(user=self.staff_user) response = self.client.get(self.endpoint) self.assertEqual(response.status_code, status.HTTP_200_OK) items = self._extract_results(response.data) self.assertEqual(len(items), 1) self.assertEqual(items[0]["id"], str(own_request.id)) def test_admin_approve_requires_confirmation_before_delete(self): delete_request = MetadataDeletionRequest.objects.create( target_type=MetadataDeletionRequest.TargetType.COLLECTION_TYPE, target_id=self.collection_type.id, target_name=str(self.collection_type), requested_by=self.staff_user, ) self.client.force_authenticate(user=self.admin_user) preview_response = self.client.post( f"{self.endpoint}{delete_request.id}/approve/", {"review_note": "Approved, duplicate metadata."}, format="json", ) self.assertEqual(preview_response.status_code, status.HTTP_409_CONFLICT) self.assertTrue(preview_response.data["requires_confirmation"]) self.assertIn("deletion_impact", preview_response.data) delete_request.refresh_from_db() self.assertEqual( delete_request.status, MetadataDeletionRequest.StatusChoices.PENDING ) self.assertTrue(CollectionType.objects.filter(id=self.collection_type.id).exists()) response = self.client.post( f"{self.endpoint}{delete_request.id}/approve/", { "review_note": "Approved, duplicate metadata.", "confirm_deletion": True, }, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) delete_request.refresh_from_db() self.assertEqual( delete_request.status, MetadataDeletionRequest.StatusChoices.APPROVED ) self.assertEqual(delete_request.reviewed_by_id, self.admin_user.id) self.assertIsNotNone(delete_request.reviewed_at) self.assertFalse(CollectionType.objects.filter(id=self.collection_type.id).exists()) def test_approve_warning_includes_related_record_count(self): facility_type = FacilityType.objects.create(name="Border Point") checkpoint = CheckPoint.objects.create( name="Warn CP", location="Kathmandu", facility_type=facility_type, ) submission_type = SubmissionType.objects.create( name="Warning Revenue", code_prefix="WRN", rank=100, ) SubmissionStatus.objects.bulk_create( [SubmissionStatus(name="Pending", code="PENDING-WARN")] ) pending_status = SubmissionStatus.objects.get(code="PENDING-WARN") RevenueFineSubmission.objects.create( submission_type=submission_type, checkpoint=checkpoint, created_by=self.staff_user, updated_by=self.staff_user, status=pending_status, collection_type=self.collection_type, payment_refrence="PAY-WARN-1", base_amount=Decimal("100"), currency="NPR", ) delete_request = MetadataDeletionRequest.objects.create( target_type=MetadataDeletionRequest.TargetType.COLLECTION_TYPE, target_id=self.collection_type.id, target_name=str(self.collection_type), requested_by=self.staff_user, ) self.client.force_authenticate(user=self.admin_user) response = self.client.post( f"{self.endpoint}{delete_request.id}/approve/", {"review_note": "Preview impact only"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_409_CONFLICT) impact = response.data["deletion_impact"] self.assertGreaterEqual(impact["related_records"], 1) self.assertGreaterEqual(impact["total_records_including_target"], 2) def test_admin_can_deny_delete_request_without_deleting_target(self): delete_request = MetadataDeletionRequest.objects.create( target_type=MetadataDeletionRequest.TargetType.COMMODITY_CATEGORY, target_id=self.commodity_category.id, target_name=str(self.commodity_category), requested_by=self.staff_user, ) self.client.force_authenticate(user=self.admin_user) response = self.client.post( f"{self.endpoint}{delete_request.id}/deny/", {"review_note": "Needed for reporting references."}, format="json", ) self.assertEqual(response.status_code, status.HTTP_200_OK) delete_request.refresh_from_db() self.assertEqual( delete_request.status, MetadataDeletionRequest.StatusChoices.DENIED ) self.assertTrue( CommodityCategory.objects.filter(id=self.commodity_category.id).exists() ) def test_staff_cannot_approve_delete_request(self): delete_request = MetadataDeletionRequest.objects.create( target_type=MetadataDeletionRequest.TargetType.COMMODITY_TYPE, target_id=self.commodity_type.id, target_name=str(self.commodity_type), requested_by=self.staff_user, ) self.client.force_authenticate(user=self.staff_user) response = self.client.post( f"{self.endpoint}{delete_request.id}/approve/", {"review_note": "Trying to self-approve"}, format="json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) delete_request.refresh_from_db() self.assertEqual( delete_request.status, MetadataDeletionRequest.StatusChoices.PENDING ) self.assertTrue(CommodityType.objects.filter(id=self.commodity_type.id).exists()) class UnitConversionAPITests(APITestCase): endpoint = "/api/v1/record/unit-conversions/" convert_endpoint = "/api/v1/record/unit-conversions/convert/" def setUp(self): self.admin_user = User.objects.create_user( email="uom-admin@example.com", password="password123", name="UOM Admin", role=User.Role.ADMIN, ) self.staff_user = User.objects.create_user( email="uom-staff@example.com", password="password123", name="UOM Staff", role=User.Role.STAFF, ) self.kg = UnitOfMeasurement.objects.create( name="Kilogram", code="KG", dimension=UnitOfMeasurement.DimensionChoices.WEIGHT, ) self.gm = UnitOfMeasurement.objects.create( name="Gram", code="GM", dimension=UnitOfMeasurement.DimensionChoices.WEIGHT, ) def test_admin_can_create_unit_conversion(self): self.client.force_authenticate(user=self.admin_user) response = self.client.post( self.endpoint, { "from_unit": str(self.kg.id), "to_unit": str(self.gm.id), "factor": "1000", }, format="json", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertTrue( UnitConversion.objects.filter( from_unit=self.kg, to_unit=self.gm, factor=Decimal("1000"), ).exists() ) def test_staff_cannot_create_unit_conversion(self): self.client.force_authenticate(user=self.staff_user) response = self.client.post( self.endpoint, { "from_unit": str(self.kg.id), "to_unit": str(self.gm.id), "factor": "1000", }, format="json", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_convert_endpoint_uses_direct_ratio(self): UnitConversion.objects.create(from_unit=self.kg, to_unit=self.gm, factor="1000") self.client.force_authenticate(user=self.staff_user) response = self.client.get( self.convert_endpoint, {"value": "2.5", "from_unit": "KG", "to_unit": "GM"}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["factor"], "1000") self.assertEqual(response.data["converted_value"], "2500") def test_convert_endpoint_uses_inverse_ratio_when_reverse_exists(self): UnitConversion.objects.create(from_unit=self.gm, to_unit=self.kg, factor="0.001") self.client.force_authenticate(user=self.staff_user) response = self.client.get( self.convert_endpoint, {"value": "2500", "from_unit": "KG", "to_unit": "GM"}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["factor"], "1000") self.assertEqual(response.data["converted_value"], "2500000") def test_convert_endpoint_returns_400_when_ratio_missing(self): self.client.force_authenticate(user=self.staff_user) response = self.client.get( self.convert_endpoint, {"value": "1", "from_unit": "KG", "to_unit": "GM"}, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("conversion", response.data["error"]["details"][0]["field"]) class AdminCheckpostReportEndpointTests(APITestCase): overview_endpoint = "/api/v1/record/admin-reports/checkpost/overview/" export_endpoint = "/api/v1/record/admin-reports/checkpost/export/" def setUp(self): self.facility_type = FacilityType.objects.create(name="Integrated Checkpost") self.nagdhunga = CheckPoint.objects.create( name="Nagdhunga", location="Kathmandu", facility_type=self.facility_type, ) self.ramnagar = CheckPoint.objects.create( name="Ramnagar", location="Parsa", facility_type=self.facility_type, ) self.admin_user = User.objects.create_user( email="admin-checkpost@example.com", password="password123", name="Admin User", role=User.Role.ADMIN, ) self.staff_user = User.objects.create_user( email="staff-checkpost@example.com", password="password123", name="Staff User", role=User.Role.STAFF, checkpoint=self.nagdhunga, ) self.submission_type = SubmissionType.objects.create( name="Admin Checkpost Import", code_prefix="ACP", rank=5, ) SubmissionStatus.objects.bulk_create( [SubmissionStatus(name="Pending", code="PENDING")] ) self.pending_status = SubmissionStatus.objects.get(code="PENDING") self.unit_heads = UnitOfMeasurement.objects.create(name="Heads", code="HEADS") self.unit_kg = UnitOfMeasurement.objects.create(name="KG", code="KG") self.commodity_category = CommodityCategory.objects.create(name="Livestock") self.buffalo_type = CommodityType.objects.create( category=self.commodity_category, name="Buffalo/Heifer", ) self.goat_type = CommodityType.objects.create( category=self.commodity_category, name="Goat/Sheep", ) self.chicken_type = CommodityType.objects.create( category=self.commodity_category, name="Ready Chicken", ) self.collection_category = CollectionCategory.objects.create(name="Fees") self.import_permit_fee = CollectionType.objects.create( category=self.collection_category, name="Import Permit Fee", ) self.late_fee = CollectionType.objects.create( category=self.collection_category, name="Late Fee", ) self._create_import_submission( checkpoint=self.nagdhunga, commodity=self.buffalo_type, quantity=1250, month=4, district="Dang", suffix="nag-buf-1", ) self._create_import_submission( checkpoint=self.nagdhunga, commodity=self.goat_type, quantity=5400, month=5, district="Kailali", suffix="nag-goat-1", ) self._create_import_submission( checkpoint=self.ramnagar, commodity=self.buffalo_type, quantity=600, month=4, district="Dang", suffix="ram-buf-1", ) self._create_import_submission( checkpoint=self.ramnagar, commodity=self.chicken_type, quantity=4100, month=6, district="Banke", suffix="ram-chk-1", ) self._create_health_inspection_submission( checkpoint=self.nagdhunga, product_name="Buffalo Meat", quantity=450, result="Pass", suffix="health-pass", ) self._create_health_inspection_submission( checkpoint=self.nagdhunga, product_name="Buffalo Meat", quantity=20, result="Fail", suffix="health-fail", ) self._create_health_inspection_submission( checkpoint=self.ramnagar, product_name="Buffalo Meat", quantity=999, result="Pass", suffix="health-other-checkpoint", ) self._create_destruction_submission( checkpoint=self.nagdhunga, commodity=self.chicken_type, quantity=500, method="Burying", suffix="destruction-burying", ) self._create_destruction_submission( checkpoint=self.ramnagar, commodity=self.chicken_type, quantity=999, method="Burning", suffix="destruction-other-checkpoint", ) self._create_revenue_submission( checkpoint=self.nagdhunga, collection_type=self.import_permit_fee, base_amount=Decimal("125000.00"), suffix="permit-fee", ) self._create_revenue_submission( checkpoint=self.nagdhunga, collection_type=self.late_fee, base_amount=Decimal("25250.00"), suffix="late-fee", ) self._create_revenue_submission( checkpoint=self.ramnagar, collection_type=self.import_permit_fee, base_amount=Decimal("999999.00"), suffix="other-checkpoint-fee", ) def _create_import_submission(self, checkpoint, commodity, quantity, month, district, suffix): return ImportSubmission.objects.create( submission_type=self.submission_type, checkpoint=checkpoint, created_by=self.admin_user, updated_by=self.admin_user, status=self.pending_status, commodity_type=commodity, detailed_description="Admin report seed data", approval_no=f"APP-{suffix}", approval_time=datetime(2026, 4, 10, 11, 0, 0), import_date=f"2082-{month:02d}-10", importer_name="Importer", importer_address=f"{district}, Nepal", exporter_name="Exporter", exporter_address="Kathmandu", total_quantity=quantity, bag_packet_count=20, manufacturing_date="2082-03-15", batch_no=f"BATCH-{suffix}", base_amount=Decimal("1000.00"), currency="NPR", health_certificate_no=f"HC-{suffix}", health_certificate_issue_date="2082-03-18", health_certificate_issue_agency="Food Dept", invoice_no=f"INV-{suffix}", invoice_date="2082-03-20", origin_of_consignment="Nepal", unit_of_measurement=self.unit_heads, remarks="Admin report testing", ) def _create_health_inspection_submission( self, checkpoint, product_name, quantity, result, suffix ): return HealthInspectionSubmission.objects.create( submission_type=self.submission_type, checkpoint=checkpoint, created_by=self.admin_user, updated_by=self.admin_user, status=self.pending_status, commodity_type=self.buffalo_type, inspection_type="Routine Check", product_name=product_name, health_date="2082-04-12", quantity=Decimal(str(quantity)), unit_of_measurement=self.unit_kg, result=result, notes=f"Admin report health seed {suffix}", ) def _create_destruction_submission( self, checkpoint, commodity, quantity, method, suffix ): return DestructionSubmission.objects.create( submission_type=self.submission_type, checkpoint=checkpoint, created_by=self.admin_user, updated_by=self.admin_user, status=self.pending_status, commodity_type=commodity, quantity=Decimal(str(quantity)), unit_of_measurement=self.unit_heads, destruction_method=method, destruction_date="2082-04-13", reason=f"Admin report destruction seed {suffix}", ) def _create_revenue_submission( self, checkpoint, collection_type, base_amount, suffix ): return RevenueFineSubmission.objects.create( submission_type=self.submission_type, checkpoint=checkpoint, created_by=self.admin_user, updated_by=self.admin_user, status=self.pending_status, collection_type=collection_type, revenue_date="2082-04-14", payment_refrence=f"PAY-{suffix}", reason="Admin report revenue seed data", base_amount=base_amount, currency="NPR", remarks="Admin report revenue testing", ) def test_admin_overview_returns_new_admin_report_payload(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.overview_endpoint, { "year": 2082, "quarter": 1, "checkpoint": str(self.nagdhunga.id), "comparison_checkpoint": str(self.ramnagar.id), }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["fiscal_year"], "2082/83") self.assertEqual(response.data["quarter"]["label"], "First Quarter") self.assertEqual( str(response.data["selected_checkpoint"]["id"]), str(self.nagdhunga.id) ) self.assertEqual( str(response.data["comparison_checkpoint"]["id"]), str(self.ramnagar.id) ) health_rows = response.data["health_inspection_data"]["rows"] self.assertEqual(len(health_rows), 1) self.assertEqual( health_rows[0], { "product": "Buffalo Meat", "unit": "KG", "passed": Decimal("450"), "failed": Decimal("20"), "total_quantity": Decimal("470"), }, ) destruction_rows = response.data["destruction_data"]["rows"] self.assertEqual(len(destruction_rows), 1) self.assertEqual( destruction_rows[0], { "commodity": "Ready Chicken", "method": "Burying", "unit": "Heads", "total_quantity": Decimal("500"), }, ) revenue_data = response.data["revenue_data"] self.assertEqual( Decimal(str(revenue_data["total_amount"])), Decimal("150250.00") ) revenue_rows = {row["category"]: row for row in revenue_data["rows"]} self.assertEqual(set(revenue_rows), {"Import Permit Fee", "Late Fee"}) self.assertEqual( Decimal(str(revenue_rows["Import Permit Fee"]["amount"])), Decimal("125000.00"), ) self.assertEqual(revenue_rows["Import Permit Fee"]["currency"], "NPR") movement_rows = response.data["movement_data"]["rows"] commodity_names = {row["commodity"] for row in movement_rows} self.assertIn("Buffalo/Heifer", commodity_names) self.assertIn("Goat/Sheep", commodity_names) self.assertNotIn("Ready Chicken", commodity_names) district_rows = response.data["district_return_comparison"]["rows"] self.assertGreaterEqual(len(district_rows), 1) self.assertIn("district", district_rows[0]) self.assertIn("selected_checkpoint", district_rows[0]) self.assertIn("comparison_checkpoint", district_rows[0]) def test_admin_overview_supports_search_filter(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.overview_endpoint, { "year": 2082, "quarter": 1, "checkpoint": str(self.nagdhunga.id), "search": "buffalo", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) movement_rows = response.data["movement_data"]["rows"] self.assertEqual(len(movement_rows), 1) self.assertEqual(movement_rows[0]["commodity"], "Buffalo/Heifer") def test_admin_overview_returns_empty_enforcement_sections_for_empty_period(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.overview_endpoint, { "year": 2082, "quarter": 2, "checkpoint": str(self.nagdhunga.id), }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["health_inspection_data"]["rows"], []) self.assertEqual(response.data["destruction_data"]["rows"], []) self.assertEqual(response.data["revenue_data"]["rows"], []) self.assertEqual( Decimal(str(response.data["revenue_data"]["total_amount"])), Decimal("0.0"), ) def test_staff_cannot_access_admin_checkpost_overview(self): self.client.force_authenticate(user=self.staff_user) response = self.client.get(self.overview_endpoint, {"year": 2082}) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_admin_checkpost_export_supports_excel(self): self.client.force_authenticate(user=self.admin_user) response = self.client.get( self.export_endpoint, { "year": 2082, "quarter": 1, "checkpoint": str(self.nagdhunga.id), "export_format": "excel", }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn("text/csv", response["Content-Type"]) csv_text = response.content.decode("utf-8") self.assertIn("Nagdhunga", csv_text) self.assertIn("Buffalo/Heifer", csv_text) self.assertNotIn("Ready Chicken", csv_text) class AdminCheckpostConversionAggregationTests(APITestCase): overview_endpoint = "/api/v1/record/admin-reports/checkpost/overview/" def setUp(self): self.facility_type = FacilityType.objects.create(name="Integrated Checkpost") self.selected_checkpoint = CheckPoint.objects.create( name="Nagdhunga", location="Kathmandu", facility_type=self.facility_type, ) self.comparison_checkpoint = CheckPoint.objects.create( name="Ramnagar", location="Parsa", facility_type=self.facility_type, ) self.admin_user = User.objects.create_user( email="admin-conversion-checkpost@example.com", password="password123", name="Admin Conversion User", role=User.Role.ADMIN, ) self.client.force_authenticate(user=self.admin_user) self.submission_type = SubmissionType.objects.create( name="Admin Checkpost Conversion", code_prefix="ACC", rank=7, ) SubmissionStatus.objects.bulk_create( [SubmissionStatus(name="Pending", code="PENDING")] ) self.pending_status = SubmissionStatus.objects.get(code="PENDING") self.unit_kg = UnitOfMeasurement.objects.create( name="Kilogram", code="KG-CHECK", dimension=UnitOfMeasurement.DimensionChoices.WEIGHT, ) self.unit_gm = UnitOfMeasurement.objects.create( name="Gram", code="GM-CHECK", dimension=UnitOfMeasurement.DimensionChoices.WEIGHT, ) UnitConversion.objects.create( from_unit=self.unit_kg, to_unit=self.unit_gm, factor=Decimal("1000"), ) self.commodity_category = CommodityCategory.objects.create(name="Livestock") self.buffalo_type = CommodityType.objects.create( category=self.commodity_category, name="Buffalo/Heifer", ) self._create_import_submission(self.unit_kg, quantity=1, suffix="kg") self._create_import_submission(self.unit_gm, quantity=500, suffix="gm") def _create_import_submission(self, unit, quantity, suffix): return ImportSubmission.objects.create( submission_type=self.submission_type, checkpoint=self.selected_checkpoint, created_by=self.admin_user, updated_by=self.admin_user, status=self.pending_status, commodity_type=self.buffalo_type, detailed_description="Admin conversion report seed data", approval_no=f"APP-{suffix}", approval_time=datetime(2026, 4, 10, 11, 0, 0), import_date="2082-04-10", importer_name="Importer", importer_address="Dang, Nepal", exporter_name="Exporter", exporter_address="Kathmandu", total_quantity=quantity, bag_packet_count=20, manufacturing_date="2082-03-15", batch_no=f"BATCH-{suffix}", base_amount=Decimal("1000.00"), currency="NPR", health_certificate_no=f"HC-{suffix}", health_certificate_issue_date="2082-03-18", health_certificate_issue_agency="Food Dept", invoice_no=f"INV-{suffix}", invoice_date="2082-03-20", origin_of_consignment="Nepal", unit_of_measurement=unit, remarks="Admin report conversion testing", ) def test_admin_overview_merges_convertible_units_without_response_shape_change(self): response = self.client.get( self.overview_endpoint, { "year": 2082, "quarter": 1, "checkpoint": str(self.selected_checkpoint.id), "comparison_checkpoint": str(self.comparison_checkpoint.id), }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) movement_rows = response.data["movement_data"]["rows"] self.assertEqual(len(movement_rows), 1) row = movement_rows[0] self.assertSetEqual( set(row.keys()), {"commodity", "unit", "monthly_quantities", "quarter_total"}, ) self.assertEqual(row["commodity"], "Buffalo/Heifer") self.assertEqual(row["unit"], "Kilogram") self.assertEqual(Decimal(str(row["quarter_total"])), Decimal("1.5")) district_row = response.data["district_return_comparison"]["rows"][0] self.assertSetEqual( set(district_row.keys()), { "district", "selected_checkpoint", "comparison_checkpoint", "combined_total", }, ) self.assertEqual( Decimal(str(district_row["selected_checkpoint"]["buffalo"])), Decimal("1.5"), )