94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
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
|