mostovik-backend/tests/apps/organizations/test_directory_import.py

"""Tests for authoritative organization directory import."""

from decimal import Decimal
from tempfile import NamedTemporaryFile

from django.test import TestCase
from openpyxl import Workbook
from organizations.directory_import import (
    DIRECTORY_SHEET,
    SOURCE_HEADERS,
    OrganizationDirectoryImportError,
    OrganizationDirectoryImportService,
)
from organizations.models import Organization


class OrganizationDirectoryImportServiceTest(TestCase):
    """Checks importing the authoritative XLSX organization directory."""

    def test_import_xlsx_creates_and_updates_organizations_by_rn(self):
        path = self._workbook_path(
            [
                self._row(
                    rn="10",
                    full_name='Акционерное общество "Тест"',
                    short_name='АО "Тест"',
                    inn="7701001001",
                    ogrn="1027700100001",
                    okpo="00123456",
                    filial=".F.",
                    kpp="770101001",
                    registration_date="01.02.2020",
                    authorized_capita=1000,
                    business_act_cod="2",
                    goz_participation=".T.",
                    opk_registry_membership=".F.",
                    _gk="1",
                    in_korp="1",
                    _k="2",
                )
            ]
        )

        first = OrganizationDirectoryImportService.import_xlsx(path)
        second = OrganizationDirectoryImportService.import_xlsx(path)

        self.assertEqual(first.scanned, 1)
        self.assertEqual(first.created, 1)
        self.assertEqual(second.updated, 1)
        self.assertEqual(Organization.objects.count(), 1)

        organization = Organization.objects.get(rn=10)
        self.assertEqual(organization.name, 'Акционерное общество "Тест"')
        self.assertEqual(organization.okpo, "00123456")
        self.assertFalse(organization.is_branch)
        self.assertEqual(organization.authorized_capital, Decimal("1000.00"))
        self.assertEqual(organization.business_activity, "Производственная")
        self.assertEqual(organization.gk_name, "Роскосмос")
        self.assertEqual(organization.in_korp_name, "Входит в состав")
        self.assertEqual(organization.state_sector_name, "Головные исполнители")
        self.assertTrue(organization.goz_participation)
        self.assertFalse(organization.opk_registry_membership)

    def test_import_xlsx_rejects_missing_required_column(self):
        path = self._workbook_path([], headers=SOURCE_HEADERS[:-1])

        with self.assertRaisesMessage(
            OrganizationDirectoryImportError,
            "Missing required columns",
        ):
            OrganizationDirectoryImportService.import_xlsx(path)

    def test_import_xlsx_treats_spaced_dot_date_marker_as_empty(self):
        path = self._workbook_path(
            [
                self._row(
                    rn="11",
                    registration_date=".  .",
                    appointment_date=".  .",
                    filial=".T.",
                )
            ]
        )

        result = OrganizationDirectoryImportService.import_xlsx(path)

        self.assertEqual(result.created, 1)
        organization = Organization.objects.get(rn=11)
        self.assertIsNone(organization.registration_date)
        self.assertIsNone(organization.appointment_date)
        self.assertTrue(organization.is_branch)

    def test_import_xlsx_recovers_embedded_tsv_organization_row(self):
        first_tail = [
            *[
                self._row(
                    rn="20",
                    pn_name_en="First EN",
                    inn="7701001020",
                    ogrn="1027700100020",
                    okpo="00000020",
                    filial=".F.",
                    kpp="770101020",
                    registration_date="01.01.2020",
                    _k="2",
                )[header]
                for header in SOURCE_HEADERS[6:-1]
            ],
            "20",
            "2",
        ]
        embedded_value = "\t".join(str(value or "") for value in first_tail)
        embedded_value = (
            embedded_value
            + "_x000D_\n"
            + "\t".join(["21", "0", "0", 'ООО "Вторая"'])
        )

        second_row = self._row(
            rn="21",
            full_name='ООО "Вторая"',
            short_name='ООО "Вторая"',
            pn_name='ООО "Вторая"',
            inn="7701001021",
            ogrn="1027700100021",
            okpo="00000021",
            filial=".F.",
            kpp="770101021",
            registration_date="02.01.2020",
            _k="2",
        )
        second_row["is"] = "21"
        path = self._workbook_path(
            [
                {
                    "__raw_row__": [
                        "20",
                        "0",
                        "0",
                        'ООО "Первая"',
                        'ООО "Первая"',
                        'ООО "Первая"',
                        embedded_value,
                        *[second_row[header] for header in SOURCE_HEADERS[4:]],
                    ]
                }
            ]
        )

        result = OrganizationDirectoryImportService.import_xlsx(path)

        self.assertEqual(result.scanned, 1)
        self.assertEqual(result.created, 2)
        self.assertEqual(Organization.objects.count(), 2)
        self.assertEqual(Organization.objects.get(rn=20).inn, "7701001020")
        second = Organization.objects.get(rn=21)
        self.assertEqual(second.inn, "7701001021")
        self.assertEqual(second.integrated_structure, "")

    def _workbook_path(
        self,
        rows: list[dict[str, object]],
        *,
        headers: list[str] | None = None,
    ) -> str:
        workbook = Workbook()
        worksheet = workbook.active
        worksheet.title = DIRECTORY_SHEET
        worksheet.append(headers or SOURCE_HEADERS)
        for values in rows:
            raw_row = values.get("__raw_row__")
            if raw_row is not None:
                worksheet.append(raw_row)
            else:
                worksheet.append([values.get(header, "") for header in SOURCE_HEADERS])

        self._append_reference_sheet(workbook, "_gk", {"1": "Роскосмос"})
        self._append_reference_sheet(workbook, "in_korp", {"1": "Входит в состав"})
        self._append_reference_sheet(
            workbook,
            "business_act_cod",
            {"2": "Производственная"},
        )
        self._append_reference_sheet(
            workbook,
            "_k - Госектор",
            {"2": "Головные исполнители"},
        )

        temp_file = NamedTemporaryFile(suffix=".xlsx", delete=False)
        temp_file.close()
        workbook.save(temp_file.name)
        return temp_file.name

    @staticmethod
    def _append_reference_sheet(
        workbook: Workbook,
        title: str,
        values: dict[str, str],
    ) -> None:
        worksheet = workbook.create_sheet(title=title)
        worksheet.append(["kod", "value"])
        for key, value in values.items():
            worksheet.append([key, value])

    @staticmethod
    def _row(**overrides) -> dict[str, object]:
        row = {header: "" for header in SOURCE_HEADERS}
        row.update(
            {
                "rn": "1",
                "full_name": 'ООО "По умолчанию"',
                "inn": "7701001000",
                "ogrn": "1027700100000",
            }
        )
        row.update(overrides)
        return row