"""Integration tests for FNS upload flow (no mocks).""" import io import os import tempfile import time from apps.parsers.models import FinancialReport, FinancialReportLine from django.core.files.uploadedfile import SimpleUploadedFile from django.test import override_settings from django.urls import reverse from openpyxl import Workbook from rest_framework import status from rest_framework.test import APITestCase from tests.apps.user.factories import UserFactory from tests.utils.fixtures import fake def _digits(length: int) -> str: return "".join(str(fake.random_int(0, 9)) for _ in range(length)) def _build_fns_excel_bytes() -> bytes: wb = Workbook() ws = wb.active year = fake.random_int(min=2020, max=2025) ws.append(["Форма №1", None, year, None]) ws.append([None, "Код", "Начало", "Конец"]) ws.append([fake.word(), _digits(4), fake.random_int(10, 999), fake.random_int(10, 999)]) buf = io.BytesIO() wb.save(buf) wb.close() return buf.getvalue() class FNSUploadIntegrationTest(APITestCase): """Tests real upload + processing of FNS files.""" def setUp(self): self.user = UserFactory.create_user() self.client.force_authenticate(self.user) self.upload_url = reverse("api_v1:fns:fns-upload") def _dirs(self, base_dir: str) -> tuple[str, str, str]: watch_dir = os.path.join(base_dir, "watch") processed_dir = os.path.join(base_dir, "processed") failed_dir = os.path.join(base_dir, "failed") return watch_dir, processed_dir, failed_dir def test_upload_processes_file_and_moves_to_processed(self): content = _build_fns_excel_bytes() external_id = _digits(5) ogrn = _digits(13) filename = f"fin_{external_id}_{ogrn}.xlsx" upload = SimpleUploadedFile( filename, content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) with tempfile.TemporaryDirectory() as tmpdir: watch_dir, processed_dir, failed_dir = self._dirs(tmpdir) with override_settings( FNS_WATCH_DIRECTORY=watch_dir, FNS_PROCESSED_DIRECTORY=processed_dir, FNS_FAILED_DIRECTORY=failed_dir, ): response = self.client.post( self.upload_url, {"files": [upload]}, format="multipart" ) self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) self.assertEqual(response.data["queued"], 1) self.assertEqual(response.data["skipped"], 0) self.assertEqual(FinancialReport.objects.count(), 1) report = FinancialReport.objects.first() self.assertEqual(report.external_id, external_id) self.assertEqual(report.ogrn, ogrn) self.assertTrue( FinancialReportLine.objects.filter(report=report).exists() ) processed_path = os.path.join(processed_dir, filename) self.assertTrue(os.path.exists(processed_path)) self.assertFalse(os.path.exists(os.path.join(watch_dir, filename))) self.assertFalse( os.path.exists(os.path.join(watch_dir, f"{filename}.lock")) ) def test_upload_duplicate_is_skipped(self): content = _build_fns_excel_bytes() external_id = _digits(3) ogrn = _digits(13) filename = f"fin_{external_id}_{ogrn}.xlsx" upload1 = SimpleUploadedFile( filename, content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) upload2 = SimpleUploadedFile( filename, content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) with tempfile.TemporaryDirectory() as tmpdir: watch_dir, processed_dir, failed_dir = self._dirs(tmpdir) with override_settings( FNS_WATCH_DIRECTORY=watch_dir, FNS_PROCESSED_DIRECTORY=processed_dir, FNS_FAILED_DIRECTORY=failed_dir, ): first = self.client.post( self.upload_url, {"files": [upload1]}, format="multipart" ) second = self.client.post( self.upload_url, {"files": [upload2]}, format="multipart" ) self.assertEqual(first.status_code, status.HTTP_202_ACCEPTED) self.assertEqual(second.status_code, status.HTTP_202_ACCEPTED) self.assertEqual(second.data["queued"], 0) self.assertEqual(second.data["skipped"], 1) self.assertEqual(FinancialReport.objects.count(), 1) self.assertFalse(os.path.exists(os.path.join(watch_dir, filename))) self.assertFalse( os.path.exists(os.path.join(watch_dir, f"{filename}.lock")) ) def test_upload_invalid_filename_rejected(self): content = _build_fns_excel_bytes() upload = SimpleUploadedFile( f"{fake.word()}_{fake.random_int()}.xlsx", content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) with tempfile.TemporaryDirectory() as tmpdir: watch_dir, processed_dir, failed_dir = self._dirs(tmpdir) with override_settings( FNS_WATCH_DIRECTORY=watch_dir, FNS_PROCESSED_DIRECTORY=processed_dir, FNS_FAILED_DIRECTORY=failed_dir, ): response = self.client.post( self.upload_url, {"files": [upload]}, format="multipart" ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(FinancialReport.objects.count(), 0) def test_upload_skips_when_lock_is_fresh(self): content = _build_fns_excel_bytes() external_id = _digits(5) ogrn = _digits(13) filename = f"fin_{external_id}_{ogrn}.xlsx" upload = SimpleUploadedFile( filename, content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) with tempfile.TemporaryDirectory() as tmpdir: watch_dir, processed_dir, failed_dir = self._dirs(tmpdir) os.makedirs(watch_dir, exist_ok=True) lock_path = os.path.join(watch_dir, f"{filename}.lock") with open(lock_path, "w") as handle: handle.write("lock") now = time.time() os.utime(lock_path, (now, now)) with override_settings( FNS_WATCH_DIRECTORY=watch_dir, FNS_PROCESSED_DIRECTORY=processed_dir, FNS_FAILED_DIRECTORY=failed_dir, FNS_LOCK_TTL_SECONDS=3600, ): response = self.client.post( self.upload_url, {"files": [upload]}, format="multipart" ) self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) self.assertEqual(response.data["queued"], 0) self.assertEqual(response.data["skipped"], 1) def test_upload_skips_when_file_already_exists(self): content = _build_fns_excel_bytes() external_id = _digits(5) ogrn = _digits(13) filename = f"fin_{external_id}_{ogrn}.xlsx" upload = SimpleUploadedFile( filename, content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) with tempfile.TemporaryDirectory() as tmpdir: watch_dir, processed_dir, failed_dir = self._dirs(tmpdir) os.makedirs(watch_dir, exist_ok=True) existing_path = os.path.join(watch_dir, filename) with open(existing_path, "wb") as handle: handle.write(b"existing") with override_settings( FNS_WATCH_DIRECTORY=watch_dir, FNS_PROCESSED_DIRECTORY=processed_dir, FNS_FAILED_DIRECTORY=failed_dir, ): response = self.client.post( self.upload_url, {"files": [upload]}, format="multipart" ) self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) self.assertEqual(response.data["queued"], 0) self.assertEqual(response.data["skipped"], 1)