Files
mostovik-backend/tests/apps/organizations/test_api_v2_source_extensions.py
Aleksandr Meshchriakov b8a18d6da4
Some checks failed
CI/CD Pipeline / Quality Gate (push) Failing after 14s
CI/CD Pipeline / Build and Push Images (push) Has been skipped
CI/CD Pipeline / Deploy Dev in Dokploy (push) Has been skipped
CI/CD Pipeline / Internal Notify (push) Successful in 0s
feat: migrate parser data to source records
2026-05-19 20:21:31 +02:00

428 lines
17 KiB
Python
Raw Permalink 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.
"""Tests for organization API v2 backed by source extensions."""
from django.core.cache import cache
from django.urls import reverse
from organizations.models import (
Organization,
OrganizationSourceRecord,
PlannedInspectionExtension,
)
from rest_framework import status
from rest_framework.test import APITestCase
from tests.apps.registers.factories import (
OrganizationFactory as RegistryOrganizationFactory,
)
from tests.apps.registers.factories import (
RegisterFactory,
RegistryMembershipPeriodFactory,
)
from tests.apps.user.factories import UserFactory
class OrganizationSourceExtensionsApiV2Test(APITestCase):
"""Checks organization-centric source extension API contract."""
def setUp(self):
cache.clear()
self.user = UserFactory.create_user()
self.client.force_authenticate(self.user)
def test_list_returns_compact_source_summaries_instead_of_embedded_data(self):
organization = Organization.objects.create(
name='ООО "API"',
inn="7707083810",
ogrn="1027700132010",
)
extension = PlannedInspectionExtension.objects.create(
organization=organization,
title="Плановые проверки Генпрокуратуры России",
records_count=1,
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-API",
title="Проверка API",
payload={"registration_number": "INSP-API"},
)
response = self.client.get(
reverse("api_v2:organizations:organizations-list"),
{"has_registry": "false"},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
item = response.data["data"][0]
self.assertNotIn("data", item)
self.assertNotIn("data_sources", item)
self.assertEqual(item["identity_status"], Organization.IdentityStatus.COMPLETE)
self.assertEqual(item["sources"][0]["uid"], str(extension.uid))
self.assertEqual(item["sources"][0]["source_group"], "planned_inspections")
self.assertEqual(item["sources"][0]["records_count"], 1)
def test_organization_sources_action_returns_extensions(self):
organization = Organization.objects.create(
name='ООО "Sources"',
inn="7707083811",
ogrn="1027700132011",
)
extension = PlannedInspectionExtension.objects.create(
organization=organization,
title="Плановые проверки Генпрокуратуры России",
)
response = self.client.get(
reverse(
"api_v2:organizations:organizations-sources",
args=[organization.uid],
)
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data[0]["uid"], str(extension.uid))
def test_source_records_endpoint_returns_extension_records(self):
organization = Organization.objects.create(
name='ООО "Records"',
inn="7707083812",
ogrn="1027700132012",
)
extension = PlannedInspectionExtension.objects.create(
organization=organization,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-RECORD",
title="Проверка records",
payload={"registration_number": "INSP-RECORD"},
)
response = self.client.get(
reverse(
"api_v2:organizations:organization-sources-records",
args=[extension.uid],
)
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["data"][0]["external_id"], "INSP-RECORD")
self.assertEqual(
response.data["data"][0]["payload"]["registration_number"],
"INSP-RECORD",
)
def test_flat_source_records_endpoint_filters_by_source_group(self):
target = Organization.objects.create(
name='ООО "Flat Records"',
inn="7707083813",
ogrn="1027700132013",
)
other = Organization.objects.create(
name='ООО "Other Records"',
inn="7707083814",
ogrn="1027700132014",
)
extension = PlannedInspectionExtension.objects.create(
organization=target,
title="Плановые проверки Генпрокуратуры России",
)
other_extension = PlannedInspectionExtension.objects.create(
organization=other,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-FLAT",
title="Проверка flat",
payload={"registration_number": "INSP-FLAT"},
)
OrganizationSourceRecord.objects.create(
extension=other_extension,
record_type="inspection",
source="inspections",
external_id="INSP-OTHER",
title="Другая проверка",
payload={"registration_number": "INSP-OTHER"},
)
response = self.client.get(
reverse("api_v2:organizations:organization-source-records-list"),
{
"source_group": "planned_inspections",
"organization": str(target.uid),
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["meta"]["pagination"]["total_count"], 1)
record = response.data["data"][0]
self.assertEqual(record["external_id"], "INSP-FLAT")
self.assertEqual(record["source_group"], "planned_inspections")
self.assertEqual(record["organization"]["uid"], str(target.uid))
def test_flat_source_records_endpoint_filters_by_has_registry(self):
with_registry = Organization.objects.create(
name='ООО "With Registry Source"',
inn="7707083815",
ogrn="1027700132015",
)
without_registry = Organization.objects.create(
name='ООО "Without Registry Source"',
inn="7707083816",
ogrn="1027700132016",
)
with_registry_extension = PlannedInspectionExtension.objects.create(
organization=with_registry,
title="Плановые проверки Генпрокуратуры России",
)
without_registry_extension = PlannedInspectionExtension.objects.create(
organization=without_registry,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=with_registry_extension,
record_type="inspection",
source="inspections",
external_id="INSP-WITH-REGISTRY",
title="Проверка организации из реестра",
)
OrganizationSourceRecord.objects.create(
extension=without_registry_extension,
record_type="inspection",
source="inspections",
external_id="INSP-WITHOUT-REGISTRY",
title="Проверка организации без реестра",
)
registry = RegisterFactory(name="Реестр source records")
registry_organization = RegistryOrganizationFactory(
mn_inn=int(with_registry.inn),
mn_ogrn=int(with_registry.ogrn),
)
RegistryMembershipPeriodFactory(
registry=registry,
organization=registry_organization,
)
only_registry = self.client.get(
reverse("api_v2:organizations:organization-source-records-list"),
{
"has_registry": "true",
"source_group": "planned_inspections",
},
)
without_registry_response = self.client.get(
reverse("api_v2:organizations:organization-source-records-list"),
{
"has_registry": "false",
"source_group": "planned_inspections",
"ordering": "extension__organization__inn",
},
)
self.assertEqual(only_registry.status_code, status.HTTP_200_OK)
self.assertEqual(only_registry.data["meta"]["pagination"]["total_count"], 1)
self.assertEqual(
only_registry.data["data"][0]["external_id"],
"INSP-WITH-REGISTRY",
)
self.assertEqual(without_registry_response.status_code, status.HTTP_200_OK)
self.assertEqual(
without_registry_response.data["meta"]["pagination"]["total_count"],
1,
)
self.assertEqual(
without_registry_response.data["data"][0]["external_id"],
"INSP-WITHOUT-REGISTRY",
)
def test_source_record_organization_uses_active_registry_identity(self):
organization = Organization.objects.create(
name="1241800009703",
ogrn="1241800009703",
)
extension = PlannedInspectionExtension.objects.create(
organization=organization,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-REGISTRY-IDENTITY",
title="Проверка организации из реестра",
)
registry_organization = RegistryOrganizationFactory(
pn_name='ООО "Реестровое имя"',
mn_inn=1800020960,
mn_ogrn=int(organization.ogrn),
in_kpp=180001001,
)
RegistryMembershipPeriodFactory(organization=registry_organization)
response = self.client.get(
reverse("api_v2:organizations:organization-source-records-list"),
{
"has_registry": "true",
"source_group": "planned_inspections",
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["meta"]["pagination"]["total_count"], 1)
response_organization = response.data["data"][0]["organization"]
self.assertEqual(response_organization["uid"], str(organization.uid))
self.assertEqual(response_organization["name"], 'ООО "Реестровое имя"')
self.assertEqual(response_organization["inn"], "1800020960")
self.assertEqual(response_organization["kpp"], "180001001")
self.assertEqual(response_organization["ogrn"], organization.ogrn)
def test_source_record_organization_keeps_canonical_inn_with_leading_zero(self):
organization = Organization.objects.create(
name='ООО "Башнефть-Добыча"',
inn="0277106840",
ogrn="1090280032699",
)
extension = PlannedInspectionExtension.objects.create(
organization=organization,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-LEADING-ZERO-INN",
title="Проверка организации с ведущим нулем",
)
registry_organization = RegistryOrganizationFactory(
pn_name='ООО "Башнефть-Добыча"',
mn_inn=277106840,
mn_ogrn=int(organization.ogrn),
in_kpp=27701001,
)
RegistryMembershipPeriodFactory(organization=registry_organization)
response = self.client.get(
reverse("api_v2:organizations:organization-source-records-list"),
{
"has_registry": "true",
"source_group": "planned_inspections",
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
response_organization = response.data["data"][0]["organization"]
self.assertEqual(response_organization["inn"], "0277106840")
self.assertEqual(response_organization["ogrn"], organization.ogrn)
def test_flat_source_records_searches_payload_values_displayed_in_tables(self):
target = Organization.objects.create(
name='ООО "Поиск по payload"',
inn="7707083817",
ogrn="1027700132017",
)
other = Organization.objects.create(
name='ООО "Другая payload"',
inn="7707083818",
ogrn="1027700132018",
)
extension = PlannedInspectionExtension.objects.create(
organization=target,
title="Плановые проверки Генпрокуратуры России",
)
other_extension = PlannedInspectionExtension.objects.create(
organization=other,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-PAYLOAD",
title="Проверка payload",
payload={
"registration_number": "INSPECTION-PAYLOAD-42",
"control_authority": "Северный Ростехнадзор",
"start_date_normalized": "2026-06-15",
},
)
OrganizationSourceRecord.objects.create(
extension=other_extension,
record_type="inspection",
source="inspections",
external_id="INSP-OTHER-PAYLOAD",
title="Другая проверка payload",
payload={
"registration_number": "INSPECTION-OTHER-42",
"control_authority": "Другой контрольный орган",
},
)
for search_value in (
"INSPECTION-PAYLOAD-42",
"Северный Ростехнадзор",
"2026-06-15",
):
with self.subTest(search_value=search_value):
response = self.client.get(
reverse("api_v2:organizations:organization-source-records-list"),
{
"has_registry": "false",
"source_group": "planned_inspections",
"search": search_value,
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["meta"]["pagination"]["total_count"], 1)
self.assertEqual(
response.data["data"][0]["external_id"],
"INSP-PAYLOAD",
)
def test_flat_source_records_searches_registry_identity_displayed_in_tables(self):
organization = Organization.objects.create(
name="1241800009704",
ogrn="1241800009704",
)
extension = PlannedInspectionExtension.objects.create(
organization=organization,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-REGISTRY-SEARCH",
title="Проверка поиска по реестру",
)
registry_organization = RegistryOrganizationFactory(
pn_name='АО "Реестровый поиск"',
mn_inn=1800020961,
mn_ogrn=int(organization.ogrn),
in_kpp=180001002,
)
RegistryMembershipPeriodFactory(organization=registry_organization)
response = self.client.get(
reverse("api_v2:organizations:organization-source-records-list"),
{
"has_registry": "true",
"source_group": "planned_inspections",
"search": "Реестровый поиск",
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["meta"]["pagination"]["total_count"], 1)
record = response.data["data"][0]
self.assertEqual(record["external_id"], "INSP-REGISTRY-SEARCH")
self.assertEqual(record["organization"]["name"], 'АО "Реестровый поиск"')