from __future__ import annotations from io import BytesIO from tempfile import TemporaryDirectory from types import SimpleNamespace from unittest.mock import patch from apps.core.models import BackgroundJob, JobStatus from apps.parsers.models import ParserLoadLog, ParserSourceArtifact, ParserStagedRecord from apps.parsers.ropk_sanctions import ( ROPK_SANCTIONS_MAX_BYTES, ROPK_SANCTIONS_MIME, ROPK_SANCTIONS_RECORD_TYPE, ROPK_SANCTIONS_SOURCE, RopkSanctionsValidationError, import_ropk_sanctions, ) from apps.parsers.tasks import parse_ropk_sanctions from django.core.files.base import ContentFile from django.core.files.storage import default_storage from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase from django.urls import reverse from django.utils import timezone from openpyxl import Workbook from organizations.models import ( Organization, OrganizationSourceRecord, SanctionsExtension, ) from rest_framework import status from rest_framework.test import APITestCase from tests.apps.user.factories import UserFactory HEADERS = [ "rn", "ogrn", "inn", "okpo", "Санкции - Великобритания HM Treasury", "Санкции - Евросоюз", "Санкции - США", "Санкции - Швейцария", "Санкции секторальные - США", "Санкции - Великобритания UKSL", "Санкции - Украина", ] def _workbook(rows: list[list[object]], *, headers: list[str] | None = None) -> BytesIO: workbook = Workbook() sheet = workbook.active sheet.append(headers or HEADERS) for row in rows: sheet.append(row) output = BytesIO() workbook.save(output) workbook.close() output.seek(0) return output def _row( rn: str, *, ogrn: str = "1027700132195", inn: str = "0012345678", okpo: str = "00123456", flags: tuple[object, ...] = (1, 0, 1, 0, 1, 0, 1), ) -> list[object]: return [rn, ogrn, inn, okpo, *flags] class RopkSanctionsImportTest(TestCase): def setUp(self): self.organization = Organization.objects.create( name="АО Санкции", inn="0012345678", ogrn="1027700132195", okpo="00123456", opk_registry_membership=True, directory_imported_at=timezone.now(), ) def test_import_enriches_by_identifiers_and_publishes_typed_payload(self): artifact, result = import_ropk_sanctions( handle=_workbook([_row("000001")]), original_name="sanctions.xlsx", load_batch=1, uploaded_by_id=None, ) self.assertFalse(result.skipped) self.assertEqual(result.parsed, 1) self.assertEqual(result.published, 1) self.assertEqual(result.quarantined, 0) self.assertEqual(artifact.status, ParserSourceArtifact.Status.PUBLISHED) record = OrganizationSourceRecord.objects.get() self.assertEqual(record.source, ROPK_SANCTIONS_SOURCE) self.assertEqual(record.record_type, ROPK_SANCTIONS_RECORD_TYPE) self.assertEqual(record.external_id, "000001") self.assertEqual(record.title, self.organization.name) self.assertEqual(record.record_date, "") self.assertEqual( record.payload, { "rn": "000001", "uk_hm_treasury": True, "european_union": False, "united_states": True, "switzerland": False, "united_states_sectoral": True, "uk_uksl": False, "ukraine": True, }, ) self.assertIsInstance(record.payload["uk_hm_treasury"], bool) self.assertEqual(record.extension.organization_id, self.organization.uid) def test_import_enriches_missing_inn_and_ogrn_by_unique_okpo(self): _, result = import_ropk_sanctions( handle=_workbook([_row("2", ogrn="", inn="")]), original_name="sanctions.xlsx", load_batch=2, uploaded_by_id=None, ) self.assertEqual(result.published, 1) record = OrganizationSourceRecord.objects.get() self.assertEqual(record.extension.organization.inn, "0012345678") self.assertEqual(record.extension.organization.ogrn, "1027700132195") def test_import_preserves_full_fourteen_digit_okpo(self): self.organization.okpo = "00123456789012" self.organization.save(update_fields=["okpo"]) _, result = import_ropk_sanctions( handle=_workbook([_row("000014", ogrn="", inn="", okpo="00123456789012")]), original_name="sanctions.xlsx", load_batch=21, uploaded_by_id=None, ) self.assertEqual(result.published, 1) record = OrganizationSourceRecord.objects.get() self.assertEqual(record.extension.organization.okpo, "00123456789012") self.assertEqual(record.external_id, "000014") def test_invalid_rows_are_quarantined_without_partial_records(self): _, result = import_ropk_sanctions( handle=_workbook( [ _row("duplicate"), _row("duplicate"), _row("bad-flag", flags=(2, 0, 0, 0, 0, 0, 0)), _row("mismatch", okpo="99999999"), ] ), original_name="sanctions.xlsx", load_batch=3, uploaded_by_id=None, ) self.assertEqual(result.published, 0) self.assertEqual(result.quarantined, 4) self.assertEqual( result.reasons, {"duplicate_rn": 2, "invalid_flag": 1, "identifier_mismatch": 1}, ) self.assertFalse(OrganizationSourceRecord.objects.exists()) self.assertEqual( ParserStagedRecord.objects.filter( disposition=ParserStagedRecord.Disposition.QUARANTINED ).count(), 4, ) def test_new_snapshot_atomically_replaces_previous_records(self): import_ropk_sanctions( handle=_workbook([_row("old")]), original_name="old.xlsx", load_batch=4, uploaded_by_id=None, ) import_ropk_sanctions( handle=_workbook([_row("new")]), original_name="new.xlsx", load_batch=5, uploaded_by_id=None, ) self.assertEqual( list( OrganizationSourceRecord.objects.filter( source=ROPK_SANCTIONS_SOURCE ).values_list("external_id", flat=True) ), ["new"], ) def test_duplicate_checksum_is_skipped_without_changing_snapshot(self): raw = _workbook([_row("same")]).getvalue() _, first = import_ropk_sanctions( handle=BytesIO(raw), original_name="first.xlsx", load_batch=6, uploaded_by_id=None, ) artifact, second = import_ropk_sanctions( handle=BytesIO(raw), original_name="second.xlsx", load_batch=7, uploaded_by_id=None, ) self.assertFalse(first.skipped) self.assertTrue(second.skipped) self.assertEqual(artifact.status, ParserSourceArtifact.Status.SKIPPED) self.assertEqual( OrganizationSourceRecord.objects.filter( source=ROPK_SANCTIONS_SOURCE ).count(), 1, ) def test_invalid_header_order_rejects_batch_and_preserves_snapshot(self): import_ropk_sanctions( handle=_workbook([_row("old")]), original_name="old.xlsx", load_batch=8, uploaded_by_id=None, ) reordered = [HEADERS[1], HEADERS[0], *HEADERS[2:]] with self.assertRaisesMessage(RopkSanctionsValidationError, "invalid_headers"): import_ropk_sanctions( handle=_workbook([_row("new")], headers=reordered), original_name="invalid.xlsx", load_batch=9, uploaded_by_id=None, ) self.assertEqual( list( OrganizationSourceRecord.objects.filter( source=ROPK_SANCTIONS_SOURCE ).values_list("external_id", flat=True) ), ["old"], ) self.assertEqual( ParserSourceArtifact.objects.filter( source=ROPK_SANCTIONS_SOURCE, status=ParserSourceArtifact.Status.REJECTED, ).count(), 1, ) def test_archive_expansion_limit_rejects_file_but_retains_raw_artifact(self): with patch( "apps.parsers.ropk_sanctions.ROPK_SANCTIONS_MAX_UNCOMPRESSED_BYTES", 1, ), self.assertRaisesMessage( RopkSanctionsValidationError, "xlsx_uncompressed_size_exceeded", ): import_ropk_sanctions( handle=_workbook([_row("zip-limit")]), original_name="sanctions.xlsx", load_batch=22, uploaded_by_id=None, ) artifact = ParserSourceArtifact.objects.get(source=ROPK_SANCTIONS_SOURCE) self.assertEqual(artifact.status, ParserSourceArtifact.Status.REJECTED) self.assertTrue(artifact.file.name) self.assertTrue(artifact.file.storage.exists(artifact.file.name)) class RopkSanctionsApiTest(APITestCase): def setUp(self): self.user = UserFactory.create_user() self.admin = UserFactory.create_user(is_staff=True) self.upload_url = reverse( "api_v1:parsers:upload-parser-data", args=[ROPK_SANCTIONS_SOURCE], ) def _upload( self, name: str, content: bytes, content_type: str = ROPK_SANCTIONS_MIME ): return self.client.post( self.upload_url, {"file": SimpleUploadedFile(name, content, content_type=content_type)}, format="multipart", ) def test_upload_is_admin_only_and_queues_new_source(self): self.client.force_authenticate(self.user) self.assertEqual( self._upload("sanctions.xlsx", b"xlsx").status_code, status.HTTP_403_FORBIDDEN, ) self.client.force_authenticate(self.admin) with patch( "apps.parsers.views._save_uploaded_parser_file", return_value="parser_uploads/sanctions.xlsx", ), patch( "apps.parsers.tasks.parse_ropk_sanctions.apply_async", side_effect=lambda **kwargs: SimpleNamespace(id=kwargs["task_id"]), ): response = self._upload("sanctions.xlsx", b"xlsx") self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) self.assertEqual(response.data["data"]["source"], ROPK_SANCTIONS_SOURCE) self.assertEqual(response.data["data"]["status"], JobStatus.PENDING) self.assertEqual( response.data["data"]["task_ids"], [response.data["data"]["task_id"]], ) def test_upload_rejects_extension_mime_and_size_before_queueing(self): self.client.force_authenticate(self.admin) with patch("apps.parsers.views._save_uploaded_parser_file") as save_file, patch( "apps.parsers.tasks.parse_ropk_sanctions.apply_async" ) as apply_async: extension = self._upload("sanctions.csv", b"csv", "text/csv") mime = self._upload("sanctions.xlsx", b"xlsx", "application/zip") with patch("apps.parsers.views.ROPK_SANCTIONS_MAX_BYTES", 1): oversized = self._upload("sanctions.xlsx", b"xx") self.assertEqual(extension.status_code, status.HTTP_415_UNSUPPORTED_MEDIA_TYPE) self.assertEqual(extension.data["errors"][0]["code"], "unsupported_media_type") self.assertEqual(mime.status_code, status.HTTP_415_UNSUPPORTED_MEDIA_TYPE) self.assertEqual(mime.data["errors"][0]["code"], "unsupported_media_type") self.assertEqual( oversized.status_code, status.HTTP_413_REQUEST_ENTITY_TOO_LARGE ) self.assertEqual(oversized.data["errors"][0]["code"], "file_too_large") save_file.assert_not_called() apply_async.assert_not_called() self.assertEqual(ROPK_SANCTIONS_MAX_BYTES, 25 * 1024 * 1024) def test_upload_requires_file_before_creating_job(self): self.client.force_authenticate(self.admin) response = self.client.post(self.upload_url, {}, format="multipart") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(response.data["errors"][0]["code"], "file_required") self.assertFalse(BackgroundJob.objects.exists()) def test_upload_rejects_concurrent_job_before_saving_file(self): BackgroundJob.objects.create( task_id="ropk-running", task_name="apps.parsers.tasks.parse_ropk_sanctions", status=JobStatus.STARTED, ) self.client.force_authenticate(self.admin) with patch("apps.parsers.views._save_uploaded_parser_file") as save_file: response = self._upload("sanctions.xlsx", b"xlsx") self.assertEqual(response.status_code, status.HTTP_409_CONFLICT) self.assertEqual(response.data["errors"][0]["code"], "upload_already_running") save_file.assert_not_called() def test_catalog_records_and_file_only_contract_are_available(self): organization = Organization.objects.create( name="АО API Санкции", inn="7707083801", ogrn="1027700132101", okpo="12345671", opk_registry_membership=True, directory_imported_at=timezone.now(), ) extension = SanctionsExtension.objects.create( organization=organization, title="РОПК — Санкции", ) OrganizationSourceRecord.objects.create( extension=extension, source=ROPK_SANCTIONS_SOURCE, record_type=ROPK_SANCTIONS_RECORD_TYPE, external_id="10", title=organization.name, status="active", payload={ "rn": "10", "uk_hm_treasury": True, "european_union": False, "united_states": False, "switzerland": False, "united_states_sectoral": False, "uk_uksl": False, "ukraine": False, }, ) self.client.force_authenticate(self.user) card = self.client.get( reverse( "api_v1:sources:source-cards-detail", kwargs={"slug": "ropk-sanctions"}, ) ) records = self.client.get( reverse("api_v2:organizations:organization-source-records-list"), { "source_group": "sanctions", "source": ROPK_SANCTIONS_SOURCE, "record_type": ROPK_SANCTIONS_RECORD_TYPE, "ordering": "payload__uk_hm_treasury", }, ) self.client.force_authenticate(self.admin) run = self.client.post( reverse( "api_v1:parsers:run-parser", args=[ROPK_SANCTIONS_SOURCE], ), {}, format="json", ) self.assertEqual(card.status_code, status.HTTP_200_OK) self.assertEqual(card.data["data"]["slug"], "ropk-sanctions") self.assertFalse(card.data["data"]["supports_refresh"]) self.assertEqual(card.data["data"]["upload_url"], self.upload_url) self.assertEqual(records.status_code, status.HTTP_200_OK) self.assertEqual(records.data["meta"]["pagination"]["total_count"], 1) self.assertEqual(records.data["data"][0]["payload"]["rn"], "10") self.assertIs(records.data["data"][0]["payload"]["uk_hm_treasury"], True) self.assertEqual(run.status_code, status.HTTP_405_METHOD_NOT_ALLOWED) self.assertEqual(run.data["errors"][0]["code"], "parser_run_not_supported") refresh = self.client.post( reverse( "api_v1:sources:source-cards-refresh", kwargs={"slug": "ropk-sanctions"}, ), {}, format="json", ) self.assertEqual(refresh.status_code, status.HTTP_405_METHOD_NOT_ALLOWED) self.assertEqual(refresh.data["errors"][0]["code"], "refresh_not_supported") def test_openapi_exposes_source_specific_upload_and_boolean_payload(self): response = self.client.get( reverse("schema-swagger-ui"), {"format": "openapi"}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) schema = response.content.decode("utf-8") self.assertIn("/api/v1/parsers/upload/ropk_sanctions/", schema) self.assertIn("XLSX-файл РОПК — Санкции", schema) self.assertIn("uk_hm_treasury", schema) self.assertIn("european_union", schema) class RopkSanctionsTaskTest(TestCase): def test_task_completes_job_and_reports_contract_counts(self): Organization.objects.create( name="АО Задача Санкции", inn="0012345678", ogrn="1027700132195", okpo="00123456", opk_registry_membership=True, directory_imported_at=timezone.now(), ) workbook = _workbook([_row("task-row")]) with TemporaryDirectory() as media_root, self.settings(MEDIA_ROOT=media_root): file_path = default_storage.save( "parser_uploads/sanctions.xlsx", ContentFile(workbook.getvalue()), ) task_result = parse_ropk_sanctions.apply( kwargs={ "file_path": file_path, "original_name": "sanctions.xlsx", }, task_id="ropk-sanctions-task-test", ) self.assertFalse(default_storage.exists(file_path)) self.assertTrue(task_result.successful()) self.assertEqual(task_result.result["raw_records_count"], 1) self.assertEqual(task_result.result["published_records_count"], 1) self.assertEqual(task_result.result["quarantine_records_count"], 0) self.assertIn("load_id", task_result.result) job = BackgroundJob.objects.get(task_id="ropk-sanctions-task-test") self.assertEqual(job.status, JobStatus.SUCCESS) self.assertEqual(job.result["published_records_count"], 1) self.assertEqual( ParserLoadLog.objects.get(source=ROPK_SANCTIONS_SOURCE).status, ParserLoadLog.Status.SUCCESS, )