"""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, GenericParserRecord, IndustrialCertificateRecord, IndustrialProductRecord, InspectionRecord, ManufacturerRecord, ParserLoadLog, ProcurementRecord, ) from django.core.management import call_command from django.test import TestCase, override_settings from organizations.models import ( Organization, OrganizationSourceExtension, OrganizationSourceFinancialLine, OrganizationSourceRecord, SourceGroup, ) 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 * 4, ) 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(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"]) 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) self.assertEqual(OrganizationSourceRecord.objects.count(), 20 * 15) @override_settings(STATE_CORP_EXCHANGE_TOKEN=TEST_EXCHANGE_TOKEN) def test_created_source_records_are_exported_to_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) 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, }, ) 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)