"""Integration tests for registers API views.""" from __future__ import annotations import io from datetime import date from unittest.mock import patch from apps.registers.models import Organization, RegistryMembershipPeriod from django.core.files.uploadedfile import SimpleUploadedFile from django.db import IntegrityError from django.urls import reverse from openpyxl import Workbook from rest_framework import status from rest_framework.test import APITestCase from tests.apps.registers.factories import ( OrganizationFactory, RegisterFactory, RegisterUploadFactory, RegistryMembershipPeriodFactory, ) from tests.apps.user.factories import UserFactory def _build_register_excel_bytes(rows: list[dict], *, with_kpp: bool = True) -> bytes: workbook = Workbook() worksheet = workbook.active headers = ["pn_name", "mn_ogrn", "mn_inn"] if with_kpp: headers.append("in_kpp") headers.append("mn_okpo") worksheet.append(headers) for row in rows: values = [row["pn_name"], row["mn_ogrn"], row["mn_inn"]] if with_kpp: values.append(row.get("in_kpp")) values.append(row["mn_okpo"]) worksheet.append(values) buffer = io.BytesIO() workbook.save(buffer) workbook.close() return buffer.getvalue() def _extract_results(response_data): if hasattr(response_data, "get"): data = response_data.get("data") if isinstance(data, list): return data results = response_data.get("results") if results is not None: return results return response_data class RegistersViewsTest(APITestCase): def setUp(self): self.user = UserFactory.create_user() self.admin = UserFactory.create_user(is_staff=True) self.client.force_authenticate(self.user) def _post_upload( self, *, registry, rows: list[dict], actual_date_value: date, with_kpp: bool = True, file_name: str = "registry.xlsx", ): self.client.force_authenticate(self.admin) content = _build_register_excel_bytes(rows, with_kpp=with_kpp) upload = SimpleUploadedFile( file_name, content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) response = self.client.post( reverse("api_v1:registers:register-upload"), { "registry": str(registry.id), "actual_date": actual_date_value.isoformat(), "file": upload, }, format="multipart", ) self.client.force_authenticate(self.user) return response def _post_v2_slug_upload( self, *, slug: str, rows: list[dict], actual_date_value: date, with_kpp: bool = True, file_name: str = "registry.xlsx", ): self.client.force_authenticate(self.admin) content = _build_register_excel_bytes(rows, with_kpp=with_kpp) upload = SimpleUploadedFile( file_name, content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) response = self.client.post( reverse(f"api_v2:registers:register-upload-{slug}"), { "actual_date": actual_date_value.isoformat(), "file": upload, }, format="multipart", ) self.client.force_authenticate(self.user) return response def test_registries_list_and_retrieve(self): registry = RegisterFactory(name="Росатом") RegisterUploadFactory(registry=registry) RegistryMembershipPeriodFactory(registry=registry) list_response = self.client.get(reverse("api_v1:registers:registries-list")) self.assertEqual(list_response.status_code, status.HTTP_200_OK) list_item = next( item for item in _extract_results(list_response.data) if item["id"] == str(registry.id) ) self.assertEqual(list_item["active_organizations"], 1) self.assertEqual(list_item["uploads_count"], 2) detail_response = self.client.get( reverse("api_v1:registers:registries-detail", args=[registry.id]) ) self.assertEqual(detail_response.status_code, status.HTTP_200_OK) self.assertEqual(detail_response.data["name"], "Росатом") self.assertEqual(detail_response.data["active_organizations"], 1) self.assertEqual(detail_response.data["uploads_count"], 2) def test_default_registries_are_seeded(self): response = self.client.get(reverse("api_v1:registers:registries-list")) self.assertEqual(response.status_code, status.HTTP_200_OK) names = {item["name"] for item in _extract_results(response.data)} self.assertIn("Реестр предприятий ОПК", names) self.assertIn("Реестр госкорпорации Роскосмос", names) self.assertIn("Реестр госкорпорации Роскосмос ГОЗ", names) self.assertIn("Реестр госкорпорации Роскосмос ОПК", names) self.assertIn("Реестр госкорпорации Росатом", names) self.assertIn("Реестр госкорпорации Росатом ГОЗ", names) self.assertIn("Реестр госкорпорации Росатом ОПК", names) def test_v2_registry_slug_upload_uses_fixed_registry_and_refreshes_snapshots(self): rows = [ { "pn_name": 'АО "Росатом ГОЗ"', "mn_ogrn": "1027600980990", "mn_inn": "7601000086", "in_kpp": "760401001", "mn_okpo": "07506197", } ] with patch("registers.views._start_snapshot_refresh_task") as refresh_task: response = self._post_v2_slug_upload( slug="rosatom-goz", rows=rows, actual_date_value=date(2026, 5, 7), file_name="rosatom_goz.xlsx", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertTrue(response.data["success"]) self.assertEqual( response.data["registry_name"], "Реестр госкорпорации Росатом ГОЗ", ) refresh_task.assert_called_once_with() membership = RegistryMembershipPeriod.objects.get(ended_at__isnull=True) self.assertEqual( membership.registry.name, "Реестр госкорпорации Росатом ГОЗ", ) def test_v2_registry_slug_upload_url_does_not_duplicate_registers_segment(self): self.assertEqual( reverse("api_v2:registers:register-upload-rosatom-goz"), "/api/v2/registers/rosatom-goz/upload/", ) def test_v2_registry_slug_upload_does_not_refresh_snapshots_after_import_error(self): rows = [ { "pn_name": 'АО "Невалидный ОКПО"', "mn_ogrn": "1027600980990", "mn_inn": "7601000086", "in_kpp": "760401001", "mn_okpo": "07A06197", } ] with patch("registers.views._start_snapshot_refresh_task") as refresh_task: response = self._post_v2_slug_upload( slug="rosatom-goz", rows=rows, actual_date_value=date(2026, 5, 7), file_name="invalid_rosatom_goz.xlsx", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) refresh_task.assert_not_called() def test_organizations_list_and_retrieve(self): organization = OrganizationFactory() list_response = self.client.get(reverse("api_v1:registers:organizations-list")) self.assertEqual(list_response.status_code, status.HTTP_200_OK) detail_response = self.client.get( reverse("api_v1:registers:organizations-detail", args=[organization.id]) ) self.assertEqual(detail_response.status_code, status.HTTP_200_OK) self.assertEqual(detail_response.data["id"], organization.id) self.assertIn("periods", detail_response.data) def test_organizations_search_by_all_fields(self): organization = OrganizationFactory( pn_name='АО "Тестовая организация"', mn_ogrn=1027600980990, mn_inn=7601000086, in_kpp=760401001, mn_okpo="07506197", ) OrganizationFactory() search_values = [ "Тестовая организация", str(organization.mn_ogrn), str(organization.mn_inn), str(organization.in_kpp), organization.mn_okpo, ] for search_value in search_values: response = self.client.get( reverse("api_v1:registers:organizations-list"), {"search": search_value}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) result_ids = [item["id"] for item in _extract_results(response.data)] self.assertIn(organization.id, result_ids) def test_organizations_filter_by_registry_and_actual_date(self): registry = RegisterFactory(name="Роскосмос") organization_old = OrganizationFactory() organization_current = OrganizationFactory() upload_start = RegisterUploadFactory( registry=registry, actual_date=date(2026, 1, 1) ) upload_end = RegisterUploadFactory( registry=registry, actual_date=date(2026, 2, 1) ) RegistryMembershipPeriodFactory( registry=registry, organization=organization_old, started_at=date(2026, 1, 1), ended_at=date(2026, 2, 1), started_by_upload=upload_start, ended_by_upload=upload_end, ) RegistryMembershipPeriodFactory( registry=registry, organization=organization_current, started_at=date(2026, 2, 1), started_by_upload=upload_end, ) response_past = self.client.get( reverse("api_v1:registers:organizations-list"), {"registry": str(registry.id), "actual_date": "2026-01-15"}, ) self.assertEqual(response_past.status_code, status.HTTP_200_OK) past_ids = {item["id"] for item in _extract_results(response_past.data)} self.assertIn(organization_old.id, past_ids) self.assertNotIn(organization_current.id, past_ids) response_latest = self.client.get( reverse("api_v1:registers:organizations-list"), {"registry": str(registry.id)}, ) self.assertEqual(response_latest.status_code, status.HTTP_200_OK) latest_ids = {item["id"] for item in _extract_results(response_latest.data)} self.assertNotIn(organization_old.id, latest_ids) self.assertIn(organization_current.id, latest_ids) def test_registry_specific_organizations_list_endpoint(self): registry = RegisterFactory(name="Росатом ОПК") organization = OrganizationFactory( mn_ogrn=1027600980990, mn_inn=7601000086, mn_okpo="07506197", ) upload = RegisterUploadFactory(registry=registry, actual_date=date(2026, 3, 1)) RegistryMembershipPeriodFactory( registry=registry, organization=organization, started_at=date(2026, 3, 1), started_by_upload=upload, ) response = self.client.get( reverse( "api_v1:registers:registry-organizations-list", args=[registry.id], ), {"actual_date": "2026-03-15"}, ) self.assertEqual(response.status_code, status.HTTP_200_OK) results = _extract_results(response.data) self.assertEqual(len(results), 1) self.assertEqual(results[0]["id"], organization.id) def test_upload_closes_and_reopens_period(self): registry = RegisterFactory(name="Реестр периодов") org_a = { "pn_name": 'АО "Орг А"', "mn_ogrn": "1027600980990", "mn_inn": "7601000086", "in_kpp": "760401001", "mn_okpo": "07506197", } org_b = { "pn_name": 'АО "Орг Б"', "mn_ogrn": "1083340004527", "mn_inn": "3329051460", "in_kpp": "332901001", "mn_okpo": "07518609", } first = self._post_upload( registry=registry, rows=[org_a], actual_date_value=date(2026, 1, 1), file_name="first.xlsx", ) self.assertEqual(first.status_code, status.HTTP_201_CREATED) self.assertTrue(first.data["success"]) self.assertEqual(first.data["message"], "Файл успешно загружен") second = self._post_upload( registry=registry, rows=[org_b], actual_date_value=date(2026, 2, 1), file_name="second.xlsx", ) self.assertEqual(second.status_code, status.HTTP_201_CREATED) self.assertTrue(second.data["success"]) self.assertEqual(second.data["message"], "Файл успешно загружен") third = self._post_upload( registry=registry, rows=[org_a], actual_date_value=date(2026, 3, 1), file_name="third.xlsx", ) self.assertEqual(third.status_code, status.HTTP_201_CREATED) self.assertTrue(third.data["success"]) self.assertEqual(third.data["message"], "Файл успешно загружен") organization_a = Organization.objects.get( mn_ogrn=1027600980990, mn_inn=7601000086 ) periods = list( RegistryMembershipPeriod.objects.filter( registry=registry, organization=organization_a, ).order_by("started_at") ) self.assertEqual(len(periods), 2) self.assertEqual(periods[0].started_at, date(2026, 1, 1)) self.assertEqual(periods[0].ended_at, date(2026, 2, 1)) self.assertEqual(periods[1].started_at, date(2026, 3, 1)) self.assertIsNone(periods[1].ended_at) def test_same_organization_can_be_in_multiple_registries(self): registry_a = RegisterFactory(name="Росатом") registry_b = RegisterFactory(name="Роскосмос") org_row = { "pn_name": 'АО "Общая организация"', "mn_ogrn": "1027600980990", "mn_inn": "7601000086", "in_kpp": "760401001", "mn_okpo": "07506197", } upload_a = self._post_upload( registry=registry_a, rows=[org_row], actual_date_value=date(2026, 1, 1), file_name="reg_a.xlsx", ) upload_b = self._post_upload( registry=registry_b, rows=[org_row], actual_date_value=date(2026, 1, 1), file_name="reg_b.xlsx", ) self.assertEqual(upload_a.status_code, status.HTTP_201_CREATED) self.assertEqual(upload_b.status_code, status.HTTP_201_CREATED) response_a = self.client.get( reverse( "api_v1:registers:registry-organizations-list", args=[registry_a.id] ) ) response_b = self.client.get( reverse( "api_v1:registers:registry-organizations-list", args=[registry_b.id] ) ) self.assertEqual(response_a.status_code, status.HTTP_200_OK) self.assertEqual(response_b.status_code, status.HTTP_200_OK) ids_a = {item["id"] for item in _extract_results(response_a.data)} ids_b = {item["id"] for item in _extract_results(response_b.data)} self.assertEqual(ids_a, ids_b) self.assertEqual(len(ids_a), 1) def test_active_membership_period_is_unique_per_registry_and_organization(self): registry = RegisterFactory(name="Уникальный период") organization = OrganizationFactory() upload = RegisterUploadFactory(registry=registry, actual_date=date(2026, 6, 1)) RegistryMembershipPeriodFactory( registry=registry, organization=organization, started_at=date(2026, 6, 1), started_by_upload=upload, ended_at=None, ) with self.assertRaises(IntegrityError): RegistryMembershipPeriod.objects.create( registry=registry, organization=organization, started_at=date(2026, 7, 1), started_by_upload=upload, ended_at=None, ) def test_upload_without_kpp_column(self): registry = RegisterFactory(name="Роскосмос") response = self._post_upload( registry=registry, rows=[ { "pn_name": 'АО "Ярославский радиозавод"', "mn_ogrn": "1027600980990", "mn_inn": "7601000086", "mn_okpo": "07506197", } ], actual_date_value=date(2026, 4, 1), with_kpp=False, file_name="without_kpp.xlsx", ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertTrue(response.data["success"]) self.assertEqual(response.data["message"], "Файл успешно загружен") organization = Organization.objects.get( mn_ogrn=1027600980990, mn_inn=7601000086 ) self.assertIsNone(organization.in_kpp) def test_upload_rejects_invalid_okpo(self): registry = RegisterFactory(name="Реестр ошибки") response = self._post_upload( registry=registry, rows=[ { "pn_name": 'АО "Невалидный ОКПО"', "mn_ogrn": "1027600980990", "mn_inn": "7601000086", "in_kpp": "760401001", "mn_okpo": "07A06197", } ], actual_date_value=date(2026, 5, 1), with_kpp=True, file_name="invalid.xlsx", ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(Organization.objects.count(), 0) def test_upload_requires_authentication(self): registry = RegisterFactory(name="Закрытый реестр") content = _build_register_excel_bytes( [ { "pn_name": 'АО "Закрытый"', "mn_ogrn": "1027600980990", "mn_inn": "7601000086", "in_kpp": "760401001", "mn_okpo": "07506197", } ] ) upload = SimpleUploadedFile( "auth.xlsx", content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) self.client.force_authenticate(user=None) response = self.client.post( reverse("api_v1:registers:register-upload"), { "registry": str(registry.id), "actual_date": "2026-01-01", "file": upload, }, format="multipart", ) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_upload_forbidden_for_regular_user(self): registry = RegisterFactory(name="Только для админа") content = _build_register_excel_bytes( [ { "pn_name": 'АО "Ограниченный доступ"', "mn_ogrn": "1027600980990", "mn_inn": "7601000086", "in_kpp": "760401001", "mn_okpo": "07506197", } ] ) upload = SimpleUploadedFile( "viewer.xlsx", content, content_type=( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ), ) self.client.force_authenticate(self.user) response = self.client.post( reverse("api_v1:registers:register-upload"), { "registry": str(registry.id), "actual_date": "2026-01-01", "file": upload, }, format="multipart", ) self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)