222 lines
8.2 KiB
Python
222 lines
8.2 KiB
Python
"""Tests for organization source backfill from legacy parser tables."""
|
||
|
||
from apps.parsers.models import (
|
||
FinancialReport,
|
||
FinancialReportLine,
|
||
GenericParserRecord,
|
||
ParserLoadLog,
|
||
)
|
||
from django.core.management import call_command
|
||
from django.test import TestCase
|
||
from django.utils import timezone
|
||
from organizations.models import (
|
||
DefenseSupplierExtension,
|
||
FinancialIndicatorsExtension,
|
||
Organization,
|
||
OrganizationSourceFinancialLine,
|
||
OrganizationSourceRecord,
|
||
PlannedInspectionExtension,
|
||
SourceGroup,
|
||
)
|
||
from organizations.source_backfill import OrganizationSourceBackfillService
|
||
|
||
from tests.apps.parsers.factories import (
|
||
InspectionRecordFactory,
|
||
ManufacturerRecordFactory,
|
||
)
|
||
|
||
|
||
class OrganizationSourceBackfillServiceTest(TestCase):
|
||
"""Checks idempotent migration from legacy source records."""
|
||
|
||
@staticmethod
|
||
def _directory_organization(**kwargs) -> Organization:
|
||
kwargs.setdefault("directory_imported_at", timezone.now())
|
||
return Organization.objects.create(**kwargs)
|
||
|
||
def test_backfills_inspection_records_into_planned_inspection_extension(self):
|
||
organization = self._directory_organization(
|
||
name='ООО "Проверка"',
|
||
inn="7707083801",
|
||
ogrn="1027700132001",
|
||
)
|
||
inspection = InspectionRecordFactory(
|
||
load_batch=77,
|
||
inn=organization.inn,
|
||
ogrn=organization.ogrn,
|
||
organisation_name=organization.name,
|
||
registration_number="INSP-77",
|
||
)
|
||
|
||
result = OrganizationSourceBackfillService.backfill(
|
||
source=ParserLoadLog.Source.INSPECTIONS,
|
||
batch_id=77,
|
||
)
|
||
|
||
self.assertEqual(result.scanned, 1)
|
||
self.assertEqual(result.created_extensions, 1)
|
||
self.assertEqual(result.created_records, 1)
|
||
extension = PlannedInspectionExtension.objects.get(organization=organization)
|
||
self.assertEqual(extension.source_group, SourceGroup.PLANNED_INSPECTIONS)
|
||
self.assertEqual(extension.records_count, 1)
|
||
record = OrganizationSourceRecord.objects.get(extension=extension)
|
||
self.assertEqual(record.source, ParserLoadLog.Source.INSPECTIONS)
|
||
self.assertEqual(record.external_id, inspection.registration_number)
|
||
self.assertEqual(record.legacy_model, "apps.parsers.InspectionRecord")
|
||
self.assertEqual(record.legacy_pk, str(inspection.pk))
|
||
self.assertEqual(record.payload["registration_number"], "INSP-77")
|
||
|
||
second_result = OrganizationSourceBackfillService.backfill(
|
||
source=ParserLoadLog.Source.INSPECTIONS,
|
||
batch_id=77,
|
||
)
|
||
|
||
self.assertEqual(second_result.created_extensions, 0)
|
||
self.assertEqual(second_result.created_records, 0)
|
||
self.assertEqual(second_result.updated_records, 1)
|
||
self.assertEqual(OrganizationSourceRecord.objects.count(), 1)
|
||
|
||
def test_backfill_payload_serializes_registry_organization_fk_as_id(self):
|
||
organization = self._directory_organization(
|
||
name='ООО "FK Payload"',
|
||
inn="7707083815",
|
||
ogrn="1027700132015",
|
||
)
|
||
InspectionRecordFactory(
|
||
load_batch=78,
|
||
inn=organization.inn,
|
||
ogrn=organization.ogrn,
|
||
organisation_name=organization.name,
|
||
registration_number="INSP-FK",
|
||
registry_organization=organization,
|
||
)
|
||
|
||
OrganizationSourceBackfillService.backfill(
|
||
source=ParserLoadLog.Source.INSPECTIONS,
|
||
batch_id=78,
|
||
)
|
||
|
||
record = OrganizationSourceRecord.objects.get(external_id="INSP-FK")
|
||
self.assertEqual(record.payload["registry_organization"], str(organization.pk))
|
||
|
||
def test_backfills_financial_report_lines(self):
|
||
organization = self._directory_organization(
|
||
name='ООО "Финансы"',
|
||
inn="7707083802",
|
||
ogrn="1027700132002",
|
||
)
|
||
report = FinancialReport.objects.create(
|
||
external_id="fns-report-1",
|
||
ogrn=organization.ogrn,
|
||
file_name="fns_report.xlsx",
|
||
file_hash="a" * 64,
|
||
load_batch=88,
|
||
status=FinancialReport.Status.SUCCESS,
|
||
source=FinancialReport.SourceType.API,
|
||
)
|
||
FinancialReportLine.objects.create(
|
||
report=report,
|
||
form_code="1",
|
||
line_code="1100",
|
||
line_name="Нематериальные активы",
|
||
year=2025,
|
||
period_start=100,
|
||
period_end=200,
|
||
)
|
||
|
||
result = OrganizationSourceBackfillService.backfill(
|
||
source=ParserLoadLog.Source.FNS_REPORTS,
|
||
batch_id=88,
|
||
)
|
||
|
||
self.assertEqual(result.created_extensions, 1)
|
||
self.assertEqual(result.created_records, 1)
|
||
extension = FinancialIndicatorsExtension.objects.get(organization=organization)
|
||
record = OrganizationSourceRecord.objects.get(extension=extension)
|
||
self.assertEqual(record.external_id, "fns-report-1")
|
||
self.assertEqual(OrganizationSourceFinancialLine.objects.count(), 1)
|
||
self.assertEqual(record.financial_lines.get().period_end, 200)
|
||
|
||
def test_backfills_generic_defense_supplier_records(self):
|
||
organization = self._directory_organization(
|
||
name='ООО "ГОЗ"',
|
||
inn="7707083803",
|
||
)
|
||
generic_record = GenericParserRecord.objects.create(
|
||
load_batch=99,
|
||
source=ParserLoadLog.Source.UNFAIR_SUPPLIERS,
|
||
external_id="unfair-1",
|
||
inn=organization.inn,
|
||
ogrn="",
|
||
organisation_name=organization.name,
|
||
title="Реестр недобросовестных поставщиков",
|
||
record_date="2026-05-18",
|
||
status="active",
|
||
payload={"reason": "test"},
|
||
)
|
||
|
||
result = OrganizationSourceBackfillService.backfill(
|
||
source=ParserLoadLog.Source.UNFAIR_SUPPLIERS,
|
||
batch_id=99,
|
||
)
|
||
|
||
self.assertEqual(result.created_extensions, 1)
|
||
self.assertEqual(result.created_records, 1)
|
||
extension = DefenseSupplierExtension.objects.get(organization=organization)
|
||
record = OrganizationSourceRecord.objects.get(extension=extension)
|
||
self.assertEqual(record.source, ParserLoadLog.Source.UNFAIR_SUPPLIERS)
|
||
self.assertEqual(record.external_id, generic_record.external_id)
|
||
self.assertEqual(record.payload["reason"], "test")
|
||
|
||
def test_backfill_skips_invalid_identity_without_directory_organization(self):
|
||
manufacturer = ManufacturerRecordFactory(
|
||
load_batch=100,
|
||
full_legal_name='ИП "Грязный ИНН"',
|
||
inn="396440000001000",
|
||
ogrn="5385025402942",
|
||
)
|
||
|
||
result = OrganizationSourceBackfillService.backfill(
|
||
source=ParserLoadLog.Source.MANUFACTURES,
|
||
batch_id=100,
|
||
)
|
||
|
||
self.assertEqual(result.scanned, 1)
|
||
self.assertEqual(result.created_organizations, 0)
|
||
self.assertEqual(result.created_records, 0)
|
||
self.assertEqual(result.unresolved, 1)
|
||
self.assertFalse(
|
||
Organization.objects.filter(name=manufacturer.full_legal_name).exists()
|
||
)
|
||
self.assertFalse(
|
||
OrganizationSourceRecord.objects.filter(
|
||
legacy_pk=str(manufacturer.pk)
|
||
).exists()
|
||
)
|
||
|
||
def test_management_command_runs_source_backfill(self):
|
||
organization = self._directory_organization(
|
||
name='ООО "Команда"',
|
||
inn="7707083804",
|
||
ogrn="1027700132004",
|
||
)
|
||
InspectionRecordFactory(
|
||
load_batch=101,
|
||
inn=organization.inn,
|
||
ogrn=organization.ogrn,
|
||
organisation_name=organization.name,
|
||
registration_number="INSP-101",
|
||
)
|
||
|
||
call_command(
|
||
"backfill_organization_sources",
|
||
source=ParserLoadLog.Source.INSPECTIONS,
|
||
batch_id=101,
|
||
verbosity=0,
|
||
)
|
||
|
||
self.assertTrue(
|
||
PlannedInspectionExtension.objects.filter(organization=organization).exists()
|
||
)
|
||
self.assertEqual(OrganizationSourceRecord.objects.get().external_id, "INSP-101")
|