From eff8917a398064a2aa0311b64590f89c13d80783 Mon Sep 17 00:00:00 2001 From: Alexander Hartung Date: Mon, 9 Nov 2026 12:41:50 +0100 Subject: [PATCH 1/3] Update Filesystem_Migration --- .../helpermodules/filesystem_migrations.py | 120 +++++++++++ .../filesystem_migrations_test.py | 202 ++++++++++++++++++ .../generate_totals_subprocess.py | 23 +- .../generate_totals_subprocess_test.py | 110 +++++++++- packages/helpermodules/setdata.py | 3 +- packages/helpermodules/update_config.py | 81 ++----- packages/helpermodules/update_config_test.py | 35 +-- 7 files changed, 470 insertions(+), 104 deletions(-) create mode 100644 packages/helpermodules/filesystem_migrations.py create mode 100644 packages/helpermodules/filesystem_migrations_test.py diff --git a/packages/helpermodules/filesystem_migrations.py b/packages/helpermodules/filesystem_migrations.py new file mode 100644 index 0000000000..8294e3f3ee --- /dev/null +++ b/packages/helpermodules/filesystem_migrations.py @@ -0,0 +1,120 @@ +# UpdateConfig startet dieses Modul als eigenen Hintergrundprozess. Eine globale +# Dateisperre verhindert parallele Worker. Nach dem Einlesen des MQTT-Zustands +# laufen fehlende upgrade_file_operation_() synchron in Versionsreihenfolge. +# Erst nach normaler Rückkehr wird die Version gespeichert; bei einer weitergegebenen +# Exception endet der Lauf. Beim nächsten Worker-Start werden abgeschlossene Versionen +# übersprungen und die fehlgeschlagene Migration erneut versucht (kein Rollback). +# Migrationen müssen deshalb nach Teilabbrüchen wiederholbar sein. Neue Migrationen +# hier ergänzen und FILE_OPERATION_VERSION anheben. +# log_data_ready betrifft nur Migration 0, nicht den Abschluss aller Migrationen. + +import copy +import fcntl +import json +import logging +from pathlib import Path + +from paho.mqtt.client import Client as MqttClient, MQTTMessage + +from helpermodules import logger +from helpermodules.broker import BrokerClient +from helpermodules.measurement_logging.generate_totals_subprocess import generate_totals +from helpermodules.messaging import MessageType, pub_system_message +from helpermodules.pub import Pub +from helpermodules.utils.topic_parser import decode_payload + + +log = logging.getLogger(__name__) +LOCK_FILE = Path(__file__).resolve().parents[2] / "data" / "filesystem_migrations.lock" + + +class FilesystemMigrations: + FILE_OPERATION_VERSION = 0 + + def __init__(self) -> None: + self.all_received_topics = {} + + def on_connect(self, client: MqttClient, userdata, flags: dict, rc: int): + client.subscribe("openWB/#", 2) + + def on_message(self, client: MqttClient, userdata, msg: MQTTMessage): + self.all_received_topics.update({msg.topic: msg.payload}) + + def _update_topic(self, topic: str, payload) -> None: + publish_info = Pub().publisher.client.publish(topic, json.dumps(payload), qos=1, retain=True) + publish_info.wait_for_publish(timeout=10) + if not publish_info.is_published(): + raise TimeoutError(f"Topic '{topic}' konnte nicht gespeichert werden.") + self.all_received_topics[topic] = copy.deepcopy(payload) + + def update_filesystem(self) -> None: + """Liest den aktuellen Broker-Zustand im Migrationsprozess ein.""" + broker = BrokerClient("filesystem-migrations", self.on_connect, self.on_message) + try: + broker.start_finite_loop() + finally: + broker.disconnect() + self.__solve_breaking_changes_filesystem() + + def __solve_breaking_changes_filesystem(self) -> None: + """Führt fehlende Migrationen synchron aus und markiert nur erfolgreiche Abschlüsse.""" + file_operation_version = decode_payload(self.all_received_topics.get("openWB/system/file_operation_version")) + if file_operation_version is None: + # Neues Topic (z.B. bei Upgrade von älteren Versionen): alle File-Operation-Upgrades ausführen. + file_operation_version = [] + self._update_topic("openWB/system/file_operation_version", file_operation_version) + + log.debug(f"current file operation version: {file_operation_version}") + log.debug(f"target file operation version: {self.FILE_OPERATION_VERSION}") + for version in list(range(self.FILE_OPERATION_VERSION+1)): + operation_required = version not in file_operation_version + if version == 0: + # Alte Installationen haben Version 0 schon beim Prozessstart eingetragen. + log_totals_generation_finished = decode_payload( + self.all_received_topics.get("openWB/system/log_data_ready")) + operation_required = operation_required or log_totals_generation_finished is not True + + if operation_required: + log.info("Dateisystem-Migration %s wird ausgeführt.", version) + getattr(self, f"upgrade_file_operation_{version}")() + self._append_file_operation_version(version) + + def upgrade_file_operation_0(self) -> None: + """Generiere die Totals-Summen für Tage und Monate.""" + + self._update_topic("openWB/system/log_data_ready", False) + generate_totals() + self.all_received_topics["openWB/system/log_data_ready"] = True + + def _append_file_operation_version(self, version: int) -> None: + file_operation_versions = list(decode_payload( + self.all_received_topics.get("openWB/system/file_operation_version"))) + if version not in file_operation_versions: + file_operation_versions.append(version) + self._update_topic("openWB/system/file_operation_version", file_operation_versions) + + +def run_migrations() -> None: + LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) + # Die Lock-Datei erst nach dem gesamten Migrationslauf entfernen, auch bei Fehlern. + with LOCK_FILE.open("a") as lock_file: + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + log.info("Dateisystem-Migrationen laufen bereits. Kein zweiter Lauf gestartet.") + return + + try: + FilesystemMigrations().update_filesystem() + except Exception: + log.exception("Dateisystem-Migrationen abgebrochen.") + pub_system_message({}, "Fehler bei der Aktualisierung der Dateisystem-Daten.", MessageType.ERROR) + raise + finally: + LOCK_FILE.unlink(missing_ok=True) + + +if __name__ == "__main__": + logger.setup_generate_totals_logging(include_process_log=True) + logging.basicConfig(level=logging.DEBUG, handlers=logging.getLogger("generate_totals").handlers) + run_migrations() diff --git a/packages/helpermodules/filesystem_migrations_test.py b/packages/helpermodules/filesystem_migrations_test.py new file mode 100644 index 0000000000..f44fc98744 --- /dev/null +++ b/packages/helpermodules/filesystem_migrations_test.py @@ -0,0 +1,202 @@ +import fcntl +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest + +from helpermodules import filesystem_migrations +from helpermodules.filesystem_migrations import FilesystemMigrations + + +@pytest.mark.parametrize("file_operation_version, finished, expected_calls", [ + ([], False, 1), + ([0], True, 0), + ([0], False, 1), + (None, None, 1), +]) +def test_file_operation_0_start_behavior(monkeypatch, file_operation_version, finished, expected_calls): + monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 0) + updater = FilesystemMigrations() + updater.all_received_topics = { + "openWB/system/file_operation_version": file_operation_version, + "openWB/system/log_data_ready": finished, + } + + with patch.object(updater, "upgrade_file_operation_0") as upgrade_mock: + updater._FilesystemMigrations__solve_breaking_changes_filesystem() + assert upgrade_mock.call_count == expected_calls + + +def test_file_operations_run_sequentially_and_resume_after_failure(monkeypatch): + fails = True + updater = FilesystemMigrations() + updater.all_received_topics = { + "openWB/system/file_operation_version": [0], + "openWB/system/log_data_ready": True, + } + monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 3) + calls = [] + + def migration_1(): + assert updater.all_received_topics["openWB/system/file_operation_version"] == [0] + calls.append(1) + + def migration_2(): + assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1] + calls.append(2) + if fails: + raise RuntimeError("migration failed") + + def migration_3(): + assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1, 2] + calls.append(3) + + monkeypatch.setattr(updater, "upgrade_file_operation_1", migration_1, raising=False) + monkeypatch.setattr(updater, "upgrade_file_operation_2", migration_2, raising=False) + monkeypatch.setattr(updater, "upgrade_file_operation_3", migration_3, raising=False) + + with pytest.raises(RuntimeError, match="migration failed"): + updater._FilesystemMigrations__solve_breaking_changes_filesystem() + assert calls == [1, 2] + assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1] + fails = False + updater._FilesystemMigrations__solve_breaking_changes_filesystem() + assert calls == [1, 2, 2, 3] + + assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1, 2, 3] + + +def test_file_operation_stops_if_version_cannot_be_published(monkeypatch, mock_pub): + monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 1) + updater = FilesystemMigrations() + updater.all_received_topics = { + "openWB/system/file_operation_version": [], + "openWB/system/log_data_ready": True, + } + monkeypatch.setattr(updater, "upgrade_file_operation_0", Mock()) + next_migration = Mock() + monkeypatch.setattr(updater, "upgrade_file_operation_1", next_migration, raising=False) + mock_pub.publisher.client.publish.return_value.is_published.return_value = False + + with pytest.raises(TimeoutError, match="konnte nicht gespeichert"): + updater._FilesystemMigrations__solve_breaking_changes_filesystem() + + assert updater.all_received_topics["openWB/system/file_operation_version"] == [] + next_migration.assert_not_called() + + +def test_next_migration_runs_after_all_totals_attempts_despite_failed_day(monkeypatch, mock_pub): + monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 1) + from helpermodules.measurement_logging import generate_totals_subprocess + + updater = FilesystemMigrations() + updater.all_received_topics = { + "openWB/system/file_operation_version": [], + "openWB/system/log_data_ready": False, + } + days = [Path("20220101.json"), Path("20220102.json")] + monkeypatch.setattr(generate_totals_subprocess, "get_all_days_to_calc", lambda: days) + daily = Mock(side_effect=[RuntimeError("broken log"), {}]) + monthly = Mock(return_value={}) + monkeypatch.setattr(generate_totals_subprocess, "save_daily_source_totals", daily) + monkeypatch.setattr(generate_totals_subprocess, "save_monthly_source_totals", monthly) + monkeypatch.setattr(filesystem_migrations, "generate_totals", generate_totals_subprocess._generate_totals) + + def next_migration(): + assert daily.call_count == 2 + monthly.assert_called_once_with("202201", saving=True) + assert updater.all_received_topics["openWB/system/log_data_ready"] is True + assert updater.all_received_topics["openWB/system/file_operation_version"] == [0] + assert any(call.args == ("openWB/system/log_data_ready", "true") + for call in mock_pub.publisher.client.publish.call_args_list) + + next_operation = Mock(side_effect=next_migration) + monkeypatch.setattr(updater, "upgrade_file_operation_1", next_operation, raising=False) + + updater._FilesystemMigrations__solve_breaking_changes_filesystem() + + next_operation.assert_called_once_with() + assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1] + + +@pytest.mark.parametrize("completed, ready", [([], True), ([0], False)]) +def test_file_operation_0_failure_leaves_logs_unready_and_blocks_next(monkeypatch, completed, ready): + monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 1) + updater = FilesystemMigrations() + updater.all_received_topics = { + "openWB/system/file_operation_version": completed, + "openWB/system/log_data_ready": ready, + } + monkeypatch.setattr( + filesystem_migrations, "generate_totals", Mock(side_effect=RuntimeError("generator failed"))) + next_migration = Mock() + monkeypatch.setattr(updater, "upgrade_file_operation_1", next_migration, raising=False) + + with pytest.raises(RuntimeError, match="generator failed"): + updater._FilesystemMigrations__solve_breaking_changes_filesystem() + + assert updater.all_received_topics["openWB/system/log_data_ready"] is False + assert updater.all_received_topics["openWB/system/file_operation_version"] == completed + next_migration.assert_not_called() + + +def test_filesystem_worker_reads_current_topics_before_migrating(monkeypatch): + updater = FilesystemMigrations() + broker = Mock() + monkeypatch.setattr(filesystem_migrations, "BrokerClient", Mock(return_value=broker)) + + def read_topics(): + updater.all_received_topics = { + "openWB/system/file_operation_version": [0], + "openWB/system/log_data_ready": True, + } + + broker.start_finite_loop.side_effect = read_topics + migration = Mock() + monkeypatch.setattr(updater, "upgrade_file_operation_0", migration) + + updater.update_filesystem() + + migration.assert_not_called() + broker.start_finite_loop.assert_called_once_with() + broker.disconnect.assert_called_once_with() + + +def test_worker_skips_broker_read_while_another_worker_holds_lock(monkeypatch, tmp_path): + lock_path = tmp_path / "filesystem_migrations.lock" + monkeypatch.setattr(filesystem_migrations, "LOCK_FILE", lock_path) + + def migrate(): + assert lock_path.exists() + + update = Mock(side_effect=migrate) + monkeypatch.setattr(FilesystemMigrations, "update_filesystem", update) + + with lock_path.open("a") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + filesystem_migrations.run_migrations() + update.assert_not_called() + assert lock_path.exists() + + filesystem_migrations.run_migrations() + update.assert_called_once_with() + assert not lock_path.exists() + + +def test_worker_reports_failure_and_releases_lock(monkeypatch, tmp_path): + lock_path = tmp_path / "filesystem_migrations.lock" + monkeypatch.setattr(filesystem_migrations, "LOCK_FILE", lock_path) + update = Mock(side_effect=RuntimeError("migration failed")) + monkeypatch.setattr(FilesystemMigrations, "update_filesystem", update) + message = Mock() + monkeypatch.setattr(filesystem_migrations, "pub_system_message", message) + + with pytest.raises(RuntimeError, match="migration failed"): + filesystem_migrations.run_migrations() + + message.assert_called_once() + assert not lock_path.exists() + update.side_effect = None + filesystem_migrations.run_migrations() + assert update.call_count == 2 + assert not lock_path.exists() diff --git a/packages/helpermodules/measurement_logging/generate_totals_subprocess.py b/packages/helpermodules/measurement_logging/generate_totals_subprocess.py index f19a3b1283..079b3a2c2e 100644 --- a/packages/helpermodules/measurement_logging/generate_totals_subprocess.py +++ b/packages/helpermodules/measurement_logging/generate_totals_subprocess.py @@ -73,32 +73,39 @@ def _generate_totals(): log.info(f"Totals-Migration abgeschlossen. Fehlerhafte Logs: {errors}.") publisher = pub.Pub() publish_info = publisher.publisher.client.publish( - "openWB/set/system/log_data_ready", json.dumps(True), qos=1, retain=True) + "openWB/system/log_data_ready", json.dumps(True), qos=1, retain=True) publish_info.wait_for_publish() def generate_totals(): LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) - with LOCK_FILE.open("w") as lock_file: + with LOCK_FILE.open("a") as lock_file: try: fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - except Exception: + except BlockingIOError: log.exception("generate_totals_subprocess läuft bereits. " "Der zweite Prozess wird beendet.") - return + raise try: current_lock_stat = LOCK_FILE.stat() - except FileNotFoundError: - return + except FileNotFoundError as exc: + raise RuntimeError( + "Lock-Datei wurde zwischenzeitlich gelöscht. Totals-Berechnung wird nicht gestartet." + ) from exc # Ein anderer Prozess kann die bereits geöffnete Lock-Datei inzwischen gelöscht und neu angelegt haben. # Unsere Sperre gilt dann nur für die alte Datei. Abbrechen, damit nicht zwei Prozesse gleichzeitig rechnen. if not os.path.samestat(current_lock_stat, os.fstat(lock_file.fileno())): - return + raise RuntimeError("Lock-Datei der Totals-Migration wurde zwischenzeitlich ausgetauscht.") try: _generate_totals() finally: - LOCK_FILE.unlink(missing_ok=True) + # Nur unsere eigene Lock-Datei entfernen, keine zwischenzeitlich neu angelegte Datei. + try: + if os.path.samestat(LOCK_FILE.stat(), os.fstat(lock_file.fileno())): + LOCK_FILE.unlink(missing_ok=True) + except FileNotFoundError: + pass if __name__ == "__main__": diff --git a/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py index 81d03cb521..f3ae1eb796 100644 --- a/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py +++ b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py @@ -1,43 +1,131 @@ +import fcntl from unittest.mock import Mock +import pytest + from helpermodules.measurement_logging import generate_totals_subprocess -def test_generate_totals_calls_generator_when_lock_is_acquired(monkeypatch, tmp_path): +@pytest.mark.parametrize("fails", [False, True]) +def test_generate_totals_cleans_up_lock_after_processing(monkeypatch, tmp_path, fails): # setup lock_file = tmp_path / "locks" / "generate_totals.lock" monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) - flock_mock = Mock() - generate_mock = Mock() + generate_mock = Mock(side_effect=RuntimeError("generator failed") if fails else None) - monkeypatch.setattr(generate_totals_subprocess.fcntl, "flock", flock_mock) monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) # execution - generate_totals_subprocess.generate_totals() + if fails: + with pytest.raises(RuntimeError, match="generator failed"): + generate_totals_subprocess.generate_totals() + else: + generate_totals_subprocess.generate_totals() # evaluation - flock_mock.assert_called_once() generate_mock.assert_called_once() assert lock_file.parent.is_dir() + assert not lock_file.exists() -def test_generate_totals_skips_generator_when_lock_is_not_acquired(monkeypatch, tmp_path): +def test_generate_totals_fails_when_lock_is_not_acquired(monkeypatch, tmp_path): # setup lock_file = tmp_path / "locks" / "generate_totals.lock" monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) - flock_mock = Mock(side_effect=RuntimeError("already locked")) generate_mock = Mock() - monkeypatch.setattr(generate_totals_subprocess.fcntl, "flock", flock_mock) monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) # execution - generate_totals_subprocess.generate_totals() + lock_file.parent.mkdir(parents=True) + with lock_file.open("a") as held_lock: + fcntl.flock(held_lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + with pytest.raises(BlockingIOError): + generate_totals_subprocess.generate_totals() # evaluation - flock_mock.assert_called_once() generate_mock.assert_not_called() assert lock_file.parent.is_dir() + assert lock_file.exists() + + +@pytest.mark.parametrize("replaced", [False, True]) +def test_generate_totals_aborts_if_lock_disappears_or_is_replaced_before_processing(monkeypatch, tmp_path, replaced): + lock_file = tmp_path / "generate_totals.lock" + monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) + + def change_lock(*args): + if replaced: + lock_file.rename(tmp_path / "old.lock") + lock_file.write_text("other process") + else: + lock_file.unlink() + + monkeypatch.setattr(generate_totals_subprocess.fcntl, "flock", change_lock) + generate_mock = Mock() + monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) + + message = "ausgetauscht" if replaced else "gelöscht" + with pytest.raises(RuntimeError, match=message) as error: + generate_totals_subprocess.generate_totals() + + generate_mock.assert_not_called() + if replaced: + assert lock_file.read_text() == "other process" + else: + assert isinstance(error.value.__cause__, FileNotFoundError) + assert not lock_file.exists() + + +@pytest.mark.parametrize("replaced", [False, True]) +def test_generate_totals_cleanup_handles_changed_lock(monkeypatch, tmp_path, replaced): + lock_file = tmp_path / "generate_totals.lock" + monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) + + def generate(): + if replaced: + lock_file.rename(tmp_path / "old.lock") + lock_file.write_text("other process") + else: + lock_file.unlink() + + monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate) + + generate_totals_subprocess.generate_totals() + + if replaced: + assert lock_file.read_text() == "other process" + else: + assert not lock_file.exists() + + +@pytest.mark.parametrize("failure", [None, RuntimeError("broken log")]) +@pytest.mark.parametrize("period", ["daily", "monthly"]) +def test_generate_totals_continues_after_failed_logs(monkeypatch, mock_pub, period, failure): + days = [generate_totals_subprocess.Path(f"20220{month}01.json") for month in range(1, 4)] + monkeypatch.setattr(generate_totals_subprocess, "get_all_days_to_calc", lambda: days) + daily = Mock(return_value={}) + monthly = Mock(return_value={}) + failed = daily if period == "daily" else monthly + failed.side_effect = [failure, {}, {}] + log_error = Mock() + monkeypatch.setattr(generate_totals_subprocess.log, "exception", log_error) + monkeypatch.setattr(generate_totals_subprocess, "save_daily_source_totals", daily) + monkeypatch.setattr(generate_totals_subprocess, "save_monthly_source_totals", monthly) + + def publish(*args, **kwargs): + assert daily.call_count == 3 + assert monthly.call_count == 3 + return Mock() + + mock_pub.publisher.client.publish.side_effect = publish + + generate_totals_subprocess._generate_totals() + + assert daily.call_args_list == [((day.stem,), {"saving": True}) for day in days] + assert monthly.call_args_list == [((f"20220{month}",), {"saving": True}) for month in range(1, 4)] + assert log_error.call_count == (1 if failure is not None else 0) + mock_pub.publisher.client.publish.assert_called_once_with( + "openWB/system/log_data_ready", "true", qos=1, retain=True) diff --git a/packages/helpermodules/setdata.py b/packages/helpermodules/setdata.py index 256e34d37c..aa975a8f51 100644 --- a/packages/helpermodules/setdata.py +++ b/packages/helpermodules/setdata.py @@ -12,6 +12,7 @@ from control import data from helpermodules import hardware_configuration, subdata from helpermodules.broker import BrokerClient +from helpermodules.filesystem_migrations import FilesystemMigrations from helpermodules.pub import Pub from helpermodules.utils.topic_parser import decode_payload, get_index, get_index_position from helpermodules.update_config import UpdateConfig @@ -1119,7 +1120,7 @@ def process_system_topic(self, msg: mqtt.MQTTMessage): elif "openWB/set/system/datastore_version" in msg.topic: self._validate_value(msg, int, [(0, UpdateConfig.DATASTORE_VERSION)], collection=list) elif "openWB/set/system/file_operation_version" in msg.topic: - self._validate_value(msg, int, [(0, UpdateConfig.FILE_OPERATION_VERSION)], collection=list) + self._validate_value(msg, int, [(0, FilesystemMigrations.FILE_OPERATION_VERSION)], collection=list) elif "openWB/set/system/GetRemoteSupport" in msg.topic: # Server-Topic enthält kein json-Payload. payload = msg.payload.decode("utf-8") diff --git a/packages/helpermodules/update_config.py b/packages/helpermodules/update_config.py index de5f64bc1a..c9aad61d10 100644 --- a/packages/helpermodules/update_config.py +++ b/packages/helpermodules/update_config.py @@ -65,8 +65,6 @@ class UpdateConfig: DATASTORE_VERSION = 156 - FILE_OPERATION_VERSION = 0 - valid_topic = [ "^openWB/bat/config/bat_control_activated$", "^openWB/bat/config/power_limit_mode$", @@ -834,11 +832,11 @@ def update(self): try: # erst breaking changes auflösen, sonst sind alte Topics schon gelöscht self.__solve_breaking_changes() - self.__solve_breaking_changes_filesystem() self.__remove_outdated_topics() self._remove_invalid_topics() self.__pub_missing_defaults() self.__update_version() + self.__start_filesystem_migrations() except Exception: log.exception("Fehler bei der Aktualisierung des Brokers.") pub_system_message({}, "Fehler bei der Aktualisierung der Konfiguration des Brokers.", MessageType.ERROR) @@ -935,66 +933,21 @@ def __solve_breaking_changes(self) -> None: pub_system_message( {}, "Fehler bei der Aktualisierung der Konfiguration des Brokers.", MessageType.ERROR) - def __solve_breaking_changes_filesystem(self) -> None: - """Führt dateisystembezogene Migrationen anhand der file_operation_version aus.""" - file_operation_version = decode_payload(self.all_received_topics.get("openWB/system/file_operation_version")) - if file_operation_version is None: - # Neues Topic (z.B. bei Upgrade von älteren Versionen): alle File-Operation-Upgrades ausführen. - file_operation_version = [] - self.__update_topic("openWB/system/file_operation_version", file_operation_version) - elif isinstance(file_operation_version, int): - # Legacy-Format (int): bereits ausgeführte Upgrades als Liste abbilden. - file_operation_version = list(range(file_operation_version)) - self.__update_topic("openWB/system/file_operation_version", file_operation_version) - - log.debug(f"current file operation version: {file_operation_version}") - log.debug(f"target file operation version: {self.FILE_OPERATION_VERSION}") - for version in list(range(self.FILE_OPERATION_VERSION+1)): - try: - operation_required = version not in file_operation_version - if version == 0: - # Version 0 erneut ausführen, solange die Hintergrund-Generierung der Tages-/Monatssummen - # noch nicht erfolgreich abgeschlossen ist (Flag ist nicht True). - log_totals_generation_finished = decode_payload( - self.all_received_topics.get("openWB/system/log_data_ready")) - operation_required = operation_required or log_totals_generation_finished is not True - - if operation_required: - log.debug(f"upgrading File Operation version '{version}'") - getattr(self, f"upgrade_file_operation_{version}")() - except AttributeError: - log.error(f"missing upgrade function! '{version}'") - except Exception: - log.exception("Fehler bei der Aktualisierung des Brokers.") - pub_system_message( - {}, "Fehler bei der Aktualisierung der Konfiguration des Brokers.", MessageType.ERROR) - - def upgrade_file_operation_0(self) -> None: + def __start_filesystem_migrations(self) -> None: """ - Generiere die Totals-Summen für Tage und Monate + Startet in einem eigen Prozess die Filesystem_Migration + -> alle folgenden upgrade_file_operation_X werden nacheinander und nicht parallel abgearbeitet """ - - self.__update_topic("openWB/system/log_data_ready", False) - try: - _generate_totals_subprocess = subprocess.Popen( - [ - sys.executable, - "-m", - "helpermodules.measurement_logging.generate_totals_subprocess" - ], - cwd="/var/www/html/openWB/packages", - start_new_session=True, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - close_fds=True - ) - - log.debug("generate_totals_subprocess gestartet, PID: %s", _generate_totals_subprocess.pid) - self._append_file_operation_version(0) - except Exception: - _generate_totals_subprocess = None - log.exception("Fehler beim Starten des generate_totals_subprocess.") + process = subprocess.Popen( + [sys.executable, "-m", "helpermodules.filesystem_migrations"], + cwd=self.base_path / "packages", + start_new_session=True, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + close_fds=True, + ) + log.debug("Dateisystem-Migrationen gestartet, PID: %s", process.pid) def _loop_all_received_topics(self, callback) -> None: modified_topics = {} @@ -1014,12 +967,6 @@ def _append_datastore_version(self, version: int) -> None: datastore_versions.append(version) self.__update_topic("openWB/system/datastore_version", datastore_versions) - def _append_file_operation_version(self, version: int) -> None: - file_operation_versions = decode_payload(self.all_received_topics.get("openWB/system/file_operation_version")) - if version not in file_operation_versions: - file_operation_versions.append(version) - self.__update_topic("openWB/system/file_operation_version", file_operation_versions) - def upgrade_datastore_0(self) -> None: def upgrade(topic: str, payload) -> Optional[dict]: modified_topics = {} diff --git a/packages/helpermodules/update_config_test.py b/packages/helpermodules/update_config_test.py index aee711d241..d2117ec280 100644 --- a/packages/helpermodules/update_config_test.py +++ b/packages/helpermodules/update_config_test.py @@ -619,23 +619,6 @@ def test_upgrade_datastore_153_migrates_legacy_modes( assert uc.all_received_topics["openWB/system/datastore_version"] == [152, 153] -@pytest.mark.parametrize("file_operation_version, finished, expected_calls", [ - ([], False, 1), # erster Start - ([0], True, 0), # bereits fertig -> kein Neustart - ([0], False, 1), # angefangen, aber nicht fertig -> Neustart -]) -def test_file_operation_0_start_behavior(file_operation_version, finished, expected_calls): - update_config = UpdateConfig() - update_config.all_received_topics = { - "openWB/system/file_operation_version": file_operation_version, - "openWB/system/log_data_ready": finished - } - - with patch.object(update_config, "upgrade_file_operation_0") as upgrade_mock: - update_config._UpdateConfig__solve_breaking_changes_filesystem() - assert upgrade_mock.call_count == expected_calls - - @pytest.mark.parametrize("provider_type, configuration, expected_configuration, updated", [ ("energycharts", {"surcharge": 0.0001}, {"surcharge": 0.0001, "price_cap": None}, True), ("energycharts", {"price_cap": 0.0002}, {"price_cap": 0.0002}, False), @@ -666,3 +649,21 @@ def test_upgrade_datastore_156_adds_energycharts_price_cap( updated_topics = [call.args[0] for call in mock_pub.pub.call_args_list] assert (topic in updated_topics) is updated assert uc.all_received_topics["openWB/system/datastore_version"] == [155, 156] + + +def test_start_filesystem_migrations_launches_one_worker(monkeypatch): + updater = UpdateConfig() + popen = Mock() + monkeypatch.setattr(update_config.subprocess, "Popen", popen) + + updater._UpdateConfig__start_filesystem_migrations() + + popen.assert_called_once_with( + [update_config.sys.executable, "-m", "helpermodules.filesystem_migrations"], + cwd=updater.base_path / "packages", + start_new_session=True, + stdin=update_config.subprocess.DEVNULL, + stdout=update_config.subprocess.DEVNULL, + stderr=update_config.subprocess.DEVNULL, + close_fds=True, + ) From 08290b54793745c328ebc44462ea3194e7353d3c Mon Sep 17 00:00:00 2001 From: Alexander Hartung Date: Mon, 9 Nov 2026 14:39:41 +0100 Subject: [PATCH 2/3] Add sourcery-ai suggestions --- .gitignore | 2 + .../helpermodules/filesystem_migrations.py | 10 +- .../filesystem_migrations_test.py | 88 ++++++++--------- .../generate_totals_subprocess.py | 15 +-- .../generate_totals_subprocess_test.py | 96 ------------------- 5 files changed, 55 insertions(+), 156 deletions(-) diff --git a/.gitignore b/.gitignore index c83c7c0d5e..d5802daa0c 100644 --- a/.gitignore +++ b/.gitignore @@ -19,6 +19,8 @@ data/daily_log/* data/daily_totals/* data/monthly_log/* data/monthly_totals/* +data/filesystem_migrations.lock +data/generate_totals.lock data/backup/*.tar.gz data/backup/*.openwb-backup* data/restore/*.tar.gz diff --git a/packages/helpermodules/filesystem_migrations.py b/packages/helpermodules/filesystem_migrations.py index 8294e3f3ee..b359a33217 100644 --- a/packages/helpermodules/filesystem_migrations.py +++ b/packages/helpermodules/filesystem_migrations.py @@ -63,6 +63,10 @@ def __solve_breaking_changes_filesystem(self) -> None: # Neues Topic (z.B. bei Upgrade von älteren Versionen): alle File-Operation-Upgrades ausführen. file_operation_version = [] self._update_topic("openWB/system/file_operation_version", file_operation_version) + elif isinstance(file_operation_version, int): + # Legacy-Zähler: Die Versionen vor diesem Wert sind bereits abgeschlossen. + file_operation_version = list(range(file_operation_version)) + self._update_topic("openWB/system/file_operation_version", file_operation_version) log.debug(f"current file operation version: {file_operation_version}") log.debug(f"target file operation version: {self.FILE_OPERATION_VERSION}") @@ -96,7 +100,9 @@ def _append_file_operation_version(self, version: int) -> None: def run_migrations() -> None: LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) - # Die Lock-Datei erst nach dem gesamten Migrationslauf entfernen, auch bei Fehlern. + # Lock-Datei beibehalten, damit alle Prozesse denselben Inode sperren. + # Löschen und Neuanlegen könnte parallele Läufe ermöglichen. + # Die Sperre wird beim Verlassen des with-Blocks automatisch freigegeben. with LOCK_FILE.open("a") as lock_file: try: fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) @@ -110,8 +116,6 @@ def run_migrations() -> None: log.exception("Dateisystem-Migrationen abgebrochen.") pub_system_message({}, "Fehler bei der Aktualisierung der Dateisystem-Daten.", MessageType.ERROR) raise - finally: - LOCK_FILE.unlink(missing_ok=True) if __name__ == "__main__": diff --git a/packages/helpermodules/filesystem_migrations_test.py b/packages/helpermodules/filesystem_migrations_test.py index f44fc98744..9005449985 100644 --- a/packages/helpermodules/filesystem_migrations_test.py +++ b/packages/helpermodules/filesystem_migrations_test.py @@ -1,8 +1,9 @@ -import fcntl +import json from pathlib import Path from unittest.mock import Mock, patch import pytest +from paho.mqtt.client import MQTTMessage from helpermodules import filesystem_migrations from helpermodules.filesystem_migrations import FilesystemMigrations @@ -27,6 +28,30 @@ def test_file_operation_0_start_behavior(monkeypatch, file_operation_version, fi assert upgrade_mock.call_count == expected_calls +@pytest.mark.parametrize("progress", [0, 1, 2]) +def test_legacy_integer_progress_is_persisted_as_list_and_pending_migrations_run(monkeypatch, mock_pub, progress): + updater = FilesystemMigrations() + updater.all_received_topics = { + "openWB/system/file_operation_version": str(progress).encode(), + "openWB/system/log_data_ready": b"true", + } + monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 2) + calls = [] + for version in range(3): + monkeypatch.setattr( + updater, f"upgrade_file_operation_{version}", + lambda version=version: calls.append(version), raising=False, + ) + + updater._FilesystemMigrations__solve_breaking_changes_filesystem() + + assert calls == list(range(progress, 3)) + assert updater.all_received_topics["openWB/system/file_operation_version"] == [0, 1, 2] + first_publish = mock_pub.publisher.client.publish.call_args_list[0] + assert first_publish.args == ("openWB/system/file_operation_version", json.dumps(list(range(progress)))) + assert first_publish.kwargs == {"qos": 1, "retain": True} + + def test_file_operations_run_sequentially_and_resume_after_failure(monkeypatch): fails = True updater = FilesystemMigrations() @@ -120,7 +145,7 @@ def next_migration(): @pytest.mark.parametrize("completed, ready", [([], True), ([0], False)]) -def test_file_operation_0_failure_leaves_logs_unready_and_blocks_next(monkeypatch, completed, ready): +def test_file_operation_0_failure_leaves_logs_unready_and_blocks_next(monkeypatch, mock_pub, completed, ready): monkeypatch.setattr(FilesystemMigrations, "FILE_OPERATION_VERSION", 1) updater = FilesystemMigrations() updater.all_received_topics = { @@ -137,19 +162,28 @@ def test_file_operation_0_failure_leaves_logs_unready_and_blocks_next(monkeypatc assert updater.all_received_topics["openWB/system/log_data_ready"] is False assert updater.all_received_topics["openWB/system/file_operation_version"] == completed + mock_pub.publisher.client.publish.assert_called_once_with( + "openWB/system/log_data_ready", "false", qos=1, retain=True) next_migration.assert_not_called() def test_filesystem_worker_reads_current_topics_before_migrating(monkeypatch): updater = FilesystemMigrations() broker = Mock() - monkeypatch.setattr(filesystem_migrations, "BrokerClient", Mock(return_value=broker)) + broker_client = Mock(return_value=broker) + monkeypatch.setattr(filesystem_migrations, "BrokerClient", broker_client) + topics = { + "openWB/system/file_operation_version": b"[0]", + "openWB/system/log_data_ready": b"true", + } def read_topics(): - updater.all_received_topics = { - "openWB/system/file_operation_version": [0], - "openWB/system/log_data_ready": True, - } + on_message = broker_client.call_args.args[2] + for topic, payload in topics.items(): + message = MQTTMessage(topic=topic.encode()) + message.payload = payload + on_message(broker.client, None, message) + assert updater.all_received_topics == topics broker.start_finite_loop.side_effect = read_topics migration = Mock() @@ -160,43 +194,3 @@ def read_topics(): migration.assert_not_called() broker.start_finite_loop.assert_called_once_with() broker.disconnect.assert_called_once_with() - - -def test_worker_skips_broker_read_while_another_worker_holds_lock(monkeypatch, tmp_path): - lock_path = tmp_path / "filesystem_migrations.lock" - monkeypatch.setattr(filesystem_migrations, "LOCK_FILE", lock_path) - - def migrate(): - assert lock_path.exists() - - update = Mock(side_effect=migrate) - monkeypatch.setattr(FilesystemMigrations, "update_filesystem", update) - - with lock_path.open("a") as lock_file: - fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - filesystem_migrations.run_migrations() - update.assert_not_called() - assert lock_path.exists() - - filesystem_migrations.run_migrations() - update.assert_called_once_with() - assert not lock_path.exists() - - -def test_worker_reports_failure_and_releases_lock(monkeypatch, tmp_path): - lock_path = tmp_path / "filesystem_migrations.lock" - monkeypatch.setattr(filesystem_migrations, "LOCK_FILE", lock_path) - update = Mock(side_effect=RuntimeError("migration failed")) - monkeypatch.setattr(FilesystemMigrations, "update_filesystem", update) - message = Mock() - monkeypatch.setattr(filesystem_migrations, "pub_system_message", message) - - with pytest.raises(RuntimeError, match="migration failed"): - filesystem_migrations.run_migrations() - - message.assert_called_once() - assert not lock_path.exists() - update.side_effect = None - filesystem_migrations.run_migrations() - assert update.call_count == 2 - assert not lock_path.exists() diff --git a/packages/helpermodules/measurement_logging/generate_totals_subprocess.py b/packages/helpermodules/measurement_logging/generate_totals_subprocess.py index 079b3a2c2e..a360523825 100644 --- a/packages/helpermodules/measurement_logging/generate_totals_subprocess.py +++ b/packages/helpermodules/measurement_logging/generate_totals_subprocess.py @@ -13,7 +13,6 @@ BASE_PATH = Path(__file__).resolve().parents[3] -logger.setup_generate_totals_logging(include_process_log=True) log = logging.getLogger("generate_totals") LOCK_FILE = BASE_PATH / "data" / "generate_totals.lock" @@ -80,6 +79,9 @@ def _generate_totals(): def generate_totals(): LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) + # Lock-Datei beibehalten, damit alle Prozesse denselben Inode sperren. + # Löschen und Neuanlegen könnte parallele Läufe ermöglichen. + # Die Sperre wird beim Verlassen des with-Blocks automatisch freigegeben. with LOCK_FILE.open("a") as lock_file: try: fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) @@ -97,16 +99,9 @@ def generate_totals(): # Unsere Sperre gilt dann nur für die alte Datei. Abbrechen, damit nicht zwei Prozesse gleichzeitig rechnen. if not os.path.samestat(current_lock_stat, os.fstat(lock_file.fileno())): raise RuntimeError("Lock-Datei der Totals-Migration wurde zwischenzeitlich ausgetauscht.") - try: - _generate_totals() - finally: - # Nur unsere eigene Lock-Datei entfernen, keine zwischenzeitlich neu angelegte Datei. - try: - if os.path.samestat(LOCK_FILE.stat(), os.fstat(lock_file.fileno())): - LOCK_FILE.unlink(missing_ok=True) - except FileNotFoundError: - pass + _generate_totals() if __name__ == "__main__": + logger.setup_generate_totals_logging(include_process_log=True) generate_totals() diff --git a/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py index f3ae1eb796..e5ab61844f 100644 --- a/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py +++ b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py @@ -1,4 +1,3 @@ -import fcntl from unittest.mock import Mock import pytest @@ -6,101 +5,6 @@ from helpermodules.measurement_logging import generate_totals_subprocess -@pytest.mark.parametrize("fails", [False, True]) -def test_generate_totals_cleans_up_lock_after_processing(monkeypatch, tmp_path, fails): - # setup - lock_file = tmp_path / "locks" / "generate_totals.lock" - monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) - - generate_mock = Mock(side_effect=RuntimeError("generator failed") if fails else None) - - monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) - - # execution - if fails: - with pytest.raises(RuntimeError, match="generator failed"): - generate_totals_subprocess.generate_totals() - else: - generate_totals_subprocess.generate_totals() - - # evaluation - generate_mock.assert_called_once() - assert lock_file.parent.is_dir() - assert not lock_file.exists() - - -def test_generate_totals_fails_when_lock_is_not_acquired(monkeypatch, tmp_path): - # setup - lock_file = tmp_path / "locks" / "generate_totals.lock" - monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) - - generate_mock = Mock() - - monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) - - # execution - lock_file.parent.mkdir(parents=True) - with lock_file.open("a") as held_lock: - fcntl.flock(held_lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - with pytest.raises(BlockingIOError): - generate_totals_subprocess.generate_totals() - - # evaluation - generate_mock.assert_not_called() - assert lock_file.parent.is_dir() - assert lock_file.exists() - - -@pytest.mark.parametrize("replaced", [False, True]) -def test_generate_totals_aborts_if_lock_disappears_or_is_replaced_before_processing(monkeypatch, tmp_path, replaced): - lock_file = tmp_path / "generate_totals.lock" - monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) - - def change_lock(*args): - if replaced: - lock_file.rename(tmp_path / "old.lock") - lock_file.write_text("other process") - else: - lock_file.unlink() - - monkeypatch.setattr(generate_totals_subprocess.fcntl, "flock", change_lock) - generate_mock = Mock() - monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) - - message = "ausgetauscht" if replaced else "gelöscht" - with pytest.raises(RuntimeError, match=message) as error: - generate_totals_subprocess.generate_totals() - - generate_mock.assert_not_called() - if replaced: - assert lock_file.read_text() == "other process" - else: - assert isinstance(error.value.__cause__, FileNotFoundError) - assert not lock_file.exists() - - -@pytest.mark.parametrize("replaced", [False, True]) -def test_generate_totals_cleanup_handles_changed_lock(monkeypatch, tmp_path, replaced): - lock_file = tmp_path / "generate_totals.lock" - monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) - - def generate(): - if replaced: - lock_file.rename(tmp_path / "old.lock") - lock_file.write_text("other process") - else: - lock_file.unlink() - - monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate) - - generate_totals_subprocess.generate_totals() - - if replaced: - assert lock_file.read_text() == "other process" - else: - assert not lock_file.exists() - - @pytest.mark.parametrize("failure", [None, RuntimeError("broken log")]) @pytest.mark.parametrize("period", ["daily", "monthly"]) def test_generate_totals_continues_after_failed_logs(monkeypatch, mock_pub, period, failure): From 7758c0e4a8954e57ed33e38ea5e9b0215eca5298 Mon Sep 17 00:00:00 2001 From: Alexander Hartung Date: Mon, 9 Nov 2026 14:56:03 +0100 Subject: [PATCH 3/3] Add Copilot Suggestion --- .../generate_totals_subprocess_test.py | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py index e5ab61844f..e7c7e0a3dd 100644 --- a/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py +++ b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py @@ -1,3 +1,4 @@ +import fcntl from unittest.mock import Mock import pytest @@ -5,6 +6,54 @@ from helpermodules.measurement_logging import generate_totals_subprocess +@pytest.mark.parametrize("fails", [False, True]) +def test_generate_totals_preserves_lock_file_and_releases_lock(monkeypatch, tmp_path, fails): + lock_path = tmp_path / "locks" / "generate_totals.lock" + monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_path) + locked_inode = [] + + def generate(): + locked_inode.append(lock_path.stat().st_ino) + with lock_path.open("a") as contender: + with pytest.raises(BlockingIOError): + fcntl.flock(contender.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + if fails: + raise RuntimeError("generator failed") + + generator = Mock(side_effect=generate) + monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generator) + + if fails: + with pytest.raises(RuntimeError, match="generator failed"): + generate_totals_subprocess.generate_totals() + else: + generate_totals_subprocess.generate_totals() + + generator.assert_called_once_with() + assert lock_path.stat().st_ino == locked_inode[0] + with lock_path.open("a") as next_owner: + fcntl.flock(next_owner.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + + +def test_generate_totals_raises_without_generating_when_lock_is_contended(monkeypatch, tmp_path): + lock_path = tmp_path / "generate_totals.lock" + monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_path) + generator = Mock() + monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generator) + + with lock_path.open("a") as owner: + original_inode = lock_path.stat().st_ino + fcntl.flock(owner.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + with pytest.raises(BlockingIOError): + generate_totals_subprocess.generate_totals() + generator.assert_not_called() + assert lock_path.stat().st_ino == original_inode + + generate_totals_subprocess.generate_totals() + generator.assert_called_once_with() + assert lock_path.stat().st_ino == original_inode + + @pytest.mark.parametrize("failure", [None, RuntimeError("broken log")]) @pytest.mark.parametrize("period", ["daily", "monthly"]) def test_generate_totals_continues_after_failed_logs(monkeypatch, mock_pub, period, failure):