diff --git a/backend/tests/test_store_apply.py b/backend/tests/test_store_apply.py new file mode 100644 index 0000000..75bdbb0 --- /dev/null +++ b/backend/tests/test_store_apply.py @@ -0,0 +1,311 @@ +"""Тесты атомарного применения: 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() diff --git a/backend/tests/test_store_audit.py b/backend/tests/test_store_audit.py new file mode 100644 index 0000000..4de1802 --- /dev/null +++ b/backend/tests/test_store_audit.py @@ -0,0 +1,145 @@ +"""Тесты журнала аудита «кто/когда/что/old→new» (issue #20).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from awg_api.store import ( + AUDIT_STATUS_FAILED, + AUDIT_STATUS_OK, + Database, + Entity, + Mode, + Rule, + Store, +) +from awg_api.store import audit + + +def make_db(tmp_path: Path) -> Database: + return Database(tmp_path / "data" / "awg-dash.db") + + +def test_record_and_read_back(tmp_path: Path) -> None: + with make_db(tmp_path) as db: + with db.transaction() as conn: + entry_id = audit.record( + conn, + actor="ui:admin", + action="update", + entity=Entity.RULE.value, + entity_id=7, + old_value={"match": "old.example.com"}, + new_value={"match": "new.example.com"}, + ) + assert entry_id > 0 + entries = audit.list_entries(db.connection) + assert len(entries) == 1 + entry = entries[0] + assert entry.actor == "ui:admin" + assert entry.entity == "rule" + assert entry.entity_id == "7" + assert entry.old_value == {"match": "old.example.com"} + assert entry.new_value == {"match": "new.example.com"} + assert entry.status == AUDIT_STATUS_OK + assert entry.ts.endswith("Z") + assert entry.as_dict()["entity_id"] == "7" + + +def test_actor_is_required(tmp_path: Path) -> None: + with make_db(tmp_path) as db: + with db.transaction() as conn, pytest.raises(ValueError, match="actor"): + audit.record( + conn, + actor="", + action="update", + entity=Entity.RULE.value, + new_value={"x": 1}, + ) + + +def test_old_and_new_can_be_absent(tmp_path: Path) -> None: + with make_db(tmp_path) as db: + with db.transaction() as conn: + audit.record(conn, actor="system", action="noop", entity=Entity.APPLY.value) + entry = audit.last_entry(db.connection) + assert entry is not None + assert entry.old_value is None + assert entry.new_value is None + assert entry.entity_id is None + assert entry.detail is None + + +def test_filters_limit_and_ordering(tmp_path: Path) -> None: + with make_db(tmp_path) as db: + store = Store(db) + store.set_mode(Mode.SPLIT, actor="ui:admin") + store.set_mode(Mode.FULL, actor="ui:admin") + store.upsert_rule(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"), + actor="ui:admin") + + mode_entries = audit.list_entries(db.connection, entity=Entity.MODE.value) + assert len(mode_entries) == 2 + assert mode_entries[0].new_value == {"mode": "full"} # свежие первыми + assert mode_entries[0].old_value == {"mode": "split"} + + by_id = audit.list_entries( + db.connection, entity=Entity.MODE.value, entity_id=1, limit=1 + ) + assert len(by_id) == 1 + + assert audit.count(db.connection) == 3 + assert audit.count(db.connection, entity=Entity.RULE.value) == 1 + assert audit.last_entry(db.connection, entity=Entity.RULE.value) is not None + + +def test_ts_can_be_overridden(tmp_path: Path) -> None: + with make_db(tmp_path) as db: + with db.transaction() as conn: + audit.record( + conn, + actor="system", + action="stage", + entity=Entity.APPLY.value, + ts="2026-01-02T03:04:05.678Z", + ) + entry = audit.last_entry(db.connection) + assert entry is not None and entry.ts == "2026-01-02T03:04:05.678Z" + + +def test_scalar_and_list_payloads_roundtrip(tmp_path: Path) -> None: + with make_db(tmp_path) as db: + with db.transaction() as conn: + audit.record( + conn, + actor="system", + action="apply", + entity=Entity.APPLY.value, + old_value=["a", "b"], + new_value="plain", + status=AUDIT_STATUS_FAILED, + detail="iptables: exit 4", + ) + entry = audit.last_entry(db.connection) + assert entry is not None + assert entry.old_value == ["a", "b"] + assert entry.new_value == "plain" + assert entry.status == AUDIT_STATUS_FAILED + assert entry.detail == "iptables: exit 4" + + +def test_no_audit_row_when_transaction_rolls_back(tmp_path: Path) -> None: + """Мутация и её аудит коммитятся вместе: откат транзакции не оставляет следа.""" + with make_db(tmp_path) as db: + with pytest.raises(RuntimeError), db.transaction() as conn: + audit.record(conn, actor="system", action="update", entity=Entity.RULE.value) + raise RuntimeError("мутация не удалась") + assert audit.count(db.connection) == 0 + + +def test_last_entry_returns_none_on_empty_log(tmp_path: Path) -> None: + with make_db(tmp_path) as db: + assert audit.last_entry(db.connection) is None + assert audit.list_entries(db.connection) == [] diff --git a/backend/tests/test_store_migrations.py b/backend/tests/test_store_migrations.py new file mode 100644 index 0000000..b00bbfb --- /dev/null +++ b/backend/tests/test_store_migrations.py @@ -0,0 +1,153 @@ +"""Тесты схемы, миграций и проверки версии при старте (issue #20).""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from awg_api.store import ( + EXPECTED_TABLES, + LATEST_VERSION, + MIGRATIONS, + Database, + Migration, + SchemaError, + StorageError, + current_version, + migrate, + resolve_db_path, + verify_schema, +) +from awg_api.store.migrations import applied_migrations, table_columns, table_names + + +def test_empty_data_dir_creates_schema(tmp_path: Path) -> None: + """GIVEN пустой /data, WHEN старт, THEN схема v1 создана автоматически.""" + db_path = tmp_path / "data" / "awg-dash.db" + assert not db_path.parent.exists() + + with Database(db_path) as db: + assert db.schema_version == LATEST_VERSION == 1 + assert db.migration_report is not None + assert db.migration_report.applied == (1,) + names = table_names(db.connection) + assert set(EXPECTED_TABLES) <= names + for table, columns in EXPECTED_TABLES.items(): + assert columns <= table_columns(db.connection, table), table + + +def test_wal_mode_is_enabled(tmp_path: Path) -> None: + with Database(tmp_path / "awg-dash.db") as db: + assert db.journal_mode.lower() == "wal" + assert db.is_wal is True + + +def test_seed_rows_present(tmp_path: Path) -> None: + with Database(tmp_path / "awg-dash.db") as db: + row = db.connection.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone() + assert row is not None and row["mode"] == "off" + geoip = db.connection.execute("SELECT cron, provider FROM geoip_schedule WHERE id = 1").fetchone() + assert geoip is not None + assert geoip["provider"] == "ripe" + + +def test_migrate_is_idempotent(tmp_path: Path) -> None: + with Database(tmp_path / "awg-dash.db") as db: + report = db.migrate() + assert report.applied == () + assert report.changed is False + assert report.from_version == report.to_version == LATEST_VERSION + + +def test_applied_migrations_map(tmp_path: Path) -> None: + with Database(tmp_path / "awg-dash.db") as db: + assert applied_migrations(db.connection) == {1: "initial-state-schema"} + + +def test_auto_migrate_false_verifies_existing_schema(tmp_path: Path) -> None: + path = tmp_path / "awg-dash.db" + with Database(path): + pass + with Database(path, auto_migrate=False) as db: + assert db.migration_report is None + assert current_version(db.connection) == LATEST_VERSION + + +def test_auto_migrate_false_on_unmigrated_db_raises(tmp_path: Path) -> None: + conn = sqlite3.connect(str(tmp_path / "empty.db")) + try: + with pytest.raises(SchemaError, match="не мигрирована"): + verify_schema(conn) + finally: + conn.close() + + +def test_verify_schema_detects_missing_table(tmp_path: Path) -> None: + with Database(tmp_path / "awg-dash.db") as db: + db.connection.execute("DROP TABLE routing_rules") + with pytest.raises(SchemaError, match="routing_rules"): + db.verify_schema() + + +def test_verify_schema_detects_missing_column(tmp_path: Path) -> None: + with Database(tmp_path / "awg-dash.db") as db: + db.connection.execute("ALTER TABLE devices DROP COLUMN address") + with pytest.raises(SchemaError, match="address"): + db.verify_schema() + + +def test_verify_schema_rejects_newer_schema(tmp_path: Path) -> None: + with Database(tmp_path / "awg-dash.db") as db: + db.connection.execute( + "INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)", + (99, "from-the-future", "2099-01-01T00:00:00.000Z"), + ) + with pytest.raises(SchemaError, match="новее"): + db.verify_schema() + with pytest.raises(SchemaError, match="новее"): + db.migrate() + + +def test_failed_migration_leaves_schema_untouched(tmp_path: Path) -> None: + broken = MIGRATIONS + ( + Migration(version=2, name="broken", statements=("CREATE TABLE t2 (id INTEGER)", "NOT SQL AT ALL")), + ) + with Database(tmp_path / "awg-dash.db") as db: + with pytest.raises(StorageError, match="миграция v2"): + migrate(db.connection, migrations=broken) + assert current_version(db.connection) == 1 + assert "t2" not in table_names(db.connection) + + +def test_migrate_rejects_downgrade(tmp_path: Path) -> None: + with Database(tmp_path / "awg-dash.db") as db: + db.connection.execute( + "INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, ?)", + (2, "newer", "2099-01-01T00:00:00.000Z"), + ) + with pytest.raises(SchemaError, match="новее"): + migrate(db.connection, migrations=MIGRATIONS) + + +def test_resolve_db_path_defaults_and_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("AWG_DASH_DB", raising=False) + assert resolve_db_path() == Path("/data/awg-dash.db") + monkeypatch.setenv("AWG_DASH_DB", str(tmp_path / "custom.db")) + assert resolve_db_path() == tmp_path / "custom.db" + assert resolve_db_path(tmp_path / "explicit.db") == tmp_path / "explicit.db" + + +def test_database_uses_env_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("AWG_DASH_DB", str(tmp_path / "nested" / "env.db")) + with Database() as db: + assert db.path == tmp_path / "nested" / "env.db" + assert db.path.exists() + + +def test_unwritable_path_raises_storage_error(tmp_path: Path) -> None: + blocker = tmp_path / "blocked" + blocker.write_text("not a directory") + with pytest.raises(StorageError, match="каталог"): + Database(blocker / "awg-dash.db")