From 03023a6b8c3a260f80588c4941340492d965e000 Mon Sep 17 00:00:00 2001 From: Colin Maudry Date: Mon, 29 Jun 2026 16:14:00 +0200 Subject: [PATCH] feat: table saved_views et CRUD #95 --- src/app.py | 4 ++ src/saved_views/__init__.py | 0 src/saved_views/db.py | 77 +++++++++++++++++++++++++++++++++++ tests/saved_views/__init__.py | 0 tests/saved_views/conftest.py | 12 ++++++ tests/saved_views/test_db.py | 77 +++++++++++++++++++++++++++++++++++ 6 files changed, 170 insertions(+) create mode 100644 src/saved_views/__init__.py create mode 100644 src/saved_views/db.py create mode 100644 tests/saved_views/__init__.py create mode 100644 tests/saved_views/conftest.py create mode 100644 tests/saved_views/test_db.py diff --git a/src/app.py b/src/app.py index 64d848a..4a2e65c 100644 --- a/src/app.py +++ b/src/app.py @@ -105,6 +105,10 @@ from src.subscriptions.setup import init_subscriptions # noqa: E402 init_subscriptions(app.server) +from src.saved_views import db as saved_views_db # noqa: E402 + +saved_views_db.init_schema() + # robots.txt @app.server.route("/robots.txt") diff --git a/src/saved_views/__init__.py b/src/saved_views/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/saved_views/db.py b/src/saved_views/db.py new file mode 100644 index 0000000..533fa8e --- /dev/null +++ b/src/saved_views/db.py @@ -0,0 +1,77 @@ +import sqlite3 +from datetime import datetime, timezone + +from src.auth.db import get_conn + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS saved_views ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + table_name TEXT NOT NULL DEFAULT 'tableau', + name TEXT NOT NULL, + query TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, + UNIQUE (user_id, table_name, name) +); +CREATE INDEX IF NOT EXISTS idx_saved_views_user + ON saved_views(user_id, table_name); +""" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def init_schema() -> None: + get_conn().executescript(SCHEMA) + + +def list_views(user_id: int, table_name: str = "tableau") -> list[sqlite3.Row]: + return ( + get_conn() + .execute( + "SELECT * FROM saved_views WHERE user_id = ? AND table_name = ? " + "ORDER BY name COLLATE NOCASE", + (user_id, table_name), + ) + .fetchall() + ) + + +def get(view_id: int, user_id: int) -> sqlite3.Row | None: + return ( + get_conn() + .execute( + "SELECT * FROM saved_views WHERE id = ? AND user_id = ?", + (view_id, user_id), + ) + .fetchone() + ) + + +def upsert(user_id: int, table_name: str, name: str, query: str) -> None: + now = _now() + get_conn().execute( + "INSERT INTO saved_views " + "(user_id, table_name, name, query, 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), + ) + + +def rename(view_id: int, user_id: int, new_name: str) -> None: + get_conn().execute( + "UPDATE saved_views SET name = ?, updated_at = ? WHERE id = ? AND user_id = ?", + (new_name, _now(), view_id, user_id), + ) + + +def delete(view_id: int, user_id: int) -> None: + get_conn().execute( + "DELETE FROM saved_views WHERE id = ? AND user_id = ?", + (view_id, user_id), + ) diff --git a/tests/saved_views/__init__.py b/tests/saved_views/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/saved_views/conftest.py b/tests/saved_views/conftest.py new file mode 100644 index 0000000..ce77c3b --- /dev/null +++ b/tests/saved_views/conftest.py @@ -0,0 +1,12 @@ +import pytest + + +@pytest.fixture +def users_db_path(monkeypatch, tmp_path): + from src.auth.db import reset_conn_for_tests + + db_path = tmp_path / "users.test.sqlite" + monkeypatch.setenv("USERS_DB_PATH", str(db_path)) + reset_conn_for_tests() + yield db_path + reset_conn_for_tests() diff --git a/tests/saved_views/test_db.py b/tests/saved_views/test_db.py new file mode 100644 index 0000000..8b71bca --- /dev/null +++ b/tests/saved_views/test_db.py @@ -0,0 +1,77 @@ +from src.auth import db as auth_db +from src.saved_views import db + + +def _make_user(email="u@ex.fr"): + auth_db.init_schema() + return auth_db.create_user(email, "hash") + + +def test_init_schema_creates_table(users_db_path): + db.init_schema() + conn = auth_db.get_conn() + tables = { + row[0] + for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'") + } + assert "saved_views" in tables + + +def test_upsert_creates_and_lists(users_db_path): + db.init_schema() + uid = _make_user() + db.upsert(uid, "tableau", "Ma vue", "filtres=foo") + views = db.list_views(uid, "tableau") + assert len(views) == 1 + assert views[0]["name"] == "Ma vue" + assert views[0]["query"] == "filtres=foo" + + +def test_upsert_same_name_overwrites(users_db_path): + db.init_schema() + uid = _make_user() + db.upsert(uid, "tableau", "Ma vue", "filtres=foo") + db.upsert(uid, "tableau", "Ma vue", "filtres=bar") + views = db.list_views(uid, "tableau") + assert len(views) == 1 + assert views[0]["query"] == "filtres=bar" + + +def test_list_views_is_isolated_per_user(users_db_path): + db.init_schema() + uid1 = _make_user("a@ex.fr") + uid2 = _make_user("b@ex.fr") + db.upsert(uid1, "tableau", "Vue A", "filtres=a") + assert db.list_views(uid2, "tableau") == [] + + +def test_rename_only_affects_owner(users_db_path): + db.init_schema() + uid1 = _make_user("a@ex.fr") + uid2 = _make_user("b@ex.fr") + db.upsert(uid1, "tableau", "Vue A", "filtres=a") + view_id = db.list_views(uid1, "tableau")[0]["id"] + db.rename(view_id, uid2, "Pirate") # mauvais propriétaire → no-op + assert db.get(view_id, uid1)["name"] == "Vue A" + db.rename(view_id, uid1, "Vue B") + assert db.get(view_id, uid1)["name"] == "Vue B" + + +def test_delete_only_affects_owner(users_db_path): + db.init_schema() + uid1 = _make_user("a@ex.fr") + uid2 = _make_user("b@ex.fr") + db.upsert(uid1, "tableau", "Vue A", "filtres=a") + view_id = db.list_views(uid1, "tableau")[0]["id"] + db.delete(view_id, uid2) # mauvais propriétaire → no-op + assert db.get(view_id, uid1) is not None + db.delete(view_id, uid1) + assert db.get(view_id, uid1) is None + + +def test_views_deleted_on_user_cascade(users_db_path): + db.init_schema() + uid = _make_user() + db.upsert(uid, "tableau", "Vue A", "filtres=a") + auth_db.delete_user(uid) + assert db.list_views(uid, "tableau") == []