Files
mostovik-backend/tests/apps/organizations/test_api_v2_source_extensions.py
Aleksandr Meshchriakov 007cecc8d5
All checks were successful
CI/CD Pipeline / Quality Gate (push) Successful in 36s
CI/CD Pipeline / Build and Push Images (push) Successful in 15s
CI/CD Pipeline / Internal Notify (push) Successful in 0s
CI/CD Pipeline / Deploy Dev via Compose (push) Successful in 24s
feat: expand registry ingestion and demo exchange
2026-07-16 12:59:23 +02:00

406 lines
16 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.
"""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.user.factories import UserFactory
def create_frontend_organization(**kwargs):
"""Create an organization visible through the frontend API by default."""
kwargs.setdefault("opk_registry_membership", True)
return Organization.objects.create(**kwargs)
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 = create_frontend_organization(
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 = create_frontend_organization(
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 = create_frontend_organization(
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 = create_frontend_organization(
name='ООО "Flat Records"',
inn="7707083813",
ogrn="1027700132013",
)
other = create_frontend_organization(
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_scope_cannot_be_lifted_by_has_registry(self):
with_registry = create_frontend_organization(
name='ООО "With Registry Source"',
inn="7707083815",
ogrn="1027700132015",
opk_registry_membership=True,
ropk_razdel_num="source-records",
)
without_registry = create_frontend_organization(
name='ООО "Without Registry Source"',
inn="7707083816",
ogrn="1027700132016",
opk_registry_membership=False,
)
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="Проверка организации без реестра",
)
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-WITH-REGISTRY",
)
def test_source_record_organization_uses_active_registry_identity(self):
organization = create_frontend_organization(
name='ООО "Реестровое имя"',
inn="1800020960",
kpp="180001001",
ogrn="1241800009703",
opk_registry_membership=True,
ropk_razdel_num="identity",
)
extension = PlannedInspectionExtension.objects.create(
organization=organization,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-REGISTRY-IDENTITY",
title="Проверка организации из реестра",
)
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 = create_frontend_organization(
name='ООО "Башнефть-Добыча"',
inn="0277106840",
ogrn="1090280032699",
opk_registry_membership=True,
ropk_razdel_num="leading-zero",
)
extension = PlannedInspectionExtension.objects.create(
organization=organization,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-LEADING-ZERO-INN",
title="Проверка организации с ведущим нулем",
)
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 = create_frontend_organization(
name='ООО "Поиск по payload"',
inn="7707083817",
ogrn="1027700132017",
)
other = create_frontend_organization(
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 = create_frontend_organization(
name='АО "Реестровый поиск"',
inn="1800020961",
kpp="180001002",
ogrn="1241800009704",
opk_registry_membership=True,
ropk_razdel_num="search",
)
extension = PlannedInspectionExtension.objects.create(
organization=organization,
title="Плановые проверки Генпрокуратуры России",
)
OrganizationSourceRecord.objects.create(
extension=extension,
record_type="inspection",
source="inspections",
external_id="INSP-REGISTRY-SEARCH",
title="Проверка поиска по реестру",
)
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"], 'АО "Реестровый поиск"')