"""Персистентность: рестарт «контейнера» сохраняет правила, конфиг и аудит (issue #20).""" from __future__ import annotations from pathlib import Path import pytest from awg_api.store import ( ConfigMeta, Database, DesiredState, Device, GeoIpSchedule, Mode, Rule, StorageError, Store, ValidationError, open_storage, rollback_to, ) from awg_api.store import validation def test_state_survives_container_restart(tmp_path: Path) -> None: """GIVEN рестарт контейнера, WHEN он завершён, THEN правила и конфиг на месте.""" db_path = tmp_path / "data" / "awg-dash.db" with Database(db_path) as first: store = Store(first) store.set_mode(Mode.SPLIT, actor="ui:admin") store.set_geoip_schedule(cron="0 3 * * *", actor="ui:admin") store.upsert_rule( Rule(name="yt", kind="domain", match="youtube.com", target="tun0", priority=10), actor="ui:admin", ) store.upsert_rule( Rule(name="lan", kind="cidr", match="10.0.0.0/8", target="direct", priority=200), actor="ui:admin", ) store.upsert_config( ConfigMeta(name="peer1", path="/config/peer1.conf", checksum="deadbeef", size_bytes=1024), actor="ui:admin", ) store.register_device(Device(mac="aa:bb:cc:dd:ee:ff", name="notebook"), actor="ui:admin") audit_before = store.counts()["audit"] assert audit_before == 6 # «рестарт контейнера»: новое соединение к тому же /data with Database(db_path) as second: store = Store(second) assert second.is_wal is True assert store.get_mode() == Mode.SPLIT assert store.get_geoip_schedule().cron == "0 3 * * *" assert [(rule.name, rule.target) for rule in store.list_rules()] == [ ("yt", "tun0"), ("lan", "direct"), ] config = store.get_config("peer1") assert config is not None assert config.checksum == "deadbeef" assert store.get_device("aa:bb:cc:dd:ee:ff") is not None assert store.counts()["audit"] == audit_before assert second.migration_report is not None # миграции уже применены, повтор не нужен assert second.migration_report.applied == () def test_wal_file_next_to_database(tmp_path: Path) -> None: db_path = tmp_path / "data" / "awg-dash.db" with Database(db_path) as db: Store(db).set_mode(Mode.FULL, actor="system") assert db_path.with_name("awg-dash.db-wal").exists() assert db_path.with_name("awg-dash.db-shm").exists() def test_open_storage_builds_full_stack(tmp_path: Path) -> None: with open_storage(tmp_path / "data" / "awg-dash.db", actor="api:token") as storage: storage.store.set_mode(Mode.SPLIT, actor="api:token") plan = storage.engine.stage( DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule()), reason="через storage", ) result = storage.engine.apply(plan) assert result.ok is True assert storage.store.counts()["snapshots"] == 1 assert storage.db.schema_version == 1 def test_rollback_helper_and_missing_snapshot(tmp_path: Path) -> None: with open_storage(tmp_path / "data" / "awg-dash.db") as storage: storage.store.set_mode(Mode.OFF, actor="system") plan = storage.engine.stage(DesiredState(mode=Mode.FULL)) storage.engine.apply(plan) assert storage.store.get_mode() == Mode.FULL snapshot = rollback_to(storage.engine, plan.plan_id, reason="ручной откат") assert snapshot.payload["mode"] == "off" assert storage.store.get_mode() == Mode.OFF with pytest.raises(StorageError, match="не найден"): storage.engine.rollback(999999) def test_transactions_are_isolated_per_connection(tmp_path: Path) -> None: db_path = tmp_path / "data" / "awg-dash.db" with Database(db_path) as writer, Database(db_path) as reader: Store(writer).set_mode(Mode.SPLIT, actor="system") assert Store(reader).get_mode() == Mode.SPLIT with pytest.raises(RuntimeError, match="откат"), writer.transaction() as conn: conn.execute("UPDATE routing_mode SET mode = 'full' WHERE id = 1") raise RuntimeError("откат") assert Store(reader).get_mode() == Mode.SPLIT # -- валидация (чистые функции) ------------------------------------------------- @pytest.mark.parametrize( ("raw", "expected"), [ ("AA:BB:CC:DD:EE:FF", "aa:bb:cc:dd:ee:ff"), ("aa-bb-cc-dd-ee-ff", "aa:bb:cc:dd:ee:ff"), ("aabbccddeeff", "aa:bb:cc:dd:ee:ff"), ], ) def test_normalize_mac(raw: str, expected: str) -> None: assert validation.normalize_mac(raw) == expected def test_normalize_mac_rejects_garbage() -> None: with pytest.raises(ValidationError, match="MAC"): validation.normalize_mac("not-a-mac") @pytest.mark.parametrize("value", ["ok", "peer_1", "rule-2", "a.b:c"]) def test_validate_name_accepts(value: str) -> None: assert validation.validate_name("name", f" {value} ") == value @pytest.mark.parametrize("value", ["", "UPPER", "-lead", "with space", "юникод", "x" * 64]) def test_validate_name_rejects(value: str) -> None: with pytest.raises(ValidationError): validation.validate_name("name", value) @pytest.mark.parametrize("value", ["/config/peer1.conf", "/etc/awg-dash/rules.conf"]) def test_validate_config_path_accepts(value: str) -> None: assert validation.validate_config_path(value) == value @pytest.mark.parametrize("value", ["relative.conf", "/config/../../etc/shadow", ""]) def test_validate_config_path_rejects(value: str) -> None: with pytest.raises(ValidationError): validation.validate_config_path(value) @pytest.mark.parametrize(("value", "expected"), [("RIPE", "ripe"), ("maxmind", "maxmind")]) def test_validate_provider(value: str, expected: str) -> None: assert validation.validate_provider(value) == expected @pytest.mark.parametrize("value", ["tunnel", "direct", "block", "default"]) def test_validate_device_policy_accepts(value: str) -> None: assert validation.validate_device_policy(value) == value def test_validate_device_policy_rejects() -> None: with pytest.raises(ValidationError, match="policy"): validation.validate_device_policy("root") @pytest.mark.parametrize("value", ["*.example.com", "youtube.com", "a.b.c.example.co.uk"]) def test_validate_match_domain(value: str) -> None: assert validation.validate_match("domain", value) == value @pytest.mark.parametrize("value", ["10.0.0.0/8", "2001:db8::/32", "192.168.1.7"]) def test_validate_match_cidr(value: str) -> None: assert validation.validate_match("cidr", value) == value def test_validate_match_ipset_and_empty() -> None: assert validation.validate_match("ipset", "awg_ru") == "awg_ru" with pytest.raises(ValidationError, match="пустым"): validation.validate_match("domain", " ")