"""Offline regression tests for per-run optimization results and atomic publication.""" from asyncio import Lock from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, Mock import pytest from akkudoktoreos.core import ems as ems_module from akkudoktoreos.core.emsettings import EnergyManagementMode from akkudoktoreos.optimization.genetic0.genetic0params import ( Genetic0OptimizationParameters, ) from akkudoktoreos.optimization.optimization import OptimizationAlgorithm from akkudoktoreos.utils.datetimeutil import to_datetime @pytest.fixture def offline_ems(monkeypatch): """Run the actual EMS orchestration with no real adapters or prediction IO.""" cls = ems_module.EnergyManagement for name in ( "_start_datetime", "_observation_datetime", "_last_run_datetime", "_plan", "_optimization_solution", "_genetic_solution", "_genetic0_solution", ): monkeypatch.setattr(cls, name, None) monkeypatch.setattr(cls, "_stage", ems_module.EnergyManagementStage.IDLE) monkeypatch.setattr(cls, "_run_lock", Lock()) monkeypatch.setattr(ems_module, "CacheEnergyManagementStore", Mock()) return SimpleNamespace( config=SimpleNamespace( general=SimpleNamespace(timezone="Europe/Berlin"), ems=SimpleNamespace(mode=EnergyManagementMode.OPTIMIZATION), optimization=SimpleNamespace( algorithm=OptimizationAlgorithm.GENETIC, genetic=SimpleNamespace(generations=3, seed=17, interval_sec=3600, individuals=31), genetic0=SimpleNamespace(generations=5, seed=29), ), server=SimpleNamespace(verbose=False), ), prediction=SimpleNamespace(update_data=AsyncMock()), adapter=SimpleNamespace(update_data=AsyncMock()), set_start_datetime=cls.set_start_datetime, ) @pytest.mark.asyncio @pytest.mark.parametrize("algorithm", list(OptimizationAlgorithm)) @pytest.mark.parametrize("explicit_start", [False, True]) async def test_implicit_genetic_start_uses_site_timezone( offline_ems, monkeypatch, set_other_timezone, algorithm, explicit_start ): set_other_timezone("UTC") now = to_datetime("2026-09-16T22:47:23Z", in_timezone="UTC") def frozen_datetime(value=None, **kwargs): return to_datetime(now if value is None else value, **kwargs) monkeypatch.setattr(ems_module, "to_datetime", frozen_datetime) offline_ems.config.optimization.genetic.interval_sec = 900 await ems_module.EnergyManagement.run( offline_ems, mode=EnergyManagementMode.PREDICTION, algorithm=algorithm, start_datetime=now if explicit_start else None, ) start = ems_module.EnergyManagement._start_datetime observed = ems_module.EnergyManagement._observation_datetime assert start is not None and observed is not None assert observed.timestamp() == now.timestamp() if algorithm == OptimizationAlgorithm.GENETIC: assert start.minute == 45 assert start.hour == (22 if explicit_start else 0) assert start.day == (16 if explicit_start else 17) assert start.timezone_name == ("UTC" if explicit_start else "Europe/Berlin") else: assert start.minute == 0 assert start.hour == 22 assert start.day == 16 assert start.timezone_name == "UTC" @pytest.mark.asyncio @pytest.mark.parametrize("algorithm", list(OptimizationAlgorithm)) @pytest.mark.parametrize("selection", ["configured", "explicit"]) @pytest.mark.parametrize("supplied", [False, True]) @pytest.mark.parametrize("timezone, expected_hour", [("UTC", 8), ("Europe/Berlin", 10)]) async def test_optimization_routes_only_selected_algorithm( monkeypatch, offline_ems, algorithm, selection, supplied, timezone, expected_hour ): """Configuration selection and explicit overrides retain isolated async paths.""" selected_name = "Genetic" if algorithm == OptimizationAlgorithm.GENETIC else "Genetic0" suffix = "genetic" if algorithm == OptimizationAlgorithm.GENETIC else "genetic0" other_suffix = "genetic0" if suffix == "genetic" else "genetic" sentinel_parameters = object() sentinel_result, sentinel_plan = object(), object() solution = SimpleNamespace( optimization_solution=AsyncMock(return_value=sentinel_result), energy_management_plan=Mock(return_value=sentinel_plan), ) constructors = {} preparers = {} for prefix in ("Genetic", "Genetic0"): constructor = Mock() constructor.return_value.optimize_ems.return_value = solution constructors[prefix] = constructor monkeypatch.setattr(ems_module, prefix + "Optimization", constructor) prepare = AsyncMock(return_value=sentinel_parameters) preparers[prefix] = prepare monkeypatch.setattr( getattr(ems_module, prefix + "OptimizationParameters"), "prepare", prepare ) kwargs: dict[str, Any] = { "start_datetime": to_datetime("2026-09-16T10:00:00+02:00", in_timezone=timezone) } if selection == "configured": offline_ems.config.optimization.algorithm = algorithm else: offline_ems.config.optimization.algorithm = OptimizationAlgorithm(other_suffix.upper()) kwargs["algorithm"] = algorithm if supplied: kwargs[suffix + "_parameters"] = sentinel_parameters kwargs[suffix + "_generations"] = 7 kwargs[suffix + "_seed"] = 43 if algorithm == OptimizationAlgorithm.GENETIC: kwargs["genetic_individuals"] = 11 run_result = await ems_module.EnergyManagement.run(offline_ems, **kwargs) assert run_result is solution selected = constructors[selected_name] expected_config = getattr(offline_ems.config.optimization, suffix) selected.assert_called_once_with( verbose=False, fixed_seed=43 if supplied else expected_config.seed ) expected_arguments: dict[str, Any] = { "start_hour": expected_hour, "parameters": sentinel_parameters, "ngen": 7 if supplied else expected_config.generations, } if algorithm == OptimizationAlgorithm.GENETIC: expected_arguments["individuals"] = 11 if supplied else None selected.return_value.optimize_ems.assert_called_once_with(**expected_arguments) other_name = "Genetic0" if selected_name == "Genetic" else "Genetic" constructors[other_name].assert_not_called() preparers[other_name].assert_not_awaited() if supplied: preparers[selected_name].assert_not_awaited() else: preparers[selected_name].assert_awaited_once_with() solution.optimization_solution.assert_awaited_once_with() solution.energy_management_plan.assert_called_once_with() cls = ems_module.EnergyManagement assert getattr(cls, "_" + suffix + "_solution") is solution assert getattr(cls, "_" + other_suffix + "_solution") is None assert cls.optimization_solution() is sentinel_result assert cls.plan() is sentinel_plan assert cls.stage() == ems_module.EnergyManagementStage.IDLE offline_ems.prediction.update_data.assert_awaited_once_with( force_enable=False, force_update=False ) assert offline_ems.adapter.update_data.await_count == 2 @pytest.mark.asyncio @pytest.mark.parametrize("mode", [EnergyManagementMode.DISABLED, EnergyManagementMode.PREDICTION]) async def test_non_optimization_modes_never_optimize(monkeypatch, offline_ems, mode): constructors = [Mock(), Mock()] monkeypatch.setattr(ems_module, "GeneticOptimization", constructors[0]) monkeypatch.setattr(ems_module, "Genetic0Optimization", constructors[1]) offline_ems.config.ems.mode = mode await ems_module.EnergyManagement.run(offline_ems) for constructor in constructors: constructor.assert_not_called() assert offline_ems.prediction.update_data.await_count == ( mode == EnergyManagementMode.PREDICTION ) assert offline_ems.adapter.update_data.await_count == (mode == EnergyManagementMode.PREDICTION) assert ems_module.EnergyManagement.plan() is None @pytest.mark.asyncio @pytest.mark.parametrize("prefix", ["Genetic", "Genetic0"]) async def test_missing_preparation_does_not_dispatch_controls(monkeypatch, offline_ems, prefix): constructor = Mock() monkeypatch.setattr(ems_module, prefix + "Optimization", constructor) prepare = AsyncMock(return_value=None) monkeypatch.setattr(getattr(ems_module, prefix + "OptimizationParameters"), "prepare", prepare) await ems_module.EnergyManagement.run( offline_ems, algorithm=OptimizationAlgorithm(prefix.upper()) ) prepare.assert_awaited_once_with() constructor.assert_not_called() assert offline_ems.adapter.update_data.await_count == 1 # acquisition only assert ems_module.EnergyManagement.plan() is None assert ems_module.EnergyManagement.stage() == ems_module.EnergyManagementStage.IDLE @pytest.mark.asyncio @pytest.mark.parametrize("start_hour", [None, 11]) async def test_legacy_optimize_explicitly_uses_genetic0(monkeypatch, start_hour): """Even with GENETIC configured, /optimize must never become the new optimizer.""" from akkudoktoreos.server import eos from akkudoktoreos.server.rest.error import EOSProblem fake = SimpleNamespace( run=AsyncMock(return_value=None), genetic0_solution=Mock(return_value=None) ) monkeypatch.setattr(eos, "get_ems", lambda: fake) parameters = Genetic0OptimizationParameters.model_validate( { "ems": { "pv_prognose_wh": [0.0, 100.0], "gesamtlast": [100.0, 100.0], "strompreis_euro_pro_wh": [0.0003, 0.0003], "einspeiseverguetung_euro_pro_wh": 0.00008, "preis_euro_pro_wh_akku": 0.0, }, "pv_akku": None, "eauto": None, "inverter": None, } ) with pytest.raises(EOSProblem): await eos.fastapi_optimize(parameters=parameters, start_hour=start_hour, ngen=2) kwargs = fake.run.await_args.kwargs assert kwargs["mode"] == EnergyManagementMode.OPTIMIZATION assert kwargs["algorithm"] == OptimizationAlgorithm.GENETIC0 assert kwargs["genetic0_parameters"] is parameters assert kwargs["genetic0_generations"] == 2 assert "genetic_parameters" not in kwargs assert ( kwargs["start_datetime"] is None if start_hour is None else kwargs["start_datetime"].hour == 11 ) @pytest.mark.asyncio @pytest.mark.parametrize("phase", ["optimizer", "conversion", "plan"]) async def test_failed_legacy_http_run_never_reports_previous_solution( monkeypatch, offline_ems, phase ): import json from pathlib import Path from types import MethodType from httpx import ASGITransport, AsyncClient from akkudoktoreos.optimization.genetic0.genetic0solution import Genetic0Solution from akkudoktoreos.server import eos cls = ems_module.EnergyManagement data = json.loads( (Path(__file__).parent / "testdata/genetic0/optimize_result_1.json").read_text() ) previous = Genetic0Solution.model_validate(data) monkeypatch.setattr(cls, "_genetic0_solution", previous) constructor = Mock() error = RuntimeError("synthetic " + phase + " failure") native = SimpleNamespace( optimization_solution=AsyncMock(return_value=object()), energy_management_plan=Mock(return_value=object()), ) constructor.return_value.optimize_ems.return_value = native if phase == "optimizer": constructor.return_value.optimize_ems.side_effect = error elif phase == "conversion": native.optimization_solution.side_effect = error else: native.energy_management_plan.side_effect = error monkeypatch.setattr(ems_module, "Genetic0Optimization", constructor) offline_ems.run = MethodType(cls.run, offline_ems) offline_ems.genetic0_solution = cls.genetic0_solution monkeypatch.setattr(eos, "get_ems", lambda: offline_ems) async with AsyncClient( transport=ASGITransport(app=eos.app, raise_app_exceptions=False), base_url="http://test" ) as client: response = await client.post("/optimize?ngen=1", json=data["parameters"]) constructor.return_value.optimize_ems.assert_called_once() # This diagnostic proves a failure is the stale-success bug, not invalid input. if response.status_code == 200: assert response.json()["start_solution"] == previous.start_solution assert response.json()["result"]["total_balance"] == previous.result.total_balance assert response.status_code >= 400, ( "The failing optimizer returned HTTP 200 with the previous solution" ) @pytest.mark.asyncio @pytest.mark.parametrize("prefix", ["Genetic", "Genetic0"]) @pytest.mark.parametrize("phase", ["conversion", "plan"]) async def test_conversion_failure_preserves_consistent_previous_results( monkeypatch, offline_ems, prefix, phase ): cls = ems_module.EnergyManagement previous_specific, previous_generic, previous_plan = object(), object(), object() suffix = prefix.lower() monkeypatch.setattr(cls, "_" + suffix + "_solution", previous_specific) monkeypatch.setattr(cls, "_optimization_solution", previous_generic) monkeypatch.setattr(cls, "_plan", previous_plan) conversion = AsyncMock(return_value=object()) solution = SimpleNamespace(optimization_solution=conversion, energy_management_plan=Mock()) if phase == "conversion": conversion.side_effect = RuntimeError("synthetic conversion failure") else: solution.energy_management_plan.side_effect = RuntimeError("synthetic plan failure") constructor = Mock() constructor.return_value.optimize_ems.return_value = solution monkeypatch.setattr(ems_module, prefix + "Optimization", constructor) supplied_parameters: dict[str, Any] = {suffix + "_parameters": object()} result = await cls.run( offline_ems, algorithm=OptimizationAlgorithm(prefix.upper()), **supplied_parameters, ) assert result is None conversion.assert_awaited_once_with() assert solution.energy_management_plan.call_count == (phase == "plan") assert offline_ems.adapter.update_data.await_count == 1 assert ( getattr(cls, "_" + suffix + "_solution"), cls.optimization_solution(), cls.plan(), cls.stage(), ) == ( previous_specific, previous_generic, previous_plan, ems_module.EnergyManagementStage.IDLE, ), ( "Failed conversion published a new algorithm result beside the old plan and left EMS in OPTIMIZATION" ) @pytest.mark.asyncio async def test_legacy_endpoint_uses_run_return_value_not_last_cached_solution(monkeypatch): import json from pathlib import Path from akkudoktoreos.optimization.genetic0.genetic0solution import Genetic0Solution from akkudoktoreos.server import eos data = json.loads( (Path(__file__).parent / "testdata/genetic0/optimize_result_1.json").read_text() ) produced = Genetic0Solution.model_validate(data) fake = SimpleNamespace( run=AsyncMock(return_value=produced), genetic0_solution=Mock(side_effect=AssertionError("must use this run's result")), ) monkeypatch.setattr(eos, "get_ems", lambda: fake) result = await eos.fastapi_optimize(parameters=produced.parameters, ngen=1) assert result.start_solution == produced.start_solution assert result.result.total_balance == produced.result.total_balance fake.genetic0_solution.assert_not_called()