→ None."""
assert eh.extract_tpl_xlsx("пусто") is None
def test_extract_tpl_xlsx_bad_zip_returns_none() -> None:
"""data-file — валидный base64, но не ZIP → None (graceful)."""
b64 = base64.b64encode(b"not a zip at all").decode("ascii")
html = f'
'
assert eh.extract_tpl_xlsx(html) is None
# ── parse_heat_sheet: column-aligned парс РЕАЛЬНОЙ сетки ЕТК ──────────────────
def test_parse_heat_sheet_column_aligned_three_systems() -> None:
"""Ровно 3 системы, читаемые по колонкам строки №5; имена stripped, нуль сохранён.
Реальные прод-пары: СТ № 56 → 0.0, СТ № 54 → 3.26, СТ № 60 → 11.28.
"""
systems, all_cells = eh.parse_heat_sheet(_heat_xlsx_bytes())
pairs = [(s["system_name"], s["reserve_gcal_h"]) for s in systems]
assert pairs == [
("СТ № 56", 0.0), # нулевой резерв — честный дефицит, СОХРАНЯЕТСЯ
("СТ № 54", 3.26), # trailing-space в имени убран .strip()
("СТ № 60", 11.28),
]
# all_cells содержит ВЕСЬ лист (7 строк) для raw.
assert len(all_cells) == 7
def test_parse_heat_sheet_ignores_anketa_rubrics_flag_row_size() -> None:
"""Регрессия прод-бага: рубрики анкеты (Flag_Row_Size 74/52 в служебной колонке)
и кнопка «Добавить систему» НЕ распознаются как системы теплоснабжения."""
systems, _ = eh.parse_heat_sheet(_heat_xlsx_bytes())
names = {s["system_name"] for s in systems}
reserves = {s["reserve_gcal_h"] for s in systems}
# ни одна рубрика/кнопка не просочилась
assert not any("Количество" in n for n in names)
assert not any("Причины" in n for n in names)
assert "Добавить систему" not in names
# значения Flag_Row_Size (74, 52, 15) не попали в резервы
assert 74.0 not in reserves
assert 52.0 not in reserves
def test_parse_heat_sheet_suffixes_duplicate_system_names() -> None:
"""Разные системы с БАЙТ-В-БАЙТ одинаковым именем не схлопываются.
Живой прод-кейс (Екатеринбургэнерго, guid fe0c8c57): col7 и col8 обе называются
«Локальные зоны теплоснабжения МУП "Екатеринбургэнерго"» с резервами 62.198 и 0 —
UPSERT по (org, sphere, system_name, period) перезаписывал 62.198 нулём. Вторым
и далее вхождениям добавляется суффикс « (N)».
"""
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "ТП"
dup = 'Локальные зоны теплоснабжения МУП "Екатеринбургэнерго"'
ws.append(
_pad(
{
4: "Централизованная система",
7: dup,
8: dup,
9: 'Локальные зоны теплоснабжения ПАО "Т Плюс"',
_ROW_SIZE_COL: 15,
}
)
)
ws.append(
_pad(
{
3: 5,
4: "Резерв мощности источников тепловой энергии, в т.ч.:",
5: "Гкал/час",
7: 62.198,
8: 0,
9: 0,
_ROW_SIZE_COL: 57,
}
)
)
buf = io.BytesIO()
wb.save(buf)
systems, _ = eh.parse_heat_sheet(buf.getvalue())
pairs = [(s["system_name"], s["reserve_gcal_h"]) for s in systems]
assert pairs == [
(dup, 62.198), # первое вхождение — без суффикса, 62.198 НЕ теряется
(f"{dup} (2)", 0.0),
('Локальные зоны теплоснабжения ПАО "Т Плюс"', 0.0),
]
def test_parse_heat_sheet_no_systems_header_returns_empty() -> None:
"""Нет строки «Централизованная система» → ([], all_cells) (graceful, не падаем)."""
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "ТП"
ws.append(_pad({4: "Какой-то посторонний заголовок", _ROW_SIZE_COL: 15}))
ws.append(_pad({3: 5, 4: "Резерв мощности источников тепловой энергии, в т.ч.:", 7: 3.26}))
buf = io.BytesIO()
wb.save(buf)
systems, all_cells = eh.parse_heat_sheet(buf.getvalue())
assert systems == []
assert len(all_cells) == 2
# ── Fake DB (SAVEPOINT + execute), как в test_eesk_reserve_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
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)
def commit(self) -> None:
self.committed = True
def rollback(self) -> None:
pass
def test_upsert_heat_rows_counters_and_params() -> None:
"""parsed/inserted счётчики; params корректны; raw содержит cells; CAST jsonb."""
db = _FakeSession(scalar_value=True) # xmax=0 → insert
systems = [
{"system_name": "СТ №54", "reserve_gcal_h": 3.26},
{"system_name": "зона Юг", "reserve_gcal_h": -1.2}, # дефицит: знак сохраняется
]
all_cells = [["СТ №54", "источник", 3.26], ["зона Юг", None, -1.2]]
counts = eh._upsert_heat_rows(db, "ЕТК", "2026-Q2", "guid-xyz", systems, all_cells)
assert counts == {"parsed": 2, "inserted": 2, "updated": 0, "skipped": 0}
assert len(db.calls) == 2
sql, first = db.calls[0]
assert first["org"] == "ЕТК"
assert first["system_name"] == "СТ №54"
assert first["reserve"] == 3.26
assert first["period"] == "2026-Q2"
assert first["source_guid"] == "guid-xyz"
# raw — JSON-строка с ключом cells (весь лист), CAST в jsonb в SQL.
assert '"cells"' in first["raw"]
assert "CAST(:raw AS jsonb)" in sql
assert "ON CONFLICT (org, sphere, system_name, period)" in sql
# negative reserve (дефицит) прокинут знаком.
assert db.calls[1][1]["reserve"] == -1.2
def test_upsert_heat_rows_update_counter() -> None:
"""scalar()=False (xmax≠0) → строка учтена как update, не insert."""
db = _FakeSession(scalar_value=False)
systems = [{"system_name": "СТ №1", "reserve_gcal_h": 1.0}]
counts = eh._upsert_heat_rows(
db, "Екатеринбургэнерго", "2026-Q2", "g1", systems, [["СТ №1", 1.0]]
)
assert counts == {"parsed": 1, "inserted": 0, "updated": 1, "skipped": 0}
def test_upsert_heat_rows_skip_on_error() -> None:
"""execute бросает → строка skipped, счётчик не растёт в inserted/updated."""
class _RaisingSession(_FakeSession):
def execute(self, stmt: Any, params: Any = None) -> _FakeResult:
raise RuntimeError("db boom")
db = _RaisingSession()
systems = [{"system_name": "СТ №9", "reserve_gcal_h": 2.0}]
counts = eh._upsert_heat_rows(db, "ЕТК", "2026-Q2", "g", systems, [["СТ №9", 2.0]])
assert counts == {"parsed": 1, "inserted": 0, "updated": 0, "skipped": 1}