322 lines
12 KiB
Python
322 lines
12 KiB
Python
"""Focused unit tests for parser service helpers and small query methods."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
from unittest.mock import patch
|
||
|
||
from apps.parsers.clients.minpromtorg.schemas import IndustrialProduct
|
||
from apps.parsers.clients.proverki.schemas import Inspection
|
||
from apps.parsers.clients.zakupki.schemas import Procurement
|
||
from apps.parsers.models import FinancialReport, ParserLoadLog
|
||
from apps.parsers.services import (
|
||
FNSReportService,
|
||
IndustrialProductService,
|
||
InspectionService,
|
||
ParserLoadLogService,
|
||
ProcurementService,
|
||
RegistryOrganizationResolver,
|
||
normalize_to_date,
|
||
normalize_to_decimal,
|
||
)
|
||
from django.db import IntegrityError
|
||
from django.test import TestCase
|
||
|
||
from tests.apps.parsers.factories import ParserLoadLogFactory
|
||
from tests.apps.registers.factories import OrganizationFactory
|
||
|
||
|
||
class NormalizeHelpersTest(TestCase):
|
||
def test_normalize_to_date_handles_direct_and_embedded_formats(self):
|
||
self.assertIsNone(normalize_to_date(None))
|
||
self.assertIsNone(normalize_to_date(" "))
|
||
self.assertEqual(
|
||
normalize_to_date("2026-03-17T10:15:30+03:00"),
|
||
date(2026, 3, 17),
|
||
)
|
||
self.assertEqual(
|
||
normalize_to_date("report created at 2026-03-17 10:15"),
|
||
date(2026, 3, 17),
|
||
)
|
||
self.assertEqual(
|
||
normalize_to_date("актуально на 17.03.2026 10:15"),
|
||
date(2026, 3, 17),
|
||
)
|
||
self.assertIsNone(normalize_to_date("not a date"))
|
||
|
||
def test_normalize_to_decimal_handles_common_formats_and_invalid_values(self):
|
||
self.assertIsNone(normalize_to_decimal(None))
|
||
self.assertIsNone(normalize_to_decimal(" "))
|
||
self.assertEqual(normalize_to_decimal("1.234,56 руб."), Decimal("1234.56"))
|
||
self.assertEqual(normalize_to_decimal("1,234.56"), Decimal("1234.56"))
|
||
self.assertIsNone(normalize_to_decimal("руб."))
|
||
self.assertIsNone(normalize_to_decimal("--1"))
|
||
|
||
|
||
class RegistryOrganizationResolverTest(TestCase):
|
||
def test_normalize_identifier_rejects_non_digits(self):
|
||
self.assertIsNone(RegistryOrganizationResolver.normalize_identifier(""))
|
||
self.assertIsNone(RegistryOrganizationResolver.normalize_identifier("12 34"))
|
||
self.assertIsNone(RegistryOrganizationResolver.normalize_identifier("abc"))
|
||
|
||
def test_build_lookup_returns_empty_indexes_when_identifiers_absent(self):
|
||
lookup = RegistryOrganizationResolver.build_lookup([(None, None), ("", "")])
|
||
self.assertEqual(lookup.by_pair, {})
|
||
self.assertEqual(lookup.by_inn, {})
|
||
self.assertEqual(lookup.by_ogrn, {})
|
||
|
||
def test_resolve_organization_id_by_unique_inn_and_ogrn(self):
|
||
org_by_inn = OrganizationFactory(
|
||
mn_inn=7_701_001_001, mn_ogrn=10_277_001_000_001
|
||
)
|
||
org_by_ogrn = OrganizationFactory(
|
||
mn_inn=7_701_001_002, mn_ogrn=10_277_001_000_002
|
||
)
|
||
lookup = RegistryOrganizationResolver.build_lookup(
|
||
[
|
||
(org_by_inn.mn_inn, None),
|
||
(None, org_by_ogrn.mn_ogrn),
|
||
]
|
||
)
|
||
|
||
self.assertEqual(
|
||
RegistryOrganizationResolver.resolve_organization_id(
|
||
lookup=lookup,
|
||
inn=str(org_by_inn.mn_inn),
|
||
ogrn=None,
|
||
),
|
||
org_by_inn.id,
|
||
)
|
||
self.assertEqual(
|
||
RegistryOrganizationResolver.resolve_organization_id(
|
||
lookup=lookup,
|
||
inn=None,
|
||
ogrn=str(org_by_ogrn.mn_ogrn),
|
||
),
|
||
org_by_ogrn.id,
|
||
)
|
||
|
||
|
||
class ParserLoadLogServiceRetryTest(TestCase):
|
||
def test_create_load_log_with_next_batch_id_retries_after_integrity_error(self):
|
||
log = ParserLoadLogFactory.build(
|
||
source=ParserLoadLog.Source.INDUSTRIAL,
|
||
batch_id=2,
|
||
)
|
||
|
||
with patch.object(
|
||
ParserLoadLogService, "get_next_batch_id", side_effect=[1, 2]
|
||
), patch.object(
|
||
ParserLoadLogService,
|
||
"create_load_log",
|
||
side_effect=[IntegrityError("duplicate"), log],
|
||
) as create_mock:
|
||
(
|
||
created_log,
|
||
batch_id,
|
||
) = ParserLoadLogService.create_load_log_with_next_batch_id(
|
||
source=ParserLoadLog.Source.INDUSTRIAL
|
||
)
|
||
|
||
self.assertEqual(created_log, log)
|
||
self.assertEqual(batch_id, 2)
|
||
self.assertEqual(create_mock.call_count, 2)
|
||
|
||
def test_create_load_log_with_next_batch_id_raises_after_max_retries(self):
|
||
with patch.object(
|
||
ParserLoadLogService, "get_next_batch_id", return_value=1
|
||
), patch.object(
|
||
ParserLoadLogService,
|
||
"create_load_log",
|
||
side_effect=IntegrityError("duplicate"),
|
||
), self.assertRaisesMessage(
|
||
RuntimeError,
|
||
"Failed to allocate unique batch_id",
|
||
):
|
||
ParserLoadLogService.create_load_log_with_next_batch_id(
|
||
source=ParserLoadLog.Source.INDUSTRIAL,
|
||
max_retries=2,
|
||
)
|
||
|
||
|
||
class SmallParserServiceQueryTest(TestCase):
|
||
def test_industrial_product_service_query_helpers(self):
|
||
IndustrialProductService.save_products(
|
||
[
|
||
IndustrialProduct(
|
||
full_organisation_name='ООО "Продукт 1"',
|
||
ogrn="1027700100001",
|
||
inn="7701001001",
|
||
registry_number="PROD-1",
|
||
product_name="Станок",
|
||
product_model="MODEL-1",
|
||
okpd2_code="28.41",
|
||
tnved_code="8457109000",
|
||
regulatory_document="ГОСТ",
|
||
)
|
||
],
|
||
batch_id=7,
|
||
)
|
||
IndustrialProductService.save_products(
|
||
[
|
||
IndustrialProduct(
|
||
full_organisation_name='ООО "Продукт 2"',
|
||
ogrn="1027700100002",
|
||
inn="7701001001",
|
||
registry_number="PROD-2",
|
||
product_name="Пресс",
|
||
product_model="MODEL-2",
|
||
okpd2_code="28.42",
|
||
tnved_code="8457209000",
|
||
regulatory_document="ТУ",
|
||
)
|
||
],
|
||
batch_id=8,
|
||
)
|
||
|
||
self.assertEqual(IndustrialProductService.find_by_inn("7701001001").count(), 2)
|
||
self.assertEqual(
|
||
IndustrialProductService.find_by_inn("7701001001", batch_id=7).count(),
|
||
1,
|
||
)
|
||
self.assertEqual(
|
||
IndustrialProductService.find_by_ogrn("1027700100001").first().external_id,
|
||
"PROD-1",
|
||
)
|
||
|
||
def test_inspection_service_has_data_for_period(self):
|
||
InspectionService.save_inspections(
|
||
[
|
||
Inspection(
|
||
registration_number="INSP-1",
|
||
inn="7701002001",
|
||
ogrn="1027700200001",
|
||
organisation_name='ООО "Проверка 1"',
|
||
control_authority="Контроль",
|
||
inspection_type="Плановая",
|
||
inspection_form="Документарная",
|
||
start_date="2026-03-01",
|
||
end_date="2026-03-15",
|
||
status="planned",
|
||
legal_basis="ФЗ",
|
||
result="",
|
||
is_federal_law_248=False,
|
||
),
|
||
],
|
||
batch_id=1,
|
||
data_year=2026,
|
||
data_month=3,
|
||
)
|
||
InspectionService.save_inspections(
|
||
[
|
||
Inspection(
|
||
registration_number="INSP-2",
|
||
inn="7701002002",
|
||
ogrn="1027700200002",
|
||
organisation_name='ООО "Проверка 2"',
|
||
control_authority="Контроль",
|
||
inspection_type="Плановая",
|
||
inspection_form="Документарная",
|
||
start_date="2026-04-01",
|
||
end_date="2026-04-15",
|
||
status="planned",
|
||
legal_basis="ФЗ",
|
||
result="",
|
||
is_federal_law_248=True,
|
||
),
|
||
],
|
||
batch_id=2,
|
||
is_federal_law_248=True,
|
||
data_year=2026,
|
||
data_month=4,
|
||
)
|
||
|
||
self.assertTrue(InspectionService.has_data_for_period(2026, 3))
|
||
self.assertFalse(InspectionService.has_data_for_period(2026, 3, True))
|
||
self.assertTrue(InspectionService.has_data_for_period(2026, 4, True))
|
||
|
||
def test_procurement_service_find_by_customer_name_with_batch(self):
|
||
ProcurementService.save_procurements(
|
||
[
|
||
Procurement(
|
||
purchase_number="PROC-1",
|
||
purchase_name="Поставка 1",
|
||
customer_inn="7701003001",
|
||
customer_kpp="770101001",
|
||
customer_ogrn="1027700300001",
|
||
customer_name="АО Тестовый заказчик",
|
||
max_price="1000",
|
||
currency_code="RUB",
|
||
placement_method="Аукцион",
|
||
publish_date="2026-03-01",
|
||
end_date="2026-03-15",
|
||
status="published",
|
||
law_type="44-FZ",
|
||
purchase_object_info="Оборудование",
|
||
href="https://example.test/proc-1",
|
||
)
|
||
],
|
||
batch_id=11,
|
||
)
|
||
ProcurementService.save_procurements(
|
||
[
|
||
Procurement(
|
||
purchase_number="PROC-2",
|
||
purchase_name="Поставка 2",
|
||
customer_inn="7701003002",
|
||
customer_kpp="770101002",
|
||
customer_ogrn="1027700300002",
|
||
customer_name="АО Тестовый заказчик",
|
||
max_price="2000",
|
||
currency_code="RUB",
|
||
placement_method="Аукцион",
|
||
publish_date="2026-04-01",
|
||
end_date="2026-04-15",
|
||
status="published",
|
||
law_type="44-FZ",
|
||
purchase_object_info="Оборудование",
|
||
href="https://example.test/proc-2",
|
||
)
|
||
],
|
||
batch_id=12,
|
||
)
|
||
|
||
self.assertEqual(
|
||
ProcurementService.find_by_customer_name("Тестовый").count(), 2
|
||
)
|
||
self.assertEqual(
|
||
ProcurementService.find_by_customer_name("Тестовый", batch_id=11).count(),
|
||
1,
|
||
)
|
||
|
||
|
||
class FNSReportServiceHelpersTest(TestCase):
|
||
def test_exists_find_and_status_helpers(self):
|
||
report = FNSReportService.save_report(
|
||
external_id="EXT-100",
|
||
ogrn="1027700111111",
|
||
file_name="fin_EXT-100_1027700111111.xlsx",
|
||
file_hash="a" * 64,
|
||
source="api",
|
||
batch_id=1,
|
||
lines_data=[],
|
||
)
|
||
|
||
self.assertTrue(FNSReportService.exists_by_external_id("EXT-100"))
|
||
self.assertFalse(FNSReportService.exists_by_external_id("EXT-404"))
|
||
self.assertEqual(FNSReportService.find_by_external_id("EXT-100").id, report.id)
|
||
|
||
FNSReportService.mark_processing(report)
|
||
report.refresh_from_db()
|
||
self.assertEqual(report.status, FinancialReport.Status.PROCESSING)
|
||
|
||
FNSReportService.mark_success(report)
|
||
report.refresh_from_db()
|
||
self.assertEqual(report.status, FinancialReport.Status.SUCCESS)
|
||
|
||
FNSReportService.mark_failed(report, "boom")
|
||
report.refresh_from_db()
|
||
self.assertEqual(report.status, FinancialReport.Status.FAILED)
|
||
self.assertEqual(report.payload["error_message"], "boom")
|