fix(scripts): actually escape LIKE pattern literals + scope count endpoint
Codex review of #36 + #37 surfaced that my prior `escape="\\"` only declared the escape character — the pattern literals themselves still contained unescaped `_` and `%`, so `parent_path="foo_bar"` continued to match `fooXbar/...`, `foo2bar/...`, etc. My earlier ESCAPE-clause assertions were tautological: they verified the SQL rendered the ESCAPE keyword without ever checking that the pattern was actually escaped. The tests passed; the leak persisted. Fix in three layers: 1. Real escape: `backend/src/backend/scripts.py` gains `_escape_like_pattern(value)` that escapes `\` → `\\`, `%` → `\%`, `_` → `\_` (in that order — the escape char MUST be escaped first). `_build_list_scripts_descendant_prefix` now returns the escaped prefix. `list_workspace_directories` and `delete_workspace_directory` also escape their server-built prefixes. `count_scripts` escapes the user subtree prefix. 2. Same bug elsewhere: `backend/src/backend/resources.py:404` had the identical `DataResources.resource_name.like(f"%{keyword}%")` pattern; a search for "100%" would match everything. Now escaped too. 3. Count endpoint scope: `count_scripts` was workspace-wide and skipped the StorageObjects JOIN. Now INNER JOINs StorageObjects (drops orphans whose current_object_id is dangling) and filters by `workspace/{user_id}/` subtree so the result matches what `list_scripts(parent_path="")` would return. Multi-member workspaces no longer over-report, and orphan rows no longer inflate the count. Frontend: `DashboardRoute` is not keyed by workspace/user (only ScriptsPage is), so without a workspace_id dep the previous workspace's count persisted across navigation. useEffect now depends on `currentWorkspace?.workspace_id`; `loadScriptCount` clears the count to null at the start of the fetch so the dashboard doesn't flash a stale number. Tests — backend/tests/test_list_scripts_parent_path.py - Rewritten with three layers of coverage: * Pure helper tests for `_escape_like_pattern` (7 cases including backslash-escape-first ordering). * SQL-contract tests asserting the COMPILED PATTERN contains the escaped form (lowercased to neutralise SQLAlchemy keyword casing). * BEHAVIORAL tests on SQLite in-memory with the same LIKE semantics — proves the fix actually prevents the wildcard leak. Includes a negative test (without escape, siblings DO match) so the fixture is verified to exercise the bug. Tests — backend/tests/test_count_scripts.py - Updated to assert the JOIN + user-scope filter. New test verifies two different users in the same workspace get different subtrees. Verified: - pytest backend/tests: 65 passed (43 baseline + 12 list_scripts + 4 count + 6 helper/SQLite behavioral) - pnpm typecheck: clean - Raw SQL on MySQL (live DB) confirms `LIKE 'workspace/.../foo\_bar/%%' ESCAPE '\\'`.
This commit is contained in:
tao.chen
@@ -1,8 +1,10 @@
|
||||
"""Unit tests for GET /api/v1/scripts/count endpoint.
|
||||
|
||||
Verifies the count endpoint returns the workspace-wide active-script total
|
||||
and does NOT depend on lazy-load semantics — the dashboard uses this
|
||||
instead of `scripts.length` to avoid underreporting.
|
||||
Verifies the count endpoint returns the same scope as
|
||||
``list_scripts(parent_path="")``: workspace + active scripts whose
|
||||
``StorageObjects.relative_path`` lives under the user's subtree. This
|
||||
avoids under/over-reporting on the dashboard — the count is the size of
|
||||
the set list_scripts would return if it weren't lazy.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -11,7 +13,6 @@ from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from backend.scripts import count_scripts
|
||||
|
||||
@@ -26,16 +27,24 @@ def _ctx(user_id: str = "U001", workspace_id: str = "W001") -> SimpleNamespace:
|
||||
)
|
||||
|
||||
|
||||
def _compile(stmt) -> str:
|
||||
from sqlalchemy.dialects import mysql as mysql_dialect
|
||||
|
||||
return str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def test_count_scripts_returns_scalar_int() -> None:
|
||||
captured = []
|
||||
|
||||
class _MockScalarResult:
|
||||
def scalar(self, _stmt):
|
||||
captured.append(_stmt)
|
||||
return 7
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.scalar = AsyncMock(side_effect=lambda stmt: (captured.append(stmt), 7)[1])
|
||||
mock_session.scalar = AsyncMock(
|
||||
side_effect=lambda stmt: (captured.append(stmt), 7)[1]
|
||||
)
|
||||
|
||||
result = await count_scripts(context=_ctx(), session=mock_session)
|
||||
assert result["data"] == {"total": 7}
|
||||
@@ -44,12 +53,14 @@ async def test_count_scripts_returns_scalar_int() -> None:
|
||||
# Exactly one COUNT(*) query issued.
|
||||
assert len(captured) == 1
|
||||
stmt = captured[0]
|
||||
# SQL must select from Scripts (the COUNT target) and filter by
|
||||
# workspace_id + status. Bind params render as :workspace_id_1 etc.
|
||||
text = str(stmt).lower()
|
||||
assert "from scripts" in text
|
||||
assert "workspace_id" in text
|
||||
assert "status" in text
|
||||
sql = _compile(stmt).lower()
|
||||
# JOIN to StorageObjects so orphaned scripts (no joinable row) are
|
||||
# excluded — matches list_scripts INNER JOIN behaviour.
|
||||
assert "inner join storage_objects" in sql
|
||||
# Scope: workspace_id + active status + user subtree.
|
||||
assert "scripts.workspace_id" in sql
|
||||
assert "scripts.status" in sql
|
||||
assert "workspace/u001/%" in sql
|
||||
|
||||
|
||||
async def test_count_scripts_handles_null_result() -> None:
|
||||
@@ -61,12 +72,33 @@ async def test_count_scripts_handles_null_result() -> None:
|
||||
assert result["data"] == {"total": 0}
|
||||
|
||||
|
||||
async def test_count_scripts_uses_user_specific_subtree() -> None:
|
||||
"""Different users in the same workspace must see different totals —
|
||||
each user's count is bounded by their own ``workspace/{user_id}/``
|
||||
subtree, NOT the whole workspace."""
|
||||
captured = []
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.scalar = AsyncMock(
|
||||
side_effect=lambda stmt: (captured.append(stmt), 3)[1]
|
||||
)
|
||||
|
||||
await count_scripts(context=_ctx(user_id="alice"), session=mock_session)
|
||||
sql_alice = _compile(captured[-1])
|
||||
|
||||
await count_scripts(context=_ctx(user_id="bob"), session=mock_session)
|
||||
sql_bob = _compile(captured[-1])
|
||||
|
||||
assert "workspace/alice/%" in sql_alice
|
||||
assert "workspace/alice/%" not in sql_bob
|
||||
assert "workspace/bob/%" in sql_bob
|
||||
|
||||
|
||||
async def test_count_scripts_route_declared_before_script_id_route() -> None:
|
||||
"""Static check: the `/api/v1/scripts/count` route MUST be declared in
|
||||
scripts.py before `/api/v1/scripts/{script_id}/...`, otherwise FastAPI's
|
||||
declaration-order matching will interpret `count` as a script_id."""
|
||||
from backend.scripts import count_scripts, get_script
|
||||
|
||||
# Both callables exist (sanity).
|
||||
assert callable(count_scripts)
|
||||
assert callable(get_script)
|
||||
@@ -1,14 +1,20 @@
|
||||
"""Unit tests for the parent_path filter clause on list_scripts.
|
||||
"""Tests for the parent_path filter clause on list_scripts, plus the
|
||||
LIKE-pattern escape contract for tree-walking queries.
|
||||
|
||||
Verifies the WHERE clause built by the endpoint encodes the intended
|
||||
"direct children of parent_path" semantics: ``relative_path LIKE 'prefix/%'``
|
||||
and ``NOT LIKE 'prefix/%/%'``. These are SQL-contract assertions (mock
|
||||
session, capture compiled SQL); the repo has no integration test layer
|
||||
for endpoints, so this is the only coverage. Brittle to SQLAlchemy/dialect
|
||||
rendering changes — review the assertions together with the endpoint if
|
||||
you upgrade SQLAlchemy.
|
||||
Three layers of coverage:
|
||||
|
||||
Mirrors the pattern in test_scripts.py (unit-level, no live DB).
|
||||
1. ``_escape_like_pattern`` unit tests — pure-function correctness.
|
||||
2. SQL-contract tests (mock session) — verifies the compiled SQL contains
|
||||
the escaped pattern AND the ``ESCAPE '\\'`` clause.
|
||||
3. Behavioral test (SQLite in-memory, real LIKE execution) — proves the
|
||||
fix actually prevents the wildcard leak that motivated the change.
|
||||
A folder named ``foo_bar`` MUST NOT match sibling paths like
|
||||
``fooXbar`` / ``foo2bar`` / ``foo/bar``.
|
||||
|
||||
The repo has no MySQL integration test layer, so SQLite stands in for
|
||||
LIKE semantics — both dialects treat ``_`` as "any single char" and
|
||||
``%`` as "any sequence" by default and honour the ``ESCAPE`` clause
|
||||
identically for the ASCII characters we care about.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -18,16 +24,17 @@ from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy import Column, MetaData, String, Table, create_engine, select, text
|
||||
from sqlalchemy.dialects import mysql as mysql_dialect
|
||||
|
||||
from backend.scripts import (
|
||||
_build_list_scripts_descendant_prefix,
|
||||
_escape_like_pattern,
|
||||
normalize_user_path,
|
||||
)
|
||||
from sqlalchemy.dialects import mysql as mysql_dialect
|
||||
|
||||
|
||||
def _ctx(user_id: str = "U001") -> SimpleNamespace:
|
||||
"""Stand-in for RequestContext — only ``user.user_id`` and ``workspace_id`` are read."""
|
||||
return SimpleNamespace(
|
||||
request_id="test",
|
||||
user=SimpleNamespace(user_id=user_id),
|
||||
@@ -37,6 +44,40 @@ def _ctx(user_id: str = "U001") -> SimpleNamespace:
|
||||
)
|
||||
|
||||
|
||||
# ─── layer 1: helper unit tests ──────────────────────────────────
|
||||
|
||||
|
||||
class TestEscapeLikePattern:
|
||||
"""The escape helper itself is the load-bearing piece — test it
|
||||
exhaustively before relying on it in SQL."""
|
||||
|
||||
def test_no_metachars_unchanged(self) -> None:
|
||||
assert _escape_like_pattern("foo/bar") == "foo/bar"
|
||||
assert _escape_like_pattern("workspace/alice") == "workspace/alice"
|
||||
assert _escape_like_pattern("") == ""
|
||||
|
||||
def test_underscore_escaped(self) -> None:
|
||||
assert _escape_like_pattern("foo_bar") == r"foo\_bar"
|
||||
|
||||
def test_percent_escaped(self) -> None:
|
||||
assert _escape_like_pattern("100%") == r"100\%"
|
||||
assert _escape_like_pattern("%foo") == r"\%foo"
|
||||
|
||||
def test_backslash_escaped_first(self) -> None:
|
||||
# Must escape the escape char first, otherwise the inserted
|
||||
# backslashes would be double-escaped by the later passes.
|
||||
assert _escape_like_pattern(r"a\b") == r"a\\b"
|
||||
assert _escape_like_pattern(r"a\%b") == r"a\\\%b"
|
||||
|
||||
def test_combined(self) -> None:
|
||||
assert _escape_like_pattern("foo_bar%baz") == r"foo\_bar\%baz"
|
||||
assert _escape_like_pattern("_%") == r"\_\%"
|
||||
assert _escape_like_pattern(r"\\_%") == r"\\\\\_\%"
|
||||
|
||||
|
||||
# ─── layer 1.5: prefix helper now escapes ─────────────────────────
|
||||
|
||||
|
||||
def test_descendant_prefix_root() -> None:
|
||||
"""Empty parent_path → descendant prefix is the scoped root + '/'."""
|
||||
prefix = _build_list_scripts_descendant_prefix(_ctx("alice"), "")
|
||||
@@ -49,14 +90,20 @@ def test_descendant_prefix_subdir() -> None:
|
||||
assert prefix == "workspace/alice/foo/bar/"
|
||||
|
||||
|
||||
def test_descendant_prefix_escapes_metachars() -> None:
|
||||
"""Folder name ``foo_bar`` MUST produce ``foo\\_bar`` in the prefix
|
||||
so the trailing ``%`` doesn't become 'match any single char before
|
||||
b'."""
|
||||
prefix = _build_list_scripts_descendant_prefix(_ctx("alice"), "foo_bar")
|
||||
assert prefix == r"workspace/alice/foo\_bar/"
|
||||
|
||||
|
||||
def test_descendant_prefix_normalizes_leading_trailing_slashes() -> None:
|
||||
"""Leading/trailing slashes on parent_path must be stripped."""
|
||||
prefix = _build_list_scripts_descendant_prefix(_ctx("alice"), "/foo/bar/")
|
||||
assert prefix == "workspace/alice/foo/bar/"
|
||||
|
||||
|
||||
def test_descendant_prefix_rejects_traversal() -> None:
|
||||
"""``..`` segments must raise (matches normalize_user_path contract)."""
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
_build_list_scripts_descendant_prefix(_ctx("alice"), "foo/../bar")
|
||||
assert exc.value.status_code == 422
|
||||
@@ -68,9 +115,19 @@ def test_normalize_user_path_strips() -> None:
|
||||
assert normalize_user_path("a\\b") == "a/b"
|
||||
|
||||
|
||||
# ─── layer 2: SQL contract ────────────────────────────────────────
|
||||
|
||||
|
||||
def _compile_sql(stmt) -> str:
|
||||
return str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper() -> None:
|
||||
"""The WHERE clause must include both LIKE prefix and NOT LIKE '%/%' filters
|
||||
so deeper descendants and prefix-siblings (foo/bar vs foo/bar2) are excluded."""
|
||||
from backend.scripts import list_scripts
|
||||
|
||||
captured_sql: list[str] = []
|
||||
@@ -82,15 +139,7 @@ async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper()
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(
|
||||
str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
)
|
||||
or _MockResult()
|
||||
captured_sql.append(_compile_sql(stmt)) or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
@@ -98,16 +147,13 @@ async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper()
|
||||
|
||||
assert len(captured_sql) == 1
|
||||
sql = captured_sql[0].lower()
|
||||
# Direct-child LIKE prefix
|
||||
assert "like 'workspace/alice/foo/bar/%%'" in sql
|
||||
# NOT-LIKE deeper
|
||||
assert "not like 'workspace/alice/foo/bar/%%/%%'" in sql
|
||||
# active scripts only (existing contract preserved)
|
||||
assert "scripts.status" in sql or "scripts.status = 'active'" in sql or "scripts.status = :status" in sql
|
||||
|
||||
|
||||
async def test_list_scripts_empty_parent_path_targets_root_descendants() -> None:
|
||||
"""parent_path='' produces root-scoped LIKE prefix only, not full scan."""
|
||||
async def test_list_scripts_where_clause_escapes_pattern_literal() -> None:
|
||||
"""Regression: parent_path containing ``_`` MUST be escaped in the
|
||||
compiled LIKE pattern, otherwise sibling-path leak returns to bite."""
|
||||
from backend.scripts import list_scripts
|
||||
|
||||
captured_sql: list[str] = []
|
||||
@@ -119,63 +165,53 @@ async def test_list_scripts_empty_parent_path_targets_root_descendants() -> None
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(
|
||||
str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
)
|
||||
or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
await list_scripts(parent_path="", context=_ctx("alice"), session=mock_session)
|
||||
|
||||
sql = captured_sql[0].lower()
|
||||
assert "like 'workspace/alice/%%'" in sql
|
||||
assert "not like 'workspace/alice/%%/%%'" in sql
|
||||
|
||||
|
||||
async def test_list_scripts_where_clause_includes_escape() -> None:
|
||||
"""Both LIKE clauses must declare ESCAPE so folder names containing ``_``
|
||||
or ``%`` do not act as SQL wildcards and match sibling paths."""
|
||||
from backend.scripts import list_scripts
|
||||
|
||||
captured_sql: list[str] = []
|
||||
|
||||
class _MockResult:
|
||||
def all(self):
|
||||
return []
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(
|
||||
str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
)
|
||||
or _MockResult()
|
||||
captured_sql.append(_compile_sql(stmt)) or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
await list_scripts(parent_path="foo_bar", context=_ctx("alice"), session=mock_session)
|
||||
sql = captured_sql[0]
|
||||
# Both patterns must carry ESCAPE; counts must match between the two LIKE
|
||||
# occurrences (one positive, one negated). SQLAlchemy doubles the escape
|
||||
# char for SQL string literals, so the rendered form is `ESCAPE '\\\\'`.
|
||||
# Normalize keyword case so we don't depend on SQLAlchemy casing.
|
||||
sql_lower = sql.lower()
|
||||
# Pattern literal must contain the ESCAPED underscore. SQLAlchemy
|
||||
# doubles the escape char inside the SQL string literal, so what
|
||||
# the helper emits as `foo\_bar` renders as `foo\\_bar` here
|
||||
# (2 backslash chars in the actual SQL string).
|
||||
assert r"like 'workspace/alice/foo\\_bar/%%'" in sql_lower
|
||||
# NOT LIKE clause also escaped.
|
||||
assert r"not like 'workspace/alice/foo\\_bar/%%/%%'" in sql_lower
|
||||
# And both declare ESCAPE '\\'.
|
||||
assert sql.count("ESCAPE '\\\\'") == 2, sql
|
||||
|
||||
|
||||
async def test_list_workspace_directories_where_clause_includes_escape() -> None:
|
||||
"""list_workspace_directories must also emit ESCAPE — the same LIKE
|
||||
pattern was already vulnerable for pre-existing endpoints; this
|
||||
endpoint is in scope for the same fix."""
|
||||
async def test_list_scripts_where_clause_escapes_percent_pattern() -> None:
|
||||
"""Same regression for ``%``."""
|
||||
from backend.scripts import list_scripts
|
||||
|
||||
captured_sql: list[str] = []
|
||||
|
||||
class _MockResult:
|
||||
def all(self):
|
||||
return []
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(_compile_sql(stmt)) or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
await list_scripts(parent_path="100%match", context=_ctx("alice"), session=mock_session)
|
||||
sql = captured_sql[0]
|
||||
sql_lower = sql.lower()
|
||||
# SQLAlchemy doubles the escape char so `%` → `\%` becomes `\\%`
|
||||
# in the SQL string literal.
|
||||
assert r"workspace/alice/100\\%%match/%%" in sql_lower
|
||||
|
||||
|
||||
async def test_list_workspace_directories_where_clause_escapes_pattern() -> None:
|
||||
"""list_workspace_directories must escape user input too (was
|
||||
pre-existing debt)."""
|
||||
from backend.scripts import list_workspace_directories
|
||||
|
||||
captured_sql: list[str] = []
|
||||
@@ -194,23 +230,109 @@ async def test_list_workspace_directories_where_clause_includes_escape() -> None
|
||||
mock_session = MagicMock()
|
||||
mock_session.execute = AsyncMock(
|
||||
side_effect=lambda stmt: (
|
||||
captured_sql.append(
|
||||
str(
|
||||
stmt.compile(
|
||||
dialect=mysql_dialect.dialect(),
|
||||
compile_kwargs={"literal_binds": True},
|
||||
)
|
||||
)
|
||||
)
|
||||
or _MockResult()
|
||||
captured_sql.append(_compile_sql(stmt)) or _MockResult()
|
||||
)
|
||||
)
|
||||
|
||||
await list_workspace_directories(
|
||||
parent_path="foo_bar", context=_ctx("alice"), session=mock_session
|
||||
)
|
||||
# Two LIKE clauses in the children query + two in the has_children
|
||||
# check per directory in the result — for an empty result set only
|
||||
# the first batch executes, so we expect at least 2 ESCAPEs.
|
||||
sql = " ".join(captured_sql)
|
||||
assert sql.count("ESCAPE '\\\\'") >= 2, sql
|
||||
assert r"workspace/alice/foo\\_bar/" in sql, sql
|
||||
|
||||
|
||||
# ─── layer 3: behavioral test on real LIKE execution ──────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sqlite_like_table():
|
||||
"""SQLite in-memory table with a single VARCHAR column. Stand-in for
|
||||
``storage_objects.relative_path`` — proves the actual LIKE executor
|
||||
behaves the way we expect with the escaped pattern."""
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
metadata = MetaData()
|
||||
table = Table(
|
||||
"paths",
|
||||
metadata,
|
||||
Column("relative_path", String(1024), nullable=False),
|
||||
)
|
||||
metadata.create_all(engine)
|
||||
with engine.begin() as conn:
|
||||
# Target row (the one a parent_path="foo_bar" search MUST return).
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/foo_bar/inner.py"},
|
||||
)
|
||||
# Decoys the buggy LIKE would match but escaped the must NOT.
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/fooXbar/decoy.py"},
|
||||
)
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/foo2bar/decoy.py"},
|
||||
)
|
||||
# A truly unrelated path.
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/baz/inner.py"},
|
||||
)
|
||||
yield engine, table
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def test_sqlite_like_with_escape_does_not_match_sibling(sqlite_like_table):
|
||||
"""Execute the actual LIKE pattern the endpoint would emit for
|
||||
parent_path='foo_bar'. Confirms only the target row matches."""
|
||||
engine, table = sqlite_like_table
|
||||
escaped_prefix = _escape_like_pattern("workspace/alice/foo_bar") + "/"
|
||||
pattern = f"{escaped_prefix}%"
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
select(table.c.relative_path).where(
|
||||
table.c.relative_path.like(pattern, escape="\\")
|
||||
)
|
||||
).fetchall()
|
||||
matched = sorted(r[0] for r in rows)
|
||||
assert matched == ["workspace/alice/foo_bar/inner.py"], matched
|
||||
|
||||
|
||||
def test_sqlite_like_without_escape_matches_siblings(sqlite_like_table):
|
||||
"""Sanity check: WITHOUT escape, the same pattern matches the
|
||||
decoys too — confirming the test setup actually exercises the
|
||||
leak. If this assertion fails the SQLite fixture is broken."""
|
||||
engine, table = sqlite_like_table
|
||||
pattern = "workspace/alice/foo_bar/%"
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
select(table.c.relative_path).where(
|
||||
table.c.relative_path.like(pattern)
|
||||
)
|
||||
).fetchall()
|
||||
matched = sorted(r[0] for r in rows)
|
||||
# Without escape, the buggy behaviour returns ALL three foo*bar rows.
|
||||
assert len(matched) >= 2, matched
|
||||
|
||||
|
||||
def test_sqlite_like_with_percent_in_name(sqlite_like_table):
|
||||
"""Folder name containing ``%`` — must be escaped too."""
|
||||
engine, table = sqlite_like_table
|
||||
with engine.begin() as conn:
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/100%off/x.py"},
|
||||
)
|
||||
conn.execute(
|
||||
table.insert(),
|
||||
{"relative_path": "workspace/alice/100Xoff/y.py"},
|
||||
)
|
||||
escaped_prefix = _escape_like_pattern("workspace/alice/100%off") + "/"
|
||||
pattern = f"{escaped_prefix}%"
|
||||
with engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
select(table.c.relative_path).where(
|
||||
table.c.relative_path.like(pattern, escape="\\")
|
||||
)
|
||||
).fetchall()
|
||||
matched = sorted(r[0] for r in rows)
|
||||
assert matched == ["workspace/alice/100%off/x.py"], matched
|
||||
Reference in New Issue
Block a user