409 lines
18 KiB
Python
409 lines
18 KiB
Python
"""Unit-тесты загрузчиков «точек подключения со свободной мощностью» (#2119).
|
||
|
||
Pure / mock-based — без реальной сети и БД. Покрывает:
|
||
- normalize_sc_name: «ПС 110/35/10 Нижне-Исетская» ↔ «Нижне-Исетская», ё, тире.
|
||
- parse_voltage_class: извлечение класса напряжения из имени ЦП.
|
||
- _map_load_index: sc_indexload_id → open/limited/closed/None.
|
||
- sanitize_tp_capacity_mva: кВА-санитайз (÷1000 для ТП > порога, РП не трогаем).
|
||
- parse_reserve_number / parse_water_number: запятая, знак (дефицит <0), «н/д».
|
||
- infer_period: «1kv.-2026» → '2026-Q1'.
|
||
- parse_docx_table_rows: forward-fill вертикального merge (vMerge) на xml-фрагменте.
|
||
- extract_water_rows: извлечение ЦСВ-записей из forward-filled матрицы.
|
||
- load_tp_rp_reserves: row-offset (данные с строки 9) на синтетическом workbook.
|
||
- load_cp_reserves: row-offset (данные с строки 8) + матч по sc_name_norm.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
from contextlib import contextmanager
|
||
from typing import Any
|
||
|
||
import openpyxl
|
||
|
||
from app.services.site_finder import rosseti_reserve_loader as rr
|
||
from app.services.site_finder import rosseti_wfs_loader as rw
|
||
from app.services.site_finder import vodokanal_reserve_loader as vk
|
||
|
||
# ── normalize_sc_name ─────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_normalize_strips_ps_and_voltage_prefix() -> None:
|
||
"""«ПС 110/35/10 Нижне-Исетская» → «нижне-исетская» (совпадает с голым именем)."""
|
||
a = rw.normalize_sc_name("ПС 110/35/10 Нижне-Исетская")
|
||
b = rw.normalize_sc_name("Нижне-Исетская")
|
||
assert a == b == "нижне-исетская"
|
||
|
||
|
||
def test_normalize_yo_to_e() -> None:
|
||
"""ё → е при нормализации."""
|
||
assert rw.normalize_sc_name("Берёзовская") == rw.normalize_sc_name("Березовская")
|
||
assert rw.normalize_sc_name("Берёзовская") == "березовская"
|
||
|
||
|
||
def test_normalize_dash_variants_collapse() -> None:
|
||
"""Разные тире (—, –, -) и пробелы вокруг схлопываются в один дефис."""
|
||
assert rw.normalize_sc_name("Нижне — Исетская") == "нижне-исетская"
|
||
assert rw.normalize_sc_name("Нижне–Исетская") == "нижне-исетская"
|
||
|
||
|
||
def test_normalize_quotes_removed() -> None:
|
||
"""«ПС "Уктус"» → «уктус» (кавычки и префикс срезаны)."""
|
||
assert rw.normalize_sc_name('ПС "Уктус"') == "уктус"
|
||
|
||
|
||
def test_normalize_empty() -> None:
|
||
assert rw.normalize_sc_name(None) == ""
|
||
assert rw.normalize_sc_name("") == ""
|
||
|
||
|
||
# ── parse_voltage_class ───────────────────────────────────────────────────────
|
||
|
||
|
||
def test_parse_voltage_class_triple() -> None:
|
||
assert rw.parse_voltage_class("ПС 110/35/10 Нижне-Исетская") == "110/35/10"
|
||
|
||
|
||
def test_parse_voltage_class_double() -> None:
|
||
assert rw.parse_voltage_class("ПС 110/10 Уктус") == "110/10"
|
||
|
||
|
||
def test_parse_voltage_class_none() -> None:
|
||
assert rw.parse_voltage_class("Уктус") is None
|
||
assert rw.parse_voltage_class(None) is None
|
||
|
||
|
||
# ── _map_load_index ───────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_map_load_index_known() -> None:
|
||
assert rw._map_load_index({"sc_indexload_id": 256184}) == "open"
|
||
assert rw._map_load_index({"sc_indexload_id": 256186}) == "limited"
|
||
assert rw._map_load_index({"sc_indexload_id": 256188}) == "closed"
|
||
|
||
|
||
def test_map_load_index_string_id_coerced() -> None:
|
||
"""id приходит строкой из json — коэрсим в int."""
|
||
assert rw._map_load_index({"sc_indexload_id": "256184"}) == "open"
|
||
|
||
|
||
def test_map_load_index_unknown_and_missing() -> None:
|
||
assert rw._map_load_index({"sc_indexload_id": 999999}) is None
|
||
assert rw._map_load_index({}) is None
|
||
assert rw._map_load_index({"sc_indexload_id": "мусор"}) is None
|
||
|
||
|
||
# ── sanitize_tp_capacity_mva (кВА-санитайз) ───────────────────────────────────
|
||
|
||
|
||
def test_kva_sanitize_tp_above_threshold_divided() -> None:
|
||
"""ТП с installed=630 (кВА в колонке МВА) → 0.63 МВА, флаг True."""
|
||
val, sanitized = rr.sanitize_tp_capacity_mva(630.0, is_rp=False)
|
||
assert sanitized is True
|
||
assert val == 0.63
|
||
|
||
|
||
def test_kva_sanitize_tp_below_threshold_kept() -> None:
|
||
"""ТП с installed=1.6 МВА (правдоподобно) → не трогаем."""
|
||
val, sanitized = rr.sanitize_tp_capacity_mva(1.6, is_rp=False)
|
||
assert sanitized is False
|
||
assert val == 1.6
|
||
|
||
|
||
def test_kva_sanitize_rp_never_divided() -> None:
|
||
"""РП с высокой мощностью легитимен → не делим даже выше порога."""
|
||
val, sanitized = rr.sanitize_tp_capacity_mva(10.0, is_rp=True)
|
||
assert sanitized is False
|
||
assert val == 10.0
|
||
|
||
|
||
def test_kva_sanitize_none() -> None:
|
||
val, sanitized = rr.sanitize_tp_capacity_mva(None, is_rp=False)
|
||
assert val is None and sanitized is False
|
||
|
||
|
||
# ── parse_reserve_number ──────────────────────────────────────────────────────
|
||
|
||
|
||
def test_parse_reserve_comma_decimal() -> None:
|
||
assert rr.parse_reserve_number("1,63") == 1.63
|
||
|
||
|
||
def test_parse_reserve_negative_kept() -> None:
|
||
"""Отрицательный резерв (дефицит) сохраняет знак."""
|
||
assert rr.parse_reserve_number("-2,5") == -2.5
|
||
|
||
|
||
def test_parse_reserve_thousands_space() -> None:
|
||
"""Неразрывный пробел-разделитель тысяч убирается."""
|
||
assert rr.parse_reserve_number("1\xa0234,5") == 1234.5
|
||
|
||
|
||
def test_parse_reserve_nodata() -> None:
|
||
assert rr.parse_reserve_number("н/д") is None
|
||
assert rr.parse_reserve_number("") is None
|
||
assert rr.parse_reserve_number(None) is None
|
||
assert rr.parse_reserve_number("текст") is None
|
||
|
||
|
||
def test_parse_reserve_numeric_passthrough() -> None:
|
||
assert rr.parse_reserve_number(5) == 5.0
|
||
assert rr.parse_reserve_number(2.5) == 2.5
|
||
|
||
|
||
# ── parse_asof_date ───────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_parse_asof_date_from_filename() -> None:
|
||
d = rr.parse_asof_date("резерв по ЦП на 30.06.2026.xlsx")
|
||
assert d is not None and d.isoformat() == "2026-06-30"
|
||
|
||
|
||
def test_parse_asof_date_none() -> None:
|
||
assert rr.parse_asof_date("no date here", None) is None
|
||
|
||
|
||
# ── water: parse_water_number / infer_period ──────────────────────────────────
|
||
|
||
|
||
def test_parse_water_negative_unicode_minus() -> None:
|
||
"""Юникод-минус −168,5 (дефицит) → -168.5."""
|
||
assert vk.parse_water_number("−168,5") == -168.5
|
||
|
||
|
||
def test_parse_water_nodata() -> None:
|
||
assert vk.parse_water_number("н/д") is None
|
||
assert vk.parse_water_number("") is None
|
||
assert vk.parse_water_number(None) is None
|
||
|
||
|
||
def test_infer_period_from_filename() -> None:
|
||
assert vk.infer_period("dlya-sajta-1kv.-2026.docx") == "2026-Q1"
|
||
assert vk.infer_period("dlya-sajta-3кв-2025.docx") == "2025-Q3"
|
||
|
||
|
||
def test_infer_period_none() -> None:
|
||
assert vk.infer_period("dlya-sajta.docx") is None
|
||
|
||
|
||
# ── DOCX vMerge forward-fill ──────────────────────────────────────────────────
|
||
|
||
_W = vk._W
|
||
|
||
|
||
def _tc(text: str = "", vmerge: str | None = None) -> str:
|
||
"""Строит XML одной ячейки <w:tc> с опциональным vMerge."""
|
||
tc_pr = ""
|
||
if vmerge == "restart":
|
||
tc_pr = '<w:tcPr><w:vMerge w:val="restart"/></w:tcPr>'
|
||
elif vmerge == "continue":
|
||
tc_pr = "<w:tcPr><w:vMerge/></w:tcPr>"
|
||
return f"<w:tc>{tc_pr}<w:p><w:r><w:t>{text}</w:t></w:r></w:p></w:tc>"
|
||
|
||
|
||
def _doc_xml(rows: list[str]) -> bytes:
|
||
"""Строит минимальный document.xml с одной таблицей из строк-<w:tr>."""
|
||
trs = "".join(f"<w:tr>{r}</w:tr>" for r in rows)
|
||
xml = f'<w:document xmlns:w="{_W}"><w:body><w:tbl>{trs}</w:tbl></w:body></w:document>'
|
||
return xml.encode("utf-8")
|
||
|
||
|
||
def test_docx_vmerge_forward_fill() -> None:
|
||
"""Ячейка колонки 0 в 'continue'-строках наследует значение из restart-строки."""
|
||
rows = [
|
||
_tc("ЗФС, ГСВ", vmerge="restart") + _tc("−168,5") + _tc("примечание"),
|
||
_tc("", vmerge="continue") + _tc("12,3") + _tc(""),
|
||
_tc("", vmerge="continue") + _tc("0,0") + _tc(""),
|
||
]
|
||
matrix = vk.parse_docx_table_rows(_doc_xml(rows))
|
||
assert len(matrix) == 3
|
||
# forward-fill: имя системы протянуто во все три строки блока.
|
||
assert matrix[0][0] == "ЗФС, ГСВ"
|
||
assert matrix[1][0] == "ЗФС, ГСВ"
|
||
assert matrix[2][0] == "ЗФС, ГСВ"
|
||
# числовая колонка не наследуется.
|
||
assert matrix[1][1] == "12,3"
|
||
|
||
|
||
def test_docx_empty_no_table() -> None:
|
||
xml = f'<w:document xmlns:w="{_W}"><w:body/></w:document>'.encode()
|
||
assert vk.parse_docx_table_rows(xml) == []
|
||
|
||
|
||
def test_extract_water_rows_from_matrix() -> None:
|
||
"""Из forward-filled матрицы: имя (текст), резерв (первое число, знак сохранён)."""
|
||
matrix = [
|
||
["система", "резерв", "примечание"], # заголовок (без числа) — пропуск
|
||
["ЗФС, ГСВ", "−168,5", "дефицит"],
|
||
["Головные, ЮАС", "42,0", "запас"],
|
||
]
|
||
recs = vk.extract_water_rows(matrix)
|
||
assert len(recs) == 2
|
||
assert recs[0]["system_name"] == "ЗФС, ГСВ"
|
||
assert recs[0]["reserve_thousand_m3_day"] == -168.5
|
||
assert recs[1]["reserve_thousand_m3_day"] == 42.0
|
||
|
||
|
||
# ── Fake DB session для upsert-тестов (SAVEPOINT + execute) ───────────────────
|
||
|
||
|
||
class _FakeResult:
|
||
def __init__(self, scalar_value: Any = True, rowcount: int = 1) -> None:
|
||
self._scalar = scalar_value
|
||
self.rowcount = rowcount
|
||
|
||
def scalar(self) -> Any:
|
||
return self._scalar
|
||
|
||
|
||
class _FakeSession:
|
||
"""Мини-заглушка Session: собирает execute-параметры, поддерживает begin_nested."""
|
||
|
||
def __init__(self, scalar_value: Any = True, rowcount: int = 1) -> None:
|
||
self.calls: list[tuple[str, Any]] = []
|
||
self._scalar = scalar_value
|
||
self._rowcount = rowcount
|
||
self.committed = False
|
||
|
||
@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, self._rowcount)
|
||
|
||
def commit(self) -> None:
|
||
self.committed = True
|
||
|
||
def rollback(self) -> None:
|
||
pass
|
||
|
||
|
||
# ── load_tp_rp_reserves: row-offset (данные с строки 9) на синтетическом xlsx ──
|
||
|
||
|
||
def _build_tp_rp_workbook() -> bytes:
|
||
"""Строит xlsx с верифицированной раскладкой ТП/РП: данные с СТРОКИ 9."""
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = "ТП РП на 30.06.2026"
|
||
# Строки 1-8 — заголовки/нумерация (шум), должны игнорироваться.
|
||
for r in range(1, 9):
|
||
ws.cell(row=r, column=1, value=f"header-{r}")
|
||
# Строка 9+ — данные. Колонки (1-based): name, РЭС, region, mun, кВ, inst, load, reserve.
|
||
ws.append(["ТП-721", "Центральный РЭС", "СО", "Екатеринбург", "10", 630.0, 100.0, 530.0])
|
||
ws.append(["РП-5", "Западный РЭС", "СО", "Екатеринбург", "10", 8.0, 2.0, 6.0])
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
return buf.getvalue()
|
||
|
||
|
||
def test_load_tp_rp_row_offset_and_kva() -> None:
|
||
"""Данные читаются с строки 9; ТП-721 630→0.63 МВА (кВА-санитайз); РП не делится."""
|
||
db = _FakeSession(scalar_value=True, rowcount=1)
|
||
result = rr.load_tp_rp_reserves(db, _build_tp_rp_workbook(), "тп рп на 30.06.2026.xlsx")
|
||
|
||
assert result["rows"] == 2, result
|
||
assert result["inserted"] == 2
|
||
assert result["kva_sanitized"] == 1 # только ТП-721
|
||
|
||
# Первый upsert — ТП-721 с санитайзнутой мощностью.
|
||
tp_params = db.calls[0][1]
|
||
assert tp_params["name"] == "ТП-721"
|
||
assert tp_params["installed"] == 0.63 # 630 кВА → 0.63 МВА
|
||
assert tp_params["reserve"] == 530.0
|
||
# РП-5 — мощность не тронута (легитимно высокая для РП, но тут 8 < порога всё равно).
|
||
rp_params = db.calls[1][1]
|
||
assert rp_params["name"] == "РП-5"
|
||
assert rp_params["installed"] == 8.0
|
||
|
||
|
||
# ── load_cp_reserves: row-offset (строка 8) + матч по sc_name_norm ─────────────
|
||
|
||
|
||
def _build_cp_workbook() -> bytes:
|
||
"""Строит xlsx с раскладкой ЦП: данные с СТРОКИ 8.
|
||
|
||
Колонки (1-based): A=№, B=ЦП name, C=ПО, D=?, E=municipality, F=voltage,
|
||
G=installed, H=current, I=reserve.
|
||
"""
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = "ЦП на 30.06.2026"
|
||
for r in range(1, 8):
|
||
ws.cell(row=r, column=1, value=f"header-{r}")
|
||
# Строка 8 — первая строка данных.
|
||
ws.append([1, "ПС 110/10 Уктус", "ПО-1", "", "Екатеринбург", "110/10", 40.0, 25.0, 15.0])
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
return buf.getvalue()
|
||
|
||
|
||
def test_load_cp_reserves_row_offset_and_norm_match() -> None:
|
||
"""Данные с строки 8; UPDATE по sc_name_norm; matched считает rowcount."""
|
||
db = _FakeSession(scalar_value=None, rowcount=1) # UPDATE matched 1 row
|
||
result = rr.load_cp_reserves(db, _build_cp_workbook(), "цп на 30.06.2026.xlsx")
|
||
|
||
assert result["rows"] == 1, result
|
||
assert result["matched"] == 1
|
||
assert result["unmatched"] == 0
|
||
|
||
params = db.calls[0][1]
|
||
# sc_name_norm совпадает с нормализацией голого имени «Уктус».
|
||
assert params["name_norm"] == rw.normalize_sc_name("Уктус") == "уктус"
|
||
assert params["installed"] == 40.0
|
||
assert params["reserve"] == 15.0
|
||
assert params["asof"] is not None and params["asof"].isoformat() == "2026-06-30"
|
||
|
||
|
||
def test_load_cp_reserves_unmatched_counted() -> None:
|
||
"""ЦП без совпадения в power_supply_centers (rowcount=0) → unmatched, не падаем."""
|
||
db = _FakeSession(scalar_value=None, rowcount=0) # UPDATE matched 0 rows
|
||
result = rr.load_cp_reserves(db, _build_cp_workbook(), "цп на 30.06.2026.xlsx")
|
||
assert result["matched"] == 0
|
||
assert result["unmatched"] == 1
|
||
|
||
|
||
# ── fetch_disclosure_links (прод-инцидент 2026-07-02: URL-энкоженные href) ────
|
||
|
||
|
||
def test_fetch_disclosure_links_decodes_urlencoded_hrefs(monkeypatch):
|
||
"""href'ы на странице раскрытия URL-энкоженные (кириллица как %D0%A1...) —
|
||
фильтр обязан работать по декодированному имени и выбирать свежайший файл.
|
||
Пермский файл отфильтровывается, из двух ЦП-дат берётся поздняя."""
|
||
from urllib.parse import quote, unquote
|
||
|
||
base = "/public/upload/Documents/2026/Transmission/"
|
||
cp_tpl = (
|
||
"Информация о наличии объема свободной мощности по ЦП ПАО «Россети Урал» "
|
||
"на территории {region} (на {d}).xlsx"
|
||
)
|
||
tp_name = (
|
||
"Информация о наличии свободной мощности для ТП по ПС и РП напряжением "
|
||
"ниже 35 кВ филиала Свердловэнерго на 30.06.2026.xlsx"
|
||
)
|
||
hrefs = [
|
||
base + quote(cp_tpl.format(region="Свердловской области", d="30.03.2026")),
|
||
base + quote(cp_tpl.format(region="Свердловской области", d="30.06.2026")),
|
||
base + quote(cp_tpl.format(region="Пермского края", d="30.06.2026")),
|
||
base + quote(tp_name),
|
||
]
|
||
html = "".join(f'<a href="{h}">x</a>' for h in hrefs)
|
||
|
||
class _Resp:
|
||
status_code = 200
|
||
text = html
|
||
|
||
def raise_for_status(self):
|
||
return None
|
||
|
||
monkeypatch.setattr(
|
||
"app.services.site_finder.rosseti_reserve_loader.httpx.get",
|
||
lambda *a, **k: _Resp(),
|
||
)
|
||
links = rr.fetch_disclosure_links()
|
||
assert "30.06.2026" in unquote(links["cp"])
|
||
assert "Пермск" not in unquote(links["cp"])
|
||
assert "%D0" in links["cp"] # скачиваем по СЫРОМУ (энкоженному) href
|
||
assert "Свердловэнерго" in unquote(links["tp_rp"])
|