From 7e10ffaf8ad436f10b4195fddc4120ce576b2c7d Mon Sep 17 00:00:00 2001 From: embeddevio Date: Wed, 21 Oct 2026 23:45:30 +0300 Subject: [PATCH] =?UTF-8?q?test(store):=20=D1=80=D0=B5=D0=BF=D0=BE=D0=B7?= =?UTF-8?q?=D0=B8=D1=82=D0=BE=D1=80=D0=B8=D0=B9=20(=D0=B0=D1=83=D0=B4?= =?UTF-8?q?=D0=B8=D1=82=20=D0=BF=D0=BE=20=D0=B2=D1=81=D0=B5=D0=BC=20=D1=81?= =?UTF-8?q?=D1=83=D1=89=D0=BD=D0=BE=D1=81=D1=82=D1=8F=D0=BC)=20=D0=B8=20?= =?UTF-8?q?=D0=BF=D0=B5=D1=80=D1=81=D0=B8=D1=81=D1=82=D0=B5=D0=BD=D1=82?= =?UTF-8?q?=D0=BD=D0=BE=D1=81=D1=82=D1=8C=20=D0=BF=D0=BE=D1=81=D0=BB=D0=B5?= =?UTF-8?q?=20=D1=80=D0=B5=D1=81=D1=82=D0=B0=D1=80=D1=82=D0=B0=20(#20)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/tests/test_store_persistence.py | 184 ++++++++++++++ backend/tests/test_store_repository.py | 312 ++++++++++++++++++++++++ 2 files changed, 496 insertions(+) create mode 100644 backend/tests/test_store_persistence.py create mode 100644 backend/tests/test_store_repository.py diff --git a/backend/tests/test_store_persistence.py b/backend/tests/test_store_persistence.py new file mode 100644 index 0000000..46d0edb --- /dev/null +++ b/backend/tests/test_store_persistence.py @@ -0,0 +1,184 @@ +"""Персистентность: рестарт «контейнера» сохраняет правила, конфиг и аудит (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", " ") diff --git a/backend/tests/test_store_repository.py b/backend/tests/test_store_repository.py new file mode 100644 index 0000000..ff54771 --- /dev/null +++ b/backend/tests/test_store_repository.py @@ -0,0 +1,312 @@ +"""Тесты репозитория: CRUD-мутации с аудитом «old→new» (issue #20).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from awg_api.store import ( + ConfigMeta, + Database, + Device, + Entity, + GeoIpSchedule, + Mode, + Rule, + RuleException, + StorageError, + Store, + ValidationError, +) +from awg_api.store import audit +from awg_api.store.validation import validate_priority + + +def make_store(tmp_path: Path) -> tuple[Database, Store]: + db = Database(tmp_path / "data" / "awg-dash.db") + return db, Store(db) + + +# -- правила ----------------------------------------------------------------- +def test_upsert_rule_creates_with_audit(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + saved = store.upsert_rule( + Rule(name="yt", kind="domain", match="youtube.com", target="tun0", priority=10), + actor="ui:admin", + ) + assert saved.id is not None and saved.created_at == saved.updated_at + assert store.get_rule(saved.id) == saved + assert store.get_rule_by_name("yt") == saved + entries = store.list_audit(entity=Entity.RULE.value) + assert len(entries) == 1 + assert entries[0].action == "create" + assert entries[0].old_value is None + assert entries[0].new_value is not None + assert entries[0].new_value["name"] == "yt" + + +def test_upsert_rule_updates_by_name_and_records_old_new(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + first = store.upsert_rule(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"), + actor="ui:admin") + second = store.upsert_rule(Rule(name="yt", kind="domain", match="youtu.be", target="tun1"), + actor="ui:admin") + assert second.id == first.id + assert second.match == "youtu.be" and second.target == "tun1" + assert store.counts()["rules"] == 1 + entries = store.list_audit(entity=Entity.RULE.value) + assert [entry.action for entry in entries] == ["update", "create"] + assert entries[0].old_value["match"] == "youtube.com" + assert entries[0].new_value["match"] == "youtu.be" + assert entries[0].old_value["target"] == "tun0" + assert entries[0].new_value["target"] == "tun1" + + +def test_upsert_rule_by_id(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + saved = store.upsert_rule(Rule(name="a", kind="cidr", match="10.0.0.0/8", target="tun0"), + actor="system") + updated = store.upsert_rule( + Rule(id=saved.id, name="a", kind="cidr", match="10.1.0.0/16", target="tun0", enabled=False), + actor="system", + ) + assert updated.id == saved.id + assert updated.match == "10.1.0.0/16" + assert updated.enabled is False + assert store.list_rules(enabled_only=True) == [] + + +def test_rule_validation_rejects_bad_input(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + cases = [ + Rule(name="Bad Name", kind="domain", match="youtube.com", target="tun0"), + Rule(name="ok", kind="wrong", match="youtube.com", target="tun0"), + Rule(name="ok2", kind="domain", match="not a domain", target="tun0"), + Rule(name="ok3", kind="cidr", match="10.0.0.0/99", target="tun0"), + Rule(name="ok4", kind="domain", match="youtube.com", target="bad target!"), + Rule(name="ok5", kind="domain", match="youtube.com", target="tun0", priority=-1), + ] + for rule in cases: + with pytest.raises(ValidationError): + store.upsert_rule(rule, actor="system") + assert store.counts()["rules"] == 0 + assert audit.count(db.connection) == 0 + + +def test_priority_must_be_int() -> None: + with pytest.raises(ValidationError, match="целое"): + validate_priority(True) # bool не является допустимым приоритетом + with pytest.raises(ValidationError, match="диапазона"): + validate_priority(70000) + + +def test_delete_rule(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + saved = store.upsert_rule(Rule(name="yt", kind="domain", match="youtube.com", target="tun0"), + actor="ui:admin") + assert saved.id is not None + assert store.delete_rule(saved.id, actor="ui:admin") is True + assert store.delete_rule(saved.id, actor="ui:admin") is False + assert store.get_rule(saved.id) is None + entries = store.list_audit(entity=Entity.RULE.value) + assert [entry.action for entry in entries] == ["delete", "create"] + assert entries[0].new_value is None + assert entries[0].old_value["name"] == "yt" + + +def test_rules_sorted_by_priority(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + store.upsert_rule(Rule(name="b", kind="domain", match="b.example.com", target="tun0", + priority=200), actor="system") + store.upsert_rule(Rule(name="a", kind="domain", match="a.example.com", target="tun0", + priority=5), actor="system") + assert [rule.name for rule in store.list_rules()] == ["a", "b"] + + +# -- исключения --------------------------------------------------------------- +def test_exception_upsert_update_delete(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + created = store.upsert_exception(RuleException(match="bank.example.com", reason="банк"), + actor="ui:admin") + assert created.id is not None + assert store.get_exception("bank.example.com") == created + + updated = store.upsert_exception(RuleException(match="bank.example.com", reason="банк2"), + actor="ui:admin") + assert updated.id == created.id and updated.reason == "банк2" + assert store.counts()["exceptions"] == 1 + + assert store.list_exceptions() == [updated] + assert store.delete_exception(updated.id or 0, actor="ui:admin") is True + assert store.delete_exception(updated.id or 0, actor="ui:admin") is False + + actions = [entry.action for entry in store.list_audit(entity=Entity.EXCEPTION.value)] + assert actions == ["delete", "update", "create"] + update_entry = store.list_audit(entity=Entity.EXCEPTION.value)[1] + assert update_entry.old_value["reason"] == "банк" + assert update_entry.new_value["reason"] == "банк2" + + +def test_exception_accepts_cidr_and_rejects_garbage(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + assert store.upsert_exception(RuleException(match="192.168.1.0/24"), actor="system").match == ( + "192.168.1.0/24" + ) + assert store.upsert_exception(RuleException(match="10.0.0.1"), actor="system").match == "10.0.0.1" + with pytest.raises(ValidationError, match="домен"): + store.upsert_exception(RuleException(match="!!!"), actor="system") + with pytest.raises(ValidationError, match="пустым"): + store.upsert_exception(RuleException(match=" "), actor="system") + + +# -- режим и GeoIP ------------------------------------------------------------- +def test_set_mode_defaults_off_and_audits(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + assert store.get_mode() == Mode.OFF + assert store.set_mode("split", actor="ui:admin") == Mode.SPLIT + assert store.get_mode() == Mode.SPLIT + assert store.set_mode(Mode("full"), actor="ui:admin") == Mode.FULL + entries = store.list_audit(entity=Entity.MODE.value) + assert len(entries) == 2 + assert entries[0].old_value == {"mode": "split"} + assert entries[0].new_value == {"mode": "full"} + assert entries[0].actor == "ui:admin" + + +def test_set_mode_rejects_unknown(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + with pytest.raises(ValidationError, match="mode"): + store.set_mode("turbo", actor="ui:admin") + assert store.get_mode() == Mode.OFF + + +def test_geoip_schedule_defaults_and_partial_update(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + default = store.get_geoip_schedule() + assert default == GeoIpSchedule(enabled=True, cron="0 4 * * *", provider="ripe", + updated_at=default.updated_at) + updated = store.set_geoip_schedule(cron="*/30 * * * *", actor="ui:admin") + assert updated.cron == "*/30 * * * *" + assert updated.provider == "ripe" and updated.enabled is True + + disabled = store.set_geoip_schedule(enabled=False, provider="MAXMIND", actor="ui:admin") + assert disabled.enabled is False and disabled.provider == "maxmind" + assert store.get_geoip_schedule() == disabled + + with pytest.raises(ValidationError, match="cron"): + store.set_geoip_schedule(cron="ежечасно", actor="ui:admin") + with pytest.raises(ValidationError, match="provider"): + store.set_geoip_schedule(provider="!", actor="ui:admin") + + entries = store.list_audit(entity=Entity.GEOIP_SCHEDULE.value) + assert [entry.action for entry in entries] == ["set_geoip_schedule"] * 2 + assert entries[0].old_value["provider"] == "ripe" + assert entries[0].new_value["provider"] == "maxmind" + + +# -- конфиги ------------------------------------------------------------------ +def test_config_upsert_and_mark_applied(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + created = store.upsert_config( + ConfigMeta(name="client-peer1", path="/config/peer1.conf", version=1, checksum="abc", + size_bytes=512), actor="ui:admin", + ) + assert created.id is not None + assert store.get_config("client-peer1") == created + + updated = store.upsert_config( + ConfigMeta(name="client-peer1", path="/config/peer1.conf", version=2, checksum="def", + size_bytes=640, applied=True), actor="ui:admin", + ) + assert updated.version == 2 and updated.applied is True + assert store.counts()["configs"] == 1 + + marked = store.mark_config_applied("client-peer1", applied=False, actor="ui:admin") + assert marked.applied is False + + entries = store.list_audit(entity=Entity.CONFIG.value) + assert [entry.action for entry in entries] == ["set_applied", "update", "create"] + assert entries[0].old_value["applied"] is True + assert entries[0].new_value["applied"] is False + assert entries[1].old_value["version"] == 1 + assert entries[1].new_value["version"] == 2 + + +def test_config_validation_and_missing(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + with pytest.raises(ValidationError, match="path"): + store.upsert_config(ConfigMeta(name="x", path="relative.conf"), actor="system") + with pytest.raises(ValidationError, match="path"): + store.upsert_config(ConfigMeta(name="x", path="/config/../etc/passwd"), actor="system") + with pytest.raises(ValidationError, match="name"): + store.upsert_config(ConfigMeta(name="Bad Name", path="/config/x.conf"), actor="system") + with pytest.raises(StorageError, match="не найден"): + store.mark_config_applied("ghost", applied=True, actor="system") + + +# -- устройства ----------------------------------------------------------------- +def test_device_register_normalizes_mac_and_audits(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + created = store.register_device( + Device(mac="AA-BB-CC-DD-EE-FF", name="notebook", address="192.168.1.50"), + actor="ui:admin", + ) + assert created.mac == "aa:bb:cc:dd:ee:ff" + assert store.get_device("AA:BB:CC:DD:EE:FF") == created + + updated = store.register_device( + Device(mac="aabbccddeeff", name="notebook", policy="direct", last_seen="2026-01-01T00:00:00.000Z"), + actor="ui:admin", + ) + assert updated.id == created.id and updated.policy == "direct" + assert store.counts()["devices"] == 1 + + assert store.delete_device("aa:bb:cc:dd:ee:ff", actor="ui:admin") is True + assert store.delete_device("aa:bb:cc:dd:ee:ff", actor="ui:admin") is False + assert store.list_devices() == [] + + actions = [entry.action for entry in store.list_audit(entity=Entity.DEVICE.value)] + assert actions == ["delete", "update", "create"] + update_entry = store.list_audit(entity=Entity.DEVICE.value)[1] + assert update_entry.old_value["policy"] == "default" + assert update_entry.new_value["policy"] == "direct" + + +def test_device_validation(tmp_path: Path) -> None: + db, store = make_store(tmp_path) + with db: + with pytest.raises(ValidationError, match="MAC"): + store.register_device(Device(mac="zz:zz", name="x"), actor="system") + with pytest.raises(ValidationError, match="policy"): + store.register_device(Device(mac="aa:bb:cc:dd:ee:ff", name="x", policy="кибератака"), + actor="system") + with pytest.raises(ValidationError, match="name"): + store.register_device(Device(mac="aa:bb:cc:dd:ee:ff", name="Not A Name"), actor="system") + + +def test_audit_actor_recorded_per_mutation(tmp_path: Path) -> None: + """GIVEN изменение через сервисный слой, THEN в audit_log есть actor и old/new.""" + db, store = make_store(tmp_path) + with db: + store.set_mode(Mode.SPLIT, actor="api:token-42") + entry = store.list_audit(entity=Entity.MODE.value)[0] + assert entry.actor == "api:token-42" + assert entry.ts.endswith("Z") + assert entry.old_value == {"mode": "off"} + assert entry.new_value == {"mode": "split"} + assert entry.status == "ok"