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:
DumiFlex
2026-04-26 15:50:20 +03:00
parent 6c29f738c5
commit 759626b914
3 changed files with 67 additions and 14 deletions
+4 -1
View File
@@ -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",
),
)
+30 -6
View File
@@ -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]
+33 -7
View File
@@ -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