from __future__ import annotations from app.pipeline.mapping import ( FIXED_REPORT_COLUMNS, _extract_tag_candidate_keys, build_fixed_row, ) def _column_with(prefix: str) -> str: for column in FIXED_REPORT_COLUMNS: if column.startswith(prefix): return column raise AssertionError(f"Колонка с префиксом '{prefix}' не найдена") def test_build_fixed_row_maps_known_fields() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=2, operation_fields={ "ИдФайл": "FILE_01", "КодОперации": "5007", "СумОперации": "100.55", }, participant_fields={ "ИННФЛ": "123456789012", "Тип": "01", }, ) assert row["Имя XML файла"] == "f.xml" assert row["Идентификатор записи"] == "R1" assert row["Уникальный номер операции"] == "2" assert row["Код вида операции"] == "5007" assert row["Сумма в валюте проведения"] == "100.55" assert row[_column_with("ИНН ")] == "123456789012" assert row["Тип участника"] == "01" def test_build_fixed_row_keeps_missing_values_empty() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={}, ) for column in FIXED_REPORT_COLUMNS: assert column in row assert row["Код вида операции"] == "" assert row[_column_with("ИНН ")] == "" def test_build_fixed_row_does_not_take_operation_inn_for_participant() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={ "ИНН": "0000000000", "Тип": "99", }, participant_fields={ "ИННФЛ": "111111111111", "Тип": "01", }, ) assert row[_column_with("ИНН ")] == "111111111111" assert row["Тип участника"] == "01" def test_fixed_report_columns_over_200() -> None: assert len(FIXED_REPORT_COLUMNS) == 204 assert FIXED_REPORT_COLUMNS[126] == "Признак принадлежности к публичным лицам" def test_build_fixed_row_uses_alternative_path_leaf_for_emitent_name() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={ "ФИОСтрока": "Эмитент ФЛ", }, participant_fields={}, ) assert row["Наименование/ ФИО эмитента ЦП"] == "Эмитент ФЛ" def test_build_fixed_row_maps_dotted_participant_keys() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={ "УчастникФЛИП.СведФЛИП.ИННФЛИП": "123456789012", "УчастникФЛИП.СведФЛИП.СведДокУдЛичн.НомДок": "451099", }, ) assert row[_column_with("ИНН ")] == "123456789012" assert row["Номер"] == "451099" def test_build_fixed_row_splits_pipe_separated_xml_tag_values() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={ "Индекс": "674500", }, participant_fields={}, ) assert row["Адрес по структуре (целый)"] == "674500" def test_extract_tag_candidate_keys_supports_comma_separator() -> None: candidates = _extract_tag_candidate_keys( "КППЮЛ, ИдентификацияФЛ, ПризнакОргФормаИНБОЮЛ" ) assert candidates == ("КППЮЛ", "ИдентификацияФЛ", "ПризнакОргФормаИНБОЮЛ") def test_build_fixed_row_uses_first_suffix_value_from_index() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={ "УчастникФЛИП.Первый.ИННФЛИП": "111111111111", "УчастникФЛИП.Второй.ИННФЛИП": "222222222222", }, ) assert row[_column_with("ИНН ")] == "111111111111" def test_build_fixed_row_skips_empty_suffix_values_in_index() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={ "УчастникФЛИП.Первый.ИННФЛИП": "", "УчастникФЛИП.Второй.ИННФЛИП": "222222222222", }, ) assert row[_column_with("ИНН ")] == "222222222222" def test_structured_address_is_built_from_right_columns() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={ "Индекс": "674500", "КодОКСМ": "643", "Пункт": "Москва", "Улица": "Тверская", "Дом": "1", }, ) assert row["Адрес по структуре (целый)"] == "674500, 643, Москва, Тверская, 1" def test_one_line_address_clears_structured_and_component_columns() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={ "АдресСтрока": "г. Москва, Тверская, 1", "Индекс": "674500", "КодОКСМ": "643", "Улица": "Тверская", "Дом": "1", }, ) assert row["Адрес в одной строке"] == "г. Москва, Тверская, 1" assert row["Адрес по структуре (целый)"] == "" assert row["Индекс"] == "" assert row["Код страны"] == "" assert row["Улица"] == "" assert row["Дом"] == "" def test_fio_columns_prefer_participant_fields_over_operation_common_fields() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={ "Фам": "Кузнецов", "Имя": "Сергей", "Отч": "Владимирович", }, participant_fields={ "УчастникФЛИП.СведФЛИП.ФИОФЛИП.Фам": "Азимов", "УчастникФЛИП.СведФЛИП.ФИОФЛИП.Имя": "Ислом", "УчастникФЛИП.СведФЛИП.ФИОФЛИП.Отч": "Каримович", }, ) assert row["Фамилия"] == "Азимов" assert row["Имя"] == "Ислом" assert row["Отчество"] == "Каримович" assert row["ФИО (по структуре полностью)"] == "Азимов Ислом Каримович" def test_one_line_fio_clears_structured_and_component_fio_columns() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={ "ФИОСтрока": "Азимов Ислом Каримович", "Фам": "Азимов", "Имя": "Ислом", "Отч": "Каримович", }, ) assert row["ФИО (в одной строке)"] == "Азимов Ислом Каримович" assert row["ФИО (по структуре полностью)"] == "" assert row["Фамилия"] == "" assert row["Имя"] == "" assert row["Отчество"] == "" assert row["Наименование ЮЛ/ФИО ( по структуре полный)"] == "" def test_structured_column_is_not_filled_directly_from_single_line_tag() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={ "АдресСтрока": "г. Москва, ул. Пример, д. 1", }, ) assert row["Адрес в одной строке"] == "г. Москва, ул. Пример, д. 1" assert ( row[ "Адрес ЮЛ/ Адрес места жительства (места нахождения) " "ЕИО/Бенефициара (по структуре полный)" ] == "" ) def test_structured_column_without_component_block_joins_candidate_values() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={ "Индекс": "121467", "КодОКСМ": "643", "Пункт": "Москва", "Улица": "Молодогвардейская", "Дом": "8", "Корп": "1", "Оф": "28", }, ) assert ( row["Место государственной регистрации ЕИО/Бенефициара (структура полностью)"] == "121467, 643, Москва, Молодогвардейская, 8, 1, 28" ) def test_structured_column_uses_non_one_line_candidate_when_mixed_tags() -> None: row = build_fixed_row( file_name="f.xml", record_id="R1", operation_index=1, operation_fields={}, participant_fields={ "АдрУчредитель": "г. Москва, ул. Пушкина, д. 1", }, ) assert row["Адрес в одной строке"] == "" assert ( row[ "Адрес ЮЛ/ Адрес места жительства (места нахождения) " "ЕИО/Бенефициара (по структуре полный)" ] == "г. Москва, ул. Пушкина, д. 1" )