Добавлены тесты для критических фиксов (lock_check, UniqueViolation, escName, idx_one_running)

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.

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

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

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

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