dzxj_dtu/backend/tests/test_config_backup.py

"""运行数据备份/快照工具 (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