From 4d3ce74c0310737145d2107392c68508577875f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Fri, 31 Jul 2026 18:28:49 +0400 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D1=8B=20=D1=82=D0=B5=D1=81=D1=82=D1=8B=20=D0=B4=D0=BB?= =?UTF-8?q?=D1=8F=20=D0=BA=D1=80=D0=B8=D1=82=D0=B8=D1=87=D0=B5=D1=81=D0=BA?= =?UTF-8?q?=D0=B8=D1=85=20=D1=84=D0=B8=D0=BA=D1=81=D0=BE=D0=B2=20(lock=5Fc?= =?UTF-8?q?heck,=20UniqueViolation,=20escName,=20idx=5Fone=5Frunning)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/README.md | 10 ++++ tests/conftest.py | 19 +++++++ tests/test_api_scenario_run.py | 88 ++++++++++++++++++++++++++++++++ tests/test_db_scenario_defs.py | 32 ++++++++++++ tests/test_static_regressions.py | 23 +++++++++ 5 files changed, 172 insertions(+) create mode 100644 tests/README.md create mode 100644 tests/conftest.py create mode 100644 tests/test_api_scenario_run.py create mode 100644 tests/test_db_scenario_defs.py create mode 100644 tests/test_static_regressions.py diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..68e5754 --- /dev/null +++ b/tests/README.md @@ -0,0 +1,10 @@ +Pytest suite for critical regressions from audit rounds. + +Covers: +- `/api/scenario/run` status mapping: `lock_check=None -> 503`, `False -> 409`. +- Unique violation on `scenario_runs` insert maps to `409`. +- `db.scenario_defs.lock_check` DB-unavailable semantics (`None`). +- Static guards for DB partial unique index and `escName` escape chain. + +This suite is intentionally lightweight and mock-heavy. +No integration DB or external API calls are required. diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..440f3ab --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,19 @@ +from pathlib import Path +import sys + +import pytest + + +# Make `site/` importable as top-level modules: app, routes, db, operations, ... +SITE_DIR = Path(__file__).resolve().parents[1] / "site" +if str(SITE_DIR) not in sys.path: + sys.path.insert(0, str(SITE_DIR)) + + +@pytest.fixture +def app_client(): + from app import app + + app.config.update(TESTING=True) + with app.test_client() as client: + yield client diff --git a/tests/test_api_scenario_run.py b/tests/test_api_scenario_run.py new file mode 100644 index 0000000..9e7920d --- /dev/null +++ b/tests/test_api_scenario_run.py @@ -0,0 +1,88 @@ +import routes.api_scenario_run as api_run + + +class DummyUniqueViolation(Exception): + pgcode = "23505" + + +class DummyCursor: + def __init__(self, fail_on_execute=None, fetchone_value=None): + self._fail_on_execute = fail_on_execute + self._fetchone_value = fetchone_value + + def execute(self, *args, **kwargs): + if self._fail_on_execute is not None: + raise self._fail_on_execute + + def fetchone(self): + return self._fetchone_value + + def close(self): + return None + + +class DummyConn: + def __init__(self, fail_on_execute=None, fetchone_value=None): + self._fail_on_execute = fail_on_execute + self._fetchone_value = fetchone_value + self.committed = False + self.rolled_back = False + + def cursor(self): + return DummyCursor(self._fail_on_execute, self._fetchone_value) + + def commit(self): + self.committed = True + + def rollback(self): + self.rolled_back = True + + +def _patch_common_happy(monkeypatch): + monkeypatch.setattr(api_run, "get_client_id", lambda: "cid") + monkeypatch.setattr(api_run, "get_stand", lambda: "test") + monkeypatch.setattr(api_run, "get_definition", lambda _id, _cid, _stand: { + "id": _id, + "name": "scenario-a", + "version": 7, + "steps": [{"service_id": 1, "operation": "create", "params": {}}], + }) + monkeypatch.setattr(api_run, "get_token_info", lambda: {"email": "u@example.com"}) + + +def test_run_returns_503_when_lock_check_is_none(app_client, monkeypatch): + _patch_common_happy(monkeypatch) + monkeypatch.setattr(api_run, "lock_check", lambda _cid, _stand: None) + + resp = app_client.post("/api/scenario/run", json={"definition_id": 1}) + + assert resp.status_code == 503 + assert resp.get_json()["error"] == "DB unavailable" + + +def test_run_returns_409_when_lock_check_is_false(app_client, monkeypatch): + _patch_common_happy(monkeypatch) + monkeypatch.setattr(api_run, "lock_check", lambda _cid, _stand: False) + + resp = app_client.post("/api/scenario/run", json={"definition_id": 1}) + + assert resp.status_code == 409 + assert "already running" in resp.get_json()["error"].lower() + + +def test_run_returns_409_on_unique_violation(app_client, monkeypatch): + _patch_common_happy(monkeypatch) + monkeypatch.setattr(api_run, "lock_check", lambda _cid, _stand: True) + + conn = DummyConn(fail_on_execute=DummyUniqueViolation()) + monkeypatch.setattr(api_run, "get_conn", lambda: conn) + monkeypatch.setattr(api_run, "put_conn", lambda _conn: None) + + # Must not start a background thread in this branch. + monkeypatch.setattr(api_run, "get_client", lambda: object()) + + resp = app_client.post("/api/scenario/run", json={"definition_id": 1}) + + assert resp.status_code == 409 + assert "already running" in resp.get_json()["error"].lower() + assert conn.rolled_back is True diff --git a/tests/test_db_scenario_defs.py b/tests/test_db_scenario_defs.py new file mode 100644 index 0000000..1abe7ff --- /dev/null +++ b/tests/test_db_scenario_defs.py @@ -0,0 +1,32 @@ +import db.scenario_defs as defs + + +def test_lock_check_returns_none_when_db_connection_missing(monkeypatch): + monkeypatch.setattr(defs, "get_conn", lambda: None) + + result = defs.lock_check("cid", "test") + + assert result is None + + +def test_lock_check_returns_none_on_db_exception(monkeypatch): + class BrokenCursor: + def execute(self, *args, **kwargs): + raise RuntimeError("db broken") + + def close(self): + return None + + class BrokenConn: + def cursor(self): + return BrokenCursor() + + put_calls = {"n": 0} + + monkeypatch.setattr(defs, "get_conn", lambda: BrokenConn()) + monkeypatch.setattr(defs, "put_conn", lambda _c: put_calls.__setitem__("n", put_calls["n"] + 1)) + + result = defs.lock_check("cid", "test") + + assert result is None + assert put_calls["n"] == 1 diff --git a/tests/test_static_regressions.py b/tests/test_static_regressions.py new file mode 100644 index 0000000..e9bb900 --- /dev/null +++ b/tests/test_static_regressions.py @@ -0,0 +1,23 @@ +from pathlib import Path + + +def test_partial_unique_index_for_one_running_exists(): + init_db_path = Path(__file__).resolve().parents[1] / "site" / "db" / "init_db.py" + content = init_db_path.read_text(encoding="utf-8") + + assert "CREATE UNIQUE INDEX IF NOT EXISTS idx_one_running" in content + assert "ON scenario_runs (client_id, stand) WHERE status = 'RUNNING'" in content + + +def test_scenario_name_escape_chain_includes_amp_backslash_quote_doublequote(): + js_path = Path(__file__).resolve().parents[1] / "site" / "static" / "js" / "scenario-list.js" + content = js_path.read_text(encoding="utf-8") + + expected = ( + "const escName = def.name" + ".replace(/&/g,'&')" + ".replace(/\\\\/g,'\\\\\\\\')" + ".replace(/'/g,\"\\\\'\")" + ".replace(/\"/g,'"');" + ) + assert expected in content