# Test data directory and known migration pairs import copy 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 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", ), # Add more pairs here: # (DIR_TESTDATA / "old_config_X.json", DIR_TESTDATA / "expected_config_X.json"), ] # Any sentinel in expected data _ANY_SENTINEL = "__ANY__" 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 # 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): """Test that a current config file returns True immediately.""" backup_file = tmp_config_file.with_suffix(".bak") # Run migration result = configmigrate.migrate_config_file(tmp_config_file, backup_file) assert result is True, "Migration should succeed even from invalid version." backup_file.unlink() assert not backup_file.exists() result = configmigrate.migrate_config_file(tmp_config_file, backup_file) 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, ( 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")) 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}" ) # --- 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) 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 @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" ), } 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