"""Tests for deterministic frontend demo company management commands.""" from io import StringIO from apps.exchange.state_corp_services import StateCorpExchangeService from apps.parsers.models import ( FinancialReport, FinancialReportLine, GenericParserRecord, IndustrialCertificateRecord, IndustrialProductRecord, InspectionRecord, ManufacturerRecord, ParserLoadLog, ProcurementRecord, ) from django.core.management import call_command from django.test import TestCase, override_settings from django.utils import timezone from organizations.models import ( Organization, OrganizationSourceExtension, OrganizationSourceFinancialLine, OrganizationSourceRecord, SourceGroup, ) from organizations.test_companies import ( TEST_BALANCE_LINE_NAMES, TEST_FINANCIAL_HISTORY_YEARS, TEST_FINANCIAL_LINES_PER_YEAR, TEST_PROFIT_LOSS_LINE_NAMES, TEST_RECORD_PREFIX, ) TEST_EXCHANGE_TOKEN = "test-exchange-token" # noqa: S105 class TestCompaniesCommandsTest(TestCase): """Checks creation, refresh, and removal of the fixed demo dataset.""" def test_create_builds_twenty_companies_with_every_source_dataset(self): call_command("create_test_companies", stdout=StringIO()) companies = Organization.objects.filter(name__startswith="Тестовая компания ") self.assertEqual(companies.count(), 20) self.assertEqual( set(companies.values_list("name", flat=True)), {f"Тестовая компания {index}" for index in range(1, 21)}, ) self.assertEqual(companies.filter(opk_registry_membership=True).count(), 20) self.assertEqual( companies.filter( gk_code="2", gk_name='Госкорпорация "Росатом"', ).count(), 10, ) self.assertEqual( companies.filter( gk_code="1", gk_name='Госкорпорация "Роскосмос"', ).count(), 10, ) expected_groups = {choice.value for choice in SourceGroup} expected_sources = {choice.value for choice in ParserLoadLog.Source} for company in companies: self.assertEqual( set(company.source_extensions.values_list("source_group", flat=True)), expected_groups, ) self.assertEqual( set( OrganizationSourceRecord.objects.filter( extension__organization=company ).values_list("source", flat=True) ), expected_sources, ) self.assertEqual( OrganizationSourceExtension.objects.filter( organization__in=companies ).count(), 20 * len(expected_groups), ) self.assertEqual( OrganizationSourceRecord.objects.filter( extension__organization__in=companies ).count(), 20 * len(expected_sources), ) self.assertEqual( OrganizationSourceFinancialLine.objects.filter( source_record__extension__organization__in=companies ).count(), 20 * TEST_FINANCIAL_LINES_PER_YEAR * TEST_FINANCIAL_HISTORY_YEARS, ) current_year = timezone.localdate().year expected_financial_years = set( range( current_year - TEST_FINANCIAL_HISTORY_YEARS + 1, current_year + 1, ) ) self.assertEqual( set( OrganizationSourceFinancialLine.objects.filter( source_record__extension__organization__in=companies ).values_list("year", flat=True) ), expected_financial_years, ) self.assertEqual(IndustrialCertificateRecord.objects.count(), 20) self.assertEqual(IndustrialProductRecord.objects.count(), 20) self.assertEqual(ManufacturerRecord.objects.count(), 20) self.assertEqual(InspectionRecord.objects.count(), 20) self.assertEqual(ProcurementRecord.objects.count(), 20) self.assertEqual(FinancialReport.objects.count(), 20) self.assertEqual( set(FinancialReport.objects.values_list("external_id", flat=True)), { f"{TEST_RECORD_PREFIX}:{index:02d}:{ParserLoadLog.Source.FNS_REPORTS}" for index in range(1, 21) }, ) self.assertEqual( FinancialReportLine.objects.count(), 20 * TEST_FINANCIAL_LINES_PER_YEAR * TEST_FINANCIAL_HISTORY_YEARS, ) self.assertEqual( set(FinancialReportLine.objects.values_list("year", flat=True)), expected_financial_years, ) self.assertEqual(GenericParserRecord.objects.count(), 20 * 9) def test_create_updates_the_fixed_dataset_without_duplicates(self): call_command("create_test_companies", stdout=StringIO()) company_uids = set( Organization.objects.filter( name__startswith="Тестовая компания " ).values_list("uid", flat=True) ) first_company = Organization.objects.get(name="Тестовая компания 1") first_company.name = "Поврежденное тестовое имя" first_company.gk_name = "Неверная корпорация" first_company.save(update_fields=["name", "gk_name"]) record = OrganizationSourceRecord.objects.filter( extension__organization=first_company, source=ParserLoadLog.Source.INSPECTIONS, ).get() record.title = "Устаревшие тестовые данные" record.payload = {"stale": True} record.save(update_fields=["title", "payload"]) financial_report = FinancialReport.objects.get( registry_organization=first_company ) financial_report.external_id = "test-company-01" financial_report.save(update_fields=["external_id"]) fns_record = OrganizationSourceRecord.objects.get( extension__organization=first_company, source=ParserLoadLog.Source.FNS_REPORTS, ) stale_year = timezone.localdate().year - TEST_FINANCIAL_HISTORY_YEARS OrganizationSourceFinancialLine.objects.create( source_record=fns_record, form_code="1", line_code="1600", line_name="Устаревший баланс", year=stale_year, period_start=1, period_end=2, ) FinancialReportLine.objects.create( report=financial_report, form_code="1", line_code="1600", line_name="Устаревший баланс", year=stale_year, period_start=1, period_end=2, ) call_command("create_test_companies", stdout=StringIO()) refreshed_uids = set( Organization.objects.filter( name__startswith="Тестовая компания " ).values_list("uid", flat=True) ) self.assertEqual(refreshed_uids, company_uids) self.assertEqual(len(refreshed_uids), 20) first_company.refresh_from_db() self.assertEqual(first_company.name, "Тестовая компания 1") self.assertEqual(first_company.gk_name, 'Госкорпорация "Росатом"') record.refresh_from_db() self.assertNotEqual(record.title, "Устаревшие тестовые данные") self.assertIn("registration_number", record.payload) financial_report.refresh_from_db() self.assertEqual( financial_report.external_id, f"{TEST_RECORD_PREFIX}:01:{ParserLoadLog.Source.FNS_REPORTS}", ) self.assertFalse( OrganizationSourceFinancialLine.objects.filter( source_record=fns_record, year=stale_year, ).exists() ) self.assertFalse( FinancialReportLine.objects.filter( report=financial_report, year=stale_year, ).exists() ) self.assertEqual(OrganizationSourceRecord.objects.count(), 20 * 17) def test_create_replaces_stale_source_record_for_same_legacy_row(self): call_command("create_test_companies", stdout=StringIO()) company = Organization.objects.get(name="Тестовая компания 1") stale_record = OrganizationSourceRecord.objects.get( extension__organization=company, source=ParserLoadLog.Source.PROCUREMENTS, ) stale_record.external_id = "0373200000000000001" stale_record.save(update_fields=["external_id"]) call_command("create_test_companies", stdout=StringIO()) canonical_record = OrganizationSourceRecord.objects.get( extension__organization=company, source=ParserLoadLog.Source.PROCUREMENTS, ) self.assertEqual( canonical_record.external_id, f"{TEST_RECORD_PREFIX}:01:{ParserLoadLog.Source.PROCUREMENTS}", ) self.assertEqual( canonical_record.legacy_pk, str( ProcurementRecord.objects.get(purchase_number="0373200000000000001").pk ), ) @override_settings(STATE_CORP_EXCHANGE_TOKEN=TEST_EXCHANGE_TOKEN) def test_created_source_records_are_included_in_state_corp_package(self): call_command("create_test_companies", stdout=StringIO()) company_inns = list( Organization.objects.filter(name__startswith="Тестовая компания ") .order_by("rn") .values_list("inn", flat=True) ) package = StateCorpExchangeService.build_package(organization_inns=company_inns) reports = StateCorpExchangeService._serialize_financial_reports( { organization.ogrn: organization.inn for organization in Organization.objects.filter(inn__in=company_inns) } ) self.assertEqual( package.payload_counts, { "organizations": 20, "industrial_certificates": 20, "manufacturers": 20, "industrial_products": 20, "prosecutor_checks": 20, "public_procurements": 60, "financial_reports": 20, "arbitration_cases": 20, "bankruptcy_procedures": 20, "defense_unreliable_suppliers": 40, "information_security_registries": 20, "labor_vacancies": 20, "electronic_document_exchange": 20, "media_mentions": 20, }, ) self.assertEqual(len(reports), 20) self.assertEqual( sum(len(report["lines"]) for report in reports), 20 * TEST_FINANCIAL_LINES_PER_YEAR * TEST_FINANCIAL_HISTORY_YEARS, ) self.assertEqual( {line["year"] for report in reports for line in report["lines"]}, set( range( timezone.localdate().year - TEST_FINANCIAL_HISTORY_YEARS + 1, timezone.localdate().year + 1, ) ), ) def test_financial_statements_are_complete_balanced_and_periodic(self): call_command("create_test_companies", stdout=StringIO()) company = Organization.objects.get(name="Тестовая компания 1") record = OrganizationSourceRecord.objects.get( extension__organization=company, source=ParserLoadLog.Source.FNS_REPORTS, ) lines = list( OrganizationSourceFinancialLine.objects.filter( source_record=record ).order_by("year", "form_code", "line_code") ) self.assertEqual( len(lines), TEST_FINANCIAL_LINES_PER_YEAR * TEST_FINANCIAL_HISTORY_YEARS, ) self.assertTrue( all( line.period_start is not None and line.period_end is not None for line in lines ) ) lines_by_year = {} for line in lines: lines_by_year.setdefault(line.year, {})[ (line.form_code, line.line_code) ] = line expected_codes = { *(("1", code) for code in TEST_BALANCE_LINE_NAMES), *(("2", code) for code in TEST_PROFIT_LOSS_LINE_NAMES), } ordered_years = sorted(lines_by_year) for year in ordered_years: year_lines = lines_by_year[year] self.assertEqual(set(year_lines), expected_codes) values = {key: line.period_end for key, line in year_lines.items()} self.assertEqual( values[("1", "1100")], sum( values[("1", code)] for code in ( "1110", "1120", "1130", "1140", "1150", "1160", "1170", "1180", "1190", ) ), ) self.assertEqual( values[("1", "1200")], sum( values[("1", code)] for code in ("1210", "1220", "1230", "1240", "1250", "1260") ), ) self.assertEqual( values[("1", "1600")], values[("1", "1100")] + values[("1", "1200")], ) self.assertEqual( values[("1", "1300")], sum( values[("1", code)] for code in ("1310", "1320", "1340", "1350", "1360", "1370") ), ) self.assertEqual( values[("1", "1400")], sum(values[("1", code)] for code in ("1410", "1420", "1430", "1450")), ) self.assertEqual( values[("1", "1500")], sum( values[("1", code)] for code in ("1510", "1520", "1530", "1540", "1550") ), ) self.assertEqual(values[("1", "1600")], values[("1", "1700")]) self.assertEqual( values[("2", "2100")], values[("2", "2110")] - values[("2", "2120")], ) self.assertEqual( values[("2", "2200")], values[("2", "2100")] - values[("2", "2210")] - values[("2", "2220")], ) self.assertEqual( values[("2", "2300")], values[("2", "2200")] + values[("2", "2310")] + values[("2", "2320")] - values[("2", "2330")] + values[("2", "2340")] - values[("2", "2350")], ) self.assertEqual( values[("2", "2410")], -values[("2", "2411")] + values[("2", "2412")], ) self.assertEqual( values[("2", "2400")], values[("2", "2300")] + values[("2", "2410")] + values[("2", "2460")], ) self.assertEqual( values[("2", "2500")], values[("2", "2400")] + values[("2", "2510")] + values[("2", "2520")] + values[("2", "2530")], ) for previous_year, current_year in zip( ordered_years, ordered_years[1:], strict=False, ): for key, current_line in lines_by_year[current_year].items(): self.assertEqual( current_line.period_start, lines_by_year[previous_year][key].period_end, ) def test_delete_removes_only_the_fixed_twenty_companies(self): untouched = Organization.objects.create( name="Тестовая компания 99", inn="7700000099", ogrn="1027700000099", opk_registry_membership=True, ) call_command("create_test_companies", stdout=StringIO()) call_command("delete_test_companies", stdout=StringIO()) self.assertFalse( Organization.objects.filter( name__in=[f"Тестовая компания {index}" for index in range(1, 21)] ).exists() ) self.assertTrue(Organization.objects.filter(uid=untouched.uid).exists()) self.assertEqual(OrganizationSourceExtension.objects.count(), 0) self.assertEqual(OrganizationSourceRecord.objects.count(), 0) self.assertEqual(GenericParserRecord.objects.count(), 0) self.assertEqual(IndustrialProductRecord.objects.count(), 0) self.assertEqual(FinancialReport.objects.count(), 0)