diff --git a/src/migrations.py b/src/migrations.py index fcbc40d..6812a83 100644 --- a/src/migrations.py +++ b/src/migrations.py @@ -74,6 +74,10 @@ _MIGRATIONS: list[tuple[str, str]] = [ "id INTEGER PRIMARY KEY, user_id INTEGER, token_id INTEGER, " "kind TEXT NOT NULL, created_at TEXT NOT NULL)", ), + ( + "0012_add_token_to_saved_views", + "ALTER TABLE saved_views ADD COLUMN token TEXT", + ), ] diff --git a/src/saved_views/db.py b/src/saved_views/db.py index 533fa8e..1931aa7 100644 --- a/src/saved_views/db.py +++ b/src/saved_views/db.py @@ -1,4 +1,6 @@ +import secrets import sqlite3 +import string from datetime import datetime, timezone from src.auth.db import get_conn @@ -10,6 +12,7 @@ CREATE TABLE IF NOT EXISTS saved_views ( table_name TEXT NOT NULL DEFAULT 'tableau', name TEXT NOT NULL, query TEXT NOT NULL, + token TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, @@ -17,15 +20,45 @@ CREATE TABLE IF NOT EXISTS saved_views ( ); CREATE INDEX IF NOT EXISTS idx_saved_views_user ON saved_views(user_id, table_name); +CREATE UNIQUE INDEX IF NOT EXISTS idx_saved_views_token + ON saved_views(token); """ +_TOKEN_ALPHABET = string.ascii_letters + string.digits # base62 + + +def generate_token(length: int = 6) -> str: + return "".join(secrets.choice(_TOKEN_ALPHABET) for _ in range(length)) + + +def _unique_token(conn) -> str: + while True: + token = generate_token() + exists = conn.execute( + "SELECT 1 FROM saved_views WHERE token = ?", (token,) + ).fetchone() + if exists is None: + return token + def _now() -> str: return datetime.now(timezone.utc).isoformat() def init_schema() -> None: - get_conn().executescript(SCHEMA) + conn = get_conn() + conn.executescript(SCHEMA) + # Backfill des lignes pré-migration (token NULL). L'index unique tolère + # plusieurs NULL transitoires ; on attribue un jeton à chacune. Idempotent : + # sans effet une fois toutes les lignes pourvues. + null_rows = conn.execute( + "SELECT id FROM saved_views WHERE token IS NULL" + ).fetchall() + for row in null_rows: + conn.execute( + "UPDATE saved_views SET token = ? WHERE id = ?", + (_unique_token(conn), row["id"]), + ) def list_views(user_id: int, table_name: str = "tableau") -> list[sqlite3.Row]: @@ -51,16 +84,33 @@ def get(view_id: int, user_id: int) -> sqlite3.Row | None: ) -def upsert(user_id: int, table_name: str, name: str, query: str) -> None: +def get_by_token(token: str) -> sqlite3.Row | None: + return ( + get_conn() + .execute("SELECT * FROM saved_views WHERE token = ?", (token,)) + .fetchone() + ) + + +def upsert(user_id: int, table_name: str, name: str, query: str) -> str: now = _now() - get_conn().execute( + conn = get_conn() + # Jeton candidat, utilisé uniquement en cas d'INSERT réel ; à l'écrasement + # (ON CONFLICT ... DO UPDATE), `token` n'est pas dans le SET → conservé. + conn.execute( "INSERT INTO saved_views " - "(user_id, table_name, name, query, created_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?) " + "(user_id, table_name, name, query, token, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?) " "ON CONFLICT(user_id, table_name, name) DO UPDATE SET " "query = excluded.query, updated_at = excluded.updated_at", - (user_id, table_name, name, query, now, now), + (user_id, table_name, name, query, _unique_token(conn), now, now), ) + row = conn.execute( + "SELECT token FROM saved_views " + "WHERE user_id = ? AND table_name = ? AND name = ?", + (user_id, table_name, name), + ).fetchone() + return row["token"] def rename(view_id: int, user_id: int, new_name: str) -> None: diff --git a/tests/saved_views/test_db.py b/tests/saved_views/test_db.py index 8b71bca..f711f79 100644 --- a/tests/saved_views/test_db.py +++ b/tests/saved_views/test_db.py @@ -1,3 +1,5 @@ +import string + from src.auth import db as auth_db from src.saved_views import db @@ -75,3 +77,70 @@ def test_views_deleted_on_user_cascade(users_db_path): db.upsert(uid, "tableau", "Vue A", "filtres=a") auth_db.delete_user(uid) assert db.list_views(uid, "tableau") == [] + + +def test_generate_token_is_base62_and_length_6(): + token = db.generate_token() + assert len(token) == 6 + alphabet = set(string.ascii_letters + string.digits) + assert set(token) <= alphabet + + +def test_upsert_returns_token_on_insert(users_db_path): + db.init_schema() + uid = _make_user() + token = db.upsert(uid, "tableau", "Ma vue", "q1") + assert token + assert db.list_views(uid, "tableau")[0]["token"] == token + + +def test_upsert_preserves_token_on_overwrite(users_db_path): + db.init_schema() + uid = _make_user() + token1 = db.upsert(uid, "tableau", "Ma vue", "q1") + token2 = db.upsert(uid, "tableau", "Ma vue", "q2") + assert token2 == token1 # écrasement → lien stable + assert db.list_views(uid, "tableau")[0]["query"] == "q2" + + +def test_get_by_token_public_lookup(users_db_path): + db.init_schema() + uid = _make_user() + token = db.upsert(uid, "tableau", "Ma vue", "q1") + row = db.get_by_token(token) + assert row is not None + assert row["name"] == "Ma vue" + assert db.get_by_token("zzzzzz") is None + + +def test_tokens_are_unique_across_views(users_db_path): + db.init_schema() + uid = _make_user() + t1 = db.upsert(uid, "tableau", "Vue A", "a") + t2 = db.upsert(uid, "tableau", "Vue B", "b") + assert t1 != t2 + + +def test_backfill_assigns_tokens_to_null_rows(users_db_path): + db.init_schema() + uid = _make_user() + conn = auth_db.get_conn() + # Simule une ligne pré-migration (token NULL) en contournant upsert. + conn.execute( + "INSERT INTO saved_views " + "(user_id, table_name, name, query, token, created_at, updated_at) " + "VALUES (?, 'tableau', 'Ancienne', 'q', NULL, '', '')", + (uid,), + ) + db.init_schema() # doit backfiller + row = conn.execute( + "SELECT token FROM saved_views WHERE name = 'Ancienne'" + ).fetchone() + assert row["token"] and len(row["token"]) == 6 + # Idempotent : un second appel ne change pas le jeton attribué. + token_after_first = row["token"] + db.init_schema() + row2 = conn.execute( + "SELECT token FROM saved_views WHERE name = 'Ancienne'" + ).fetchone() + assert row2["token"] == token_after_first