"""Tests for encrypted exchange package upload and import.""" from __future__ import annotations import base64 import hashlib import json import struct import tempfile import zlib from datetime import date from decimal import Decimal from io import BytesIO from zipfile import ZIP_DEFLATED, ZipFile from apps.exchange.models import ExchangeDeliveryChannel, ExchangePackageImport from apps.exchange.services import ExchangePackageImportService from apps.external_data.models import ( ArbitrationCase, BankruptcyProcedure, DefenseUnreliableSupplier, FinancialReport, FinancialReportLine, IndustrialCertificate, IndustrialProduct, InformationSecurityRegistryEntry, LaborVacancy, ManufacturerRegistryEntry, ProsecutorCheck, PublicProcurement, ) from apps.organization.models import Organization from apps.user.models import User from cryptography.hazmat.primitives.ciphers.aead import AESGCM from django.conf import settings from django.core.files.uploadedfile import SimpleUploadedFile from django.core.management import call_command from django.urls import reverse from django.utils.crypto import get_random_string from rest_framework import status from rest_framework.test import APITestCase TEST_TOKEN = settings.EXCHANGE_SHARED_TOKEN def _b64url(data: bytes) -> str: return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=") def build_exchange_archive( *, package_id: str = "pkg-20260407-001", archive_name: str = "exchange_package_20260407.zip", bin_name: str = "exchange_package_20260407.bin", data: dict[str, list[dict[str, object]]] | None = None, schema_version: int = ExchangePackageImportService.SUPPORTED_SCHEMA_VERSION, ) -> SimpleUploadedFile: """Build encrypted exchange archive compatible with import service.""" payload = { "format": ExchangePackageImportService.PAYLOAD_FORMAT, "schema_version": schema_version, "manifest": { "package_id": package_id, "source_system": "mostovik-dev", "produced_at": "2026-04-07T12:00:00+00:00", "schema_version": schema_version, "sections": list((data or {}).keys()), }, "data": data or {}, } payload_bytes = json.dumps( payload, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") compressed_payload = zlib.compress(payload_bytes, level=9) nonce = b"sc-exch-0001" aad = ExchangePackageImportService.AAD raw_key = hashlib.sha256(TEST_TOKEN.encode("utf-8")).digest() encrypted_payload = AESGCM(raw_key).encrypt(nonce, compressed_payload, aad) header = { "format": ExchangePackageImportService.BIN_FORMAT, "version": 1, "key_id": "test-shared-token", "nonce": _b64url(nonce), "aad": _b64url(aad), "package_id": package_id, "schema_version": schema_version, "plaintext_sha256": hashlib.sha256(payload_bytes).hexdigest(), "compressed_sha256": hashlib.sha256(compressed_payload).hexdigest(), "ciphertext_sha256": hashlib.sha256(encrypted_payload).hexdigest(), } header_bytes = json.dumps( header, ensure_ascii=False, separators=(",", ":"), ).encode("utf-8") bin_bytes = ( ExchangePackageImportService.MAGIC + bytes([1]) + struct.pack(">I", len(header_bytes)) + header_bytes + encrypted_payload ) archive_bytes = BytesIO() with ZipFile(archive_bytes, "w", compression=ZIP_DEFLATED) as archive: archive.writestr(bin_name, bin_bytes) archive.writestr( f"{bin_name}.sha256", f"{hashlib.sha256(bin_bytes).hexdigest()} {bin_name}\n", ) return SimpleUploadedFile( archive_name, archive_bytes.getvalue(), content_type="application/zip", ) def build_exchange_payload() -> dict[str, list[dict[str, object]]]: """Create a representative payload for import tests.""" return { "organizations": [ { "mostovik_uid": "11111111-1111-4111-8111-111111111111", "rn": 1001, "inn": "7707083893", "name": "АО Альфа Обновленная", "full_name": "Акционерное общество Альфа Обновленная", "short_name": "АО Альфа", "pn_name": "АО Альфа", "pn_name_en": "Alpha JSC", "ogrn": "1027700132195", "kpp": "770701001", "okpo": "12345678", "ogrip": "", "identity_status": "complete", "primary_identity": "inn:7707083893|ogrn:1027700132195|kpp:770701001", "gk_code": "1", "gk_name": "Росатом", "in_korp_code": "head", "in_korp_name": "Головная организация", "filial": ".F.", "is_branch": False, "registration_date": "2024-02-15", "create_date": "2024", "organizational_legal_form": "12267", "organizational_legal_form1": "Акционерное общество", "ownership_form": "61", "ownership_form1": "Федеральная собственность", "authorized_capital": "1500000.50", "legal_address": "г. Москва, ул. Тверская, д. 1", "business_act_cod": "26.11", "business_activity": "Производство электронных компонентов", "general_director": "Иванов Иван Иванович", "general_director_tax_id": "123456789012", "uk": "АО Управляющая компания", "inn_uk": "7707000000", "appointment_date": "2025-01-10", "cf_fl_rn": "director-001", "akc_fs": "100", "akc_sf": "0", "re_za": True, "re_zasf": False, "goz_participation": True, "opk_registry_membership": True, "ropk_num": "РОПК-001", "ropk_razdel_num": "opk-1", "ropk_razdel_name": "Росатом ОПК", "min": "Минпромторг", "dep": "Департамент радиоэлектроники", "otr": "Радиоэлектроника", "integrated_structure": "Интегрированная структура Альфа", "state_sector_code": "10", "state_sector_name": "Государственный сектор", }, { "mostovik_uid": "22222222-2222-4222-8222-222222222222", "rn": 1002, "inn": "7707083894", "name": "АО Бета", "full_name": "Акционерное общество Бета", "short_name": "АО Бета", "pn_name": "АО Бета", "pn_name_en": "Beta JSC", "ogrn": "1027700132196", "kpp": "770701002", "okpo": "12345679", "ogrip": "", "identity_status": "complete", "primary_identity": "inn:7707083894|ogrn:1027700132196|kpp:770701002", "gk_code": "2", "gk_name": "Роскосмос", "in_korp_code": "member", "in_korp_name": "Организация внутри ГК", "filial": ".F.", "is_branch": False, "registration_date": "2023-09-01", "create_date": "2023", "organizational_legal_form": "12267", "organizational_legal_form1": "Акционерное общество", "ownership_form": "61", "ownership_form1": "Федеральная собственность", "authorized_capital": None, "legal_address": "", "business_act_cod": "", "business_activity": "", "general_director": "", "general_director_tax_id": "", "uk": "", "inn_uk": "", "appointment_date": None, "cf_fl_rn": "", "akc_fs": "", "akc_sf": "", "re_za": None, "re_zasf": None, "goz_participation": False, "opk_registry_membership": False, "ropk_num": "", "ropk_razdel_num": "", "ropk_razdel_name": "", "min": "", "dep": "", "otr": "", "integrated_structure": "", "state_sector_code": "", "state_sector_name": "", }, ], "industrial_products": [ { "organization_inn": "7707083893", "product_name": "Система связи М-1", "product_class": "Связь", "okpd2_code": "26.30.11", "tnved_code": "8517620000", "registry_number": "prod-001", } ], "industrial_certificates": [ { "organization_inn": "7707083893", "certificate_number": "CERT-001", "issue_date": "2026-01-10", "expiry_date": "2027-01-10", "certificate_file_url": "https://minpromtorg.gov.ru/cert/001", "organisation_name": "АО Альфа Обновленная", "ogrn": "1027700132195", } ], "manufacturers": [ { "organization_inn": "7707083893", "full_legal_name": "АО Альфа Обновленная", "inn": "7707083893", "ogrn": "1027700132195", "address": "г. Москва, ул. Тверская, д. 1", } ], "prosecutor_checks": [ { "organization_inn": "7707083893", "registration_number": "check-001", "law_type": "294-ФЗ", "control_authority": "Минпромторг", "prosecutor_office": "Прокуратура г. Москвы", "start_date": "2026-03-10", "status": "active", } ], "public_procurements": [ { "organization_inn": "7707083893", "purchase_number": "purchase-001", "law_type": "223-ФЗ", "status": "executing", "contract_amount": "4500000.75", "contract_date": "2026-02-15", "execution_start_date": "2026-02-20", "execution_end_date": "2026-11-30", "purchase_name": "Поставка специализированного оборудования", } ], "financial_reports": [ { "organization_inn": "7707083893", "external_id": "fin-001", "ogrn": "1027700132195", "file_name": "fin_001_1027700132195.xlsx", "file_hash": "f" * 64, "load_batch": 7, "status": "success", "source": "api", "lines": [ { "form_code": "1", "line_code": "1600", "line_name": "Баланс", "year": 2025, "period_start": 1000, "period_end": 1500, } ], } ], "arbitration_cases": [ { "organization_inn": "7707083893", "case_number": "А40-12345/2026", "court_name": "Арбитражный суд города Москвы", "party_role": "ответчик", "status": "in_progress", "decision_date": "2026-03-25", } ], "bankruptcy_procedures": [ { "organization_inn": "7707083893", "external_id": "fedresurs:001", "message_type": "Сообщение о намерении", "message_date": "2026-03-26", "case_number": "А40-555/2026", "status": "published", "source_url": "https://fedresurs.ru/message/001", } ], "defense_unreliable_suppliers": [ { "organization_inn": "7707083893", "external_id": "fas-goz:001", "registry_source": "fas_goz", "registry_number": "ГОЗ-001", "supplier_name": "АО Альфа Обновленная", "reason": "Уклонение от заключения контракта", "included_at": "2026-02-20", "status": "active", "source_url": "https://fas.gov.ru/register/001", } ], "information_security_registries": [ { "organization_inn": "7707083893", "external_id": "fstec:001", "registry_name": "Реестр лицензий ФСТЭК", "presence_status": "present", "entry_number": "77-001234", "issued_at": "2026-01-10", "expires_at": "2027-01-10", } ], "labor_vacancies": [ { "organization_inn": "7707083893", "external_id": "trudvsem:001", "vacancy_source": "trudvsem", "title": "Инженер-испытатель", "status": "open", "published_at": "2026-04-01", "salary_amount": "175000.00", "source_url": "https://trudvsem.ru/vacancy/001", } ], } class ExchangePackageApiTest(APITestCase): """Integration tests for exchange API and CLI import.""" def setUp(self): self.url = reverse("api_v1:exchange:package-upload") self.latest_url = reverse("api_v1:exchange:package-latest") password = get_random_string(16) self.user = User.objects.create_user( username="exchange-admin", email="exchange@example.com", password=password, ) def test_latest_requires_authentication(self): response = self.client.get(self.latest_url) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_latest_returns_null_when_no_packages_exist(self): self.client.force_authenticate(self.user) response = self.client.get(self.latest_url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual( response.data, {"success": True, "data": None, "errors": None, "meta": None}, ) def test_latest_returns_newest_package_metadata(self): older = ExchangePackageImport.objects.create( package_id="older-package", package_name="older.zip", package_hash="a" * 64, source_system="mostovik", schema_version=1, delivery_channel=ExchangeDeliveryChannel.API, status="failed", ) latest = ExchangePackageImport.objects.create( package_id="latest-package", package_name="latest.zip", package_hash="b" * 64, source_system="mostovik", schema_version=1, delivery_channel=ExchangeDeliveryChannel.API, status="success", ) ExchangePackageImport.objects.filter(pk=older.pk).update( created_at=older.created_at.replace(year=older.created_at.year - 1) ) self.client.force_authenticate(self.user) response = self.client.get(self.latest_url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["data"]["id"], str(latest.id)) self.assertEqual(response.data["data"]["package_id"], "latest-package") self.assertEqual(response.data["data"]["package_name"], "latest.zip") self.assertEqual(response.data["data"]["status"], "success") self.assertIn("created_at", response.data["data"]) def test_upload_imports_package_and_upserts_models(self): Organization.objects.create( inn="7707083893", name="АО Альфа", ogrn="1027700132000", kpp="770701000", ) archive = build_exchange_archive(data=build_exchange_payload()) response = self.client.post( self.url, {"file": archive}, format="multipart", HTTP_X_EXCHANGE_TOKEN=TEST_TOKEN, ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertFalse(response.data["result"]["duplicate"]) self.assertEqual(response.data["result"]["organizations"]["created"], 1) self.assertEqual(response.data["result"]["organizations"]["updated"], 1) self.assertNotIn("registry_memberships", response.data["result"]) self.assertEqual(Organization.objects.count(), 2) self.assertEqual(IndustrialCertificate.objects.count(), 1) self.assertEqual(ManufacturerRegistryEntry.objects.count(), 1) self.assertEqual(IndustrialProduct.objects.count(), 1) self.assertEqual(ProsecutorCheck.objects.count(), 1) self.assertEqual(PublicProcurement.objects.count(), 1) self.assertEqual(FinancialReport.objects.count(), 1) self.assertEqual(FinancialReportLine.objects.count(), 1) self.assertEqual(ArbitrationCase.objects.count(), 1) self.assertEqual(BankruptcyProcedure.objects.count(), 1) self.assertEqual(DefenseUnreliableSupplier.objects.count(), 1) self.assertEqual(InformationSecurityRegistryEntry.objects.count(), 1) self.assertEqual(LaborVacancy.objects.count(), 1) self.assertEqual( response.data["result"]["bankruptcy_procedures"]["created"], 1, ) self.assertEqual( response.data["result"]["financial_reports"]["created_lines"], 1, ) self.assertEqual( response.data["result"]["defense_unreliable_suppliers"]["created"], 1, ) self.assertEqual( response.data["result"]["information_security_registries"]["created"], 1, ) self.assertEqual(response.data["result"]["labor_vacancies"]["created"], 1) organization = Organization.objects.get(inn="7707083893") self.assertEqual(organization.name, "АО Альфа Обновленная") self.assertEqual( organization.full_name, "Акционерное общество Альфа Обновленная" ) self.assertEqual(organization.short_name, "АО Альфа") self.assertEqual( str(organization.mostovik_uid), "11111111-1111-4111-8111-111111111111" ) self.assertEqual(organization.rn, 1001) self.assertEqual(organization.gk_name, "Росатом") self.assertEqual(organization.in_korp_name, "Головная организация") self.assertEqual(organization.registration_date, date(2024, 2, 15)) self.assertEqual(organization.appointment_date, date(2025, 1, 10)) self.assertEqual( organization.business_activity, "Производство электронных компонентов" ) self.assertEqual( organization.organizational_legal_form1, "Акционерное общество" ) self.assertEqual(organization.ownership_form1, "Федеральная собственность") self.assertEqual(organization.authorized_capital, Decimal("1500000.50")) self.assertEqual(organization.general_director, "Иванов Иван Иванович") self.assertEqual(organization.general_director_tax_id, "123456789012") self.assertTrue(organization.goz_participation) self.assertTrue(organization.opk_registry_membership) self.assertEqual(organization.ropk_razdel_name, "Росатом ОПК") self.assertEqual( organization.get_active_registry_names(), ["Росатом", "Участие в ГОЗ", "Росатом ОПК"], ) package_import = ExchangePackageImport.objects.get() self.assertEqual(package_import.delivery_channel, ExchangeDeliveryChannel.API) self.assertEqual(package_import.status, "success") def test_upload_rejects_invalid_exchange_token(self): archive = build_exchange_archive(data=build_exchange_payload()) invalid_token = get_random_string(24) response = self.client.post( self.url, {"file": archive}, format="multipart", HTTP_X_EXCHANGE_TOKEN=invalid_token, ) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) self.assertEqual(ExchangePackageImport.objects.count(), 0) self.assertEqual(Organization.objects.count(), 0) def test_upload_rejects_legacy_schema_version(self): archive = build_exchange_archive( package_id="pkg-legacy-schema-version", schema_version=2, data=build_exchange_payload(), ) response = self.client.post( self.url, {"file": archive}, format="multipart", HTTP_X_EXCHANGE_TOKEN=TEST_TOKEN, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("schema_version 3", response.data["file"][0]) self.assertEqual(Organization.objects.count(), 0) def test_upload_rejects_legacy_registry_memberships_section(self): payload = build_exchange_payload() payload["registry_memberships"] = [ { "organization_inn": "7707083893", "registry_name": "Реестр госкорпорации Росатом", "started_at": "2026-01-01", "ended_at": None, } ] archive = build_exchange_archive( package_id="pkg-legacy-registry-memberships", data=payload, ) response = self.client.post( self.url, {"file": archive}, format="multipart", HTTP_X_EXCHANGE_TOKEN=TEST_TOKEN, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("registry_memberships", response.data["file"][0]) self.assertEqual(Organization.objects.count(), 0) def test_upload_rejects_legacy_organization_row_schema(self): archive = build_exchange_archive( package_id="pkg-legacy-organization-row", data={ "organizations": [ { "inn": "7707083893", "name": "АО Альфа", "ogrn": "1027700132195", } ], }, ) response = self.client.post( self.url, {"file": archive}, format="multipart", HTTP_X_EXCHANGE_TOKEN=TEST_TOKEN, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("schema_version 3", response.data["file"][0]) self.assertEqual(Organization.objects.count(), 0) def test_upload_rejects_external_rows_for_organization_absent_from_package(self): Organization.objects.create( inn="7707083893", name="АО Альфа", ogrn="1027700132195", kpp="770701001", ) archive = build_exchange_archive( package_id="pkg-missing-package-organization", data={ "organizations": [], "industrial_products": [ { "organization_inn": "7707083893", "product_name": "Система связи М-1", "registry_number": "prod-001", } ], }, ) response = self.client.post( self.url, {"file": archive}, format="multipart", HTTP_X_EXCHANGE_TOKEN=TEST_TOKEN, ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("отсутствует в разделе organizations", response.data["file"][0]) self.assertEqual(Organization.objects.count(), 1) self.assertEqual(IndustrialProduct.objects.count(), 0) package_import = ExchangePackageImport.objects.get( package_id="pkg-missing-package-organization" ) self.assertEqual(package_import.status, "failed") def test_upload_is_idempotent_for_duplicate_package(self): archive = build_exchange_archive( package_id="pkg-duplicate-001", data=build_exchange_payload(), ) first_response = self.client.post( self.url, {"file": archive}, format="multipart", HTTP_X_EXCHANGE_TOKEN=TEST_TOKEN, ) self.assertEqual(first_response.status_code, status.HTTP_201_CREATED) second_archive = build_exchange_archive( package_id="pkg-duplicate-001", data=build_exchange_payload(), ) second_response = self.client.post( self.url, {"file": second_archive}, format="multipart", HTTP_X_EXCHANGE_TOKEN=TEST_TOKEN, ) self.assertEqual(second_response.status_code, status.HTTP_201_CREATED) self.assertTrue(second_response.data["result"]["duplicate"]) self.assertEqual(Organization.objects.count(), 2) self.assertEqual(IndustrialProduct.objects.count(), 1) self.assertEqual(ExchangePackageImport.objects.count(), 2) duplicate_import = ExchangePackageImport.objects.order_by("-created_at").first() self.assertIsNotNone(duplicate_import.duplicate_of) def test_cli_import_uses_same_pipeline(self): archive = build_exchange_archive( package_id="pkg-cli-001", data=build_exchange_payload(), ) with tempfile.NamedTemporaryFile(suffix=".zip") as temp_file: temp_file.write(archive.read()) temp_file.flush() call_command("import_exchange_package", temp_file.name, "--channel=cli") imported = ExchangePackageImport.objects.get(package_id="pkg-cli-001") self.assertEqual(imported.delivery_channel, ExchangeDeliveryChannel.CLI) self.assertEqual(Organization.objects.count(), 2)