277 lines
12 KiB
Python
277 lines
12 KiB
Python
"""Unit-тесты загрузчика резервов АО «ЕЭСК» (#2119 Phase B2).
|
||
|
||
Pure / mock-based — без реальной сети и БД. Покрывает:
|
||
- select_freshest_links: 2 серии (ПС 35-220 / ТП «ниже 35»), выбор свежайшего
|
||
квартала из HTML-фрагмента; классификация по тексту ссылки.
|
||
- _classify_series / _quarter_from_text / quarter_period / quarter_last_day.
|
||
- _pct_share_to_percent: доля 0.41 → 41.0.
|
||
- load_ps_35_220: шапка r4, данные r7+/r65+, строка-разделитель, доля→41.0,
|
||
матч ЕЭСК-only (dzo_name ILIKE %ЕЭСК%) / анматч.
|
||
- load_tp_rp_sub35: данные с r4, name «ТП/РП <A>», municipality='Екатеринбург'.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
from contextlib import contextmanager
|
||
from typing import Any
|
||
|
||
import openpyxl
|
||
|
||
from app.services.site_finder import eesk_reserve_loader as ee
|
||
from app.services.site_finder import rosseti_wfs_loader as rw
|
||
|
||
# ── select_freshest_links / классификация серий ──────────────────────────────
|
||
|
||
|
||
def _link(href: str, text: str) -> str:
|
||
return f'<a href="{href}">{text}</a>'
|
||
|
||
|
||
def test_select_freshest_links_two_series_quarters() -> None:
|
||
"""Из HTML-хаба выбирается свежайший квартал каждой серии; серии различаются."""
|
||
base = "/upload/site1/document_file/"
|
||
html = "".join(
|
||
[
|
||
_link(
|
||
base + "aaa.xlsx",
|
||
"Резерв мощности по центрам питания ПС 35-220 кВ 1 квартал 2026",
|
||
),
|
||
_link(
|
||
base + "bbb.xlsx",
|
||
"Резерв мощности по центрам питания ПС 35-220 кВ 2 квартала 2026",
|
||
),
|
||
_link(
|
||
base + "ccc.xls",
|
||
"Наличие свободной мощности ТП и РП ниже 35 кВ 1 квартал 2026",
|
||
),
|
||
_link(
|
||
base + "ddd.xls",
|
||
"Наличие свободной мощности ТП и РП ниже 35 кВ 2 квартала 2026",
|
||
),
|
||
]
|
||
)
|
||
links = ee.select_freshest_links(html)
|
||
|
||
assert set(links) == {"ps_35_220", "tp_rp_sub35"}
|
||
# Свежайший ПС — 2 квартал 2026 (bbb), абсолютный URL достроен.
|
||
assert links["ps_35_220"]["url"].endswith("bbb.xlsx")
|
||
assert links["ps_35_220"]["url"].startswith("https://www.eesk.ru")
|
||
assert (links["ps_35_220"]["year"], links["ps_35_220"]["quarter"]) == (2026, 2)
|
||
# Свежайший ТП — 2 квартал 2026 (ddd).
|
||
assert links["tp_rp_sub35"]["url"].endswith("ddd.xls")
|
||
assert (links["tp_rp_sub35"]["year"], links["tp_rp_sub35"]["quarter"]) == (2026, 2)
|
||
|
||
|
||
def test_select_freshest_picks_latest_year_over_quarter() -> None:
|
||
"""4 квартал 2025 старее, чем 1 квартал 2026 — берём 2026."""
|
||
base = "/upload/site1/document_file/"
|
||
html = "".join(
|
||
[
|
||
_link(base + "old.xlsx", "Резерв по центрам питания 35-220 4 квартал 2025"),
|
||
_link(base + "new.xlsx", "Резерв по центрам питания 35-220 1 квартал 2026"),
|
||
]
|
||
)
|
||
links = ee.select_freshest_links(html)
|
||
assert links["ps_35_220"]["url"].endswith("new.xlsx")
|
||
assert (links["ps_35_220"]["year"], links["ps_35_220"]["quarter"]) == (2026, 1)
|
||
|
||
|
||
def test_classify_series_below_wins_over_cp_phrase() -> None:
|
||
"""«ниже 35» классифицирует как ТП даже если рядом «по центрам питания»."""
|
||
assert ee._classify_series("резерв по центрам питания ниже 35 кв") == "tp_rp_sub35"
|
||
assert ee._classify_series("резерв мощности по центрам питания 35-220") == "ps_35_220"
|
||
assert ee._classify_series("что-то постороннее") is None
|
||
|
||
|
||
def test_quarter_from_text_and_helpers() -> None:
|
||
assert ee._quarter_from_text("… 3 квартал 2025 …") == (2025, 3)
|
||
assert ee._quarter_from_text("… 2 квартала 2026 …") == (2026, 2)
|
||
assert ee._quarter_from_text("без квартала 2026") is None
|
||
assert ee.quarter_period(2026, 2) == "2026-Q2"
|
||
assert ee.quarter_last_day(2026, 2).isoformat() == "2026-06-30"
|
||
assert ee.quarter_last_day(2026, 1).isoformat() == "2026-03-31"
|
||
assert ee.quarter_last_day(2025, 4).isoformat() == "2025-12-31"
|
||
|
||
|
||
def test_pct_share_to_percent() -> None:
|
||
"""Доля 0.41 → 41.0; уже-проценты не трогаем; мусор → None."""
|
||
assert ee._pct_share_to_percent(0.41) == 41.0
|
||
assert ee._pct_share_to_percent("0,41") == 41.0
|
||
assert ee._pct_share_to_percent(1.0) == 100.0
|
||
assert ee._pct_share_to_percent(41.0) == 41.0 # >1 → уже проценты
|
||
assert ee._pct_share_to_percent("н/д") is None
|
||
|
||
|
||
# ── Fake DB session (SAVEPOINT + execute), как в test_connection_capacity ────
|
||
|
||
|
||
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_ps_35_220: r4 шапка, данные r7+/r65+, разделитель, доля→41.0 ─────────
|
||
|
||
|
||
def _build_ps_workbook() -> bytes:
|
||
"""xlsx ПС 35-220: лист «Лист1», шапка r4, данные с r7, строка-разделитель.
|
||
|
||
Колонки 0-based: A=№, B=адм.район, C=имя ПС, D=макс полн. МВА, E=доля загрузки,
|
||
F=степень МУ250, G=свободная МВт.
|
||
"""
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = "Лист1"
|
||
# Строки 1-6 — заголовки/шапка (r4 — реальная шапка), должны игнорироваться.
|
||
for r in range(1, 7):
|
||
ws.cell(row=r, column=1, value=f"header-{r}")
|
||
# Строка 7 — первая строка данных.
|
||
ws.append([1, "Ленинский", "ПС 110/10 Уктус", 40.0, 0.41, "МУ250", 15.0])
|
||
# Строка 8 — вторая ПС.
|
||
ws.append([2, "Кировский", "ПС 35/6 Изоплит", 16.0, 0.60, "МУ250", 4.0])
|
||
# Строка 9 — РАЗДЕЛИТЕЛЬ секций: C пусто И D не число → скип.
|
||
ws.append([None, None, None, None, None, None, None])
|
||
# Строка 10 — ещё одна ПС (эмулирует секцию r65+).
|
||
ws.append([3, "Октябрьский", "ПС 110/10 Сибирь", 25.0, 0.50, "МУ250", 8.0])
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
return buf.getvalue()
|
||
|
||
|
||
def test_load_ps_35_220_parse_and_match() -> None:
|
||
"""Данные с r7, разделитель пропущен, доля→%, матч по sc_name_norm."""
|
||
from datetime import date
|
||
|
||
db = _FakeSession(scalar_value=None, rowcount=1) # UPDATE matched 1 row
|
||
result = ee.load_ps_35_220(db, _build_ps_workbook(), date(2026, 6, 30))
|
||
|
||
# 3 строки данных (разделитель отсеян), все смэтчились.
|
||
assert result["rows"] == 3, result
|
||
assert result["matched"] == 3
|
||
assert result["unmatched"] == 0
|
||
|
||
first = db.calls[0][1]
|
||
assert first["name_norm"] == rw.normalize_sc_name("Уктус") == "уктус"
|
||
assert first["installed"] == 40.0
|
||
assert first["reserve"] == 15.0
|
||
assert first["district"] == "Ленинский"
|
||
assert first["load_pct"] == 41.0 # доля 0.41 → 41.0%
|
||
assert first["asof"] == date(2026, 6, 30)
|
||
|
||
|
||
def test_load_ps_35_220_unmatched_counted() -> None:
|
||
"""ПС без совпадения (rowcount=0 — напр. не ЕЭСК) → unmatched, не падаем."""
|
||
from datetime import date
|
||
|
||
db = _FakeSession(scalar_value=None, rowcount=0) # UPDATE matched 0 rows
|
||
result = ee.load_ps_35_220(db, _build_ps_workbook(), date(2026, 6, 30))
|
||
assert result["rows"] == 3
|
||
assert result["matched"] == 0
|
||
assert result["unmatched"] == 3
|
||
|
||
|
||
def test_load_ps_35_220_sql_scopes_to_eesk() -> None:
|
||
"""UPDATE ограничен dzo_name ILIKE '%ЕЭСК%' — не затирает Россети-строки."""
|
||
from datetime import date
|
||
|
||
db = _FakeSession(scalar_value=None, rowcount=1)
|
||
ee.load_ps_35_220(db, _build_ps_workbook(), date(2026, 6, 30))
|
||
sql = db.calls[0][0]
|
||
assert "ILIKE '%ЕЭСК%'" in sql
|
||
assert "capacity_source = 'eesk_35_220'" in sql
|
||
assert "reserve_unit = 'МВт'" in sql
|
||
|
||
|
||
# ── load_tp_rp_sub35: данные с r4, name «ТП/РП <A>», municipality ────────────
|
||
|
||
|
||
def _build_tp_workbook() -> bytes:
|
||
"""xlsx ТП/РП «ниже 35»: лист «Публикация», r1 title, r3 шапка, данные r4+.
|
||
|
||
Колонки 0-based: A=номер, B=предельно доп. МВт, C=загрузка МВт, D=резерв МВт.
|
||
"""
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = "Публикация"
|
||
ws.cell(row=1, column=1, value="Заголовок публикации") # r1 title
|
||
# r2 пустая, r3 — шапка.
|
||
ws.cell(row=3, column=1, value="№ ТП/РП")
|
||
ws.cell(row=3, column=2, value="предельно допустимая, МВт")
|
||
# Строка 4+ — данные.
|
||
ws.cell(row=4, column=1, value=10003) # int-номер (Excel часто float)
|
||
ws.cell(row=4, column=2, value=1.6)
|
||
ws.cell(row=4, column=3, value=1.0)
|
||
ws.cell(row=4, column=4, value=0.6)
|
||
ws.cell(row=5, column=1, value="10004") # уже строка
|
||
ws.cell(row=5, column=2, value=2.5)
|
||
ws.cell(row=5, column=3, value=2.0)
|
||
ws.cell(row=5, column=4, value=0.5)
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
return buf.getvalue()
|
||
|
||
|
||
def test_load_tp_rp_sub35_parse() -> None:
|
||
"""Данные с r4; name «ТП/РП 10003», name_norm='10003', municipality='Екатеринбург'."""
|
||
from datetime import date
|
||
|
||
db = _FakeSession(scalar_value=True, rowcount=1)
|
||
result = ee.load_tp_rp_sub35(db, _build_tp_workbook(), date(2026, 6, 30))
|
||
|
||
assert result["rows"] == 2, result
|
||
assert result["inserted"] == 2
|
||
|
||
first = db.calls[0][1]
|
||
# float-номер 10003.0 из Excel нормализован к «10003» (без .0).
|
||
assert first["name"] == "ТП/РП 10003"
|
||
assert first["name_norm"] == "10003"
|
||
assert first["municipality"] == "Екатеринбург"
|
||
assert first["installed"] == 1.6
|
||
assert first["current"] == 1.0
|
||
assert first["reserve"] == 0.6
|
||
assert first["asof"] == date(2026, 6, 30)
|
||
|
||
second = db.calls[1][1]
|
||
assert second["name"] == "ТП/РП 10004"
|
||
assert second["name_norm"] == "10004"
|
||
|
||
|
||
def test_load_tp_rp_sub35_voltage_null_no_kva_sanitize() -> None:
|
||
"""voltage_kv NULL в SQL; installed НЕ делится на 1000 (без кВА-санитайза)."""
|
||
from datetime import date
|
||
|
||
db = _FakeSession(scalar_value=True, rowcount=1)
|
||
ee.load_tp_rp_sub35(db, _build_tp_workbook(), date(2026, 6, 30))
|
||
sql = db.calls[0][0]
|
||
assert "voltage_kv" in sql and "NULL" in sql
|
||
# installed сохранён как есть (1.6), без деления.
|
||
assert db.calls[0][1]["installed"] == 1.6
|