312 lines
13 KiB
Python
312 lines
13 KiB
Python
"""Тесты атомарного применения: 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()
|