fix(db): id entropy + like escaping + ms timestamps + sentinel typing
- Bump _gen_id suffix from token_hex(2) to token_hex(4) for 32-bit collision resistance - Escape LIKE wildcards (%, _, \) in list(query=...) with ESCAPE clause - Switch _now() and migrations.py to millisecond-precision ISO 8601 timestamps - Add _Unset typed sentinel replacing object-typed ... default on update() - Support offset without limit via LIMIT -1 OFFSET ? in list() - Remove time.sleep() calls from module repository tests (ms precision makes them unnecessary) - Add 3 new tests: percent/underscore literal search, offset-without-limit
This commit is contained in:
@@ -57,5 +57,8 @@ def migrate(conn: sqlite3.Connection) -> None:
|
||||
conn.execute(stmt)
|
||||
conn.execute(
|
||||
"INSERT INTO migrations(version, applied_at) VALUES(?, ?);",
|
||||
(version, _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")),
|
||||
(
|
||||
version,
|
||||
_dt.datetime.now(_dt.UTC).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
|
||||
),
|
||||
)
|
||||
|
||||
@@ -17,6 +17,19 @@ _TYPE_PREFIX = {"wildcard": "wc", "fixed_values": "fv"}
|
||||
_SLUG_RE = re.compile(r"[^a-z0-9]+")
|
||||
|
||||
|
||||
class _Unset:
|
||||
"""Sentinel for 'argument not provided' where ``None`` is a valid value."""
|
||||
_instance: _Unset | None = None
|
||||
|
||||
def __new__(cls) -> _Unset:
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
return cls._instance
|
||||
|
||||
|
||||
_UNSET = _Unset()
|
||||
|
||||
|
||||
class ModuleNotFound(LookupError):
|
||||
"""Raised when a requested module id does not exist."""
|
||||
|
||||
@@ -27,7 +40,7 @@ def _slug(name: str) -> str:
|
||||
|
||||
|
||||
def _now() -> str:
|
||||
return _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
return _dt.datetime.now(_dt.UTC).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
|
||||
|
||||
|
||||
def _row_to_module(row: sqlite3.Row) -> dict[str, Any]:
|
||||
@@ -52,7 +65,7 @@ class ModuleRepository:
|
||||
|
||||
def _gen_id(self, type: str, name: str) -> str:
|
||||
prefix = _TYPE_PREFIX.get(type, "mod")
|
||||
return f"{prefix}_{_slug(name)}_{secrets.token_hex(2)}"
|
||||
return f"{prefix}_{_slug(name)}_{secrets.token_hex(4)}"
|
||||
|
||||
def create(
|
||||
self,
|
||||
@@ -95,7 +108,7 @@ class ModuleRepository:
|
||||
*,
|
||||
name: str | None = None,
|
||||
description: str | None = None,
|
||||
category_id: str | None | object = ...,
|
||||
category_id: str | None | _Unset = _UNSET,
|
||||
tags: list[str] | None = None,
|
||||
payload: dict[str, Any] | None = None,
|
||||
is_favorite: bool | None = None,
|
||||
@@ -104,7 +117,9 @@ class ModuleRepository:
|
||||
new = {
|
||||
"name": existing["name"] if name is None else name,
|
||||
"description": existing["description"] if description is None else description,
|
||||
"category_id": existing["category_id"] if category_id is ... else category_id,
|
||||
"category_id": (
|
||||
existing["category_id"] if isinstance(category_id, _Unset) else category_id
|
||||
),
|
||||
"tags": existing["tags"] if tags is None else tags,
|
||||
"payload": existing["payload"] if payload is None else payload,
|
||||
"is_favorite": existing["is_favorite"] if is_favorite is None else is_favorite,
|
||||
@@ -149,8 +164,13 @@ class ModuleRepository:
|
||||
clauses.append("category_id = ?")
|
||||
params.append(category_id)
|
||||
if query:
|
||||
clauses.append("name LIKE ? COLLATE NOCASE")
|
||||
params.append(f"%{query}%")
|
||||
escaped = (
|
||||
query.replace("\\", "\\\\")
|
||||
.replace("%", "\\%")
|
||||
.replace("_", "\\_")
|
||||
)
|
||||
clauses.append("name LIKE ? ESCAPE '\\' COLLATE NOCASE")
|
||||
params.append(f"%{escaped}%")
|
||||
if favorites_only:
|
||||
clauses.append("is_favorite = 1")
|
||||
sql = "SELECT * FROM modules"
|
||||
@@ -160,5 +180,9 @@ class ModuleRepository:
|
||||
if limit is not None:
|
||||
sql += " LIMIT ? OFFSET ?"
|
||||
params.extend([limit, offset])
|
||||
elif offset:
|
||||
# SQLite uses LIMIT -1 to mean "unlimited" while still honoring OFFSET.
|
||||
sql += " LIMIT -1 OFFSET ?"
|
||||
params.append(offset)
|
||||
rows = self._conn.execute(sql, params).fetchall()
|
||||
return [_row_to_module(r) for r in rows]
|
||||
|
||||
@@ -40,17 +40,15 @@ def test_get_raises_module_not_found(wp_db):
|
||||
|
||||
|
||||
def test_update_bumps_version_and_updated_at(wp_db):
|
||||
import time
|
||||
repo = ModuleRepository(wp_db)
|
||||
created = repo.create(
|
||||
type="wildcard", name="colors", description="",
|
||||
category_id=None, tags=[], payload=_new_payload(),
|
||||
)
|
||||
time.sleep(1.05) # ensure ISO-second-precision updated_at differs
|
||||
updated = repo.update(created["id"], name="palette", payload=_new_payload())
|
||||
assert updated["name"] == "palette"
|
||||
assert updated["version"] == 2
|
||||
assert updated["updated_at"] >= created["updated_at"]
|
||||
assert updated["updated_at"] > created["updated_at"]
|
||||
|
||||
|
||||
def test_delete_removes_row(wp_db):
|
||||
@@ -108,11 +106,9 @@ def test_list_search_matches_name_case_insensitive(wp_db):
|
||||
|
||||
|
||||
def test_list_orders_by_updated_at_desc(wp_db):
|
||||
import time
|
||||
repo = ModuleRepository(wp_db)
|
||||
a = repo.create(type="wildcard", name="a", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
time.sleep(1.05) # ISO-second-precision: ensure b > a
|
||||
b = repo.create(type="wildcard", name="b", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
rows = repo.list()
|
||||
@@ -131,12 +127,10 @@ def test_list_favorites_only(wp_db):
|
||||
|
||||
|
||||
def test_list_limit_offset(wp_db):
|
||||
import time
|
||||
repo = ModuleRepository(wp_db)
|
||||
for n in ["a", "b", "c"]:
|
||||
repo.create(type="wildcard", name=n, description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
time.sleep(1.05)
|
||||
rows = repo.list(limit=1, offset=1)
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["name"] == "b"
|
||||
@@ -161,3 +155,35 @@ def test_create_id_uses_fv_prefix_for_fixed_values(wp_db):
|
||||
category_id=None, tags=[], payload={"values": []},
|
||||
)
|
||||
assert row["id"].startswith("fv_lens_")
|
||||
|
||||
|
||||
def test_list_search_treats_percent_literally(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
repo.create(type="wildcard", name="50%_off", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
repo.create(type="wildcard", name="other", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
rows = repo.list(query="50%")
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["name"] == "50%_off"
|
||||
|
||||
|
||||
def test_list_search_treats_underscore_literally(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
repo.create(type="wildcard", name="my_module", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
repo.create(type="wildcard", name="myXmodule", description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
rows = repo.list(query="my_")
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["name"] == "my_module"
|
||||
|
||||
|
||||
def test_list_offset_without_limit(wp_db):
|
||||
repo = ModuleRepository(wp_db)
|
||||
for n in ["a", "b", "c"]:
|
||||
repo.create(type="wildcard", name=n, description="",
|
||||
category_id=None, tags=[], payload=_new_payload())
|
||||
rows = repo.list(offset=1)
|
||||
# offset=1 skips the first (most-recent) row → expect 2 rows back
|
||||
assert len(rows) == 2
|
||||
|
||||
Reference in New Issue
Block a user