Files
mostovik-backend/tests/apps/registers/test_views.py
Aleksandr Meshchriakov 89607356b7
All checks were successful
CI/CD Pipeline / Quality Gate (push) Successful in 24s
CI/CD Pipeline / Build and Push Images (push) Successful in 10s
CI/CD Pipeline / Internal Notify (push) Successful in 0s
CI/CD Pipeline / Deploy Dev in Dokploy (push) Successful in 1s
Add organization stats endpoint
2026-05-12 17:48:54 +02:00

632 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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, Register, 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_stat_organizations_endpoint_returns_registry_cards(self):
opk_registry, _ = Register.objects.get_or_create(name="Реестр предприятий ОПК")
rosatom_registry, _ = Register.objects.get_or_create(
name="Реестр госкорпорации Росатом"
)
roscosmos_goz_registry, _ = Register.objects.get_or_create(
name="Реестр госкорпорации Роскосмос ГОЗ"
)
opk_organization = OrganizationFactory()
shared_organization = OrganizationFactory()
inactive_organization = OrganizationFactory()
RegistryMembershipPeriodFactory(
registry=opk_registry,
organization=opk_organization,
)
RegistryMembershipPeriodFactory(
registry=rosatom_registry,
organization=opk_organization,
)
RegistryMembershipPeriodFactory(
registry=rosatom_registry,
organization=shared_organization,
)
RegistryMembershipPeriodFactory(
registry=roscosmos_goz_registry,
organization=inactive_organization,
ended_at=date(2026, 6, 1),
)
response = self.client.get("/api/v1/stat/organizations/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
payload = response.data["data"]
self.assertEqual(payload["total_organizations"], 3)
self.assertEqual(payload["active_registry_organizations"], 2)
self.assertEqual(payload["counts"]["total"], 3)
self.assertEqual(payload["counts"]["opk"], 1)
self.assertEqual(payload["counts"]["rosatom"], 2)
self.assertEqual(payload["counts"]["roscosmos-goz"], 0)
cards_by_slug = {item["slug"]: item for item in payload["cards"]}
self.assertEqual(
cards_by_slug["total"]["title"],
"Общее количество организаций",
)
self.assertEqual(
cards_by_slug["rosatom"]["registry_name"],
"Реестр госкорпорации Росатом",
)
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)