457 lines
19 KiB
Python
457 lines
19 KiB
Python
"""Unit-тесты загрузчика точек выхода ГРС «Екатеринбурггаз» (ekgas.ru) (#2119 B2, PR-2).
|
||
|
||
Pure / mock-based — без реальной сети и БД. Покрывает:
|
||
- extract_form6_links + select_latest_file: выбор свежайшего файла из синтетического
|
||
каталога с ПРИОРИТЕТОМ ФАКТА («за») над прогнозом («на») — даже когда «на» новее по
|
||
(год, месяц) (отдельный тест); годовой .xls / не-форма-6 отфильтрованы;
|
||
- parse_outlet_sheet: РЕАЛЬНАЯ сетка (подписи row 10, нумерация row 11, данные с row 12),
|
||
forward-fill block-marker ГРС, «0.000*»→needs_calc (НЕ ноль), дефицит −0.084 сохранён,
|
||
сноска-скип single-cell;
|
||
- extract_address: скобочный/хвостовой адрес извлечён, юрлицо без адреса → None;
|
||
- _dedup_outlet_names: суффикс « (N)»; _upsert_outlet_rows: счётчики (mock-DB).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
from contextlib import contextmanager
|
||
from datetime import date
|
||
from typing import Any
|
||
|
||
import openpyxl
|
||
|
||
from app.services.site_finder import ekgas_outlet_loader as ek
|
||
|
||
# ── Каталог: синтетический HTML со ссылками на формы 6 ────────────────────────
|
||
|
||
|
||
def _form6_name(kind: str, month: str, year: int) -> str:
|
||
"""Имя файла формы 6 приложения №4 в реальном формате ekgas."""
|
||
return (
|
||
f"Форма 6 приложения № 4 к Приказу ФАС России № 960 от 08.12.2022г. "
|
||
f"{kind} {month} {year}г..xlsx"
|
||
)
|
||
|
||
|
||
def _catalog_link(kind: str, month: str, year: int, hex_dir: str, asof: str = "") -> str:
|
||
"""<a> каталога: путь /uploads/default/<hex>/<имя> + опц. дата публикации рядом."""
|
||
name = _form6_name(kind, month, year)
|
||
href = f"/uploads/default/{hex_dir}/{name}"
|
||
asof_html = f"<span>{asof}</span>" if asof else ""
|
||
return f'<div class="row"><a href="{href}">{name}</a>{asof_html}</div>'
|
||
|
||
|
||
def test_extract_form6_links_parses_period_and_asof() -> None:
|
||
"""Период «(за|на) <месяц> <год>» + asof (DD.MM.YYYY рядом) извлекаются; URL к apex."""
|
||
html = (
|
||
"<html><body>"
|
||
+ _catalog_link("за", "май", 2026, "a1b2c3", asof="03.06.2026")
|
||
+ "</body></html>"
|
||
)
|
||
|
||
links = ek.extract_form6_links(html)
|
||
|
||
assert len(links) == 1
|
||
link = links[0]
|
||
assert link["period_month"] == date(2026, 5, 1)
|
||
assert link["is_forecast"] is False # «за» = факт
|
||
assert link["asof"] == date(2026, 6, 3)
|
||
assert link["url"].startswith("https://ekgas.ru/uploads/default/a1b2c3/")
|
||
|
||
|
||
def test_extract_form6_links_na_is_forecast() -> None:
|
||
"""«на <месяц>» → is_forecast=True (прогноз)."""
|
||
html = "<html>" + _catalog_link("на", "июль", 2026, "deadbeef") + "</html>"
|
||
links = ek.extract_form6_links(html)
|
||
assert len(links) == 1
|
||
assert links[0]["is_forecast"] is True
|
||
assert links[0]["period_month"] == date(2026, 7, 1)
|
||
|
||
|
||
def test_extract_form6_links_skips_yearly_xls_and_foreign() -> None:
|
||
"""Годовой «Форма №6 (Приложение №2) … план на <год>» (.xls) и прочие файлы — скип."""
|
||
html = (
|
||
"<html>"
|
||
# годовой план — приложение №2, .xls → НЕ форма 6 приложения №4
|
||
'<a href="/uploads/default/ff00/Форма №6 (Приложение №2) план на 2026г..xls">'
|
||
"годовой план</a>"
|
||
# посторонний файл
|
||
'<a href="/uploads/default/aa11/Прайс-лист.pdf">прайс</a>'
|
||
# валидная помесячная форма 6
|
||
+ _catalog_link("за", "апрель", 2026, "cc22")
|
||
+ "</html>"
|
||
)
|
||
|
||
links = ek.extract_form6_links(html)
|
||
|
||
assert len(links) == 1
|
||
assert links[0]["period_month"] == date(2026, 4, 1)
|
||
|
||
|
||
def test_select_latest_file_fact_beats_newer_forecast() -> None:
|
||
"""ПРИОРИТЕТ ФАКТА: «за май 2026» выбирается, хотя «на июль 2026» новее по (год,месяц).
|
||
|
||
Документирует правило select_latest_file: если есть хоть один файл «за» (факт),
|
||
берём максимальный (год, месяц) СРЕДИ «за», даже если прогноз «на» новее. Прогноз
|
||
берётся ТОЛЬКО когда файлов «за» нет вообще.
|
||
"""
|
||
html = (
|
||
"<html>"
|
||
+ _catalog_link("за", "апрель", 2026, "h1")
|
||
+ _catalog_link("за", "май", 2026, "h2") # свежайший ФАКТ
|
||
+ _catalog_link("на", "июль", 2026, "h3") # прогноз новее по (год,месяц)
|
||
+ "</html>"
|
||
)
|
||
|
||
chosen = ek.select_latest_file(ek.extract_form6_links(html))
|
||
|
||
assert chosen is not None
|
||
assert chosen["period_month"] == date(2026, 5, 1) # «за май», не «на июль»
|
||
assert chosen["is_forecast"] is False
|
||
|
||
|
||
def test_select_latest_file_forecast_only_fallback() -> None:
|
||
"""Файлов «за» нет вообще → берём свежайший «на» (прогноз, is_forecast=True)."""
|
||
html = (
|
||
"<html>"
|
||
+ _catalog_link("на", "июнь", 2026, "f1")
|
||
+ _catalog_link("на", "август", 2026, "f2") # свежайший прогноз
|
||
+ "</html>"
|
||
)
|
||
|
||
chosen = ek.select_latest_file(ek.extract_form6_links(html))
|
||
|
||
assert chosen is not None
|
||
assert chosen["period_month"] == date(2026, 8, 1)
|
||
assert chosen["is_forecast"] is True
|
||
|
||
|
||
def test_select_latest_file_empty_returns_none() -> None:
|
||
assert ek.select_latest_file([]) is None
|
||
|
||
|
||
# ── parse_outlet_sheet: РЕАЛЬНАЯ сетка (row 10 подписи, 11 нумерация, 12+ данные) ──
|
||
|
||
|
||
def _pad(cells: dict[int, object], width: int = 7) -> list[object]:
|
||
"""Строка листа из width колонок: dict{col_index: value} → выровненный ряд."""
|
||
row: list[object] = [None] * width
|
||
for idx, value in cells.items():
|
||
row[idx] = value
|
||
return row
|
||
|
||
|
||
def _outlet_xlsx_bytes() -> bytes:
|
||
"""XLSX формы 6 — сокращённая РЕАЛЬНАЯ сетка (собрана openpyxl'ом как на проде).
|
||
|
||
Строки 1-9 пустые/шапка, row 10 = подписи, row 11 = нумерация 1..7, данные с row 12:
|
||
* 2 блока ГРС (block-marker в col0 только в 1-й строке блока → forward-fill);
|
||
* «0.000*» в col6 → needs_calc (НЕ ноль);
|
||
* отрицательный free (дефицит −0.084) сохраняется знаком;
|
||
* адрес в «Точке выхода» → address_extracted; юрлицо без адреса → None;
|
||
* последняя строка — сноска про «*» (single-cell) → скип.
|
||
"""
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = "Форма 6"
|
||
|
||
# Строки 1-9: пустая шапка формы (title/период/пусто) — до подписей.
|
||
for _ in range(9):
|
||
ws.append(_pad({}))
|
||
# row 10 — подписи колонок.
|
||
ws.append(
|
||
_pad(
|
||
{
|
||
0: "Точка входа",
|
||
1: "Точка выхода",
|
||
2: "Тип потребителя",
|
||
3: "Группа потребления",
|
||
4: "Объём заявок, млн.куб.м",
|
||
5: "Удовлетворено, млн.куб.м",
|
||
6: "Свободная мощность, млн.куб.м",
|
||
}
|
||
)
|
||
)
|
||
# row 11 — нумерация столбцов 1..7.
|
||
ws.append(_pad({0: 1, 1: 2, 2: 3, 3: 4, 4: 5, 5: 6, 6: 7}))
|
||
|
||
# row 12 — блок 1, первая строка (block-marker ГРС в col0).
|
||
ws.append(
|
||
_pad(
|
||
{
|
||
0: "ГРС Арамиль",
|
||
1: 'ООО "Тепло" (г.Екатеринбург, Складской проезд,4а)',
|
||
2: "котельная",
|
||
3: "5 - 0,1-1млн.куб.м",
|
||
4: 1.2,
|
||
5: 0.8,
|
||
6: 0.4,
|
||
}
|
||
)
|
||
)
|
||
# row 13 — блок 1, вторая строка (col0 ПУСТ → forward-fill «ГРС Арамиль»),
|
||
# свободная мощность «0.000*» → needs_calc.
|
||
ws.append(
|
||
_pad(
|
||
{
|
||
1: "ИП Рошканяну Владимир Антонович", # юрлицо без адреса → address None
|
||
2: "промплощадка",
|
||
4: 0.05,
|
||
5: 0.05,
|
||
6: "0.000*", # НЕ ноль — требует гидрорасчёта
|
||
}
|
||
)
|
||
)
|
||
# row 14 — блок 2, первая строка (новый block-marker), дефицит −0.084.
|
||
ws.append(
|
||
_pad(
|
||
{
|
||
0: "ГРС Кольцово",
|
||
1: 'ЗАО "Луч", пер. Базовый, 17', # адрес хвостом
|
||
2: "ТЭЦ",
|
||
3: "4",
|
||
4: 2.5,
|
||
5: 2.5,
|
||
6: -0.084, # дефицит — знак сохраняется
|
||
}
|
||
)
|
||
)
|
||
# row 15 — сноска про «*» (single-cell) → скип.
|
||
ws.append(_pad({0: "* требуется гидравлический расчёт свободной мощности"}))
|
||
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
return buf.getvalue()
|
||
|
||
|
||
def test_parse_outlet_sheet_forward_fill_and_markers() -> None:
|
||
"""Forward-fill ГРС, «0.000*»→needs_calc, дефицит сохранён, адрес извлечён, сноска-скип."""
|
||
records = ek.parse_outlet_sheet(_outlet_xlsx_bytes())
|
||
|
||
# 3 строки данных (сноска single-cell отброшена).
|
||
assert len(records) == 3
|
||
|
||
r0, r1, r2 = records
|
||
|
||
# Блок 1, строка 1.
|
||
assert r0["grs_name"] == "ГРС Арамиль"
|
||
assert r0["free_capacity_mln_m3"] == 0.4
|
||
assert r0["free_capacity_needs_calc"] is False
|
||
assert r0["address_extracted"] == "г.Екатеринбург, Складской проезд,4а"
|
||
|
||
# Блок 1, строка 2 — forward-fill той же ГРС + «0.000*» → needs_calc, NULL free.
|
||
assert r1["grs_name"] == "ГРС Арамиль" # col0 был пуст → протянут
|
||
assert r1["free_capacity_mln_m3"] is None
|
||
assert r1["free_capacity_needs_calc"] is True # «0.000*» — НЕ ноль
|
||
assert r1["address_extracted"] is None # ИП без адреса
|
||
|
||
# Блок 2 — новый block-marker + дефицит −0.084 (знак сохранён) + адрес хвостом.
|
||
assert r2["grs_name"] == "ГРС Кольцово"
|
||
assert r2["free_capacity_mln_m3"] == -0.084
|
||
assert r2["free_capacity_needs_calc"] is False
|
||
assert r2["address_extracted"] == "пер. Базовый, 17"
|
||
|
||
# raw_cells — сырая строка списком (для jsonb raw).
|
||
assert r0["raw_cells"][0] == "ГРС Арамиль"
|
||
|
||
|
||
def test_parse_outlet_sheet_volumes_and_meta() -> None:
|
||
"""Объёмы заявок/удовлетворено и тип/группа парсятся из своих колонок."""
|
||
records = ek.parse_outlet_sheet(_outlet_xlsx_bytes())
|
||
r0 = records[0]
|
||
assert r0["volume_requested_mln_m3"] == 1.2
|
||
assert r0["volume_approved_mln_m3"] == 0.8
|
||
assert r0["consumer_type"] == "котельная"
|
||
assert r0["consumption_group"] == "5 - 0,1-1млн.куб.м"
|
||
|
||
|
||
# ── extract_address: консервативный адрес-экстрактор ─────────────────────────
|
||
|
||
|
||
def test_extract_address_parenthetical() -> None:
|
||
"""«(г.Екатеринбург, Складской проезд,4а)» → адрес извлечён."""
|
||
name = 'ООО "Тепло" (г.Екатеринбург, Складской проезд,4а)'
|
||
assert ek.extract_address(name) == "г.Екатеринбург, Складской проезд,4а"
|
||
|
||
|
||
def test_extract_address_trailing() -> None:
|
||
"""«, пер. Базовый, 17» хвостом → адрес извлечён."""
|
||
assert ek.extract_address('ЗАО "Луч", пер. Базовый, 17') == "пер. Базовый, 17"
|
||
|
||
|
||
def test_extract_address_street_with_house() -> None:
|
||
"""«г.…, ул. К.Маркса,31» → адрес с улицей и домом."""
|
||
got = ek.extract_address("Котельная, г. Реж, ул. К.Маркса,31")
|
||
assert got is not None
|
||
assert "К.Маркса" in got
|
||
assert got.rstrip().endswith("31")
|
||
|
||
|
||
def test_extract_address_legal_entity_returns_none() -> None:
|
||
"""Юрлицо без адресного сигнала → None (не выдумываем)."""
|
||
assert ek.extract_address("ИП Рошканяну Владимир Антонович") is None
|
||
|
||
|
||
def test_extract_address_no_street_signal_returns_none() -> None:
|
||
"""Название с числом, но без улицы/города («ООО ТЭЦ-2») → None (число ≠ дом)."""
|
||
assert ek.extract_address('ООО "ТЭЦ-2"') is None
|
||
assert ek.extract_address("") is None
|
||
|
||
|
||
# ── _parse_free_capacity: «0.000*» vs настоящий 0 vs дефицит ─────────────────
|
||
|
||
|
||
def test_parse_free_capacity_needs_calc_marker() -> None:
|
||
"""«0.000*» → (None, True). Вариации с запятой/пробелом тоже маркер."""
|
||
assert ek._parse_free_capacity("0.000*") == (None, True)
|
||
assert ek._parse_free_capacity("0,000 *") == (None, True)
|
||
|
||
|
||
def test_parse_free_capacity_real_zero_and_deficit() -> None:
|
||
"""Настоящий 0 (без «*») → (0.0, False); дефицит −0.084 → (−0.084, False)."""
|
||
assert ek._parse_free_capacity(0.0) == (0.0, False)
|
||
assert ek._parse_free_capacity(-0.084) == (-0.084, False)
|
||
assert ek._parse_free_capacity("—") == (None, False)
|
||
|
||
|
||
# ── _dedup_outlet_names: повторы outlet_name внутри файла ─────────────────────
|
||
|
||
|
||
def test_dedup_outlet_names_suffix() -> None:
|
||
"""Одинаковые outlet_name у одной ГРС → 2-му и далее суффикс « (N)» (не схлопывать)."""
|
||
records = [
|
||
{"grs_name": "ГРС Арти", "outlet_name": "Котельная №1"},
|
||
{"grs_name": "ГРС Арти", "outlet_name": "Котельная №1"}, # дубль
|
||
{"grs_name": "ГРС Арти", "outlet_name": "Котельная №1"}, # ещё дубль
|
||
{"grs_name": "ГРС Кольцово", "outlet_name": "Котельная №1"}, # другая ГРС → не дубль
|
||
]
|
||
|
||
ek._dedup_outlet_names(records)
|
||
|
||
assert [r["outlet_name"] for r in records] == [
|
||
"Котельная №1",
|
||
"Котельная №1 (2)",
|
||
"Котельная №1 (3)",
|
||
"Котельная №1", # другая ГРС — суффикс не нужен
|
||
]
|
||
|
||
|
||
# ── Fake DB (SAVEPOINT + execute), как в test_eias_heat_loader.py ─────────────
|
||
|
||
|
||
class _FakeResult:
|
||
def __init__(self, scalar_value: Any = True) -> None:
|
||
self._scalar = scalar_value
|
||
|
||
def scalar(self) -> Any:
|
||
return self._scalar
|
||
|
||
|
||
class _FakeSession:
|
||
"""Мини-заглушка Session: собирает execute-параметры, поддерживает begin_nested."""
|
||
|
||
def __init__(self, scalar_value: Any = True) -> None:
|
||
self.calls: list[tuple[str, Any]] = []
|
||
self._scalar = scalar_value
|
||
|
||
@contextmanager
|
||
def begin_nested(self): # type: ignore[no-untyped-def]
|
||
yield
|
||
|
||
def execute(self, stmt: Any, params: Any = None) -> _FakeResult:
|
||
self.calls.append((str(stmt), params))
|
||
return _FakeResult(self._scalar)
|
||
|
||
|
||
def _record(
|
||
grs: str,
|
||
outlet: str,
|
||
free: float | None,
|
||
needs_calc: bool = False,
|
||
address: str | None = None,
|
||
) -> dict:
|
||
return {
|
||
"grs_name": grs,
|
||
"outlet_name": outlet,
|
||
"consumer_type": "котельная",
|
||
"consumption_group": "4",
|
||
"volume_requested_mln_m3": 1.0,
|
||
"volume_approved_mln_m3": 0.9,
|
||
"free_capacity_mln_m3": free,
|
||
"free_capacity_needs_calc": needs_calc,
|
||
"address_extracted": address,
|
||
"raw_cells": [grs, outlet, "котельная"],
|
||
}
|
||
|
||
|
||
_META = {
|
||
"period_month": date(2026, 5, 1),
|
||
"is_forecast": False,
|
||
"asof": date(2026, 6, 3),
|
||
"source_url": "https://ekgas.ru/uploads/default/h1/f.xlsx",
|
||
}
|
||
|
||
|
||
def test_upsert_outlet_rows_counters_and_params() -> None:
|
||
"""Счётчики inserted/needs_calc/with_address; params корректны; CAST jsonb + ON CONFLICT."""
|
||
db = _FakeSession(scalar_value=True) # xmax=0 → insert
|
||
records = [
|
||
_record("ГРС Арамиль", "ООО Тепло", 0.4, address="пер. Базовый, 17"),
|
||
_record("ГРС Арамиль", "ИП Рошканяну", None, needs_calc=True),
|
||
_record("ГРС Кольцово", "ЗАО Луч", -0.084), # дефицит: знак сохраняется
|
||
]
|
||
|
||
counts = ek._upsert_outlet_rows(db, records, {**_META, "source_url": _META["source_url"]})
|
||
|
||
assert counts == {
|
||
"rows": 3,
|
||
"inserted": 3,
|
||
"updated": 0,
|
||
"skipped": 0,
|
||
"needs_calc": 1,
|
||
"with_address": 1,
|
||
}
|
||
assert len(db.calls) == 3
|
||
|
||
sql, first = db.calls[0]
|
||
assert first["grs_name"] == "ГРС Арамиль"
|
||
assert first["grs_name_norm"] == "грс арамиль" # нормализация из B1
|
||
assert first["free"] == 0.4
|
||
assert first["period_month"] == date(2026, 5, 1)
|
||
assert first["is_forecast"] is False
|
||
assert first["address"] == "пер. Базовый, 17"
|
||
assert '"cells"' in first["raw"]
|
||
assert "CAST(:raw AS jsonb)" in sql
|
||
assert "ON CONFLICT (grs_name_norm, outlet_name, period_month, is_forecast)" in sql
|
||
# needs_calc-строка: free NULL, needs_calc True прокинуты.
|
||
assert db.calls[1][1]["free"] is None
|
||
assert db.calls[1][1]["needs_calc"] is True
|
||
# дефицит сохранён знаком.
|
||
assert db.calls[2][1]["free"] == -0.084
|
||
|
||
|
||
def test_upsert_outlet_rows_update_counter() -> None:
|
||
"""scalar()=False (xmax≠0) → строка учтена как update, не insert."""
|
||
db = _FakeSession(scalar_value=False)
|
||
counts = ek._upsert_outlet_rows(db, [_record("ГРС X", "outlet", 1.0)], _META)
|
||
assert counts["updated"] == 1
|
||
assert counts["inserted"] == 0
|
||
|
||
|
||
def test_upsert_outlet_rows_skip_on_error() -> None:
|
||
"""execute бросает → строка skipped, счётчики insert/update не растут."""
|
||
|
||
class _RaisingSession(_FakeSession):
|
||
def execute(self, stmt: Any, params: Any = None) -> _FakeResult:
|
||
raise RuntimeError("db boom")
|
||
|
||
db = _RaisingSession()
|
||
counts = ek._upsert_outlet_rows(db, [_record("ГРС Y", "outlet", 2.0)], _META)
|
||
assert counts == {
|
||
"rows": 1,
|
||
"inserted": 0,
|
||
"updated": 0,
|
||
"skipped": 1,
|
||
"needs_calc": 0,
|
||
"with_address": 0,
|
||
}
|