Files
EOS/tests/test_configmigrate.py
T

624 lines
28 KiB
Python
Raw Normal View History

# Test data directory and known migration pairs
import copy
2025-10-28 02:50:31 +01:00
import json
import shutil
from pathlib import Path
from typing import Any
import pytest
from akkudoktoreos.config import configmigrate
from akkudoktoreos.config.config import ConfigEOS, SettingsEOSDefaults
from akkudoktoreos.config.configmigrate import migrate_config_data
2025-10-28 02:50:31 +01:00
from akkudoktoreos.core.version import __version__
DIR_TESTDATA = Path(__file__).absolute().parent.joinpath("testdata")
MIGRATION_PAIRS = [
(
DIR_TESTDATA / "eos_config_minimal_0_1_0.json",
DIR_TESTDATA / "eos_config_minimal_now.json",
),
(
DIR_TESTDATA / "eos_config_andreas_0_1_0.json",
DIR_TESTDATA / "eos_config_andreas_now.json",
),
(
DIR_TESTDATA / "eos_config_unstripped.json",
DIR_TESTDATA / "eos_config_stripped.json",
),
2025-10-28 02:50:31 +01:00
# Add more pairs here:
# (DIR_TESTDATA / "old_config_X.json", DIR_TESTDATA / "expected_config_X.json"),
]
# Any sentinel in expected data
_ANY_SENTINEL = "__ANY__"
2025-10-28 02:50:31 +01:00
def _dict_contains(superset: Any, subset: Any, path="") -> list[str]:
"""Recursively verify that all key-value pairs from a subset dictionary or list exist in a superset.
Supports nested dictionaries and lists. Extra keys in superset are allowed.
Numeric values (int/float) are compared with tolerance.
Args:
superset (Any): The dictionary or list that should contain all items from `subset`.
subset (Any): The expected dictionary or list.
path (str, optional): Current nested path used for error reporting. Defaults to "".
Returns:
list[str]: A list of strings describing mismatches or missing keys. Empty list if all subset items are present.
"""
errors = []
if isinstance(subset, dict) and isinstance(superset, dict):
for key, sub_value in subset.items():
full_path = f"{path}/{key}" if path else key
if key not in superset:
errors.append(f"Missing key: {full_path}")
continue
errors.extend(_dict_contains(superset[key], sub_value, full_path))
elif isinstance(subset, list) and isinstance(superset, list):
for i, elem in enumerate(subset):
if i >= len(superset):
full_path = f"{path}[{i}]" if path else f"[{i}]"
errors.append(f"List too short at {full_path}: expected element {elem}")
continue
errors.extend(_dict_contains(superset[i], elem, f"{path}[{i}]" if path else f"[{i}]"))
else:
# "__ANY__" in expected means "accept whatever value the migration produces"
if subset == _ANY_SENTINEL:
return errors
2025-10-28 02:50:31 +01:00
# Compare values (with numeric tolerance)
if isinstance(subset, (int, float)) and isinstance(superset, (int, float)):
if abs(float(subset) - float(superset)) > 1e-6:
errors.append(f"Value mismatch at {path}: expected {subset}, got {superset}")
elif subset != superset:
errors.append(f"Value mismatch at {path}: expected {subset}, got {superset}")
return errors
class TestConfigMigration:
"""Tests for migrate_config_file()"""
@pytest.fixture
def tmp_config_file(self, config_default_dirs) -> Path:
"""Create a temporary valid config file with an invalid version."""
config_default_dir_user, _, _, _ = config_default_dirs
config_file_user = config_default_dir_user.joinpath(ConfigEOS.CONFIG_FILE_NAME)
# Create a default config object (simulates the latest schema)
default_config = SettingsEOSDefaults()
# Dump to JSON
config_json = json.loads(default_config.model_dump_json())
# Corrupt the version (simulate outdated config)
config_json["general"]["version"] = "0.0.0-old"
# Write file
with config_file_user.open("w", encoding="utf-8") as f:
json.dump(config_json, f, indent=4)
return config_file_user
def test_migrate_config_file_from_invalid_version(self, tmp_config_file: Path):
"""Test that migration updates an outdated config version successfully."""
backup_file = tmp_config_file.with_suffix(".bak")
# Run migration
result = configmigrate.migrate_config_file(tmp_config_file, backup_file)
# Verify success
assert result is True, "Migration should succeed even from invalid version."
# Verify backup exists
assert backup_file.exists(), "Backup file should be created before migration."
# Verify version updated
with tmp_config_file.open("r", encoding="utf-8") as f:
migrated_data = json.load(f)
assert migrated_data["general"]["version"] == __version__, \
"Migrated config should have updated version."
# Verify it still matches the structure of SettingsEOSDefaults
new_model = SettingsEOSDefaults(**migrated_data)
assert isinstance(new_model, SettingsEOSDefaults)
def test_migrate_config_file_already_current(self, tmp_config_file: Path):
2025-10-28 02:50:31 +01:00
"""Test that a current config file returns True immediately."""
backup_file = tmp_config_file.with_suffix(".bak")
2025-10-28 02:50:31 +01:00
# Run migration
result = configmigrate.migrate_config_file(tmp_config_file, backup_file)
assert result is True, "Migration should succeed even from invalid version."
2025-10-28 02:50:31 +01:00
backup_file.unlink()
assert not backup_file.exists()
result = configmigrate.migrate_config_file(tmp_config_file, backup_file)
2025-10-28 02:50:31 +01:00
assert result is True
assert not backup_file.exists(), "No backup should be made if config is already current."
@pytest.mark.parametrize("old_file, expected_file", MIGRATION_PAIRS)
def test_migrate_old_version_config(self, tmp_path: Path, old_file: Path, expected_file: Path):
"""Ensure migration from old → new schema produces the expected output."""
# --- Prepare temporary working file based on expected file name ---
working_file = expected_file.with_suffix(".new")
shutil.copy(old_file, working_file)
# Backup file path (inside tmp_path to avoid touching repo files)
backup_file = tmp_path / f"{old_file.stem}.bak"
failed = False
try:
assert working_file.exists(), f"Working config file is missing: {working_file}"
# --- Perform migration ---
result = configmigrate.migrate_config_file(working_file, backup_file)
# --- Assertions ---
assert result is True, f"Migration failed for {old_file.name}"
assert configmigrate.mapped_count >= 1, f"No mapped migrations for {old_file.name}"
assert configmigrate.auto_count >= 1, f"No automatic migrations for {old_file.name}"
assert len(configmigrate.skipped_paths) <= 3, (
2025-10-28 02:50:31 +01:00
f"Too many skipped paths in {old_file.name}: {configmigrate.skipped_paths}"
)
assert backup_file.exists(), f"Backup file not created for {old_file.name}"
# --- Compare migrated result with expected output ---
old_data = json.loads(old_file.read_text(encoding="utf-8"))
2025-10-28 02:50:31 +01:00
new_data = json.loads(working_file.read_text(encoding="utf-8"))
expected_data = json.loads(expected_file.read_text(encoding="utf-8"))
# Check version
assert new_data["general"]["version"] == __version__, (
f"Expected version {__version__}, got {new_data['general']['version']}"
)
# Recursive subset comparison
errors = _dict_contains(new_data, expected_data)
assert not errors, (
f"Migrated config for {old_file.name} is missing or mismatched fields:\n" +
"\n".join(errors) + f"\n{new_data}"
2025-10-28 02:50:31 +01:00
)
# --- Compare migrated result with migration map ---
# Ensure all expected mapped fields are actually migrated and correct
missing_migrations = []
mismatched_values = []
for old_path, mapping in configmigrate.MIGRATION_MAP.items():
if mapping is None:
continue # skip intentionally dropped fields
# Determine new path (string or tuple)
new_path = mapping[0] if isinstance(mapping, tuple) else mapping
# Get value from expected data (if present)
expected_value = configmigrate._get_json_nested_value(expected_data, new_path)
if expected_value is None:
continue # new field not present in expected config
# Check that migration recorded this old path
if old_path.strip("/") not in configmigrate.migrated_source_paths:
missing_migrations.append(f"{old_path} → {new_path}")
continue
# Verify the migrated value matches the expected one
new_value = configmigrate._get_json_nested_value(new_data, new_path)
value_errors = _dict_contains(new_value, expected_value, new_path)
if value_errors:
# Check if this mapping uses _KEEP_DEFAULT and the old value was None/missing
old_value = configmigrate._get_json_nested_value(old_data, old_path)
keep_default = (
isinstance(mapping, tuple)
and configmigrate._KEEP_DEFAULT in mapping
)
if keep_default and old_value is None:
continue # acceptable: old was None, new model keeps its default
mismatched_values.extend(value_errors)
2025-10-28 02:50:31 +01:00
assert not missing_migrations, (
"Some expected migration map entries were not migrated:\n"
+ "\n".join(missing_migrations)
)
assert not mismatched_values, (
"Migrated values differ from expected results:\n"
+ "\n".join(mismatched_values)
)
# Validate migrated config with schema
new_model = SettingsEOSDefaults(**new_data)
assert isinstance(new_model, SettingsEOSDefaults)
except Exception:
# mark failure and re-raise so pytest records the error and the working_file is kept
failed = True
raise
finally:
# Remove the .new working file only if the test passed (failed == False)
if not failed and working_file.exists():
working_file.unlink(missing_ok=True)
class TestDeviceMigration:
@pytest.mark.parametrize("as_list", [False, True])
def test_migration_preserves_lcos_and_input(self, as_list):
battery = {"device_id": "house", "capacity_wh": 12000,
"levelized_cost_of_storage_kwh": 0.123}
raw = {"devices": {"batteries": [battery] if as_list else {"house": battery}}}
original = copy.deepcopy(raw)
migrated = migrate_config_data(raw)
assert migrated.devices.batteries is not None
assert migrated.devices.batteries["house"].levelized_cost_of_storage_amt_kwh == 0.123
assert raw == original
2026-11-09 19:11:15 +00:00
@pytest.mark.parametrize("devices_key", ["batteries", "electric_vehicles"])
@pytest.mark.parametrize("as_list", [False, True])
def test_migration_of_legacy_battery_fields(self, devices_key, as_list):
"""Legacy battery fields migrate to the profile based settings."""
battery = {
"device_id": "house",
"capacity_wh": 12000,
"charging_efficiency": 0.9,
"discharging_efficiency": 0.9,
"max_charge_power_w": 3000,
}
raw = {"devices": {devices_key: [battery] if as_list else {"house": battery}}}
migrated = migrate_config_data(raw)
devices = getattr(migrated.devices, devices_key)
assert devices is not None
settings = devices["house"]
assert settings.profile_adjustment.nominal_capacity_wh == 12000
# The configured charge power limit of the installation still applies.
assert settings.profile_adjustment.max_charge_power_w == 3000
assert settings.effective_max_charge_power_w <= 3000
dumped = settings.model_dump(exclude_computed_fields=True)
for legacy in ("capacity_wh", "charging_efficiency", "discharging_efficiency",
"max_charge_power_w"):
assert legacy not in dumped
def test_migration_writes_former_defaults_of_a_battery_before_profiles(self):
"""Former defaults were left out of files; the battery keeps them."""
battery = {"device_id": "house", "min_soc_percentage": 10}
migrated = migrate_config_data({"devices": {"batteries": {"house": battery}}})
assert migrated.devices.batteries is not None
adjustment = migrated.devices.batteries["house"].profile_adjustment
assert adjustment.nominal_capacity_wh == 8000
assert adjustment.max_charge_power_w == 5000
# The former charge power limit also limited discharging.
assert adjustment.max_discharge_power_w == 5000
assert migrated.devices.batteries["house"].min_soc_percentage == 10
def test_migration_keeps_the_charge_limit_as_discharge_limit(self):
"""Before battery profiles, max_charge_power_w also capped discharging."""
battery = {"device_id": "house", "capacity_wh": 8000, "max_charge_power_w": 3000}
migrated = migrate_config_data({"devices": {"batteries": {"house": battery}}})
assert migrated.devices.batteries is not None
settings = migrated.devices.batteries["house"]
assert settings.profile_adjustment.max_discharge_power_w == 3000
assert settings.effective_max_discharge_power_w is not None
assert settings.effective_max_discharge_power_w <= 3000
def test_migration_logs_dropped_efficiencies(self):
"""Efficiencies are no setting anymore; dropping them is logged as error."""
from loguru import logger
messages: list[str] = []
sink_id = logger.add(lambda message: messages.append(message.record["message"]), level="ERROR")
try:
battery = {"device_id": "house", "charging_efficiency": 0.9, "discharging_efficiency": 0.85}
migrate_config_data({"devices": {"batteries": {"house": battery}}})
finally:
logger.remove(sink_id)
assert len(messages) == 2
assert all("Battery 'house'" in message for message in messages)
assert any("'charging_efficiency' 0.9" in message for message in messages)
assert any("'discharging_efficiency' 0.85" in message for message in messages)
@pytest.mark.parametrize(
"legacy_name, adjustment_name",
[
("nominal_capacity_wh", "nominal_capacity_wh"),
("max_charge_power_limit_w", "max_charge_power_w"),
("max_discharge_power_limit_w", "max_discharge_power_w"),
],
)
def test_migration_of_interim_profile_fields(self, legacy_name, adjustment_name):
"""Interim fields of the settings with profiles move into the adjustment.
Such a battery is not from before battery profiles, so it gets no former
defaults.
"""
battery = {"device_id": "house", legacy_name: 4000}
migrated = migrate_config_data({"devices": {"batteries": {"house": battery}}})
assert migrated.devices.batteries is not None
adjustment = migrated.devices.batteries["house"].profile_adjustment
assert getattr(adjustment, adjustment_name) == 4000
for name, value in adjustment.model_dump().items():
if name != adjustment_name:
assert value is None
def test_migration_of_full_charge_settings(self):
"""Full charge interval and threshold move into the adjustment, chemistry is dropped."""
battery = {
"device_id": "house",
"cell_chemistry": "LFP",
"full_charge_interval_days": 21,
"full_charge_detect_percentage": 98,
"full_charge_soc_percentage": 100,
"full_charge_allow_grid": True,
}
migrated = migrate_config_data({"devices": {"batteries": {"house": battery}}})
assert migrated.devices.batteries is not None
settings = migrated.devices.batteries["house"]
assert settings.profile_adjustment.full_charge_interval_days == 21
assert settings.profile_adjustment.full_charge_detect_percentage == 98
assert settings.full_charge_soc_percentage == 100
assert settings.full_charge_allow_grid is True
dumped = settings.model_dump(exclude_computed_fields=True)
for legacy in ("cell_chemistry", "full_charge_interval_days",
"full_charge_detect_percentage"):
assert legacy not in dumped
def test_migration_drops_home_assistant_config_path_of_cell_chemistry(self):
raw = {
"adapter": {
"homeassistant": {
"config_entity_ids": {
"devices/batteries/house/cell_chemistry": "sensor.chemistry",
"devices/batteries/house/full_charge_interval_days": "sensor.interval",
}
}
}
}
migrated = migrate_config_data(raw)
assert migrated.adapter.homeassistant.config_entity_ids == {
"devices/batteries/house/profile_adjustment/full_charge_interval_days": (
"sensor.interval"
),
}
def test_migration_maps_a_charge_limit_entity_to_both_limits(self):
"""Before battery profiles, one max_charge_power_w entity also capped discharging."""
raw = {
"adapter": {
"homeassistant": {
"config_entity_ids": {
"devices/batteries/house/max_charge_power_w": "sensor.limit",
"devices/batteries/garage/max_charge_power_w": "sensor.charge",
"devices/batteries/garage/profile_adjustment/max_discharge_power_w": (
"sensor.discharge"
),
}
}
}
}
migrated = migrate_config_data(raw)
assert migrated.adapter.homeassistant.config_entity_ids == {
"devices/batteries/house/profile_adjustment/max_charge_power_w": "sensor.limit",
"devices/batteries/house/profile_adjustment/max_discharge_power_w": "sensor.limit",
"devices/batteries/garage/profile_adjustment/max_charge_power_w": "sensor.charge",
"devices/batteries/garage/profile_adjustment/max_discharge_power_w": (
"sensor.discharge"
),
}
2026-11-10 17:56:13 +00:00
def test_migration_logs_dropped_inverter_efficiencies_as_error(self):
"""Like battery efficiencies, dropped inverter efficiencies are logged as error."""
from loguru import logger
messages: list[str] = []
sink_id = logger.add(lambda message: messages.append(message.record["message"]), level="ERROR")
try:
inverter = {
"device_id": "inv1",
"max_power_w": 8000,
"dc_to_ac_efficiency_curve": [[0.1, 0.9], [1.0, 0.97]],
}
migrate_config_data({"devices": {"inverters": {"inv1": inverter}}})
finally:
logger.remove(sink_id)
assert len(messages) == 1
assert "Inverter 'inv1'" in messages[0]
assert "dc_to_ac_efficiency_curve" in messages[0]
def test_migration_of_legacy_inverter_fields(self):
"""Inverter power limits move into the adjustment, efficiencies are dropped."""
inverter = {
"device_id": "inv1",
"max_power_w": 8000,
"max_ac_charge_power_w": 3000,
"ac_to_dc_efficiency": 0.95,
"dc_to_ac_efficiency": 0.96,
"dc_to_ac_efficiency_curve": [[0.1, 0.9], [1.0, 0.96]],
"ac_charge_limits_total_charge": True,
"battery_id": "bat1",
}
migrated = migrate_config_data({"devices": {"inverters": {"inv1": inverter}}})
assert migrated.devices.inverters is not None
settings = migrated.devices.inverters["inv1"]
assert settings.profile_adjustment.nominal_power_w == 8000
assert settings.profile_adjustment.max_ac_charge_power_w == 3000
assert settings.profile_adjustment.max_power_w is None
assert settings.battery_id == "bat1"
dumped = settings.model_dump(exclude_computed_fields=True)
for legacy in ("max_power_w", "max_ac_charge_power_w", "ac_to_dc_efficiency",
"dc_to_ac_efficiency", "dc_to_ac_efficiency_curve",
"ac_charge_limits_total_charge"):
assert legacy not in dumped
def test_migration_keeps_ac_charging_disabled(self):
"""A legacy AC→DC efficiency of 0 disabled AC charging."""
inverter = {"device_id": "inv1", "max_power_w": 8000, "ac_to_dc_efficiency": 0.0}
migrated = migrate_config_data({"devices": {"inverters": [inverter]}})
assert migrated.devices.inverters is not None
assert migrated.devices.inverters["inv1"].profile_adjustment.max_ac_charge_power_w == 0
def test_migration_keeps_present_inverter_adjustment(self):
inverter = {
"device_id": "inv1",
"max_power_w": 8000,
"profile_adjustment": {"nominal_power_w": 6000},
}
migrated = migrate_config_data({"devices": {"inverters": {"inv1": inverter}}})
assert migrated.devices.inverters is not None
assert migrated.devices.inverters["inv1"].profile_adjustment.nominal_power_w == 6000
def test_migration_of_inverter_home_assistant_config_paths(self):
raw = {
"adapter": {
"homeassistant": {
"config_entity_ids": {
"devices/inverters/inv1/max_power_w": "sensor.inverter_power",
"devices/inverters/inv1/dc_to_ac_efficiency": "sensor.efficiency",
"devices/inverters/inv1/battery_id": "sensor.battery",
}
}
}
}
migrated = migrate_config_data(raw)
assert migrated.adapter.homeassistant.config_entity_ids == {
"devices/inverters/inv1/profile_adjustment/nominal_power_w": "sensor.inverter_power",
"devices/inverters/inv1/battery_id": "sensor.battery",
}
2026-11-09 19:11:15 +00:00
def test_migration_keeps_a_migrated_battery(self):
"""A battery in the current layout is not changed, e.g. on a version bump."""
battery = {
"device_id": "house",
"battery_profile": "my-battery",
"profile_adjustment": {"max_discharge_power_w": 3000},
}
migrated = migrate_config_data({"devices": {"batteries": {"house": battery}}})
assert migrated.devices.batteries is not None
settings = migrated.devices.batteries["house"]
assert settings.battery_profile == "my-battery"
assert settings.profile_adjustment.max_discharge_power_w == 3000
assert settings.profile_adjustment.nominal_capacity_wh is None
assert settings.profile_adjustment.max_charge_power_w is None
@pytest.mark.parametrize(
"legacy_name, adjustment_name",
[
("capacity_wh", "nominal_capacity_wh"),
("max_charge_power_w", "max_charge_power_w"),
("nominal_capacity_wh", "nominal_capacity_wh"),
("max_charge_power_limit_w", "max_charge_power_w"),
],
)
def test_migration_keeps_present_adjustment(self, legacy_name, adjustment_name):
"""A value already in the profile adjustment wins over its legacy field."""
battery = {
"device_id": "house",
legacy_name: 12000,
"profile_adjustment": {adjustment_name: 10000},
}
migrated = migrate_config_data({"devices": {"batteries": {"house": battery}}})
assert migrated.devices.batteries is not None
adjustment = migrated.devices.batteries["house"].profile_adjustment
assert getattr(adjustment, adjustment_name) == 10000
def test_migration_of_home_assistant_config_paths(self):
"""Config paths of moved battery fields follow them into the adjustment."""
raw = {
"adapter": {
"homeassistant": {
"config_entity_ids": {
"devices/batteries/house/capacity_wh": "sensor.capacity",
"devices/electric_vehicles/car/max_charge_power_w": "sensor.ev_limit",
"devices/batteries/house/charging_efficiency": "sensor.efficiency",
"devices/batteries/house/min_soc_percentage": "sensor.min_soc",
"general/latitude": "sensor.latitude",
}
}
}
}
migrated = migrate_config_data(raw)
assert migrated.adapter.homeassistant.config_entity_ids == {
"devices/batteries/house/profile_adjustment/nominal_capacity_wh": "sensor.capacity",
"devices/electric_vehicles/car/profile_adjustment/max_charge_power_w": (
"sensor.ev_limit"
),
# The former charge limit also capped discharging.
"devices/electric_vehicles/car/profile_adjustment/max_discharge_power_w": (
"sensor.ev_limit"
),
# charging_efficiency is computed now and cannot be set.
"devices/batteries/house/min_soc_percentage": "sensor.min_soc",
"general/latitude": "sensor.latitude",
}
def test_migrate_config_file_keeps_charge_power_limit(self, tmp_path: Path):
"""The legacy charge power limit survives the file migration."""
config_file = tmp_path / "EOS.config.json"
battery = {"device_id": "house", "capacity_wh": 12000, "max_charge_power_w": 3000}
config_file.write_text(
json.dumps(
{
"general": {"version": "0.0.0-old"},
"devices": {"batteries": {"house": battery}},
}
),
encoding="utf-8",
)
assert configmigrate.migrate_config_file(config_file, tmp_path / "EOS.config.bak")
migrated = json.loads(config_file.read_text(encoding="utf-8"))
house = migrated["devices"]["batteries"]["house"]
assert house["profile_adjustment"] == {
"nominal_capacity_wh": 12000,
"max_charge_power_w": 3000,
"max_discharge_power_w": 3000,
}
assert "max_charge_power_w" not in house
assert "capacity_wh" not in house
class TestSlashKeyedDictMigration:
"""Dict entries whose keys contain '/' must survive migration unchanged."""
ENTITY_IDS = {
"devices/electric_vehicles/ev11/max_soc_percentage": "sensor.ev_max_soc",
"devices/electric_vehicles/ev11/min_soc_deadline_datetime": "sensor.ev_departure",
"devices/electric_vehicles/ev11/min_soc_percentage": "sensor.ev_target_soc",
}
def test_migrate_config_data_keeps_slash_keys(self):
raw = {"adapter": {"homeassistant": {"config_entity_ids": dict(self.ENTITY_IDS)}}}
migrated = migrate_config_data(raw)
assert migrated.adapter.homeassistant.config_entity_ids == self.ENTITY_IDS
assert not any(p.startswith("adapter/homeassistant") for p in configmigrate.skipped_paths)
def test_migrate_config_file_keeps_slash_keys(self, config_default_dirs):
config_default_dir_user, _, _, _ = config_default_dirs
config_file = config_default_dir_user.joinpath(ConfigEOS.CONFIG_FILE_NAME)
config_file.write_text(
json.dumps(
{
"general": {"version": "0.0.0-old"},
"adapter": {"homeassistant": {"config_entity_ids": self.ENTITY_IDS}},
}
),
encoding="utf-8",
)
assert configmigrate.migrate_config_file(config_file, config_file.with_suffix(".bak"))
migrated = json.loads(config_file.read_text(encoding="utf-8"))
assert migrated["general"]["version"] == __version__
assert migrated["adapter"]["homeassistant"]["config_entity_ids"] == self.ENTITY_IDS