Aleksandr Meshchriakov
bd8e1a8400
All checks were successful
CI/CD Pipeline / Code Quality Checks (push) Successful in 2m41s
CI/CD Pipeline / Run Tests (push) Successful in 2m47s
CI/CD Pipeline / Build Docker Images (push) Successful in 2m24s
CI/CD Pipeline / Push to Gitea Registry (push) Successful in 0s
CI/CD Pipeline / Deploy to Server (push) Successful in 0s
381 lines
14 KiB
Python
381 lines
14 KiB
Python
"""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,
|
||
IndustrialProduct,
|
||
InformationSecurityRegistryEntry,
|
||
LaborVacancy,
|
||
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,
|
||
) -> 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",
|
||
}
|
||
],
|
||
"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": "Поставка специализированного оборудования",
|
||
}
|
||
],
|
||
"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(Organization.objects.count(), 2)
|
||
self.assertEqual(IndustrialProduct.objects.count(), 1)
|
||
self.assertEqual(ProsecutorCheck.objects.count(), 1)
|
||
self.assertEqual(PublicProcurement.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"]["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)
|
||
|
||
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_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)
|