"""运行数据备份/快照工具 (config_backup) 单元测试。""" import json import os import sys import pytest sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from modules import config_backup # noqa: E402 @pytest.fixture() def workdir(tmp_path): cfg = tmp_path / 'config' back = tmp_path / 'backups' cfg.mkdir() return cfg, back def write_text(path, text): with open(path, 'w', encoding='utf-8') as f: f.write(text) class TestWriteJsonAtomic: def test_write_and_read(self, workdir): cfg, _ = workdir path = str(cfg / 'demo.json') config_backup.write_json_atomic(path, {'a': 1, '中文': 'ok'}) with open(path) as f: assert json.load(f) == {'a': 1, '中文': 'ok'} assert not os.path.exists(path + '.tmp') def test_read_json_file_missing_raises(self, workdir): cfg, _ = workdir with pytest.raises(FileNotFoundError): config_backup.read_json_file(str(cfg / 'nope.json')) class TestSnapshot: def test_snapshot_created_and_deduped(self, workdir): cfg, back = workdir path = str(cfg / 'dtu_config.json') write_text(path, '{"a": 1}') name = config_backup.snapshot_before_write(path, str(back), min_interval_s=0) assert name and os.path.exists(os.path.join(str(back), name)) # 内容未变化 -> 去重 assert config_backup.snapshot_before_write(path, str(back), min_interval_s=0) is None def test_snapshot_missing_file(self, workdir): cfg, back = workdir assert config_backup.snapshot_before_write(str(cfg / 'none.json'), str(back)) is None def test_snapshot_throttled(self, workdir): cfg, back = workdir path = str(cfg / 'dtu_config.json') write_text(path, '{"a": 1}') assert config_backup.snapshot_before_write(path, str(back), min_interval_s=3600) is not None write_text(path, '{"a": 2}') # 距上一份快照不足节流时间 -> 跳过 assert config_backup.snapshot_before_write(path, str(back), min_interval_s=3600) is None def test_snapshot_rotation(self, workdir): cfg, back = workdir path = str(cfg / 'devices.json') for i in range(5): write_text(path, json.dumps({'round': i})) config_backup.snapshot_before_write(path, str(back), keep=2, min_interval_s=0) snaps = config_backup.list_snapshots(str(back), stem='devices') assert len(snaps) == 2 def test_snapshot_stem_and_filter(self, workdir): cfg, back = workdir p1 = str(cfg / 'dtu_config.json') p2 = str(cfg / 'mqtt_config.json') write_text(p1, '{"a": 1}') write_text(p2, '{"b": 2}') n1 = config_backup.snapshot_before_write(p1, str(back), min_interval_s=0) n2 = config_backup.snapshot_before_write(p2, str(back), min_interval_s=0) assert config_backup.snapshot_stem(n1) == 'dtu_config' assert config_backup.snapshot_stem(n2) == 'mqtt_config' assert len(config_backup.list_snapshots(str(back), stem='dtu_config')) == 1 def test_same_second_collision_suffix(self, workdir): cfg, back = workdir path = str(cfg / 'dtu_config.json') write_text(path, '{"a": 1}') n1 = config_backup.snapshot_before_write(path, str(back), min_interval_s=0, now=1000.0) write_text(path, '{"a": 2}') n2 = config_backup.snapshot_before_write(path, str(back), min_interval_s=0, now=1000.0) assert n1 != n2 assert config_backup.snapshot_stem(n2) == 'dtu_config' def test_read_and_delete_snapshot(self, workdir): cfg, back = workdir path = str(cfg / 'serial_config.json') write_text(path, '{"port": "COM1"}') name = config_backup.snapshot_before_write(path, str(back), min_interval_s=0) data, err = config_backup.read_snapshot(name, str(back)) assert err is None and data == {'port': 'COM1'} ok, _ = config_backup.delete_snapshot(name, str(back)) assert ok and not os.path.exists(os.path.join(str(back), name)) def test_reject_path_traversal(self, workdir): _, back = workdir for bad in ('../etc/passwd', 'a/b.json', '..{}', 'x.json'): assert config_backup.read_snapshot(bad, str(back))[1] is not None assert config_backup.delete_snapshot(bad, str(back))[0] is False def test_latest_snapshot_content(self, workdir): cfg, back = workdir path = str(cfg / 'devices.json') write_text(path, '{"aa": 1}') config_backup.snapshot_before_write(path, str(back), min_interval_s=0) data, name = config_backup.latest_snapshot_content(path, str(back)) assert data == {'aa': 1} and name class TestBundle: def _bundle(self): return config_backup.build_bundle( { 'dtu_config': {'dtu_id': 'dtu_001', 'port_mappings': {'PANEL_dtu_001_1': {'1': 'AABBCCDDEEFF'}}}, 'serial_config': {'port': '/dev/ttyS8', 'baudrate': 9600}, 'mqtt_config': {'broker': 'xt.wenhq.top', 'port': 8581}, 'devices': {'18' + '00' * 10 + '59': 1}, 'system_env': 'DHT11_ENABLED=true\n', }, device_meta={'dtu_id': 'dtu_001'}, uid_prefix='a1b2', ) def test_build_and_validate_ok(self): bundle = self._bundle() assert bundle['format'] == config_backup.BUNDLE_FORMAT assert bundle['effective_uid_prefix'] == 'A1B2' ok, msg = config_backup.validate_bundle(bundle) assert ok, msg def test_validate_rejects_wrong_format(self): assert config_backup.validate_bundle({'format': 'other'})[0] is False assert config_backup.validate_bundle([])[0] is False def test_validate_rejects_unknown_file_key(self): bundle = self._bundle() bundle['files']['shadow'] = {'x': 1} ok, msg = config_backup.validate_bundle(bundle) assert not ok and '未知文件项' in msg def test_validate_rejects_bad_devices(self): bundle = self._bundle() bundle['files']['devices'] = {'not-a-uid': 1} assert config_backup.validate_bundle(bundle)[0] is False bundle = self._bundle() bundle['files']['devices'] = {'a' * 24: 999} assert config_backup.validate_bundle(bundle)[0] is False def test_validate_rejects_bad_address_range(self): bundle = self._bundle() bundle['files']['dtu_config']['address_range'] = {'start': 50, 'end': 1} assert config_backup.validate_bundle(bundle)[0] is False bundle['files']['dtu_config']['address_range'] = {'start': 1, 'end': 50} assert config_backup.validate_bundle(bundle)[0] is True def test_validate_accepts_partial_files(self): bundle = config_backup.build_bundle({'devices': {'a' * 24: 2}}) assert config_backup.validate_bundle(bundle)[0] is True def test_validate_rejects_empty_files(self): bundle = config_backup.build_bundle({'dtu_config': None, 'devices': {}, 'system_env': ''}) assert config_backup.validate_bundle(bundle)[0] is False