""" Adversarial tests — написаны ДО исправлений. Часть из них должна упасть и показать реальные баги. Запуск: python3 -m pytest test_adversarial.py -v """ from __future__ import annotations import os from pathlib import Path from unittest.mock import MagicMock, patch import pytest import openpyxl import main from main import _normalize_kn, parse_pdf_extract, rename_extract, process_all, FIASSearcher from selenium.common.exceptions import WebDriverException FOUND_DATA = { "full_name": "ул. Тестовая, 1", "status": "Актуальный", "obj_type": "Здание", "fias_id": "aaaa-bbbb-cccc-dddd-eeee", "pdf_url": None, } def make_xlsx(tmp_path, rows, headers=("КН", "Вид ОН")) -> str: p = tmp_path / "in.xlsx" wb = openpyxl.Workbook() ws = wb.active ws.append(list(headers)) for r in rows: ws.append(list(r)) wb.save(str(p)) return str(p) def get_result_rows(tmp_path): files = list(Path(tmp_path).rglob("Результаты_ФИАС_*.xlsx")) assert files return list(openpyxl.load_workbook(str(files[0])).active.iter_rows(min_row=2, values_only=True)) def run(tmp_path, rows, headers=("КН", "Вид ОН"), searcher=None): xlsx = make_xlsx(tmp_path, rows, headers) if searcher is None: searcher = MagicMock() searcher.search.return_value = None with patch("main.FIASSearcher", return_value=searcher): process_all(xlsx, str(tmp_path), log=lambda _: None) return get_result_rows(tmp_path) def pdf_mock(text): page = MagicMock() page.extract_text.return_value = text ctx = MagicMock() ctx.__enter__ = lambda s: ctx ctx.__exit__ = MagicMock(return_value=False) ctx.pages = [page] return ctx # ───────────────────────────────────────────────────────────────────────────── # rename_extract — path safety # ───────────────────────────────────────────────────────────────────────────── class TestRenameExtractPathSafety: def _src(self, tmp_path): f = tmp_path / "_tmp.pdf" f.write_bytes(b"%PDF") return str(f) def test_slash_in_kn_does_not_escape_output_dir(self, tmp_path): """ Слэш в КН создаёт путь вида output_dir/ГАР_01/01_... — подпапки не существует → FileNotFoundError. БАГ: kn.replace(':', '_') не трогает '/'. """ result = rename_extract(self._src(tmp_path), "01/01:0001001:1", "25.03.2024", str(tmp_path)) assert Path(result).parent == tmp_path, ( f"Файл создан вне output_dir: {Path(result).parent}" ) def test_dotdot_in_kn_does_not_escape_output_dir(self, tmp_path): """ КН с '../' может вывести файл за пределы сессионной папки. """ result = rename_extract(self._src(tmp_path), "../evil:1:1:1", "25.03.2024", str(tmp_path)) assert tmp_path in Path(result).parents or Path(result).parent == tmp_path, ( f"Path traversal: {result}" ) def test_backslash_in_kn_does_not_break_path(self, tmp_path): """Обратный слэш (Windows-данные в Excel) не должен ломать путь.""" result = rename_extract(self._src(tmp_path), "77\\01:0001001:1", "25.03.2024", str(tmp_path)) assert Path(result).exists() # ───────────────────────────────────────────────────────────────────────────── # _normalize_kn — граничные случаи # ───────────────────────────────────────────────────────────────────────────── class TestNormalizeKnEdge: def test_only_colons(self): """':::' остаётся ':::' — не падает.""" assert _normalize_kn(":::") == ":::" def test_numeric_value_from_excel(self): """ Excel иногда хранит числа без двоеточий (авто-формат). Функция не должна добавлять двоеточия — это не КН. """ result = _normalize_kn(770100010001) # int, без двоеточий assert ":" not in result assert result == "770100010001" def test_float_from_excel(self): """Число с плавающей точкой из Excel — не должно превращаться в 'None'.""" result = _normalize_kn(77.0) assert result != "None" assert "." in result or result.isdigit() def test_empty_string(self): assert _normalize_kn("") == "" def test_newlines_inside_kn_removed(self): """Перенос строки внутри значения ячейки.""" assert _normalize_kn("77:01\n:0001001:1") == "77:01:0001001:1" # ───────────────────────────────────────────────────────────────────────────── # parse_pdf_extract — регрессии и граничные случаи # ───────────────────────────────────────────────────────────────────────────── class TestParsePdfExtractEdge: UUID = "12345678-abcd-abcd-abcd-123456789abc" def _parse(self, tmp_path, text): p = str(tmp_path / "t.pdf") Path(p).touch() with patch("pdfplumber.open", return_value=pdf_mock(text)): return parse_pdf_extract(p, log=lambda _: None) def test_multiple_uuids_keyword_match_beats_fallback(self, tmp_path): """ В тексте два UUID; у второго есть ключевое слово 'номер'. Должен быть выбран второй, а не первый. """ uuid_noise = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" uuid_correct = "11111111-2222-3333-4444-555555555555" _, num = self._parse(tmp_path, f"Прочее {uuid_noise}\nномер выписки {uuid_correct}") assert num == uuid_correct, f"Выбран неправильный UUID: {num}" def test_short_uuid_not_matched(self, tmp_path): """UUID с неправильным числом символов не должен матчиться.""" short = "12345678-abcd-abcd-abcd-12345678" # последняя группа слишком короткая _, num = self._parse(tmp_path, f"номер {short}") assert num != short def test_page_extract_text_returns_none(self, tmp_path): """ pdfplumber может вернуть None для страницы без текста. Код делает `or ""` — не должен падать. """ page = MagicMock() page.extract_text.return_value = None # явный None ctx = MagicMock() ctx.__enter__ = lambda s: ctx ctx.__exit__ = MagicMock(return_value=False) ctx.pages = [page] p = str(tmp_path / "t.pdf") Path(p).touch() with patch("pdfplumber.open", return_value=ctx): date, num = parse_pdf_extract(p, log=lambda _: None) assert date is None assert num is None def test_date_regex_does_not_validate_calendar(self, tmp_path): """ Регулярка проверяет только цифровой формат ДД.ММ.ГГГГ, а не реальность даты — '99.99.9999' пройдёт. Тест документирует это поведение (не баг, но нужно знать). """ date, _ = self._parse(tmp_path, "дата выписки 99.99.9999") assert date == "99.99.9999" # regexp не валидирует, принимает любые цифры # ───────────────────────────────────────────────────────────────────────────── # process_all — нештатные входные данные # ───────────────────────────────────────────────────────────────────────────── class TestProcessAllEdge: def test_short_data_rows_no_crash(self, tmp_path): """ Строка данных короче заголовка (vid_col существует, но ячейка пустая). openpyxl дополняет None — не должно быть IndexError. """ wb = openpyxl.Workbook() ws = wb.active ws.append(["КН", "Вид ОН"]) ws.append(["77:01:0001001:1"]) # только 1 ячейка, нет vid_on p = str(tmp_path / "in.xlsx") wb.save(p) ms = MagicMock() ms.search.return_value = None with patch("main.FIASSearcher", return_value=ms): process_all(p, str(tmp_path), log=lambda _: None) rows = get_result_rows(tmp_path) assert len(rows) == 1 assert rows[0][1] is None # vid_on должен быть None, не IndexError def test_header_at_row5_boundary_found(self, tmp_path): """Заголовок на строке 5 (граница поиска) — должен найтись.""" wb = openpyxl.Workbook() ws = wb.active for _ in range(4): ws.append(["мусор"]) ws.append(["КН"]) # строка 5 ws.append(["77:01:0001001:1"]) p = str(tmp_path / "in.xlsx") wb.save(p) ms = MagicMock() ms.search.return_value = FOUND_DATA with patch("main.FIASSearcher", return_value=ms): process_all(p, str(tmp_path), log=lambda _: None) ms.search.assert_called() def test_header_at_row6_not_found_logs_error(self, tmp_path): """Заголовок на строке 6 (за пределами поиска) — должен залогировать ошибку.""" wb = openpyxl.Workbook() ws = wb.active for _ in range(5): ws.append(["мусор"]) ws.append(["КН"]) # строка 6 — вне окна поиска ws.append(["77:01:0001001:1"]) p = str(tmp_path / "in.xlsx") wb.save(p) logs = [] ms = MagicMock() with patch("main.FIASSearcher", return_value=ms): process_all(p, str(tmp_path), log=logs.append) assert any("не найдена" in l for l in logs) ms.search.assert_not_called() def test_differently_formatted_duplicates_searched_once(self, tmp_path): """ '77:01:0001001:1' и '77 : 01 : 0001001 : 1' нормализуются одинаково — должны считаться дублями и вести к одному поиску (+ retry = max 2 вызовов). """ ms = MagicMock() ms.search.return_value = FOUND_DATA xlsx = make_xlsx(tmp_path, [ ("77:01:0001001:1",), ("77 : 01 : 0001001 : 1",), ], headers=("КН",)) with patch("main.FIASSearcher", return_value=ms): process_all(xlsx, str(tmp_path), log=lambda _: None) assert ms.search.call_count == 1, ( f"Ожидался 1 вызов (дубли дедуплицированы), получено {ms.search.call_count}" ) def test_exception_on_first_attempt_not_retried_marks_error(self, tmp_path): """ WebDriverException на 1-й попытке НЕ вызывает повтор — КН сразу пишется как 'ошибка'. Это ограничение текущей логики (retry работает только для None). """ attempts = {"n": 0} def boom(kn): attempts["n"] += 1 if attempts["n"] == 1: raise WebDriverException("stale") return FOUND_DATA # вторая попытка дала бы результат, но не дойдёт ms = MagicMock() ms.search.side_effect = boom rows = run(tmp_path, [("77:01:0001001:1", None)], searcher=ms) assert rows[0][2] == "ошибка" assert ms.search.call_count == 1 # второй шанс не даётся def test_kn_with_slash_in_excel_does_not_crash_process_all(self, tmp_path): """ Если ячейка Excel содержит КН со слэшем (некорректные данные), download_pdf не должен завершаться path traversal или крашем. Ошибка должна быть поймана и записана в Excel. """ data_with_pdf = {**FOUND_DATA, "pdf_url": "https://fias.nalog.ru/Export/ExportPdfStatement?id=1"} ms = MagicMock() ms.search.return_value = data_with_pdf ms.download_pdf.return_value = False # скачивание не удалось — не важно # Главное — не должно быть необработанного исключения rows = run(tmp_path, [("01/01:0001001:1", None)], searcher=ms) # Строка должна быть в результатах (найдена, пусть без PDF) assert len(rows) == 1 def test_output_xlsx_created_even_when_no_kns(self, tmp_path): """ При пустом списке КН результирующий файл всё равно должен существовать. До исправления: wb_out.save() вызывался только внутри цикла → файл не создавался. """ run(tmp_path, []) files = list(Path(tmp_path).rglob("Результаты_ФИАС_*.xlsx")) assert files, "Файл результатов не создан при пустом входном xlsx" def test_search_result_fias_id_empty_string_not_crashes(self, tmp_path): """ fias_id = '' (сайт вернул пустую ячейку) — не должно падать. openpyxl хранит пустую строку как пустую ячейку и читает обратно как None — это поведение библиотеки, не баг кода. """ ms = MagicMock() ms.search.return_value = {**FOUND_DATA, "fias_id": ""} rows = run(tmp_path, [("77:01:0001001:1", None)], searcher=ms) # Пустая строка → пустая ячейка → None при чтении (openpyxl behavior) assert rows[0][2] in ("", None) # Другие поля должны быть на месте assert rows[0][4] == "ул. Тестовая, 1" def test_vid_on_value_preserved_in_output(self, tmp_path): """Значение «Вид ОН» из Excel должно попасть в итоговую таблицу.""" ms = MagicMock() ms.search.return_value = FOUND_DATA rows = run(tmp_path, [("77:01:0001001:1", "Здание")], searcher=ms) assert rows[0][1] == "Здание" def test_none_kn_cell_skipped(self, tmp_path): """Пустые ячейки КН не попадают в список и не ищутся.""" wb = openpyxl.Workbook() ws = wb.active ws.append(["КН"]) ws.append([None]) ws.append([""]) ws.append([" "]) ws.append(["77:01:0001001:1"]) # единственный валидный p = str(tmp_path / "in.xlsx") wb.save(p) ms = MagicMock() ms.search.return_value = None with patch("main.FIASSearcher", return_value=ms): process_all(p, str(tmp_path), log=lambda _: None) assert ms.search.call_count <= 2 # только 1 КН × 2 retry # ───────────────────────────────────────────────────────────────────────────── # FIASSearcher — парсинг результатов, граничные случаи # ───────────────────────────────────────────────────────────────────────────── def cells_of(texts): return [MagicMock(text=t) for t in texts] def tr(texts, html=""): row = MagicMock() row.find_elements.side_effect = lambda by, tag: cells_of(texts) if tag == "td" else [] row.get_attribute.return_value = html return row def th_row(): row = MagicMock() row.find_elements.return_value = [] return row def run_search_rows(rows): """Прогоняет логику парсинга search() с замоканным Selenium.""" s = FIASSearcher(log=lambda _: None) driver = MagicMock() driver.find_elements.side_effect = lambda by, sel: ( rows if "ResultPagerDataPlaceholder" in sel else [] ) driver.find_element.return_value = MagicMock() s.driver = driver s.wait = MagicMock() s.wait.until.return_value = MagicMock() with patch("main.WebDriverWait") as wdw, patch("main.time"): inst = MagicMock() wdw.return_value = inst calls = [0] def until_se(cond): calls[0] += 1 if calls[0] <= 2: from selenium.common.exceptions import TimeoutException raise TimeoutException() return MagicMock() inst.until.side_effect = until_se return s.search("77:01:0001001:1") DATA_CELLS = ["Адрес", "Актуальный", "Здание", "x", "y", "fias-id-here"] class TestFIASSearcherEdge: def test_whitespace_only_cells_stripped(self): """Ячейки с пробелами должны возвращаться как пустые строки после strip().""" cells = [" Адрес ", " ", "Здание", "x", "y", " fias "] row = tr(cells) result = run_search_rows([row]) assert result is not None assert result["full_name"] == "Адрес" assert result["status"] == "" assert result["fias_id"] == "fias" def test_six_cells_exactly_accepted(self): """Ровно 6 ячеек — граница проверки len(cells) >= 6.""" row = tr(DATA_CELLS) # ровно 6 result = run_search_rows([row]) assert result is not None def test_five_cells_returns_none(self): """5 ячеек — меньше минимума, must return None.""" row = tr(DATA_CELLS[:5]) result = run_search_rows([row]) assert result is None def test_many_header_rows_then_data(self): """Несколько заголовочных строк перед строкой данных — пропускаются все.""" result = run_search_rows([th_row(), th_row(), th_row(), tr(DATA_CELLS)]) assert result is not None assert result["full_name"] == "Адрес" def test_pdf_url_with_single_quotes_in_html(self): """URL в одинарных кавычках — rstrip должен убрать кавычку.""" html = "href='/Export/ExportPdfStatement?id=42&t=pdf'" row = tr(DATA_CELLS, html=html) result = run_search_rows([row]) assert result["pdf_url"] == "https://fias.nalog.ru/Export/ExportPdfStatement?id=42&t=pdf" def test_pdf_url_ampersand_decoded(self): """& в HTML должен быть раскодирован в & в URL.""" html = 'href="/Export/ExportPdfStatement?a=1&b=2"' row = tr(DATA_CELLS, html=html) result = run_search_rows([row]) assert "&" not in (result["pdf_url"] or "") assert "a=1&b=2" in (result["pdf_url"] or "") def test_no_rows_returns_none(self): assert run_search_rows([]) is None def test_only_header_rows_returns_none(self): result = run_search_rows([th_row(), th_row()]) assert result is None # ───────────────────────────────────────────────────────────────────────────── # tmp_pdf path safety # ───────────────────────────────────────────────────────────────────────────── class TestTmpPdfPathSafety: def test_slash_in_kn_tmp_pdf_stays_in_session_dir(self, tmp_path): """ КН со слэшем не должен создавать tmp_pdf вне session_dir. До исправления: f"_tmp_{kn.replace(':', '_')}.pdf" оставлял '/' → path traversal. """ data_with_pdf = {**FOUND_DATA, "pdf_url": "https://fias.nalog.ru/Export/ExportPdfStatement?id=1"} ms = MagicMock() ms.search.return_value = data_with_pdf captured_paths = [] original_download = ms.download_pdf def capture_path(url, path): captured_paths.append(path) return False # не скачиваем реально ms.download_pdf.side_effect = capture_path with patch("main.FIASSearcher", return_value=ms): process_all( make_xlsx(tmp_path, [("01/999:0001001:1", None)]), str(tmp_path), log=lambda _: None, ) # Хотя бы один путь должен был быть передан, и все они должны быть внутри tmp_path assert captured_paths, "download_pdf не был вызван" session_dir = str(tmp_path) for p in captured_paths: resolved = str(Path(p).resolve()) assert resolved.startswith(str(Path(session_dir).resolve())), ( f"tmp_pdf вышел за пределы session_dir: {p}" ) def test_dotdot_in_kn_tmp_pdf_stays_in_session_dir(self, tmp_path): """../ в КН не должен выводить tmp_pdf за пределы session_dir.""" data_with_pdf = {**FOUND_DATA, "pdf_url": "https://fias.nalog.ru/Export/ExportPdfStatement?id=1"} ms = MagicMock() ms.search.return_value = data_with_pdf captured_paths = [] def capture_path(url, path): captured_paths.append(path) return False ms.download_pdf.side_effect = capture_path with patch("main.FIASSearcher", return_value=ms): process_all( make_xlsx(tmp_path, [("../evil:0001001:1", None)]), str(tmp_path), log=lambda _: None, ) assert captured_paths, "download_pdf не был вызван" for p in captured_paths: resolved = str(Path(p).resolve()) assert resolved.startswith(str(Path(str(tmp_path)).resolve())), ( f"tmp_pdf вышел за пределы session_dir: {p}" )