""" Тесты для FNS парсера бухгалтерской отчетности. """ from pathlib import Path from tempfile import TemporaryDirectory from apps.parsers.clients.fns.parser import FNSExcelParser, FNSParserError from apps.parsers.clients.fns.schemas import ParsedReport, ReportLine from apps.parsers.models import FinancialReport from apps.parsers.services import ( FNSReportOrganizationResolutionSkipped, FNSReportService, ) from django.test import TestCase from openpyxl import Workbook from organizations.models import Organization from tests.apps.parsers.organization_helpers import create_directory_organization from tests.utils.fixtures import fake def _digits(length: int) -> str: return "".join(str(fake.random_int(0, 9)) for _ in range(length)) def _year() -> int: return fake.random_int(min=2020, max=2026) def _line_code() -> str: return _digits(4) def _form_code() -> str: return str(fake.random_int(min=1, max=6)) def _create_registry_organization(*, inn: str, ogrn: str) -> Organization: return create_directory_organization( name=fake.company(), ogrn=ogrn, inn=inn, kpp=_digits(9), okpo=_digits(8), ) class TestFNSExcelParserFilename(TestCase): """Тесты парсинга имени файла.""" def test_parse_valid_filename(self): """Корректное имя файла.""" external_id = _digits(7) ogrn = _digits(13) external_id_parsed, ogrn_parsed = FNSExcelParser.parse_filename( f"fin_{external_id}_{ogrn}.xlsx" ) self.assertEqual(external_id_parsed, external_id) self.assertEqual(ogrn_parsed, ogrn) def test_parse_filename_with_long_id(self): """Имя файла с длинным ID.""" external_id = _digits(8) ogrn = _digits(13) external_id_parsed, ogrn_parsed = FNSExcelParser.parse_filename( f"fin_{external_id}_{ogrn}.xlsx" ) self.assertEqual(external_id_parsed, external_id) self.assertEqual(ogrn_parsed, ogrn) def test_parse_filename_with_15_digit_ogrn(self): """Имя файла с 15-значным ОГРН (ИП).""" external_id = _digits(3) ogrn = _digits(15) external_id_parsed, ogrn_parsed = FNSExcelParser.parse_filename( f"fin_{external_id}_{ogrn}.xlsx" ) self.assertEqual(external_id_parsed, external_id) self.assertEqual(ogrn_parsed, ogrn) def test_parse_invalid_filename_no_fin_prefix(self): """Неверный формат - без префикса fin.""" with self.assertRaises(FNSParserError): FNSExcelParser.parse_filename( f"{fake.word()}_{_digits(3)}_{_digits(13)}.xlsx" ) def test_parse_invalid_filename_wrong_extension(self): """Неверный формат - неправильное расширение.""" with self.assertRaises(FNSParserError): FNSExcelParser.parse_filename(f"fin_{_digits(3)}_{_digits(13)}.xls") def test_parse_invalid_filename_short_ogrn(self): """Неверный формат - короткий ОГРН.""" with self.assertRaises(FNSParserError): FNSExcelParser.parse_filename(f"fin_{_digits(3)}_{_digits(12)}.xlsx") class TestReportLineSchema(TestCase): """Тесты схемы ReportLine.""" def test_create_report_line(self): """Создание строки отчета.""" form_code = _form_code() line_code = _line_code() line_name = fake.word() year = _year() period_start = fake.random_int(min=10, max=1000) period_end = fake.random_int(min=10, max=1000) line = ReportLine( form_code=form_code, line_code=line_code, line_name=line_name, year=year, period_start=period_start, period_end=period_end, ) self.assertEqual(line.form_code, form_code) self.assertEqual(line.line_code, line_code) self.assertEqual(line.year, year) self.assertEqual(line.period_start, period_start) self.assertEqual(line.period_end, period_end) def test_report_line_with_none_values(self): """Строка отчета с пустыми значениями.""" line = ReportLine( form_code=_form_code(), line_code=_line_code(), line_name=fake.word(), year=_year(), ) self.assertIsNone(line.period_start) self.assertIsNone(line.period_end) class TestParsedReportSchema(TestCase): """Тесты схемы ParsedReport.""" def test_create_parsed_report(self): """Создание отчета.""" external_id = _digits(3) ogrn = _digits(13) year_a = _year() year_b = year_a - 1 if year_a > 2020 else year_a + 1 year_c = year_a + 1 if year_a < 2026 else year_a - 1 report = ParsedReport( external_id=external_id, ogrn=ogrn, lines=[ ReportLine(_form_code(), _line_code(), fake.word(), year_a, 100, 200), ReportLine(_form_code(), _line_code(), fake.word(), year_b, 50, 100), ReportLine(_form_code(), _line_code(), fake.word(), year_c, 1000, 1500), ], ) self.assertEqual(report.external_id, external_id) self.assertEqual(report.ogrn, ogrn) self.assertEqual(len(report.lines), 3) def test_report_years(self): """Получение годов из отчета.""" year_a = _year() year_b = year_a - 1 if year_a > 2020 else year_a + 1 year_c = year_a + 1 if year_a < 2026 else year_a - 1 report = ParsedReport( external_id=_digits(3), ogrn=_digits(13), lines=[ ReportLine(_form_code(), _line_code(), fake.word(), year_a, 100, 200), ReportLine(_form_code(), _line_code(), fake.word(), year_b, 50, 100), ReportLine(_form_code(), _line_code(), fake.word(), year_c, 30, 50), ], ) self.assertEqual(report.years, {year_a, year_b, year_c}) def test_report_forms(self): """Получение форм из отчета.""" form_a = _form_code() form_b = _form_code() form_c = _form_code() report = ParsedReport( external_id=_digits(3), ogrn=_digits(13), lines=[ ReportLine(form_a, _line_code(), fake.word(), _year(), 100, 200), ReportLine(form_b, _line_code(), fake.word(), _year(), 1000, 1500), ReportLine(form_c, _line_code(), fake.word(), _year(), 500, 600), ], ) self.assertEqual(report.forms, {form_a, form_b, form_c}) def test_get_lines_by_form(self): """Фильтрация строк по форме.""" form_target = _form_code() form_other = _form_code() while form_other == form_target: form_other = _form_code() report = ParsedReport( external_id=_digits(3), ogrn=_digits(13), lines=[ ReportLine(form_target, _line_code(), fake.word(), _year(), 100, 200), ReportLine(form_other, _line_code(), fake.word(), _year(), 1000, 1500), ReportLine(form_target, _line_code(), fake.word(), _year(), 300, 400), ], ) form1_lines = report.get_lines_by_form(form_target) self.assertEqual(len(form1_lines), 2) self.assertTrue(all(line.form_code == form_target for line in form1_lines)) def test_get_lines_by_year(self): """Фильтрация строк по году.""" year_target = _year() year_other = year_target - 1 if year_target > 2020 else year_target + 1 report = ParsedReport( external_id=_digits(3), ogrn=_digits(13), lines=[ ReportLine( _form_code(), _line_code(), fake.word(), year_target, 100, 200 ), ReportLine( _form_code(), _line_code(), fake.word(), year_other, 50, 100 ), ReportLine( _form_code(), _line_code(), fake.word(), year_target, 150, 250 ), ], ) year_lines = report.get_lines_by_year(year_target) self.assertEqual(len(year_lines), 2) self.assertTrue(all(line.year == year_target for line in year_lines)) class TestFNSParserParseValue(TestCase): """Тесты метода _parse_value.""" def test_parse_integer(self): """Парсинг целого числа.""" value = fake.random_int(min=1, max=1000) self.assertEqual(FNSExcelParser._parse_value(value), value) def test_parse_float(self): """Парсинг числа с плавающей точкой.""" value = fake.random_int(min=1, max=1000) self.assertEqual(FNSExcelParser._parse_value(value + 0.5), value) def test_parse_string_integer(self): """Парсинг строкового числа.""" value = fake.random_int(min=1, max=1000) self.assertEqual(FNSExcelParser._parse_value(str(value)), value) def test_parse_string_with_comma(self): """Парсинг числа с запятой.""" value = fake.random_int(min=1, max=1000) self.assertEqual(FNSExcelParser._parse_value(f"{value},5"), value) def test_parse_string_with_spaces(self): """Парсинг числа с пробелами.""" value = fake.random_int(min=1000, max=999999) formatted = f"{value:,}".replace(",", " ") self.assertEqual(FNSExcelParser._parse_value(formatted), value) def test_parse_none(self): """Парсинг None.""" self.assertIsNone(FNSExcelParser._parse_value(None)) def test_parse_empty_string(self): """Парсинг пустой строки.""" self.assertIsNone(FNSExcelParser._parse_value("")) def test_parse_dash(self): """Парсинг прочерка.""" self.assertIsNone(FNSExcelParser._parse_value("-")) def test_parse_invalid_string(self): """Парсинг некорректной строки.""" self.assertIsNone(FNSExcelParser._parse_value(fake.word())) class TestFNSExcelParserWorkbook(TestCase): """Регрессии поддерживаемых структур бухгалтерских XLSX.""" def test_parses_workbook_without_line_code_column(self): with TemporaryDirectory() as temp_directory: path = Path(temp_directory) / "fin_0130160_1187700006273.xlsx" workbook = Workbook() worksheet = workbook.active worksheet.append(["Форма №1", 2020, None, 2021, None]) worksheet.append( ["Бухгалтерский баланс", "Начало", "Конец", "Начало", "Конец"] ) worksheet.append(["Баланс", 60_232, 234_841, 234_841, 244_479]) worksheet.append(["Внеоборотные активы", None, None, None, None]) worksheet.append(["Итого", 1_878, 7_071, 7_071, 18_709]) worksheet.append(["Форма №3", 2020, None, 2021, None]) worksheet.append( [ "Отчет об изменении капитала", "Начало", "Конец", "Начало", "Конец", ] ) worksheet.append(["Добавочный капитал", None, None, None, None]) worksheet.append(["Увеличение - всего", 0, 10, 10, 20]) worksheet.append(["переоценка имущества", 0, 10, 10, 20]) worksheet.append(["Уменьшение - всего", 0, 5, 5, 6]) worksheet.append(["реорганизация ЮЛ", 0, 5, 5, 6]) worksheet.append(["Резервный капитал", None, None, None, None]) worksheet.append(["переоценка имущества", 0, -5, -5, -6]) worksheet.append(["Форма №4", 2020, None, 2021, None]) worksheet.append( [ "Отчет о движении денежных средств", "Начало", "Конец", "Начало", "Конец", ] ) worksheet.append(["Денежные потоки от инвестиционных операций", None, None]) worksheet.append(["Поступления – всего", 0, 100, 100, 200]) worksheet.append(["Сальдо", 0, 100, 100, 200]) worksheet.append(["Форма №6", 2020, None, 2021, None]) worksheet.append( [ "Отчет о целевом использовании полученных средств", "Начало", "Конец", "Начало", "Конец", ] ) worksheet.append(["Поступило средств", None, None, None, None]) worksheet.append(["Итого", 0, 10, 10, 20]) worksheet.append(["Использовано средств", None, None, None, None]) worksheet.append(["Итого", 0, 5, 5, 6]) workbook.save(path) report = FNSExcelParser.parse_file(path) self.assertEqual(report.external_id, "0130160") self.assertEqual(report.ogrn, "1187700006273") self.assertEqual(report.years, {2020, 2021}) self.assertEqual( {line.line_code for line in report.lines}, { "1100", "1600", "3310", "3312", "3320", "3322", "3326", "4200", "4210", "6200", "6300", }, ) def test_rejects_codeless_value_row_without_known_accounting_code(self): with TemporaryDirectory() as temp_directory: path = Path(temp_directory) / "fin_0000001_1187700006273.xlsx" workbook = Workbook() worksheet = workbook.active worksheet.append(["Форма №1", 2021, None]) worksheet.append(["Бухгалтерский баланс", "Начало", "Конец"]) worksheet.append(["Неизвестная строка", 1, 2]) workbook.save(path) with self.assertRaisesMessage(FNSParserError, "не определен код строки"): FNSExcelParser.parse_file(path) class TestFNSReportServiceIntegration(TestCase): """Интеграционные тесты сервиса FNSReportService.""" def test_save_report(self): """Сохранение отчета.""" external_id = _digits(6) ogrn = _digits(13) _create_registry_organization(inn=_digits(10), ogrn=ogrn) file_hash = fake.sha1(raw_output=False) lines_data = [ { "form_code": _form_code(), "line_code": _line_code(), "line_name": fake.word(), "year": _year(), "period_start": fake.random_int(min=10, max=1000), "period_end": fake.random_int(min=10, max=1000), }, { "form_code": _form_code(), "line_code": _line_code(), "line_name": fake.word(), "year": _year(), "period_start": fake.random_int(min=10, max=1000), "period_end": fake.random_int(min=10, max=1000), }, ] report = FNSReportService.save_report( external_id=external_id, ogrn=ogrn, file_name=f"fin_{external_id}_{ogrn}.xlsx", file_hash=file_hash, source=FinancialReport.SourceType.API, batch_id=1, lines_data=lines_data, ) self.assertIsNotNone(report.id) self.assertEqual(report.external_id, external_id) self.assertEqual(report.ogrn, ogrn) self.assertEqual(report.status, FinancialReport.Status.SUCCESS) self.assertEqual(report.lines.count(), 2) def test_save_report_links_registry_organization_when_exists(self): """Отчет ФНС должен связываться с организацией из справочника по ОГРН.""" ogrn = _digits(13) organization = _create_registry_organization(inn=_digits(10), ogrn=ogrn) external_id = _digits(6) report = FNSReportService.save_report( external_id=external_id, ogrn=ogrn, file_name=f"fin_{external_id}_{ogrn}.xlsx", file_hash=fake.sha1(raw_output=False), source=FinancialReport.SourceType.API, batch_id=1, lines_data=[], ) self.assertEqual(report.registry_organization_id, organization.id) def test_save_report_raises_skipped_when_organization_is_ambiguous(self): ogrn = _digits(13) _create_registry_organization(inn=_digits(10), ogrn=ogrn) _create_registry_organization(inn=_digits(10), ogrn=ogrn) with self.assertRaises(FNSReportOrganizationResolutionSkipped) as exc: FNSReportService.save_report( external_id=_digits(6), ogrn=ogrn, file_name=f"fin_{_digits(6)}_{ogrn}.xlsx", file_hash=fake.sha1(raw_output=False), source=FinancialReport.SourceType.API, batch_id=1, lines_data=[], ) self.assertEqual(exc.exception.reason, "ambiguous") def test_exists_by_hash(self): """Проверка существования по хешу.""" unique_hash = fake.sha1(raw_output=False) ogrn = _digits(13) _create_registry_organization(inn=_digits(10), ogrn=ogrn) FNSReportService.save_report( external_id=_digits(6), ogrn=ogrn, file_name=fake.file_name(extension="xlsx"), file_hash=unique_hash, source=FinancialReport.SourceType.API, batch_id=1, lines_data=[], ) self.assertTrue(FNSReportService.exists_by_hash(unique_hash)) self.assertFalse(FNSReportService.exists_by_hash(fake.sha1(raw_output=False))) def test_find_by_ogrn(self): """Поиск по ОГРН.""" ogrn = _digits(13) _create_registry_organization(inn=_digits(10), ogrn=ogrn) FNSReportService.save_report( external_id=_digits(6), ogrn=ogrn, file_name=fake.file_name(extension="xlsx"), file_hash=fake.sha1(raw_output=False), source=FinancialReport.SourceType.FILE_WATCH, batch_id=1, lines_data=[], ) FNSReportService.save_report( external_id=_digits(6), ogrn=ogrn, file_name=fake.file_name(extension="xlsx"), file_hash=fake.sha1(raw_output=False), source=FinancialReport.SourceType.FILE_WATCH, batch_id=1, lines_data=[], ) reports = FNSReportService.find_by_ogrn(ogrn) self.assertEqual(reports.count(), 2)