From c4661c9db97ce4ed2c2c3471af06b18b69d99162 Mon Sep 17 00:00:00 2001 From: embeddevio Date: Wed, 28 Oct 2026 21:12:35 +0300 Subject: [PATCH] =?UTF-8?q?test(#20):=20=D1=80=D0=B5=D0=B3=D1=80=D0=B5?= =?UTF-8?q?=D1=81=D1=81=D0=B8=D0=BE=D0=BD=D0=BD=D1=8B=D0=B5=20=D1=82=D0=B5?= =?UTF-8?q?=D1=81=D1=82=D1=8B=20=D0=BF=D0=BE=D1=82=D0=BE=D0=BA=D0=BE=D0=B1?= =?UTF-8?q?=D0=B5=D0=B7=D0=BE=D0=BF=D0=B0=D1=81=D0=BD=D0=BE=D1=81=D1=82?= =?UTF-8?q?=D0=B8=20(#54)=20=D0=B8=20BaseException/TOCTOU=20(#55)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Новый файл backend/tests/test_store_concurrency.py: * N потоков × M upsert с чтениями → 0 ошибок, rules == audit_rows == N*M; * чтение во время открытой транзакции другого потока не видит незафиксированных строк (ждёт коммита); * чтение во время падающего apply не видит состояния, которое будет откачено (AC «не частично применённая конфигурация»); * BaseException (KeyboardInterrupt/SystemExit/CancelledError) из Applier → revert вызван, failed в audit_log, БД восстановлена; * два конкурентных set_geoip_schedule → связная цепочка old→new в журнале. На коде до правки все 7 тестов падают; после — проходят. Refs #20, Refs #54, Refs #55 --- backend/tests/test_store_concurrency.py | 308 ++++++++++++++++++++++++ 1 file changed, 308 insertions(+) create mode 100644 backend/tests/test_store_concurrency.py diff --git a/backend/tests/test_store_concurrency.py b/backend/tests/test_store_concurrency.py new file mode 100644 index 0000000..1b879cf --- /dev/null +++ b/backend/tests/test_store_concurrency.py @@ -0,0 +1,308 @@ +"""Тесты потокобезопасности слоя хранения (#54, D-1/D-2 из #55; issue #20). + +Контракт (задокументирован в ``backend/awg_api/store/README.md``): одно +соединение SQLite, все операции — чтения и записи — сериализуются одним +``RLock``. Читатель никогда не видит незафиксированных строк и состояния, +которое затем будет откачено, а конкурентные мутации дают согласованный +журнал аудита. +""" + +from __future__ import annotations + +import asyncio +import threading +import time +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +import pytest + +from awg_api.store import ( + AUDIT_STATUS_FAILED, + ApplyEngine, + ApplyError, + Database, + DesiredState, + Entity, + Mode, + Rule, + Store, + capture_state, +) + +THREADS = 4 +PER_THREAD = 25 +JOIN_TIMEOUT = 30.0 +# окно наблюдения: за это время незаблокированное чтение гарантированно успело бы вернуться +OBSERVE_SECONDS = 0.2 + +RULE_INSERT = ( + "INSERT INTO routing_rules (name, kind, match, target, priority, enabled," + " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)" +) + + +def open_store(tmp_path: Path) -> tuple[Database, Store]: + db = Database(tmp_path / "data" / "awg-dash.db") + return db, Store(db) + + +def rule(tid: int, idx: int) -> Rule: + return Rule( + name=f"r{tid}x{idx}", + kind="domain", + match=f"t{tid}r{idx}.example.com", + target="tun0", + ) + + +def run_threads(workers: list[Any]) -> list[threading.Thread]: + threads = [threading.Thread(target=worker) for worker in workers] + for thread in threads: + thread.start() + for thread in threads: + thread.join(JOIN_TIMEOUT) + assert not thread.is_alive(), f"поток {thread.name} не завершился" + return threads + + +# -- #54: конкурентные мутации вместе с чтениями --------------------------------- +def test_concurrent_upserts_and_reads_are_consistent(tmp_path: Path) -> None: + """N потоков × M upsert с чтениями: без ошибок, rules == audit == N*M.""" + db, store = open_store(tmp_path) + with db: + errors: list[str] = [] + barrier = threading.Barrier(THREADS) + + def worker(tid: int) -> None: + try: + barrier.wait(timeout=JOIN_TIMEOUT) + for idx in range(PER_THREAD): + store.upsert_rule(rule(tid, idx), actor=f"thread{tid}") + store.list_rules() + store.get_mode() + store.get_rule_by_name(f"r{tid}x{idx}") + except BaseException as exc: # noqa: BLE001 - фиксируем любое падение потока + errors.append(f"thread{tid}: {type(exc).__name__}: {exc}") + + run_threads([lambda tid=tid: worker(tid) for tid in range(THREADS)]) + + total = THREADS * PER_THREAD + assert errors == [] + assert store.counts()["rules"] == total + assert store.counts()["audit"] == total + assert len(store.list_rules()) == total + assert len(store.list_audit(limit=total * 2)) == total + + +# -- #54: чтение не видит незафиксированных строк ------------------------------- +def test_reads_never_see_uncommitted_rows(tmp_path: Path) -> None: + """Пока транзакция другого потока открыта, читатель ждёт коммита, а не читает грязь.""" + db, store = open_store(tmp_path) + with db: + store.upsert_rule( + Rule(name="baseline", kind="domain", match="baseline.example.com", target="tun0"), + actor="system", + ) + transaction_open = threading.Event() + reader_started = threading.Event() + release = threading.Event() + observed: list[list[str]] = [] + writer_errors: list[str] = [] + + def writer() -> None: + try: + with db.transaction() as conn: + conn.execute( + RULE_INSERT, + ( + "uncommitted", + "domain", + "uncommitted.example.com", + "tun0", + 100, + 1, + "2026-01-01T00:00:00Z", + "2026-01-01T00:00:00Z", + ), + ) + transaction_open.set() + release.wait(JOIN_TIMEOUT) + except BaseException as exc: # noqa: BLE001 - фиксируем падение писателя + writer_errors.append(f"{type(exc).__name__}: {exc}") + + def reader() -> None: + reader_started.set() + observed.append([item.name for item in store.list_rules()]) + + writer_thread = threading.Thread(target=writer) + writer_thread.start() + assert transaction_open.wait(JOIN_TIMEOUT) + reader_thread = threading.Thread(target=reader) + reader_thread.start() + assert reader_started.wait(JOIN_TIMEOUT) + time.sleep(OBSERVE_SECONDS) + # незафиксированная строка наружу не видна: чтение ждёт завершения транзакции + assert observed == [] + + release.set() + writer_thread.join(JOIN_TIMEOUT) + reader_thread.join(JOIN_TIMEOUT) + assert writer_errors == [] + assert observed == [["baseline", "uncommitted"]] + + +# -- #54 п.4: во время падающего apply наружу не отдаётся откаченное состояние ---- +class BlockingApplier: + """Применитель, который «висит» внутри apply и затем падает как iptables.""" + + def __init__(self) -> None: + self.entered = threading.Event() + self.release = threading.Event() + self.reverted: list[dict[str, Any]] = [] + + def apply(self, payload: Mapping[str, Any]) -> None: + self.entered.set() + assert self.release.wait(JOIN_TIMEOUT) + raise RuntimeError("mock iptables: exit 4") + + def revert(self, payload: Mapping[str, Any]) -> None: + self.reverted.append(dict(payload)) + + +def test_read_during_failing_apply_never_sees_discarded_state(tmp_path: Path) -> None: + """AC #20: снаружи не видно «полуприменённой» конфигурации, которую затем откатят.""" + db, store = open_store(tmp_path) + applier = BlockingApplier() + engine = ApplyEngine(store, applier, actor="ui:admin") + with db: + store.upsert_rule( + Rule(name="keep", kind="domain", match="keep.example.com", target="tun0"), + actor="system", + ) + before = capture_state(db.connection) + plan = engine.stage( + DesiredState( + mode=Mode.FULL, + rules=(Rule(name="new", kind="domain", match="new.example.com", target="tun0"),), + ), + reason="сломается", + ) + errors: list[BaseException] = [] + observed: list[list[str]] = [] + reader_started = threading.Event() + + def apply_worker() -> None: + try: + engine.apply(plan) + except BaseException as exc: # noqa: BLE001 - ожидаем ApplyError + errors.append(exc) + + def reader() -> None: + reader_started.set() + observed.append([item.name for item in store.list_rules()]) + + apply_thread = threading.Thread(target=apply_worker) + apply_thread.start() + assert applier.entered.wait(JOIN_TIMEOUT) + reader_thread = threading.Thread(target=reader) + reader_thread.start() + assert reader_started.wait(JOIN_TIMEOUT) + time.sleep(OBSERVE_SECONDS) + assert observed == [], "чтение вернуло состояние, которого ещё нет в БД" + + applier.release.set() + apply_thread.join(JOIN_TIMEOUT) + reader_thread.join(JOIN_TIMEOUT) + + assert len(errors) == 1 + error = errors[0] + assert isinstance(error, ApplyError) + assert error.restored is True + assert observed == [["keep"]] + assert [item.name for item in store.list_rules()] == ["keep"] + assert capture_state(db.connection) == before + assert applier.reverted == [before] + + +# -- #55 D-1: BaseException не должен оставлять staged без исхода --------------- +class InterruptingApplier: + """Применитель, прерывающий apply исключением из ``BaseException``.""" + + def __init__(self, exc_type: type[BaseException]) -> None: + self.exc_type = exc_type + self.applied = 0 + self.reverted: list[dict[str, Any]] = [] + + def apply(self, payload: Mapping[str, Any]) -> None: + self.applied += 1 + raise self.exc_type("внешний применитель прерван") + + def revert(self, payload: Mapping[str, Any]) -> None: + self.reverted.append(dict(payload)) + + +@pytest.mark.parametrize("exc_type", [KeyboardInterrupt, SystemExit, asyncio.CancelledError]) +def test_base_exception_from_applier_reverts_and_records_failure( + tmp_path: Path, exc_type: type[BaseException] +) -> None: + """Отмена/прерывание apply → revert внешней подсистемы + failed в audit_log.""" + db, store = open_store(tmp_path) + applier = InterruptingApplier(exc_type) + engine = ApplyEngine(store, applier, actor="ui:admin") + with db: + store.upsert_rule( + Rule(name="keep", kind="domain", match="keep.example.com", target="tun0"), + actor="system", + ) + before = capture_state(db.connection) + plan = engine.stage( + DesiredState( + mode=Mode.FULL, + rules=(Rule(name="new", kind="domain", match="new.example.com", target="tun0"),), + ), + reason="отмена", + ) + + with pytest.raises(exc_type): + engine.apply(plan) + + assert applier.applied == 1 + assert applier.reverted == [before] + assert capture_state(db.connection) == before + assert [item.name for item in store.list_rules()] == ["keep"] + + entries = store.list_audit(entity=Entity.APPLY.value, entity_id=plan.plan_id, limit=10) + assert entries[0].status == AUDIT_STATUS_FAILED + assert exc_type.__name__ in (entries[0].detail or "") + assert entries[0].new_value == before + + +# -- #55 D-2: конкурентные set_geoip_schedule дают связный журнал --------------- +def test_concurrent_geoip_updates_keep_audit_chain_consistent(tmp_path: Path) -> None: + """Каждое изменение пишет корректный old_value: цепочка old→new не рвётся.""" + db, store = open_store(tmp_path) + with db: + initial = store.get_geoip_schedule() + crons = ("0 1 * * *", "0 2 * * *") + errors: list[str] = [] + barrier = threading.Barrier(len(crons)) + + def worker(cron: str) -> None: + try: + barrier.wait(timeout=JOIN_TIMEOUT) + store.set_geoip_schedule(cron=cron, actor=f"ui:{cron}") + except BaseException as exc: # noqa: BLE001 - фиксируем падение потока + errors.append(f"{cron}: {type(exc).__name__}: {exc}") + + run_threads([lambda cron=cron: worker(cron) for cron in crons]) + + assert errors == [] + entries = list(reversed(store.list_audit(entity=Entity.GEOIP_SCHEDULE.value, limit=10))) + assert len(entries) == len(crons) + first, second = entries + assert first.old_value["cron"] == initial.cron + assert first.new_value["cron"] == second.old_value["cron"] + assert second.new_value["cron"] == store.get_geoip_schedule().cron + assert {first.new_value["cron"], second.new_value["cron"]} == set(crons)