"""Тесты ФНС opendata loader'а + lookup (app/services/fns_opendata_loader.py,
app/services/fns_lookup.py, мигр. 296/295, "bulk-дампы ФНС по юрлицам").
Coverage:
- resolve_dataset_file_url — резолв .zip-ссылки из литерального куска HTML каталога
(несколько ссылок → берём максимальную дату публикации); ValueError, если ссылок нет.
- harvest_element — generic-харвестер: ИНН-подобный атрибут даёт identity, числовые
атрибуты (кроме id/name-подобных) становятся FnsRecord; элемент без ИНН → [].
- iter_dataset_records — round-trip через реально собранный .zip с мини-XML фикстурой.
- load_dataset — HTTP замокан через httpx.MockTransport (каталог + zip); dry_run не
пишет; повторная загрузка той же версии — skip без сети на .zip.
- Статические asserts по SQL: CAST-дисциплина, ON CONFLICT, IS DISTINCT FROM.
- upsert_records: `db.begin_nested()` в исходнике, модуль НЕ коммитит (коммитит caller).
- fns_lookup.get_facts_by_inn на MagicMock db: группировка по dataset, пустой ИНН.
"""
from __future__ import annotations
import inspect
import io
import os
import re
import zipfile
from datetime import date
from unittest.mock import MagicMock
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
import httpx
import pytest
from lxml import etree
from app.services import fns_lookup as fl
from app.services import fns_opendata_loader as fol
# ─────────────────────────────────────────────────────────────────────────────
# resolve_dataset_file_url
# ─────────────────────────────────────────────────────────────────────────────
_CATALOG_HTML = """
"""
def test_resolve_dataset_file_url_picks_latest_publication() -> None:
url, published = fol.resolve_dataset_file_url(
_CATALOG_HTML, base_url="https://www.nalog.gov.ru/opendata/7707329152-revexp/"
)
assert (
url
== "https://file.nalog.ru/opendata/7707329152-revexp/data-20260825-structure-20180110.zip"
)
assert published == date(2026, 8, 25)
def test_resolve_dataset_file_url_relative_href_resolved_against_base() -> None:
html = 'v'
url, published = fol.resolve_dataset_file_url(
html, base_url="https://www.nalog.gov.ru/opendata/7707329152-snr/"
)
assert url == "https://www.nalog.gov.ru/opendata/7707329152-snr/data-20260301-structure-1.zip"
assert published == date(2026, 3, 1)
def test_resolve_dataset_file_url_raises_when_no_zip_links() -> None:
with pytest.raises(ValueError, match="data-YYYYMMDD"):
fol.resolve_dataset_file_url("empty", base_url="https://x/")
# ─────────────────────────────────────────────────────────────────────────────
# harvest_element — generic-харвестер
# ─────────────────────────────────────────────────────────────────────────────
def test_harvest_element_extracts_numeric_attrs_excluding_ids() -> None:
elem = etree.fromstring(
'<СвНП ИННЮЛ="7707329152" КПП="770701001" НаимОрг="ООО РОМАШКА" '
'СумДоход="1234567.89" СумРасход="654321,00"/>'
)
records = fol.harvest_element(elem, dataset="revexp", period=date(2026, 8, 25))
by_series = {r.series: r for r in records}
assert set(by_series) == {"СумДоход", "СумРасход"}
assert by_series["СумДоход"].inn == "7707329152"
assert by_series["СумДоход"].value == pytest.approx(1234567.89)
assert by_series["СумРасход"].value == pytest.approx(654321.00) # запятая — decimal separator
assert by_series["СумДоход"].org_name == "ООО РОМАШКА"
assert by_series["СумДоход"].dataset == "revexp"
assert by_series["СумДоход"].period == date(2026, 8, 25)
def test_harvest_element_without_inn_returns_empty() -> None:
elem = etree.fromstring('<Прочее КПП="770701001" Значение="1"/>')
assert fol.harvest_element(elem, dataset="revexp", period=date(2026, 1, 1)) == []
def test_harvest_element_skips_non_numeric_attrs() -> None:
elem = etree.fromstring('<СвНП ИННЮЛ="123" Комментарий="текст, не число"/>')
assert fol.harvest_element(elem, dataset="snr", period=date(2026, 1, 1)) == []
# ─────────────────────────────────────────────────────────────────────────────
# iter_dataset_records — round-trip через собранный .zip
# ─────────────────────────────────────────────────────────────────────────────
def _zip_bytes(xml_by_name: dict[str, str]) -> bytes:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
for name, content in xml_by_name.items():
zf.writestr(name, content)
return buf.getvalue()
_SAMPLE_XML = """
<Файл>
<Документ>
<СвНП ИННЮЛ="7707329152" НаимОрг="ООО РОМАШКА" СумДоход="1000"/>
<СвНП ИННЮЛ="6660000001" НаимОрг="ООО ВАСИЛЁК" СумДоход="2000" СумРасход="500"/>
"""
def test_iter_dataset_records_round_trip_from_zip() -> None:
data = _zip_bytes({"revexp_66.xml": _SAMPLE_XML})
records = list(fol.iter_dataset_records(data, dataset="revexp", period=date(2026, 8, 25)))
assert len(records) == 3 # 1000 + (2000, 500)
inns = {r.inn for r in records}
assert inns == {"7707329152", "6660000001"}
assert all(r.dataset == "revexp" and r.period == date(2026, 8, 25) for r in records)
def test_iter_dataset_records_raises_when_zip_has_no_xml() -> None:
data = _zip_bytes({"readme.txt": "no xml here"})
with pytest.raises(ValueError, match="\\.xml"):
list(fol.iter_dataset_records(data, dataset="revexp", period=date(2026, 1, 1)))
# ─────────────────────────────────────────────────────────────────────────────
# load_dataset — HTTP замокан (никакой живой сети)
# ─────────────────────────────────────────────────────────────────────────────
def _make_client(zip_bytes: bytes, *, file_url: str, catalog_html: str) -> httpx.Client:
def handler(request: httpx.Request) -> httpx.Response:
if str(request.url) == file_url:
return httpx.Response(200, content=zip_bytes)
if "opendata/7707329152-" in str(request.url):
return httpx.Response(200, text=catalog_html)
return httpx.Response(404)
return httpx.Client(transport=httpx.MockTransport(handler))
def test_load_dataset_dry_run_does_not_write(monkeypatch: pytest.MonkeyPatch) -> None:
file_url = "https://file.nalog.ru/opendata/7707329152-revexp/data-20260825-structure-1.zip"
catalog_html = f'zip'
client = _make_client(b"unused", file_url=file_url, catalog_html=catalog_html)
db = MagicMock()
db.execute.return_value.first.return_value = None # get_loaded_version → None
result = fol.load_dataset(db, "revexp", client=client, dry_run=True)
assert result == {"dataset": "revexp", "skipped": 0, "records": 0, "upserted": 0}
# dry_run: единственный execute — SELECT версии (get_loaded_version), UPSERT не звался.
assert db.execute.call_count == 1
def test_load_dataset_skips_when_version_unchanged() -> None:
file_url = "https://file.nalog.ru/opendata/7707329152-revexp/data-20260825-structure-1.zip"
catalog_html = f'zip'
client = _make_client(b"should-not-be-fetched", file_url=file_url, catalog_html=catalog_html)
db = MagicMock()
db.execute.return_value.first.return_value = (file_url, date(2026, 8, 25))
result = fol.load_dataset(db, "revexp", client=client)
assert result == {"dataset": "revexp", "skipped": 1, "records": 0, "upserted": 0}
def test_load_dataset_unknown_slug_raises() -> None:
db = MagicMock()
with pytest.raises(ValueError, match="неизвестный slug"):
fol.load_dataset(db, "not-a-real-dataset")
# ─────────────────────────────────────────────────────────────────────────────
# Статические asserts по SQL (без БД)
# ─────────────────────────────────────────────────────────────────────────────
def test_upsert_facts_sql_uses_cast_never_double_colon() -> None:
sql = re.sub(r"\s+", " ", str(fol._UPSERT_FACTS_SQL.text))
assert not re.search(r":\w+::", sql)
assert "CAST(:inn AS text)" in sql
assert "CAST(:period AS date)" in sql
assert "CAST(:value AS numeric)" in sql
assert "ON CONFLICT (inn, dataset, series, period) DO UPDATE SET" in sql
assert "IS DISTINCT FROM" in sql
def test_upsert_version_sql_uses_cast_and_conflict_target() -> None:
sql = re.sub(r"\s+", " ", str(fol._UPSERT_VERSION_SQL.text))
assert not re.search(r":\w+::", sql)
assert "ON CONFLICT (dataset) DO UPDATE SET" in sql
def test_upsert_records_uses_savepoint_and_module_does_not_commit() -> None:
src = inspect.getsource(fol.upsert_records)
assert "with db.begin_nested():" in src
module_src = inspect.getsource(fol)
assert "db.commit()" not in module_src
# ─────────────────────────────────────────────────────────────────────────────
# fns_lookup.get_facts_by_inn — MagicMock db
# ─────────────────────────────────────────────────────────────────────────────
def test_get_facts_by_inn_groups_by_dataset() -> None:
db = MagicMock()
org = "ООО РОМАШКА"
p2026 = date(2026, 8, 25)
def row(dataset: str, series: str, period: date, value: int) -> dict[str, object]:
return {
"dataset": dataset,
"series": series,
"period": period,
"value": value,
"org_name": org,
}
db.execute.return_value.mappings.return_value.all.return_value = [
row("revexp", "СумДоход", p2026, 1000),
row("revexp", "СумРасход", p2026, 500),
row("sshr2019", "ССЧ", date(2019, 1, 1), 42),
]
result = fl.get_facts_by_inn(db, "7707329152")
assert set(result) == {"revexp", "sshr2019"}
assert len(result["revexp"]) == 2
assert result["sshr2019"][0]["series"] == "ССЧ"
def test_get_facts_by_inn_empty_inn_returns_empty_without_query() -> None:
db = MagicMock()
assert fl.get_facts_by_inn(db, "") == {}
assert fl.get_facts_by_inn(db, " ") == {}
db.execute.assert_not_called()
def test_get_facts_by_inn_no_rows_returns_empty_dict() -> None:
db = MagicMock()
db.execute.return_value.mappings.return_value.all.return_value = []
assert fl.get_facts_by_inn(db, "0000000000") == {}
_REAL_SHAPE_XML = """
<Файл ИдФайл="VO_OTKRDAN_5_9965_9965_20260825_cf1c08a2" ВерсФорм="4.01" \
ТипИнф="ОТКРДАННЫЕ5" КолДок="2">
<ИдОтпр><ФИООтв Фамилия="_" Имя="_"/>
<Документ ИдДок="dea28452-05f3-42bf-a08f-7a648c593473" ДатаДок="25.08.2026" \
ДатаСост="31.12.2025">
<СведНП НаимОрг="ОБЩЕСТВО С ОГРАНИЧЕННОЙ ОТВЕТСТВЕННОСТЬЮ "СПК"" \
ИННЮЛ="4205406898"/>
<СведДохРасх СумДоход="341864000.00" СумРасход="282224000.00"/>
<Документ ИдДок="28be840b-5db9-422b-b80c-3b678b808c96" ДатаДок="25.08.2026" \
ДатаСост="31.12.2025">
<СведНП НаимОрг="ООО ВАСИЛЁК" ИННЮЛ="6660000001"/>
<СведДохРасх СумДоход="1000.00" СумРасход="500.00"/>
"""
def test_iter_dataset_records_parses_real_fns_document_shape() -> None:
"""Боевая форма ФНС: ИНН на `СведНП`, показатели на СОСЕДНЕМ `СведДохРасх`.
Регресс на реальный дефект: харвестер требовал ИНН и числа на ОДНОМ элементе,
поэтому на настоящем дампе (revexp, data-20260825, 2118 XML) извлекал НОЛЬ
записей, а тесты на выдуманной фикстуре этого не показывали. Плюс безусловный
`elem.clear()` вычищал детей до закрытия `<Документ>`.
"""
data = _zip_bytes({"revexp_1.xml": _REAL_SHAPE_XML})
records = list(fol.iter_dataset_records(data, dataset="revexp", period=date(2026, 8, 25)))
assert len(records) == 4 # 2 организации × (СумДоход, СумРасход)
assert {r.inn for r in records} == {"4205406898", "6660000001"}
assert {r.series for r in records} == {"СумДоход", "СумРасход"}
# период берётся из ДатаСост документа, а не из переданного fallback'а
assert {r.period for r in records} == {date(2025, 12, 31)}
first = next(r for r in records if r.inn == "4205406898" and r.series == "СумДоход")
assert first.value == 341864000.0
assert first.org_name is not None and "СПК" in first.org_name
def test_service_ids_are_not_harvested_as_facts() -> None:
"""ВерсФорм/КолДок/ДатаДок — служебные, фактами быть не должны."""
data = _zip_bytes({"revexp_1.xml": _REAL_SHAPE_XML})
series = {
r.series for r in fol.iter_dataset_records(data, dataset="revexp", period=date(2026, 8, 25))
}
assert not series & {"ВерсФорм", "КолДок", "ДатаДок", "ДатаСост", "ИННЮЛ"}