gendesign/backend/tests/test_eesk_reserve_loader.py

"""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["asof"] == date(2026, 6, 30)


def test_load_ps_35_220_does_not_write_load_percent() -> None:
    """#2464-B: степень загрузки НЕ уходит в UPDATE и не попадает в load_index.

    Раньше значение колонки E писалось как
    `load_index = COALESCE(load_index, CAST(:load_pct AS text))`. load_index —
    категориальная колонка ('open'|'limited'|'closed'|NULL,
    data/sql/180_connection_capacity.sql:35): число строкой фронт отбрасывает
    в «неизвестно» (classifyLoadIndex), а в power_summary.by_load_index
    появлялся бы бакет с именем вроде "41.0".

    На проде не стреляло только потому, что load_index заполнен у всех строк
    (open 2741 / limited 346 / closed 329, NULL 0 — замер верификации 13.08),
    и COALESCE не проваливался.
    """
    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 убираем: слово load_index встречается в пояснении,
    # а проверять надо ИСПОЛНЯЕМЫЙ текст, а не прозу вокруг него.
    sql_code = "\n".join(line.split("--", 1)[0] for line in str(db.calls[0][0]).splitlines())
    assert "load_index" not in sql_code, sql_code
    for _sql, params in db.calls:
        assert "load_pct" not in params, params


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