This commit is contained in:
jze9
2026-06-01 17:05:07 +05:00
parent 89c3500151
commit 070261e947
7 changed files with 1288 additions and 0 deletions

17
.gitignore vendored Normal file
View File

@@ -0,0 +1,17 @@
# Портативный Firefox + geckodriver (большие, скачиваются fetch_vendor.py)
vendor/
# Сборка
build/
dist/
*.spec
__pycache__/
*.pyc
# Результаты прогонов (папки вида ДД.ММ.ГГГГ_ЧЧ-ММ-СС)
[0-9][0-9].[0-9][0-9].[0-9][0-9][0-9][0-9]_*/
# Прочее
.pytest_cache/
.venv/
python-3.8.10-win32.exe

Binary file not shown.

121
fetch_vendor.py Normal file
View File

@@ -0,0 +1,121 @@
#!/usr/bin/env python3
"""
Скачивает портативный Firefox и geckodriver (Windows) в папку vendor/,
чтобы build.py / build_wine.sh встроили их в .exe. Тогда конечному
пользователю нужно установить только КриптоПро — Firefox и драйвер уже внутри.
Запуск (на машине сборки, можно на Linux):
python fetch_vendor.py # win64 (по умолчанию)
python fetch_vendor.py --arch win32
python fetch_vendor.py --lang en-US
Для распаковки Firefox нужен 7-Zip:
Linux: sudo apt install p7zip-full
Windows: установите 7-Zip и добавьте 7z в PATH
"""
from __future__ import annotations
import argparse
import io
import json
import shutil
import subprocess
import sys
import urllib.request
import zipfile
from pathlib import Path
ROOT = Path(__file__).parent
VENDOR = ROOT / "vendor"
GECKO_API = "https://api.github.com/repos/mozilla/geckodriver/releases/latest"
def _download(url: str) -> bytes:
print(f"{url}")
req = urllib.request.Request(url, headers={"User-Agent": "fetch-vendor"})
with urllib.request.urlopen(req, timeout=120) as resp:
return resp.read()
def _which_7z() -> str | None:
for name in ("7z", "7za", "7zr"):
if shutil.which(name):
return name
return None
def fetch_geckodriver(arch: str) -> None:
print("geckodriver:")
data = json.loads(_download(GECKO_API).decode("utf-8"))
asset = next(
(a for a in data["assets"] if f"{arch}.zip" in a["name"] and a["name"].endswith(".zip")),
None,
)
if not asset:
names = [a["name"] for a in data["assets"]]
sys.exit(f" Не найден ассет для {arch}. Доступно: {names}")
blob = _download(asset["browser_download_url"])
with zipfile.ZipFile(io.BytesIO(blob)) as z:
member = next(n for n in z.namelist() if n.endswith("geckodriver.exe"))
VENDOR.mkdir(exist_ok=True)
with z.open(member) as src, open(VENDOR / "geckodriver.exe", "wb") as dst:
shutil.copyfileobj(src, dst)
print(f"{VENDOR / 'geckodriver.exe'} ({data['tag_name']})")
def fetch_firefox(arch: str, lang: str) -> None:
print("Firefox (портативный):")
sevenzip = _which_7z()
if not sevenzip:
print(
" ✗ 7-Zip не найден — не могу распаковать установщик Firefox.\n"
" Установите 7-Zip (Linux: sudo apt install p7zip-full), либо вручную\n"
" распакуйте Firefox так, чтобы получилось vendor/firefox/firefox.exe"
)
return
os_param = "win64" if arch == "win64" else "win"
url = f"https://download.mozilla.org/?product=firefox-latest&os={os_param}&lang={lang}"
installer = VENDOR / "_firefox_setup.exe"
VENDOR.mkdir(exist_ok=True)
installer.write_bytes(_download(url))
extract_dir = VENDOR / "_ff_extract"
if extract_dir.exists():
shutil.rmtree(extract_dir)
# Установщик Firefox — это 7-Zip SFX; внутри папка core/ с firefox.exe.
subprocess.run(
[sevenzip, "x", "-y", f"-o{extract_dir}", str(installer)],
check=True, stdout=subprocess.DEVNULL,
)
core = extract_dir / "core"
if not (core / "firefox.exe").exists():
sys.exit(f" В установщике не найден core/firefox.exe (распаковано в {extract_dir})")
target = VENDOR / "firefox"
if target.exists():
shutil.rmtree(target)
shutil.move(str(core), str(target))
installer.unlink(missing_ok=True)
shutil.rmtree(extract_dir, ignore_errors=True)
print(f"{target / 'firefox.exe'}")
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--arch", choices=("win64", "win32"), default="win64")
ap.add_argument("--lang", default="ru")
args = ap.parse_args()
print(f"Папка: {VENDOR} (arch={args.arch}, lang={args.lang})\n")
fetch_geckodriver(args.arch)
print()
fetch_firefox(args.arch, args.lang)
print("\nГотово. Теперь соберите: python build.py")
if __name__ == "__main__":
main()

521
test_adversarial.py Normal file
View File

@@ -0,0 +1,521 @@
"""
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&amp;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):
"""&amp; в HTML должен быть раскодирован в & в URL."""
html = 'href="/Export/ExportPdfStatement?a=1&amp;b=2"'
row = tr(DATA_CELLS, html=html)
result = run_search_rows([row])
assert "&amp;" 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}"
)

629
test_main.py Normal file
View File

@@ -0,0 +1,629 @@
"""
Автотесты для 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:
"""Заголовочная строка — <th> вместо <td>."""
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):
"""Регрессия: <thead><tr><th>...</th></tr> не должен давать 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&amp;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="<td>нет ссылки</td>")
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"<html>login</html>")
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()