""" Автотесты для main.py. Запуск: python3 -m pytest test_main.py -v Браузер и настоящий ФИАС не нужны — всё замокано. """ from __future__ import annotations import os import re from pathlib import Path from unittest.mock import MagicMock, patch, call import pytest import openpyxl import main from main import ( _normalize_kn, parse_pdf_extract, rename_extract, process_all, FIASSearcher, ) from selenium.common.exceptions import TimeoutException, WebDriverException # ───────────────────────────────────────────────────────────────────────────── # Helpers # ───────────────────────────────────────────────────────────────────────────── def make_xlsx(tmp_path: Path, rows, headers=("КН", "Вид ОН")) -> str: path = tmp_path / "input.xlsx" wb = openpyxl.Workbook() ws = wb.active ws.append(list(headers)) for row in rows: ws.append(list(row)) wb.save(str(path)) return str(path) def get_result_rows(output_dir: Path) -> list: """Читает первый Результаты_ФИАС_*.xlsx из папки сессии (без заголовка).""" files = list(output_dir.rglob("Результаты_ФИАС_*.xlsx")) assert files, "Результирующий xlsx не найден" wb = openpyxl.load_workbook(str(files[0])) return list(wb.active.iter_rows(min_row=2, values_only=True)) def fake_searcher(search_map: dict | None = None, *, side_effect=None): """Возвращает мок FIASSearcher с заданными результатами поиска.""" m = MagicMock() if side_effect is not None: m.search.side_effect = side_effect elif search_map is not None: m.search.side_effect = lambda kn: search_map.get(kn) else: m.search.return_value = None return m def make_pdf_mock(text: str): """Мок pdfplumber.open(path) как контекстный менеджер с одной страницей.""" 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 # ───────────────────────────────────────────────────────────────────────────── # _normalize_kn # ───────────────────────────────────────────────────────────────────────────── class TestNormalizeKn: def test_already_clean(self): assert _normalize_kn("77:01:0001001:1") == "77:01:0001001:1" def test_spaces_around_colons(self): assert _normalize_kn("77 : 01 : 0001001 : 1") == "77:01:0001001:1" def test_tabs_around_colons(self): assert _normalize_kn("77\t:\t01\t:\t0001001\t:\t1") == "77:01:0001001:1" def test_mixed_whitespace(self): assert _normalize_kn(" 77 :01: 0001001:1 ") == "77:01:0001001:1" def test_trailing_newline(self): assert _normalize_kn("77:01:0001001:1\n") == "77:01:0001001:1" def test_none_becomes_string_none(self): # raw=None приходит если ячейка пустая, но это отфильтровывается раньше assert _normalize_kn("None") == "None" # ───────────────────────────────────────────────────────────────────────────── # parse_pdf_extract # ───────────────────────────────────────────────────────────────────────────── class TestParsePdfExtract: UUID = "12345678-abcd-abcd-abcd-123456789abc" def _parse(self, tmp_path, text): path = str(tmp_path / "test.pdf") Path(path).touch() with patch("pdfplumber.open", return_value=make_pdf_mock(text)): return parse_pdf_extract(path, log=lambda _: None) def test_date_near_keyword(self, tmp_path): date, _ = self._parse(tmp_path, "дата выписки 25.03.2024 прочее") assert date == "25.03.2024" def test_date_near_vipiska_keyword(self, tmp_path): date, _ = self._parse(tmp_path, "выписка от 01.01.2023") assert date == "01.01.2023" def test_date_fallback_first_in_text(self, tmp_path): date, _ = self._parse(tmp_path, "Документ 11.11.2021 без ключевых слов") assert date == "11.11.2021" def test_date_prefers_keyword_over_first(self, tmp_path): # "01.01.2000" стоит до ключевого слова, "25.03.2024" — рядом с ним date, _ = self._parse(tmp_path, "01.01.2000 текст дата документа 25.03.2024") assert date == "25.03.2024" def test_uuid_near_keyword(self, tmp_path): _, num = self._parse(tmp_path, f"номер идентификатор {self.UUID}") assert num == self.UUID def test_uuid_fallback(self, tmp_path): _, num = self._parse(tmp_path, f"Сведения {self.UUID} конец") assert num == self.UUID def test_uuid_case_insensitive(self, tmp_path): uuid_upper = self.UUID.upper() _, num = self._parse(tmp_path, f"Идентификатор {uuid_upper}") assert num is not None assert num.lower() == self.UUID.lower() def test_nothing_found(self, tmp_path): date, num = self._parse(tmp_path, "Текст без дат и UUID") assert date is None assert num is None def test_both_found(self, tmp_path): date, num = self._parse(tmp_path, f"дата 31.12.2023 номер {self.UUID}") assert date == "31.12.2023" assert num == self.UUID def test_corrupt_pdf_returns_none_none(self, tmp_path): path = str(tmp_path / "bad.pdf") Path(path).write_bytes(b"garbage") with patch("pdfplumber.open", side_effect=Exception("corrupt")): date, num = parse_pdf_extract(path, log=lambda _: None) assert date is None assert num is None def test_empty_text_returns_none_none(self, tmp_path): date, num = self._parse(tmp_path, "") assert date is None assert num is None def test_multipage_text_joined(self, tmp_path): """Текст со всех страниц объединяется.""" page1, page2 = MagicMock(), MagicMock() page1.extract_text.return_value = "стр 1" page2.extract_text.return_value = f"дата 15.06.2022 номер {self.UUID}" ctx = MagicMock() ctx.__enter__ = lambda s: ctx ctx.__exit__ = MagicMock(return_value=False) ctx.pages = [page1, page2] path = str(tmp_path / "multi.pdf") Path(path).touch() with patch("pdfplumber.open", return_value=ctx): date, num = parse_pdf_extract(path, log=lambda _: None) assert date == "15.06.2022" assert num == self.UUID # ───────────────────────────────────────────────────────────────────────────── # rename_extract # ───────────────────────────────────────────────────────────────────────────── class TestRenameExtract: KN = "77:01:0001001:1" DATE = "25.03.2024" EXPECTED_STEM = "ГАР_77_01_0001001_1_25.03.2024" def _make_src(self, tmp_path): src = tmp_path / "_tmp.pdf" src.write_bytes(b"%PDF-test") return str(src) def test_normal_rename(self, tmp_path): src = self._make_src(tmp_path) result = rename_extract(src, self.KN, self.DATE, str(tmp_path)) assert Path(result).exists() assert Path(result).name == f"{self.EXPECTED_STEM}.pdf" assert not Path(src).exists() def test_colons_replaced_with_underscores(self, tmp_path): src = self._make_src(tmp_path) result = rename_extract(src, self.KN, self.DATE, str(tmp_path)) assert ":" not in Path(result).name def test_conflict_appends_counter(self, tmp_path): (tmp_path / f"{self.EXPECTED_STEM}.pdf").write_bytes(b"existing") src = self._make_src(tmp_path) result = rename_extract(src, self.KN, self.DATE, str(tmp_path)) assert Path(result).name == f"{self.EXPECTED_STEM}_1.pdf" def test_multiple_conflicts_increment(self, tmp_path): for suffix in ("", "_1", "_2"): (tmp_path / f"{self.EXPECTED_STEM}{suffix}.pdf").write_bytes(b"x") src = self._make_src(tmp_path) result = rename_extract(src, self.KN, self.DATE, str(tmp_path)) assert Path(result).name == f"{self.EXPECTED_STEM}_3.pdf" def test_no_date_uses_today(self, tmp_path): src = self._make_src(tmp_path) result = rename_extract(src, self.KN, None, str(tmp_path)) assert Path(result).exists() assert re.search(r"\d{2}\.\d{2}\.\d{4}", Path(result).name) # ───────────────────────────────────────────────────────────────────────────── # process_all — интеграционные тесты с замоканным FIASSearcher # ───────────────────────────────────────────────────────────────────────────── FOUND_DATA = { "full_name": "ул. Тестовая, 1", "status": "Актуальный", "obj_type": "Здание", "fias_id": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", "pdf_url": None, } class TestProcessAll: def _run(self, tmp_path, rows, headers=("КН", "Вид ОН"), *, mock_searcher=None): xlsx = make_xlsx(tmp_path, rows, headers) if mock_searcher is None: mock_searcher = fake_searcher() with patch("main.FIASSearcher", return_value=mock_searcher): process_all(xlsx, str(tmp_path), log=lambda _: None) return get_result_rows(tmp_path) # ── столбцы ────────────────────────────────────────────────────────────── def test_missing_kn_column_returns_early(self, tmp_path): xlsx = make_xlsx(tmp_path, [("val",)], headers=("НЕТ",)) mock_s = fake_searcher() logs = [] with patch("main.FIASSearcher", return_value=mock_s): process_all(xlsx, str(tmp_path), log=logs.append) mock_s.search.assert_not_called() assert any("не найдена" in l for l in logs) def test_kn_column_case_insensitive(self, tmp_path): # "Кадастровый номер" (смешанный регистр) rows = self._run(tmp_path, [("77:01:0001001:1",)], headers=("Кадастровый номер",)) assert len(rows) == 1 # ── нормальные сценарии ─────────────────────────────────────────────────── def test_kn_found_writes_correct_row(self, tmp_path): ms = fake_searcher({"77:01:0001001:1": FOUND_DATA}) rows = self._run(tmp_path, [("77:01:0001001:1", "Здание")], mock_searcher=ms) assert rows[0][0] == "77:01:0001001:1" assert rows[0][3] == "Актуальный" assert rows[0][4] == "ул. Тестовая, 1" assert rows[0][2] == "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" def test_kn_not_found_writes_absent_row(self, tmp_path): rows = self._run(tmp_path, [("99:99:9999999:9", None)]) assert rows[0][2] == "отсутствует" assert rows[0][3] == "отсутствует" def test_multiple_kns_all_processed(self, tmp_path): ms = fake_searcher({ "77:01:0001001:1": FOUND_DATA, "77:01:0001001:2": None, }) rows = self._run( tmp_path, [("77:01:0001001:1", None), ("77:01:0001001:2", None)], mock_searcher=ms, ) assert len(rows) == 2 def test_empty_xlsx_produces_no_data_rows(self, tmp_path): ms = fake_searcher() rows = self._run(tmp_path, [], mock_searcher=ms) assert rows == [] ms.search.assert_not_called() # Output file must exist even with zero entries assert list((tmp_path).rglob("Результаты_ФИАС_*.xlsx")) # ── дедупликация ────────────────────────────────────────────────────────── def test_duplicate_kn_searched_once(self, tmp_path): # Each unique KN gets up to 2 attempts (retry logic). # A duplicate must not add extra attempts beyond that. ms = fake_searcher({"77:01:0001001:1": FOUND_DATA}) # returns on first try self._run( tmp_path, [("77:01:0001001:1", None), ("77:01:0001001:1", None)], mock_searcher=ms, ) # Exactly 1 call: found on first attempt, duplicate skipped entirely assert ms.search.call_count == 1 def test_duplicate_kn_writes_one_row(self, tmp_path): ms = fake_searcher() rows = self._run( tmp_path, [("77:01:0001001:1", None), ("77:01:0001001:1", None)], mock_searcher=ms, ) assert len(rows) == 1 # ── нормализация КН ─────────────────────────────────────────────────────── def test_spaces_in_kn_normalized_before_search(self, tmp_path): # search() is called with the normalised KN (spaces stripped around colons). # It may be called twice due to retry logic; assert_any_call covers both. ms = fake_searcher() self._run(tmp_path, [("77 : 01 : 0001001 : 1", None)], mock_searcher=ms) ms.search.assert_any_call("77:01:0001001:1") # Must never be called with the raw un-normalised form for c in ms.search.call_args_list: assert c.args[0] == "77:01:0001001:1" def test_normalized_kn_written_to_output(self, tmp_path): ms = fake_searcher() rows = self._run(tmp_path, [("77 : 01 : 0001001 : 1", None)], mock_searcher=ms) assert rows[0][0] == "77:01:0001001:1" # ── ошибки при поиске ───────────────────────────────────────────────────── def test_webdriver_error_writes_exactly_one_error_row(self, tmp_path): """Регрессия: двойная запись 'ошибка' + 'отсутствует' для одного КН.""" ms = fake_searcher(side_effect=WebDriverException("crash")) rows = self._run(tmp_path, [("77:01:0001001:1", None)], mock_searcher=ms) assert len(rows) == 1 assert rows[0][2] == "ошибка" # Убеждаемся что нет второй строки «отсутствует» statuses = [r[2] for r in rows] assert "отсутствует" not in statuses def test_generic_exception_writes_exactly_one_error_row(self, tmp_path): ms = fake_searcher(side_effect=RuntimeError("unexpected")) rows = self._run(tmp_path, [("77:01:0001001:1", None)], mock_searcher=ms) assert len(rows) == 1 assert rows[0][2] == "ошибка" def test_error_kn_still_marked_done(self, tmp_path): """После ошибки дубликат того же КН не должен искаться повторно.""" ms = fake_searcher(side_effect=WebDriverException("crash")) self._run( tmp_path, [("77:01:0001001:1", None), ("77:01:0001001:1", None)], mock_searcher=ms, ) assert ms.search.call_count == 1 # ── повтор поиска ───────────────────────────────────────────────────────── def test_retry_on_none_result(self, tmp_path): """Первая попытка даёт None, вторая — данные: КН должен быть найден.""" results = [None, FOUND_DATA] ms = fake_searcher(side_effect=lambda kn: results.pop(0)) rows = self._run(tmp_path, [("77:01:0001001:1", None)], mock_searcher=ms) assert rows[0][3] == "Актуальный" assert ms.search.call_count == 2 def test_both_retries_none_marks_absent(self, tmp_path): ms = fake_searcher(side_effect=lambda kn: None) rows = self._run(tmp_path, [("77:01:0001001:1", None)], mock_searcher=ms) assert rows[0][2] == "отсутствует" assert ms.search.call_count == 2 # ───────────────────────────────────────────────────────────────────────────── # FIASSearcher — парсинг строк таблицы результатов # ───────────────────────────────────────────────────────────────────────────── def make_cell(text: str) -> MagicMock: c = MagicMock() c.text = text return c def make_tr(cells_text: list[str], inner_html: str = "") -> MagicMock: cells = [make_cell(t) for t in cells_text] row = MagicMock() row.find_elements.side_effect = lambda by, tag: cells if tag == "td" else [] row.get_attribute.return_value = inner_html return row def make_th_row() -> MagicMock: """Заголовочная строка — вместо .""" row = MagicMock() row.find_elements.return_value = [] # td → пусто return row def _call_search_on_rows(rows: list) -> dict | None: """ Прогоняет только ту часть search(), которая читает строки из DOM, полностью обходя WebDriverWait / Selenium ожидания. """ searcher = 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() searcher.driver = driver wait = MagicMock() # wait.until всегда возвращает мок-поле для send_keys wait.until.return_value = MagicMock() searcher.wait = wait with ( patch("main.WebDriverWait") as mock_wdw, patch("main.time") as mock_time, ): # WebDriverWait(driver, n).until(...) → всегда успешно (кроме visibility) instance = MagicMock() mock_wdw.return_value = instance # invisibility → TimeoutException (div остаётся видимым) # staleness → TimeoutException (обновление на месте) # visibility → успех (возвращает мок) call_count = [0] def until_side_effect(condition): call_count[0] += 1 # Первый вызов — invisibility → TimeoutException # Второй — staleness → TimeoutException if call_count[0] <= 2: raise TimeoutException() return MagicMock() instance.until.side_effect = until_side_effect return searcher.search("77:01:0001001:1") class TestFIASSearcherRowParsing: DATA_CELLS = ["ул. Тестовая, 1", "Актуальный", "Здание", "col3", "col4", "aaaa-bbbb-cccc-dddd-eeee"] def test_normal_data_row_parsed(self): row = make_tr(self.DATA_CELLS, inner_html="") result = _call_search_on_rows([row]) assert result is not None assert result["full_name"] == "ул. Тестовая, 1" assert result["status"] == "Актуальный" assert result["fias_id"] == "aaaa-bbbb-cccc-dddd-eeee" def test_header_row_skipped(self): """Регрессия: ... не должен давать None.""" header = make_th_row() data = make_tr(self.DATA_CELLS) result = _call_search_on_rows([header, data]) assert result is not None assert result["full_name"] == "ул. Тестовая, 1" def test_only_header_row_returns_none(self): result = _call_search_on_rows([make_th_row()]) assert result is None def test_no_rows_returns_none(self): result = _call_search_on_rows([]) assert result is None def test_pdf_url_extracted_from_html(self): html = 'href="/Export/ExportPdfStatement?id=123&type=abc"' row = make_tr(self.DATA_CELLS, inner_html=html) result = _call_search_on_rows([row]) assert result is not None assert result["pdf_url"] == "https://fias.nalog.ru/Export/ExportPdfStatement?id=123&type=abc" def test_pdf_url_none_when_absent(self): row = make_tr(self.DATA_CELLS, inner_html="нет ссылки") result = _call_search_on_rows([row]) assert result is not None assert result["pdf_url"] is None def test_pdf_url_extracted_from_second_row_when_header_first(self): """PDF URL берётся из строки данных, а не из rows[0] (был баг).""" html = 'href="/Export/ExportPdfStatement?id=CORRECT"' header = make_th_row() data = make_tr(self.DATA_CELLS, inner_html=html) result = _call_search_on_rows([header, data]) assert result is not None assert "CORRECT" in (result["pdf_url"] or "") def test_row_with_less_than_6_cells_skipped(self): short_row = make_tr(["a", "b", "c"]) # только 3 ячейки data = make_tr(self.DATA_CELLS) result = _call_search_on_rows([short_row, data]) assert result is not None assert result["full_name"] == "ул. Тестовая, 1" # ───────────────────────────────────────────────────────────────────────────── # FIASSearcher.download_pdf # ───────────────────────────────────────────────────────────────────────────── class TestDownloadPdf: PDF_BYTES = b"%PDF-1.4 fake pdf content" URL = "https://fias.nalog.ru/Export/ExportPdfStatement?id=1" def _searcher(self): s = FIASSearcher(log=lambda _: None) s.driver = MagicMock() s.driver.get_cookies.return_value = [{"name": "session", "value": "abc"}] s.driver.execute_script.return_value = "Mozilla/5.0" return s def _mock_resp(self, content=None, status=200): resp = MagicMock() resp.content = content or self.PDF_BYTES resp.status_code = status resp.raise_for_status = MagicMock( side_effect=None if status < 400 else Exception(f"HTTP {status}") ) resp.headers = {"Content-Type": "application/pdf"} return resp @patch("main.requests.Session") def test_success_writes_file(self, mock_session_cls, tmp_path): mock_session_cls.return_value.get.return_value = self._mock_resp() s = self._searcher() dest = str(tmp_path / "out.pdf") assert s.download_pdf(self.URL, dest) is True assert Path(dest).read_bytes() == self.PDF_BYTES @patch("main.requests.Session") def test_non_pdf_response_returns_false(self, mock_session_cls, tmp_path): mock_session_cls.return_value.get.return_value = self._mock_resp(b"login") s = self._searcher() assert s.download_pdf(self.URL, str(tmp_path / "out.pdf")) is False assert not (tmp_path / "out.pdf").exists() @patch("main.requests.Session") def test_http_error_returns_false(self, mock_session_cls, tmp_path): resp = self._mock_resp(status=403) resp.raise_for_status.side_effect = Exception("403 Forbidden") mock_session_cls.return_value.get.return_value = resp s = self._searcher() assert s.download_pdf(self.URL, str(tmp_path / "out.pdf")) is False @patch("main.requests.Session") def test_network_exception_returns_false(self, mock_session_cls, tmp_path): mock_session_cls.return_value.get.side_effect = ConnectionError("timeout") s = self._searcher() assert s.download_pdf(self.URL, str(tmp_path / "out.pdf")) is False @patch("main.requests.Session") def test_cookies_transferred_to_session(self, mock_session_cls, tmp_path): mock_session = MagicMock() mock_session_cls.return_value = mock_session mock_session.get.return_value = self._mock_resp() s = self._searcher() s.download_pdf(self.URL, str(tmp_path / "out.pdf")) mock_session.cookies.set.assert_called_with("session", "abc") class TestFIASSearcherSetup: """Тесты для обработки ошибок при запуске браузера.""" @patch("main.webdriver.Firefox") @patch("main.FirefoxService") def test_setup_geckodriver_timeout_raises_webdriver_exception(self, mock_service, mock_ff): """Таймаут при запуске geckodriver должен привести к понятной ошибке.""" # Симулируем таймаут подключения к geckodriver mock_ff.side_effect = WebDriverException("HTTPConnectionPool timeout") s = FIASSearcher(log=lambda _: None) with pytest.raises(WebDriverException) as exc_info: s.setup() assert "браузер" in str(exc_info.value).lower() @patch("main.WebDriverWait") @patch("main.webdriver.Firefox") @patch("main.FirefoxService") def test_setup_fias_page_unreachable_raises_webdriver_exception(self, mock_service, mock_ff, mock_wdw): """Таймаут при открытии страницы ФИАС должен привести к понятной ошибке.""" mock_driver = MagicMock() mock_ff.return_value = mock_driver # Симулируем таймаут при driver.get(FIAS_URL) mock_driver.get.side_effect = WebDriverException("net::ERR_CONNECTION_TIMED_OUT") s = FIASSearcher(log=lambda _: None) with pytest.raises(WebDriverException) as exc_info: s.setup() assert "fias.nalog.ru" in str(exc_info.value).lower() assert "интернет" in str(exc_info.value).lower() @patch("main.WebDriverWait") @patch("main.webdriver.Firefox") @patch("main.FirefoxService") def test_setup_form_not_loading_raises_webdriver_exception(self, mock_service, mock_ff, mock_wdw): """Если форма поиска не загружается, должна быть понятная ошибка.""" mock_driver = MagicMock() mock_ff.return_value = mock_driver mock_driver.get.return_value = None # Симулируем TimeoutException при ожидании элементов формы mock_wait_instance = MagicMock() mock_wdw.return_value = mock_wait_instance mock_wait_instance.until.side_effect = TimeoutException("Element not found") s = FIASSearcher(log=lambda _: None) with pytest.raises(WebDriverException) as exc_info: s.setup() assert "форма" in str(exc_info.value).lower() or "загрузилась" in str(exc_info.value).lower()