import json import os import time import logging from target_simulator.utils.config_manager import ConfigManager def test_skip_write_when_empty_preserves_existing(tmp_path, caplog): caplog.set_level(logging.INFO) settings_path = str(tmp_path / "settings.json") scenarios_path = str(tmp_path / "scenarios.json") # Create an existing scenarios file with open(scenarios_path, "w", encoding="utf-8") as f: json.dump({"existing": 1}, f) cm = ConfigManager(filename=settings_path, scenarios_filename=scenarios_path) # Force in-memory scenarios empty and call save cm._scenarios = {} cm._save_scenarios() # File should remain unchanged with open(scenarios_path, "r", encoding="utf-8") as f: data = json.load(f) assert data == {"existing": 1} # Ensure we logged skipping behavior assert any("Skipping saving scenarios" in r.message for r in caplog.records) def test_atomic_write_and_backup_created(tmp_path): settings_path = str(tmp_path / "settings.json") scenarios_path = str(tmp_path / "scenarios.json") # Start with an existing scenarios file with open(scenarios_path, "w", encoding="utf-8") as f: json.dump({"old": True}, f) cm = ConfigManager(filename=settings_path, scenarios_filename=scenarios_path) # Now update scenarios and save - should create a backup and then write cm._scenarios = {"new": 123} cm._save_scenarios() # scenarios.json should now contain the new content with open(scenarios_path, "r", encoding="utf-8") as f: data = json.load(f) assert data == {"new": 123} # A backup should exist with the previous content bak1 = scenarios_path + ".bak1" assert os.path.exists(bak1) with open(bak1, "r", encoding="utf-8") as f: bak = json.load(f) assert bak == {"old": True} def test_rotation_keeps_last_n_backups(tmp_path): settings_path = str(tmp_path / "settings.json") scenarios_path = str(tmp_path / "scenarios.json") cm = ConfigManager(filename=settings_path, scenarios_filename=scenarios_path) # create multiple writes to generate backups # create initial file with open(scenarios_path, "w", encoding="utf-8") as f: json.dump({"v": 0}, f) cm._scenarios = {"v": 1} cm._save_scenarios() time.sleep(0.01) cm._scenarios = {"v": 2} cm._save_scenarios() time.sleep(0.01) cm._scenarios = {"v": 3} cm._save_scenarios() time.sleep(0.01) cm._scenarios = {"v": 4} cm._save_scenarios() time.sleep(0.01) cm._scenarios = {"v": 5} cm._save_scenarios() # bak1 should exist and contain v=4 (previous before last write) bak1 = scenarios_path + ".bak1" assert os.path.exists(bak1) with open(bak1, "r", encoding="utf-8") as f: bak = json.load(f) assert isinstance(bak, dict) # bak5 may exist depending on rotates; ensure we don't have more than 5 backups backups = [scenarios_path + f".bak{i}" for i in range(1, 7)] existing = [p for p in backups if os.path.exists(p)] assert len(existing) <= 5