"""Integration tests for parsers API views (no mocks).""" from __future__ import annotations import io import os import tempfile from unittest.mock import Mock, patch from apps.parsers.models import FinancialReport, FinancialReportLine, ProcurementRecord from django.core.files.uploadedfile import SimpleUploadedFile from django.urls import reverse from openpyxl import Workbook from rest_framework import status from rest_framework.test import APITestCase from tests.apps.parsers.factories import ( IndustrialCertificateRecordFactory, IndustrialProductRecordFactory, InspectionRecordFactory, ManufacturerRecordFactory, ParserLoadLogFactory, ) 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(["Form", None, year, None]) ws.append([None, "Code", "Start", "End"]) 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() def _create_procurement_record() -> ProcurementRecord: return ProcurementRecord.objects.create( load_batch=fake.random_int(min=1, max=1000), purchase_number=_digits(19), purchase_name=fake.sentence(nb_words=6), customer_inn=_digits(10), customer_kpp=_digits(9), customer_ogrn=_digits(13), customer_name=fake.company(), max_price=str(fake.pydecimal(left_digits=7, right_digits=2, positive=True)), status=fake.word(), law_type="44-FZ", href=fake.url(), region_code=f"{fake.random_int(min=1, max=99):02d}", ) class ParsersViewSetTest(APITestCase): def setUp(self): self.user = UserFactory.create_user() self.admin = UserFactory.create_superuser() def test_certificates_list_and_retrieve(self): record = IndustrialCertificateRecordFactory() self.client.force_authenticate(self.user) url = reverse("api_v1:minpromtorg:certificates-list") response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) detail = self.client.get( reverse("api_v1:minpromtorg:certificates-detail", args=[record.id]) ) self.assertEqual(detail.status_code, status.HTTP_200_OK) def test_manufacturers_list_and_retrieve(self): record = ManufacturerRecordFactory() self.client.force_authenticate(self.user) url = reverse("api_v1:minpromtorg:manufacturers-list") response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) detail = self.client.get( reverse("api_v1:minpromtorg:manufacturers-detail", args=[record.id]) ) self.assertEqual(detail.status_code, status.HTTP_200_OK) def test_products_list_and_retrieve(self): record = IndustrialProductRecordFactory() self.client.force_authenticate(self.user) url = reverse("api_v1:minpromtorg:industrial-products-list") response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) detail = self.client.get( reverse("api_v1:minpromtorg:industrial-products-detail", args=[record.id]) ) self.assertEqual(detail.status_code, status.HTTP_200_OK) def test_inspections_list_and_retrieve(self): record = InspectionRecordFactory() self.client.force_authenticate(self.user) url = reverse("api_v1:proverki:inspections-list") response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) detail = self.client.get( reverse("api_v1:proverki:inspections-detail", args=[record.id]) ) self.assertEqual(detail.status_code, status.HTTP_200_OK) def test_procurements_list_and_retrieve(self): record = _create_procurement_record() self.client.force_authenticate(self.user) url = reverse("api_v1:zakupki:procurements-list") response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) detail = self.client.get( reverse("api_v1:zakupki:procurements-detail", args=[record.id]) ) self.assertEqual(detail.status_code, status.HTTP_200_OK) def test_dashboard_data_exposes_source_groups_for_page(self): self.client.force_authenticate(self.user) response = self.client.get("/api/v1/parsers/dashboard/") self.assertEqual(response.status_code, status.HTTP_200_OK) payload = response.data["data"] self.assertIn("groups", payload) self.assertIn("api", payload["groups"]) self.assertIn("uploads", payload["groups"]) self.assertEqual(payload["api_sources"], payload["groups"]["api"]) self.assertEqual(payload["file_sources"], payload["groups"]["uploads"]) def test_financial_reports_list_and_retrieve(self): report = FinancialReport.objects.create( external_id=_digits(5), ogrn=_digits(13), file_name=f"fin_{_digits(5)}_{_digits(13)}.xlsx", file_hash=fake.sha256(raw_output=False), load_batch=fake.random_int(min=1, max=1000), status=FinancialReport.Status.SUCCESS, source=FinancialReport.SourceType.API, ) FinancialReportLine.objects.create( report=report, form_code="1", line_code=_digits(4), line_name=fake.word(), year=fake.random_int(min=2020, max=2025), period_start=fake.random_int(min=1, max=999), period_end=fake.random_int(min=1, max=999), ) self.client.force_authenticate(self.user) url = reverse("api_v1:fns:fns-reports-list") response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["data"][0]["lines_count"], 1) detail = self.client.get( reverse("api_v1:fns:fns-reports-detail", args=[report.id]) ) self.assertEqual(detail.status_code, status.HTTP_200_OK) self.assertIn("lines", detail.data) def test_system_logs_admin_only(self): log = ParserLoadLogFactory() url_logs = reverse("api_v1:system:parser-logs-list") response = self.client.get(url_logs) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) self.client.force_authenticate(self.user) response = self.client.get(url_logs) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.client.force_authenticate(self.admin) response = self.client.get(url_logs) self.assertEqual(response.status_code, status.HTTP_200_OK) detail = self.client.get( reverse("api_v1:system:parser-logs-detail", args=[log.id]) ) self.assertEqual(detail.status_code, status.HTTP_200_OK) def test_system_logs_support_search_and_organizations_count(self): search_marker = "manufactures-unique-search-marker" first_log = ParserLoadLogFactory( source="manufactures", batch_id=101, status="success", error_message=search_marker, ) ParserLoadLogFactory( source="inspections", batch_id=202, status="failed", error_message="timeout", ) ManufacturerRecordFactory(load_batch=101, inn="7701000001") ManufacturerRecordFactory(load_batch=101, inn="7701000002") self.client.force_authenticate(self.admin) response = self.client.get( reverse("api_v1:system:parser-logs-list"), {"search": search_marker}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) rows = response.data["results"] self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["id"], first_log.id) self.assertEqual(rows[0]["organizations_count"], 2) def test_system_logs_export_returns_csv(self): ParserLoadLogFactory( source="manufactures", batch_id=333, status="success", records_count=4, ) ManufacturerRecordFactory(load_batch=333, inn="7701000001") self.client.force_authenticate(self.admin) response = self.client.get( reverse("api_v1:system:parser-logs-export"), {"search": "333"}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response["Content-Type"], "text/csv; charset=utf-8") content = response.content.decode("utf-8") self.assertIn("organizations_count", content) self.assertIn("333", content) def test_system_logs_support_search_by_source_label(self): ParserLoadLogFactory( source="fns_reports", batch_id=909, status="success", ) self.client.force_authenticate(self.admin) response = self.client.get( reverse("api_v1:system:parser-logs-list"), {"search": "финансово"}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data["results"]), 1) self.assertEqual(response.data["results"][0]["source"], "financial-indicators") def test_fns_upload_invalid_filename(self): self.client.force_authenticate(self.admin) with tempfile.TemporaryDirectory() as tmpdir: watch_dir = os.path.join(tmpdir, "watch") processed_dir = os.path.join(tmpdir, "processed") failed_dir = os.path.join(tmpdir, "failed") content = _build_fns_excel_bytes() upload = SimpleUploadedFile( f"bad_{fake.random_int()}.xlsx", content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) with self.settings( FNS_WATCH_DIRECTORY=watch_dir, FNS_PROCESSED_DIRECTORY=processed_dir, FNS_FAILED_DIRECTORY=failed_dir, ): url = reverse("api_v1:fns:fns-upload") response = self.client.post( url, {"files": [upload]}, format="multipart" ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_fns_upload_forbidden_for_regular_user(self): self.client.force_authenticate(self.user) with tempfile.TemporaryDirectory() as tmpdir: watch_dir = os.path.join(tmpdir, "watch") processed_dir = os.path.join(tmpdir, "processed") failed_dir = os.path.join(tmpdir, "failed") content = _build_fns_excel_bytes() upload = SimpleUploadedFile( f"fin_{_digits(5)}_{_digits(13)}.xlsx", content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) with self.settings( FNS_WATCH_DIRECTORY=watch_dir, FNS_PROCESSED_DIRECTORY=processed_dir, FNS_FAILED_DIRECTORY=failed_dir, ): url = reverse("api_v1:fns:fns-upload") response = self.client.post( url, {"files": [upload]}, format="multipart" ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def test_fns_upload_accepts_single_file_payload(self): self.client.force_authenticate(self.admin) with tempfile.TemporaryDirectory() as tmpdir: watch_dir = os.path.join(tmpdir, "watch") processed_dir = os.path.join(tmpdir, "processed") failed_dir = os.path.join(tmpdir, "failed") content = _build_fns_excel_bytes() upload = SimpleUploadedFile( f"fin_{_digits(5)}_{_digits(13)}.xlsx", content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) with self.settings( FNS_WATCH_DIRECTORY=watch_dir, FNS_PROCESSED_DIRECTORY=processed_dir, FNS_FAILED_DIRECTORY=failed_dir, ): url = reverse("api_v1:fns:fns-upload") response = self.client.post(url, {"file": upload}, format="multipart") self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data["success"], True) self.assertIn("message", response.data) def test_parsing_settings_get_and_patch(self): self.client.force_authenticate(self.admin) url = reverse("api_v1:parsing:parsing-settings") initial = self.client.get(url) self.assertEqual(initial.status_code, status.HTTP_200_OK) self.assertEqual(initial.data["planned_inspections"], "monthly") updated = self.client.patch( url, {"planned_inspections": "weekly"}, format="json", ) self.assertEqual(updated.status_code, status.HTTP_200_OK) self.assertEqual(updated.data["planned_inspections"], "weekly") def test_run_sync_inspections_accepts_limited_sync_params(self): self.client.force_authenticate(self.user) url = reverse("api_v1:parsers:run-parser", args=["sync_inspections"]) payload = { "max_months_per_law": 1, "start_year": 2026, "start_month": 4, "include_fz294": True, "include_fz248": False, "current_year": 2026, "current_month": 4, } with patch( "apps.parsers.views.tasks.sync_inspections.apply_async", return_value=Mock(id="task-123"), ) as apply_async_mock: response = self.client.post(url, payload, format="json") self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) task_kwargs = apply_async_mock.call_args.kwargs["kwargs"] for key, value in payload.items(): self.assertEqual(task_kwargs[key], value) self.assertEqual(task_kwargs["requested_by_id"], self.user.id)