avm/state-corp-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e09002d3af4d13bf2229283944a2bfb842bcee1b
state-corp-backend/tests/apps/exchange/test_api.py
Aleksandr Meshchriakov e09002d3af
All checks were successful
CI/CD Pipeline / Code Quality Checks (push) Successful in 2m47s
Details
CI/CD Pipeline / Run Tests (push) Successful in 2m49s
Details
CI/CD Pipeline / Build Docker Images (push) Successful in 41s
Details
CI/CD Pipeline / Push to Gitea Registry (push) Successful in 1s
Details
CI/CD Pipeline / Deploy to Server (push) Successful in 1s
Details
feat: import exchange registry memberships
2026-05-28 12:11:53 +02:00

497 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 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.registers.models import RegisterUpload, RegistryMembershipPeriod
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,
) -> SimpleUploadedFile:
"""Build encrypted exchange archive compatible with import service."""
payload = {
"format": ExchangePackageImportService.PAYLOAD_FORMAT,
"schema_version": 1,
"manifest": {
"package_id": package_id,
"source_system": "mostovik-dev",
"produced_at": "2026-04-07T12:00:00+00:00",
"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": 1,
"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": [
{
"inn": "7707083893",
"name": "АО Альфа Обновленная",
"short_name": "АО Альфа",
"organization_type": "ao",
"cluster": "radioelectronics",
"ogrn": "1027700132195",
"kpp": "770701001",
"okpo": "12345678",
"registration_date": "2024-02-15",
"legal_address": "г. Москва, ул. Тверская, д. 1",
"activity_type": "Производство электронных компонентов",
"founder_name": "Госкорпорация Пример",
"ownership_type": "Федеральная собственность",
"legal_form": "Акционерное общество",
"charter_capital_amount": "1500000.50",
"general_director_name": "Иванов Иван Иванович",
"general_director_inn": "123456789012",
"general_director_appointment_date": "2025-01-10",
"executors_count": 175,
"financial_reports_available": True,
"tax_reports_available": True,
"in_defense_unreliable_suppliers_registry": False,
"in_275_fz_registry": True,
"bankruptcy_messages_found": False,
},
{
"inn": "7707083894",
"name": "АО Бета",
"short_name": "АО Бета",
"organization_type": "pao",
"cluster": "space",
"ogrn": "1027700132196",
"kpp": "770701002",
"okpo": "12345679",
"registration_date": "2023-09-01",
},
],
"industrial_products": [
{
"organization_inn": "7707083893",
"product_name": "Система связи М-1",
"product_class": "Связь",
"okpd2_code": "26.30.11",
"tnved_code": "8517620000",
"registry_number": "prod-001",
}
],
"registry_memberships": [
{
"organization_inn": "7707083893",
"registry_name": "Реестр госкорпорации Росатом",
"started_at": "2026-01-01",
"ended_at": None,
},
{
"organization_inn": "7707083894",
"registry_name": "Реестр госкорпорации Роскосмос",
"started_at": "2026-02-01",
"ended_at": None,
},
],
"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")
password = get_random_string(16)
self.user = User.objects.create_user(
username="exchange-admin",
email="exchange@example.com",
password=password,
)
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.assertEqual(
response.data["result"]["registry_memberships"]["opened"],
2,
)
self.assertEqual(Organization.objects.count(), 2)
self.assertEqual(RegisterUpload.objects.count(), 2)
self.assertEqual(RegistryMembershipPeriod.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.short_name, "АО Альфа")
self.assertEqual(organization.organization_type, "ao")
self.assertEqual(organization.cluster, "radioelectronics")
self.assertEqual(organization.registration_date, date(2024, 2, 15))
self.assertEqual(organization.executors_count, 175)
self.assertTrue(organization.tax_reports_available)
self.assertTrue(organization.in_275_fz_registry)
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_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)
Reference in New Issue View Git Blame Copy Permalink