"""Тесты атомарного применения: validate -> stage -> apply -> rollback (issue #20).""" from __future__ import annotations from collections.abc import Mapping from pathlib import Path from typing import Any import pytest from awg_api.store import ( AUDIT_STATUS_FAILED, AUDIT_STATUS_OK, AUDIT_STATUS_STAGED, ApplyEngine, ApplyError, Database, DesiredState, Entity, GeoIpSchedule, Mode, NullApplier, Rule, RuleException, Store, ValidationError, capture_state, ) class MockIptables: """Мок внешнего применителя: умеет «падать» как реальный iptables.""" def __init__(self, *, fail_apply: bool = False, fail_revert: bool = False) -> None: self.applied: list[dict[str, Any]] = [] self.reverted: list[dict[str, Any]] = [] self.fail_apply = fail_apply self.fail_revert = fail_revert def apply(self, payload: Mapping[str, Any]) -> None: if self.fail_apply: self.applied.append(dict(payload)) raise RuntimeError("mock iptables: exit 4") self.applied.append(dict(payload)) def revert(self, payload: Mapping[str, Any]) -> None: if self.fail_revert: raise RuntimeError("mock iptables: revert failed") self.reverted.append(dict(payload)) def make_engine( tmp_path: Path, applier: MockIptables | None = None ) -> tuple[Database, Store, ApplyEngine, MockIptables]: db = Database(tmp_path / "data" / "awg-dash.db") store = Store(db) mock = applier if applier is not None else MockIptables() return db, store, ApplyEngine(store, mock, actor="ui:admin"), mock def desired(*rules: Rule, mode: Mode = Mode.SPLIT) -> DesiredState: return DesiredState(mode=mode, rules=rules, geoip=GeoIpSchedule()) # -- validate ------------------------------------------------------------------ def test_validate_accepts_good_state(tmp_path: Path) -> None: db, _store, engine, _mock = make_engine(tmp_path) with db: state = desired(Rule(name="yt", kind="domain", match="youtube.com", target="tun0")) assert engine.validate(state) is state def test_validate_rejects_bad_state(tmp_path: Path) -> None: db, _store, engine, _mock = make_engine(tmp_path) with db: cases: list[DesiredState] = [ DesiredState(mode="turbo"), # type: ignore[arg-type] desired(Rule(name="dup", kind="domain", match="a.example.com", target="tun0"), Rule(name="dup", kind="domain", match="b.example.com", target="tun0")), desired(Rule(name="x", kind="cidr", match="10.0.0.0/99", target="tun0")), DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(cron="всякий раз")), DesiredState( mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(), exceptions=(RuleException(match="!!!"),), ), DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(provider="!")), ] for case in cases: with pytest.raises(ValidationError): engine.validate(case) def test_validate_does_not_touch_database(tmp_path: Path) -> None: db, store, engine, _mock = make_engine(tmp_path) with db: with pytest.raises(ValidationError): engine.apply_payload({"mode": "split", "rules": [{"name": "x", "kind": "cidr", "match": "bad", "target": "tun0"}]}) assert store.counts() == { "rules": 0, "exceptions": 0, "configs": 0, "devices": 0, "audit": 0, "snapshots": 0, } # -- stage --------------------------------------------------------------------- def test_stage_saves_snapshot_and_staged_audit(tmp_path: Path) -> None: db, store, engine, _mock = make_engine(tmp_path) with db: store.upsert_rule(Rule(name="old", kind="domain", match="old.example.com", target="tun0"), actor="system") store.set_mode(Mode.OFF, actor="system") before = capture_state(db.connection) plan = engine.stage(desired(Rule(name="new", kind="domain", match="new.example.com", target="tun0")), reason="тест") assert plan.previous == before assert plan.desired.mode == Mode.SPLIT # состояние ещё не изменено assert capture_state(db.connection) == before assert [rule.name for rule in store.list_rules()] == ["old"] snapshots = store.counts()["snapshots"] assert snapshots == 1 staged = store.list_audit(entity=Entity.APPLY.value)[0] assert staged.status == AUDIT_STATUS_STAGED assert staged.action == "stage" assert staged.detail == "тест" assert staged.old_value == before def test_stage_rejects_invalid_state_without_snapshot(tmp_path: Path) -> None: db, store, engine, _mock = make_engine(tmp_path) with db: with pytest.raises(ValidationError): engine.stage(DesiredState(mode=Mode.SPLIT, rules=(), geoip=GeoIpSchedule(cron="bad"))) assert store.counts()["snapshots"] == 0 # -- apply --------------------------------------------------------------------- def test_apply_writes_state_and_audit(tmp_path: Path) -> None: db, store, engine, mock = make_engine(tmp_path) with db: plan = engine.stage(desired(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"), mode=Mode.FULL), reason="включение") result = engine.apply(plan) assert result.ok is True assert result.plan_id == plan.plan_id assert result.audit_id > 0 assert store.get_mode() == Mode.FULL assert [rule.name for rule in store.list_rules()] == ["yt"] assert mock.applied == [plan.desired.to_payload()] entry = store.list_audit(entity=Entity.APPLY.value)[0] assert entry.status == AUDIT_STATUS_OK assert entry.action == "apply" assert entry.old_value == plan.previous assert entry.new_value == plan.desired.to_payload() def test_failed_apply_restores_previous_state(tmp_path: Path) -> None: """GIVEN ошибочный apply (мок падения iptables), THEN состояние прежнее + ApplyError.""" db, store, engine, mock = make_engine(tmp_path, MockIptables(fail_apply=True)) with db: store.upsert_rule(Rule(name="keep", kind="domain", match="keep.example.com", target="tun0"), actor="system") store.set_mode(Mode.SPLIT, actor="system") store.set_geoip_schedule(cron="0 5 * * *", actor="system") before = capture_state(db.connection) plan = engine.stage(desired(Rule(name="new", kind="domain", match="new.example.com", target="tun0"), mode=Mode.FULL), reason="сломается") with pytest.raises(ApplyError) as excinfo: engine.apply(plan) error = excinfo.value assert error.restored is True assert error.audit_id is not None and error.audit_id > 0 # состояние, а не «частично применённая конфигурация» assert capture_state(db.connection) == before assert [rule.name for rule in store.list_rules()] == ["keep"] assert store.get_mode() == Mode.SPLIT assert store.get_geoip_schedule().cron == "0 5 * * *" assert mock.reverted == [before] failed = store.list_audit(entity=Entity.APPLY.value)[0] assert failed.status == AUDIT_STATUS_FAILED assert failed.id == error.audit_id assert failed.new_value == before assert "mock iptables" in (failed.detail or "") def test_failed_apply_with_failing_revert_still_restores_db(tmp_path: Path) -> None: db, store, engine, _mock = make_engine( tmp_path, MockIptables(fail_apply=True, fail_revert=True) ) with db: before = capture_state(db.connection) plan = engine.stage(desired(Rule(name="x", kind="domain", match="x.example.com", target="tun0"))) with pytest.raises(ApplyError) as excinfo: engine.apply(plan) assert excinfo.value.restored is True assert capture_state(db.connection) == before def test_apply_can_succeed_after_failed_apply(tmp_path: Path) -> None: db, store, engine, mock = make_engine(tmp_path, MockIptables(fail_apply=True)) with db: bad = engine.stage(desired(Rule(name="bad", kind="domain", match="bad.example.com", target="tun0"))) with pytest.raises(ApplyError): engine.apply(bad) mock.fail_apply = False good_rule = Rule(name="good", kind="domain", match="good.example.com", target="tun0") result = engine.apply(engine.stage(desired(good_rule, mode=Mode.FULL))) assert result.ok is True assert [rule.name for rule in store.list_rules()] == ["good"] assert store.get_mode() == Mode.FULL # -- rollback ------------------------------------------------------------------ def test_rollback_returns_previous_state(tmp_path: Path) -> None: db, store, engine, _mock = make_engine(tmp_path) with db: store.upsert_rule(Rule(name="old", kind="domain", match="old.example.com", target="tun0"), actor="ui:admin") before = capture_state(db.connection) result = engine.apply( engine.stage(desired(Rule(name="new", kind="domain", match="new.example.com", target="tun0"), mode=Mode.FULL), reason="применение"), ) assert result.ok is True snapshot = engine.rollback(result.plan_id, reason="откат оператором") assert snapshot.id == result.plan_id assert snapshot.payload == before assert capture_state(db.connection) == before assert [rule.name for rule in store.list_rules()] == ["old"] assert store.get_mode() == Mode.OFF rollback_entry = store.list_audit(entity=Entity.APPLY.value)[0] assert rollback_entry.action == "rollback" assert rollback_entry.detail == "откат оператором" assert rollback_entry.new_value == before def test_apply_payload_sugar_and_null_applier(tmp_path: Path) -> None: db, store, engine, _mock = make_engine(tmp_path) with db: result = engine.apply_payload( { "mode": "split", "rules": [{"name": "yt", "kind": "domain", "match": "youtube.com", "target": "tun0"}], "geoip": {"enabled": False, "cron": "0 3 * * *", "provider": "ripe"}, "exceptions": [{"match": "bank.example.com", "reason": "банк"}], }, reason="api: применить", ) assert result.ok is True assert store.get_mode() == Mode.SPLIT assert store.get_geoip_schedule().enabled is False assert store.list_exceptions()[0].match == "bank.example.com" engine_null = ApplyEngine(store, NullApplier(), actor="system") assert engine_null.apply_payload({"mode": "off"}).ok is True assert store.get_mode() == Mode.OFF def test_desired_state_payload_roundtrip() -> None: state = DesiredState( mode=Mode.FULL, rules=(Rule(name="yt", kind="domain", match="youtube.com", target="tun0", priority=7),), geoip=GeoIpSchedule(enabled=False, cron="0 1 * * *", provider="maxmind"), exceptions=(RuleException(match="bank.example.com", reason="банк", enabled=False),), ) restored = DesiredState.from_payload(state.to_payload()) assert restored.mode == Mode.FULL assert restored.rules == state.rules assert restored.geoip == state.geoip assert restored.exceptions == state.exceptions @pytest.mark.parametrize( "payload", [ {"rules": "не список"}, {"rules": [{"kind": "domain", "match": "a.example.com", "target": "tun0"}]}, {"rules": ["строка"]}, {"exceptions": "не список"}, {"exceptions": [{"reason": "нет match"}]}, {"geoip": "не объект"}, {"mode": "turbo"}, ], ) def test_desired_state_from_payload_rejects_bad_payload(payload: dict[str, Any]) -> None: with pytest.raises(ValidationError): DesiredState.from_payload(payload) def test_apply_engine_requires_store(tmp_path: Path) -> None: db, _store, engine, _mock = make_engine(tmp_path) with db: assert engine.validate(desired()) == desired()