Files
ems/backend/tests/test_control_export_plan_guard.py
Dusan Vojacek a7dff75e58
Some checks failed
CI and deploy / migration-check (push) Failing after 39s
CI and deploy / deploy (push) Has been skipped
Add export plan guard to block Deye export against plan.
Force PASSIVE/no-export when sell is negative or export_mode is NONE,
and alert NEG_SELL_EXPORT in plan_actual_slot_guard when export still occurs.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-29 00:14:52 +02:00

107 lines
3.1 KiB
Python

"""Exekuční pojistka exportu podle plánu (Plan 3)."""
from __future__ import annotations
import unittest
from services.control.exporter_monolith import (
ControlSetpoints,
_apply_export_plan_guard,
get_deye_mode,
)
from services.control.models import OperatingModeInfo
from services.control.setpoints import _DictRecord
def _auto_mode() -> OperatingModeInfo:
return OperatingModeInfo(
mode_code="AUTO",
battery_mode="AUTO",
grid_mode="AUTO",
ev_enabled=True,
heat_pump_enabled_def=True,
loxone_mode_value=0,
)
def _sp(**kwargs: object) -> ControlSetpoints:
base = dict(
battery_w=0,
grid_export_limit=8000,
ev1_current_a=0,
ev2_current_a=0,
heat_pump_enable=False,
grid_setpoint_w=-8000,
ev1_power_w=0,
ev2_power_w=0,
target_soc_pct=50,
deye_physical_mode="SELL",
export_mode="BATTERY_SELL",
export_ban=False,
)
base.update(kwargs)
return ControlSetpoints(**base) # type: ignore[arg-type]
class ExportPlanGuardTests(unittest.TestCase):
def test_neg_sell_forces_passive_no_export(self) -> None:
sp = _sp()
pi = _DictRecord(
{
"grid_setpoint_w": -8000,
"effective_sell_price": -0.5,
"export_mode": "NONE",
}
)
out = _apply_export_plan_guard(1, _auto_mode(), pi, sp)
self.assertEqual(get_deye_mode(out), "PASSIVE")
self.assertTrue(out.export_ban)
self.assertEqual(out.grid_export_limit, 0)
self.assertGreaterEqual(out.grid_setpoint_w, 0)
self.assertEqual(out.export_mode, "NONE")
def test_export_mode_none_with_non_negative_grid(self) -> None:
sp = _sp(grid_setpoint_w=0, battery_w=-5000, export_mode="BATTERY_SELL")
pi = _DictRecord(
{
"grid_setpoint_w": 0,
"effective_sell_price": 2.5,
"export_mode": "NONE",
}
)
out = _apply_export_plan_guard(1, _auto_mode(), pi, sp)
self.assertEqual(get_deye_mode(out), "PASSIVE")
self.assertEqual(out.battery_w, 0)
self.assertTrue(out.export_ban)
def test_profitable_export_unchanged(self) -> None:
sp = _sp()
pi = _DictRecord(
{
"grid_setpoint_w": -8000,
"effective_sell_price": 9.5,
"export_mode": "BATTERY_SELL",
}
)
out = _apply_export_plan_guard(1, _auto_mode(), pi, sp)
self.assertIs(out, sp)
self.assertEqual(get_deye_mode(out), "SELL")
def test_non_auto_mode_skipped(self) -> None:
sp = _sp()
pi = _DictRecord({"effective_sell_price": -1.0, "export_mode": "NONE"})
mode = OperatingModeInfo(
mode_code="SELF_SUSTAIN",
battery_mode="PASSIVE",
grid_mode="PASSIVE",
ev_enabled=False,
heat_pump_enabled_def=False,
loxone_mode_value=1,
)
out = _apply_export_plan_guard(1, mode, pi, sp)
self.assertIs(out, sp)
if __name__ == "__main__":
unittest.main()