"""Тесты ГАР-loader'а знаменателя (app/services/gar_flats_loader.py, мигр. 143). Структура (mirror'ит test_cadastral_geo_match): - ДЕТЕРМИНИРОВАННОЕ ядро (без БД): синтетические крошечные ГАР-XML фикстуры в tmp_path → парс → assert apartment-count на дом, сборка адреса, фильтрация ISACTUAL/ISACTIVE, дом с 0 квартир → flat_count 0, мульти-сегментная сборка адреса. - СТАТИЧЕСКИЕ asserts по SQL (_INSERT_SQL / _MATCH_SQL): ON CONFLICT, psycopg-v3 CAST, DISTINCT ON tie-break, идемпотентный gate. - ОПЦИОНАЛЬНЫЙ real-Postgres тест upsert+match → houses.gar_flat_count (self-SKIP без БД, создаёт TEMP-таблицы + fn в транзакции и откатывает). """ from __future__ import annotations import inspect import os import re import sys from pathlib import Path from typing import Any import pytest # settings нужен DSN на импорте (как в sibling-тестах); ядро статично/без БД. os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") from app.services import gar_flats_loader as gfl # ───────────────────────────────────────────────────────────────────────────── # Фикстуры: крошечный синтетический ГАР-дамп # ───────────────────────────────────────────────────────────────────────────── # Иерархия: # 1 регион «обл Свердловская» # └─ 2 город «г Екатеринбург» # ├─ 3 улица «ул Шаумяна» # │ ├─ 100 дом «10» → квартиры 1001,1002,1003 (3 активные) + 1004(ISACTIVE=0) # │ └─ 200 дом «5 к1» → квартиры 2001,2002 (2 активные) # └─ 30 улица «ул Ленина» # └─ 300 дом «7» → 0 квартир # inactive: адр-объект 9 (ISACTIVE=0), дом 900 (ISACTUAL=0), кв 1005 (ISACTUAL=0) _ADDR_OBJ_XML = """ """ _HOUSES_XML = """ """ _APARTMENTS_XML = """ """ # Иерархия: пара (OBJECTID лежит под PARENTOBJID). _MUN_HIER_XML = """ """ @pytest.fixture def gar_dir(tmp_path: Path) -> str: """Пишет синтетический ГАР-каталог, возвращает путь.""" (tmp_path / "AS_ADDR_OBJ_20260601_x.XML").write_text(_ADDR_OBJ_XML, encoding="utf-8") (tmp_path / "AS_HOUSES_20260601_x.XML").write_text(_HOUSES_XML, encoding="utf-8") (tmp_path / "AS_APARTMENTS_20260601_x.XML").write_text(_APARTMENTS_XML, encoding="utf-8") (tmp_path / "AS_MUN_HIERARCHY_20260601_x.XML").write_text(_MUN_HIER_XML, encoding="utf-8") return str(tmp_path) def _rows_by_guid(rows: list[gfl.GarHouseRow]) -> dict[str, gfl.GarHouseRow]: return {r.house_guid: r for r in rows} # ───────────────────────────────────────────────────────────────────────────── # Парс: отдельные файлы # ───────────────────────────────────────────────────────────────────────────── def test_parse_addr_objects_filters_inactive(gar_dir: str) -> None: paths = gfl._find_files(gar_dir, gfl.ADDR_OBJ_PREFIX) addr = gfl.parse_addr_objects(paths) assert set(addr.keys()) == {1, 2, 3, 30} # снесённая (9, ISACTIVE=0) исключена assert addr[3].name == "Шаумяна" and addr[3].typename == "ул" def test_parse_houses_filters_non_actual(gar_dir: str) -> None: paths = gfl._find_files(gar_dir, gfl.HOUSES_PREFIX) houses = gfl.parse_houses(paths) assert set(houses.keys()) == {100, 200, 300} # 900 (ISACTUAL=0) исключён assert houses[200].housenum == "5" and houses[200].addnum1 == "1" def test_parse_apartment_ids_excludes_inactive_and_non_actual(gar_dir: str) -> None: paths = gfl._find_files(gar_dir, gfl.APARTMENTS_PREFIX) apts = gfl.parse_apartment_ids(paths) # 1004 (ISACTIVE=0) и 1005 (ISACTUAL=0) исключены. assert apts == {1001, 1002, 1003, 2001, 2002} def test_parse_hierarchy_counts_apartments_per_house(gar_dir: str) -> None: addr = gfl.parse_addr_objects(gfl._find_files(gar_dir, gfl.ADDR_OBJ_PREFIX)) houses = gfl.parse_houses(gfl._find_files(gar_dir, gfl.HOUSES_PREFIX)) apts = gfl.parse_apartment_ids(gfl._find_files(gar_dir, gfl.APARTMENTS_PREFIX)) hier_paths = gfl._find_files(gar_dir, gfl.MUN_HIER_PREFIX) apt_count, child_to_parent = gfl.parse_hierarchy(hier_paths, apts, set(houses), set(addr)) # ISACTIVE=0 строка иерархии (9999) не учитывается; 1004 не в apartment_ids → не считается. assert apt_count.get(100) == 3 assert apt_count.get(200) == 2 assert apt_count.get(300, 0) == 0 # дом без квартир # child_to_parent заполнен ТОЛЬКО для домов и адр-объектов (не для квартир). assert child_to_parent[100] == 3 assert child_to_parent[3] == 2 assert 1001 not in child_to_parent # ───────────────────────────────────────────────────────────────────────────── # Сборка адреса + строки # ───────────────────────────────────────────────────────────────────────────── def test_build_rows_apartment_counts_and_zero_house(gar_dir: str) -> None: rows, stats = gfl.parse_gar_region(gar_dir, "66", "2026-06-01") by_guid = _rows_by_guid(rows) assert by_guid["h-a"].flat_count == 3 assert by_guid["h-b"].flat_count == 2 assert by_guid["h-c"].flat_count == 0 # дом с 0 квартир → flat_count 0 assert stats.houses == 3 assert stats.apartments == 5 # суммарно учтено под домами assert stats.rows == 3 def test_multi_segment_address_assembly(gar_dir: str) -> None: rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01") a = _rows_by_guid(rows)["h-a"] # street_name = непосредственный родитель (улица), « ». assert a.street_name == "ул Шаумяна" assert a.house_num == "10" # full_address — мульти-сегмент регион→город→улица + номер (регион — name-first). assert a.full_address == "Свердловская обл, г Екатеринбург, ул Шаумяна, 10" # norm_src (вход нормализатора) — «улица, номер». assert a.norm_src == "ул Шаумяна, 10" def test_house_num_with_korpus(gar_dir: str) -> None: rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01") b = _rows_by_guid(rows)["h-b"] # ADDNUM1=1 ADDTYPE1=1 (Корпус) → «5 к1». assert b.house_num == "5 к1" assert b.street_name == "ул Шаумяна" def test_region_and_version_propagated(gar_dir: str) -> None: rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01") assert all(r.region_code == "66" for r in rows) assert all(r.gar_version == "2026-06-01" for r in rows) def test_render_house_num_unit() -> None: h = gfl.GarHouse( object_id=1, guid="x", housenum="12", addnum1="3", addnum2="", addtype1="2", addtype2="", ) assert gfl.render_house_num(h) == "12 стр3" # ADDTYPE1=2 = Строение def test_adm_hierarchy_fallback(tmp_path: Path) -> None: """Без AS_MUN_HIERARCHY использует AS_ADM_HIERARCHY (fallback).""" (tmp_path / "AS_ADDR_OBJ_x.XML").write_text(_ADDR_OBJ_XML, encoding="utf-8") (tmp_path / "AS_HOUSES_x.XML").write_text(_HOUSES_XML, encoding="utf-8") (tmp_path / "AS_APARTMENTS_x.XML").write_text(_APARTMENTS_XML, encoding="utf-8") (tmp_path / "AS_ADM_HIERARCHY_x.XML").write_text(_MUN_HIER_XML, encoding="utf-8") rows, _ = gfl.parse_gar_region(str(tmp_path), "66", "2026-06-01") assert _rows_by_guid(rows)["h-a"].flat_count == 3 # ───────────────────────────────────────────────────────────────────────────── # Статические asserts по SQL (psycopg v3, ON CONFLICT, tie-break) # ───────────────────────────────────────────────────────────────────────────── _INSERT_SQL = str(gfl._INSERT_SQL.text) _MATCH_SQL = str(gfl._MATCH_SQL.text) def test_insert_is_idempotent_on_conflict() -> None: flat = re.sub(r"\s+", " ", _INSERT_SQL) assert "INSERT INTO gar_house_flats" in flat assert "ON CONFLICT (house_guid) DO UPDATE SET" in flat # norm_address считает SQL-fn в INSERT (= «UPDATE после insert»). assert "tradein_normalize_short_addr(CAST(:norm_src AS text))" in flat def test_match_distinct_on_tiebreak_max_flat_count() -> None: flat = re.sub(r"\s+", " ", _MATCH_SQL) # Канонический ключ (мигр. 144), а не точное равенство norm_address. assert "DISTINCT ON (canon)" in flat assert "tradein_canon_addr(norm_address) AS canon" in flat # Tie-break: на canon — строка с макс flat_count, затем меньший house_guid. assert "ORDER BY canon, flat_count DESC, house_guid" in flat # houses-сторона: тот же канон поверх COALESCE(short/full/address). houses_side = ( "tradein_canon_addr( COALESCE(h.short_address, h.full_address, h.address) ) = gp.canon" ) assert houses_side in flat # ЕКБ-ограничение по full_address (защита от cross-town коллизий). assert "full_address ILIKE '%' || CAST(:city AS text) || '%'" in flat # Идемпотентность повторного прогона. assert "h.gar_flat_count IS DISTINCT FROM gp.flat_count" in flat assert "gar_match_method = 'canon_addr'" in flat # Только дома с известным знаменателем матчим. assert "flat_count > 0" in flat def test_no_psycopg_v3_colon_colon_cast() -> None: # Только исполняемый SQL (модульный docstring специально содержит анти-паттерн как памятку). assert not re.search(r":\w+::", _INSERT_SQL) assert not re.search(r":\w+::", _MATCH_SQL) def test_upsert_uses_savepoint_per_batch() -> None: src = inspect.getsource(gfl.upsert_gar_houses) assert "with db.begin_nested():" in src # upsert НЕ коммитит сам (коммитит caller). assert "db.commit()" not in src # ───────────────────────────────────────────────────────────────────────────── # Канонический ключ адреса tradein_canon_addr (мигр. 144) — DB-free мирроринг # ───────────────────────────────────────────────────────────────────────────── # Чистый Python-зеркало SQL-функции tradein_canon_addr (мигр. 144) для юнит-теста БЕЗ БД. # Канонический источник — SQL-функция в data/sql/144; зеркало точно повторяет её regexp-шаги # и проверено на тех же кейсах (порядок альтернатив «длинная|короткая» совпадает с POSIX # longest-match Postgres; границы [^а-я] идентичны после lower()+ё→е). Для ground-truth на # живой БД те же утверждения дублируются в real-Postgres-тесте ниже. _CANON_TYPES = ( "улица|ул|переулок|пер|проспект|пркт|пр|бульвар|бр|шоссе|ш|проезд|" "набережная|наб|площадь|пл|тупик|туп|аллея|микрорайон|мкр" ) _CANON_RN_RE = re.compile(r"р-н[^,]*,") _CANON_TYPE_RE = re.compile(r"(^|[^а-я])(" + _CANON_TYPES + r")([^а-я]|$)") _CANON_KEEP_RE = re.compile(r"[^а-я0-9]") def _canon_addr_py(s: str | None) -> str: """Python-зеркало SQL tradein_canon_addr(text) (мигр. 144).""" s = (s or "").lower().replace("ё", "е") s = _CANON_RN_RE.sub("", s) s = _CANON_TYPE_RE.sub(r"\1\3", s) s = _CANON_KEEP_RE.sub("", s) return s def test_canon_addr_street_type_and_punct_collapse() -> None: # «ул. Шаумяна,20» и «ул. Шаумяна, 20» канон-равны (тип улицы + запятая/пробел схлопнуты). assert _canon_addr_py("ул. Шаумяна,20") == _canon_addr_py("ул. Шаумяна, 20") assert _canon_addr_py("ул. Шаумяна,20") == "шаумяна20" def test_canon_addr_strips_rayon_and_korpus_slash() -> None: # «р-н Ленинский, улица Цвиллинга, 7/6» → «цвиллинга76» (район срезан, слэш выброшен). assert _canon_addr_py("р-н Ленинский, улица Цвиллинга, 7/6") == "цвиллинга76" def test_canon_addr_keeps_letter_suffix_distinct() -> None: # Литера корпуса «6Б» — ДРУГОЕ здание: не должно схлопываться до «6». a = _canon_addr_py("улица Машиностроителей, 6Б") b = _canon_addr_py("ул. Машиностроителей, 6") assert a != b assert a == "машиностроителей6б" assert b == "машиностроителей6" def test_canon_addr_handles_none_and_empty() -> None: assert _canon_addr_py(None) == "" assert _canon_addr_py("") == "" # Строка из одного типа улицы без названия схлопывается в пустую. assert _canon_addr_py("улица") == "" # ───────────────────────────────────────────────────────────────────────────── # CLI: --match-only флаг (app/tasks/gar_flats_load.py) # ───────────────────────────────────────────────────────────────────────────── def test_match_only_does_not_require_dir() -> None: """--match-only парсится без --dir (парс/загрузка пропускаются).""" from app.tasks import gar_flats_load as task args = task.build_parser().parse_args(["--match-only", "--region", "66"]) assert args.match_only is True assert args.dir is None assert args.region == "66" def test_main_match_only_skips_load(monkeypatch: pytest.MonkeyPatch) -> None: """main(--match-only) вызывает только run_gar_match_only, НЕ run_gar_flats_load.""" from app.tasks import gar_flats_load as task calls: dict[str, Any] = {} def fake_match_only(region_code: str, *, city_filter: str | None = None) -> dict[str, int]: calls["match_only"] = (region_code, city_filter) return {"houses_matched": 5} def fake_load(*args: Any, **kwargs: Any) -> dict[str, int]: calls["load"] = True return {} monkeypatch.setattr(task, "run_gar_match_only", fake_match_only) monkeypatch.setattr(task, "run_gar_flats_load", fake_load) monkeypatch.setattr(sys, "argv", ["gar_flats_load", "--match-only", "--region", "66"]) task.main() assert "load" not in calls # парс/загрузка пропущены assert calls["match_only"] == ("66", "Екатеринбург") def test_main_match_only_empty_city_disables_filter(monkeypatch: pytest.MonkeyPatch) -> None: """Пустой --city → city_filter=None (фильтр отключён, матч за пределами ЕКБ).""" from app.tasks import gar_flats_load as task captured: dict[str, Any] = {} def fake_match_only(region_code: str, *, city_filter: str | None = None) -> dict[str, int]: captured["city_filter"] = city_filter return {"houses_matched": 0} monkeypatch.setattr(task, "run_gar_match_only", fake_match_only) monkeypatch.setattr(sys, "argv", ["gar_flats_load", "--match-only", "--city", ""]) task.main() assert captured["city_filter"] is None def test_main_requires_dir_without_match_only(monkeypatch: pytest.MonkeyPatch) -> None: """Без --match-only и без --dir — parser.error → SystemExit.""" from app.tasks import gar_flats_load as task monkeypatch.setattr(sys, "argv", ["gar_flats_load", "--region", "66"]) with pytest.raises(SystemExit): task.main() # ───────────────────────────────────────────────────────────────────────────── # Опциональный real-Postgres тест upsert+match (self-skip без БД) # ───────────────────────────────────────────────────────────────────────────── def _live_session() -> Any | None: try: from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "") if not dsn or "localhost:5432/test" in dsn: return None engine = create_engine(dsn, future=True) conn = engine.connect() conn.execute(text("SELECT 1")) conn.close() return sessionmaker(bind=engine, future=True)() except Exception: return None # Тот же нормализатор, что в мигр. 063 — создаём в транзакции теста, откатываем. _NORMALIZER_FN = """ CREATE OR REPLACE FUNCTION tradein_normalize_short_addr(addr text) RETURNS text LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $$ SELECT trim(both ' ,.' FROM regexp_replace( regexp_replace( regexp_replace( regexp_replace( regexp_replace( regexp_replace(addr, '^\\s*(?:Россия|РФ|Российская\\s+Федерация)\\s*,\\s*', '', 'i'), '^[А-ЯЁа-яё][А-ЯЁа-яё\\s-]+\\s+обл(?:асть|\\.)\\s*,\\s*', '', 'i'), '^г\\.?\\s*[А-ЯЁ][А-ЯЁа-яё-]+\\s*,\\s*', '', 'i'), '\\s*·\\s*.+$', '', 'i'), ',\\s*(?:кв\\.?|корп\\.?|к\\.?|оф\\.?|пом\\.?|подъезд)\\s*\\d+.*$', '', 'i'), '\\s{2,}', ' ', 'g')); $$; """ # Канон-fn мигр. 144 — создаём в транзакции теста, откатываем. raw-строка: \1\3 литеральны. _CANON_FN = r""" CREATE OR REPLACE FUNCTION tradein_canon_addr(s text) RETURNS text AS $func$ SELECT regexp_replace( regexp_replace( regexp_replace( translate(lower(coalesce(s, '')), 'ё', 'е'), 'р-н[^,]*,', '', 'g'), '(^|[^а-я])(улица|ул|переулок|пер|проспект|пркт|пр|бульвар|бр|шоссе|ш|проезд|набережная|наб|площадь|пл|тупик|туп|аллея|микрорайон|мкр)([^а-я]|$)', '\1\3', 'g'), '[^а-я0-9]', '', 'g'); $func$ LANGUAGE sql IMMUTABLE; """ @pytest.mark.skipif(_live_session() is None, reason="нет доступной Postgres test-БД") def test_upsert_and_canon_match_populates_gar_flat_count(gar_dir: str) -> None: """С реальной Postgres: upsert ГАР + КАНОН-match → houses.gar_flat_count (мигр. 144). Проверяет: канон бридит формат-расхождение, которого точный матч не видел («ул. Шаумяна,10» ↔ ГАР «ул Шаумяна, 10»); ЕКБ-фильтр исключает same-canon не-ЕКБ строку; идемпотентность повторного прогона; метод 'canon_addr'. """ from sqlalchemy import text db = _live_session() assert db is not None try: db.execute(text(_NORMALIZER_FN)) db.execute(text(_CANON_FN)) db.execute( text( "CREATE TEMP TABLE gar_house_flats (" " house_guid text PRIMARY KEY, object_id bigint, region_code text NOT NULL," " flat_count int, full_address text, norm_address text, street_name text," " house_num text, loaded_at timestamptz NOT NULL DEFAULT now(), gar_version text" ") ON COMMIT DROP" ) ) db.execute( text( "CREATE TEMP TABLE houses (" " id serial PRIMARY KEY, short_address text, full_address text, address text," " gar_house_guid text, gar_flat_count int, gar_matched_at timestamptz," " gar_match_method text" ") ON COMMIT DROP" ) ) # Format-divergent адреса: «ул. Шаумяна,10» (точка/без пробела) канон-совпадёт с ГАР # h-a «ул Шаумяна, 10» (flat=3), хотя точный матч их РАЗЛИЧИЛ бы. «ул Ленина, 7» — h-c # (flat=0) не матч. «ул. Мира,5» — для теста ЕКБ-фильтра (см. ниже). db.execute( text( "INSERT INTO houses (short_address) VALUES " "('ул. Шаумяна,10'), ('ул Ленина, 7'), ('ул. Мира,5')" ) ) rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01") upserted = gfl.upsert_gar_houses(db, rows, batch_size=2) assert upserted == 3 # norm_address посчитан SQL-fn = «ул Шаумяна, 10» (точный матч с houses провалился бы). norm = db.execute( text("SELECT norm_address FROM gar_house_flats WHERE house_guid='h-a'") ).scalar() assert norm == "ул Шаумяна, 10" # Канон-матч (ЕКБ default): Шаумяна канон-совпал, хотя формат адреса различался. matched = gfl.match_houses_to_gar(db, region_code="66") assert matched == 1 # Идемпотентность: повторный матч ничего не трогает. assert gfl.match_houses_to_gar(db, region_code="66") == 0 shaumyana = db.execute( text( "SELECT gar_flat_count, gar_match_method FROM houses " "WHERE short_address='ул. Шаумяна,10'" ) ).fetchone() assert shaumyana is not None and shaumyana[0] == 3 assert shaumyana[1] == "canon_addr" # Дом без квартир (flat_count=0) НЕ матчится. lenina = db.execute( text("SELECT gar_flat_count FROM houses WHERE short_address='ул Ленина, 7'") ).scalar() assert lenina is None # ── ЕКБ-фильтр: не-ЕКБ ГАР-строка с тем же каноном «мира5» (Берёзовский, flat=4) ── db.execute( text( "INSERT INTO gar_house_flats " "(house_guid, region_code, flat_count, full_address, norm_address) VALUES " "('h-berez', '66', 4, 'обл Свердловская, г Берёзовский, ул Мира, 5', 'ул Мира, 5')" ) ) # Default ЕКБ-фильтр: full_address без «Екатеринбург» исключён → Мира НЕ матчится. assert gfl.match_houses_to_gar(db, region_code="66") == 0 mira_before = db.execute( text("SELECT gar_flat_count FROM houses WHERE short_address='ул. Мира,5'") ).scalar() assert mira_before is None # city_filter=None отключает фильтр → не-ЕКБ строка матчится (canon «мира5»). assert gfl.match_houses_to_gar(db, region_code="66", city_filter=None) == 1 mira_after = db.execute( text( "SELECT gar_flat_count, gar_match_method FROM houses " "WHERE short_address='ул. Мира,5'" ) ).fetchone() assert mira_after is not None and mira_after[0] == 4 assert mira_after[1] == "canon_addr" finally: db.rollback() db.close()