fix: harden state corp exchange imports

tests/apps/core/test_startup_checks.py

@@ -0,0 +1,38 @@
+"""Tests for fail-fast dependency checks."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from apps.core.startup_checks import _check_db
+from django.test import SimpleTestCase
+
+
+class DatabaseStartupCheckTest(SimpleTestCase):
+    """Require a Unicode-safe PostgreSQL database before startup."""
+
+    @staticmethod
+    def _connection_with_encoding(encoding: str) -> MagicMock:
+        connection = MagicMock()
+        cursor = connection.cursor.return_value.__enter__.return_value
+        cursor.fetchone.side_effect = [(1,), (encoding,)]
+        return connection
+
+    @patch("apps.core.startup_checks.psycopg2.connect")
+    def test_check_db_accepts_utf8_database(self, connect):
+        connect.return_value = self._connection_with_encoding("UTF8")
+
+        ok, message = _check_db(timeout_seconds=3)
+
+        self.assertTrue(ok)
+        self.assertEqual(message, "OK")
+
+    @patch("apps.core.startup_checks.psycopg2.connect")
+    def test_check_db_rejects_non_utf8_database(self, connect):
+        connect.return_value = self._connection_with_encoding("SQL_ASCII")
+
+        ok, message = _check_db(timeout_seconds=3)
+
+        self.assertFalse(ok)
+        self.assertIn("database encoding SQL_ASCII is unsupported", message)
+        self.assertIn("UTF8 is required", message)

tests/apps/exchange/test_api.py

@@ -11,6 +11,7 @@ import zlib
 from datetime import date
 from decimal import Decimal
 from io import BytesIO
+from unittest.mock import patch
 from zipfile import ZIP_DEFLATED, ZipFile
 
 from apps.exchange.models import ExchangeDeliveryChannel, ExchangePackageImport
@@ -35,6 +36,7 @@ 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.db import models
 from django.urls import reverse
 from django.utils.crypto import get_random_string
 from rest_framework import status
@@ -531,6 +533,71 @@ class ExchangePackageApiTest(APITestCase):
         self.assertEqual(ExchangePackageImport.objects.count(), 0)
         self.assertEqual(Organization.objects.count(), 0)
 
+    def test_upload_logs_unexpected_import_error_with_package_context(self):
+        archive = build_exchange_archive(
+            package_id="pkg-unexpected-import-error",
+            data=build_exchange_payload(),
+        )
+
+        with patch.object(
+            ExchangePackageImportService,
+            "_import_payload",
+            side_effect=RuntimeError("database failure"),
+        ), patch("apps.exchange.services.logger.exception") as log_exception:
+            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.assertEqual(
+            response.data,
+            {"file": ["Не удалось импортировать пакет обмена"]},
+        )
+        package_import = ExchangePackageImport.objects.get(
+            package_id="pkg-unexpected-import-error"
+        )
+        log_exception.assert_called_once_with(
+            "Unexpected exchange package import failure "
+            "(package_id=%s, package_hash=%s, delivery_channel=%s)",
+            "pkg-unexpected-import-error",
+            package_import.package_hash,
+            ExchangeDeliveryChannel.API,
+        )
+
+    def test_upload_preserves_long_public_procurement_name(self):
+        payload = build_exchange_payload()
+        long_purchase_name = (
+            "Поставка специализированного оборудования " * 16
+        ).strip()
+        payload["public_procurements"][0]["purchase_name"] = long_purchase_name
+        archive = build_exchange_archive(
+            package_id="pkg-long-public-procurement-name",
+            data=payload,
+        )
+
+        response = self.client.post(
+            self.url,
+            {"file": archive},
+            format="multipart",
+            HTTP_X_EXCHANGE_TOKEN=TEST_TOKEN,
+        )
+
+        self.assertGreater(len(long_purchase_name), 500)
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(
+            PublicProcurement.objects.get(
+                purchase_number="purchase-001"
+            ).purchase_name,
+            long_purchase_name,
+        )
+        self.assertIsInstance(
+            PublicProcurement._meta.get_field("purchase_name"),
+            models.TextField,
+        )
+
     def test_upload_rejects_legacy_schema_version(self):
         archive = build_exchange_archive(
             package_id="pkg-legacy-schema-version",