bruno
fc8548194a
Templates (v7.71.x) : - registre unifié \ emplates\ (migrations 48-49) + TemplateService.instantiate unique (UI, API v2, agent, scheduler) - sélecteur (pilule page vide, menu •••, commande /template), gestionnaire /templates, menu New ▾, From template, base inline dans un document - 141 presets système (59 pages, 42 bases, 15 blocs, 25 lignes), titre auto depuis le template, variables title réservée - récurrences RRULE + scheduler dédupliqué, agent apply_template/list_templates, API /api/templates + /api/v2/fd-templates - correctifs : bouton Templates, centrage fenêtre, filtres CSP, flux de création, variable title - tests : tests/test_fd_templates.py (19) et e2e/templates_picker.spec.js (8) Inclut le travail déjà présent dans le working tree (vues Notion : view_query/view_aggregate/form_projection/geocoding, property_types, database_table, docs agents-skills) et ignore .playwright-mcp/.
387 lines
17 KiB
Python
387 lines
17 KiB
Python
"""Templates unifiés façon Notion — tests de la refonte (§17.2 du cahier des charges).
|
|
|
|
Couvre : migrations 48/49, backfill idempotent, presets système,
|
|
résolution des variables (aucun {{token}} survivant), dégradation sur
|
|
propriété renommée, défaut exclusif, idempotence, récurrence dédupliquée,
|
|
API interne et annulation (undo).
|
|
"""
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from app.db import get_conn
|
|
from app.services.fd_templates import TemplateService, registry_ready
|
|
|
|
|
|
@pytest.fixture
|
|
def actor():
|
|
return {"id": 1, "login": "tester", "full_name": "Tester", "is_admin": 1}
|
|
|
|
|
|
def _ensure_ws(conn, actor):
|
|
"""Workspace valide (FK) pour les inserts directs des tests."""
|
|
from app.services.fd_templates import _ensure_workspace
|
|
return _ensure_workspace(conn, actor)
|
|
|
|
|
|
def _db_version():
|
|
with get_conn() as conn:
|
|
return conn.execute("SELECT COALESCE(MAX(version),0) FROM schema_version").fetchone()[0]
|
|
|
|
|
|
def test_migrations_48_49_applied(client):
|
|
assert registry_ready(get_conn().__enter__()) or True
|
|
with get_conn() as conn:
|
|
assert registry_ready(conn)
|
|
tables = {r[0] for r in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
|
|
assert {"templates", "template_runs", "template_recurrences"} <= tables
|
|
assert _db_version() >= 49
|
|
|
|
|
|
def test_backfill_idempotent_and_presets(client, actor):
|
|
with get_conn() as conn:
|
|
first = TemplateService.backfill_registry(conn)
|
|
n1 = conn.execute("SELECT COUNT(*) FROM templates WHERE deleted_at IS NULL").fetchone()[0]
|
|
second = TemplateService.backfill_registry(conn)
|
|
n2 = conn.execute("SELECT COUNT(*) FROM templates WHERE deleted_at IS NULL").fetchone()[0]
|
|
assert n1 == n2
|
|
assert sum(second.values()) == 0
|
|
# Silos legacy catalogués, jamais détruits.
|
|
legacy_db = conn.execute("SELECT COUNT(*) FROM database_templates").fetchone()[0]
|
|
reg_db = conn.execute(
|
|
"SELECT COUNT(*) FROM templates WHERE source_kind='legacy_database'").fetchone()[0]
|
|
assert reg_db == legacy_db and legacy_db > 0
|
|
# Galerie système : au moins 5 presets utilisables.
|
|
n_presets = TemplateService.ensure_system_presets(conn)
|
|
sys_rows = conn.execute(
|
|
"SELECT COUNT(*) FROM templates WHERE scope='system' AND deleted_at IS NULL").fetchone()[0]
|
|
assert sys_rows >= 5
|
|
assert TemplateService.ensure_system_presets(conn) == 0
|
|
assert first["legacy_database"] >= 0
|
|
|
|
|
|
def test_instantiate_page_resolves_variables(client, actor):
|
|
with get_conn() as conn:
|
|
dto = TemplateService.create_template(conn, actor, {
|
|
"kind": "page", "name": "Revue", "scope": "personal",
|
|
"title_template": "Revue du {{today}} — {{me}}",
|
|
"blocks": [{"id": "b1", "type": "paragraph", "content": "vu le {{today}}"}],
|
|
})
|
|
conn.commit()
|
|
result = TemplateService.instantiate(conn, actor, dto["id"], {}, {})
|
|
assert result["created_type"] == "page"
|
|
assert "{{" not in result["url"] and result["url"].startswith("/pages/")
|
|
page = conn.execute("SELECT * FROM pages WHERE id=?",
|
|
(int(result["created_id"]),)).fetchone()
|
|
assert "{{" not in (page["title"] or "")
|
|
assert "{{" not in (page["content"] or "")
|
|
assert "Revue du 20" in page["title"] # date du jour résolue
|
|
runs = conn.execute("SELECT * FROM template_runs WHERE template_id=?",
|
|
(dto["id"],)).fetchall()
|
|
assert len(runs) == 1 and runs[0]["status"] == "ok"
|
|
|
|
|
|
def test_required_variable_blocks_write(client, actor):
|
|
with get_conn() as conn:
|
|
dto = TemplateService.create_template(conn, actor, {
|
|
"kind": "page", "name": "Spec", "scope": "personal",
|
|
"title_template": "Spec — {{sujet}}",
|
|
"variables": [{"name": "sujet", "type": "text", "required": True}],
|
|
"blocks": [],
|
|
})
|
|
conn.commit()
|
|
before = conn.execute("SELECT COUNT(*) FROM pages").fetchone()[0]
|
|
with pytest.raises(ValueError, match="sujet"):
|
|
TemplateService.instantiate(conn, actor, dto["id"], {}, {})
|
|
after = conn.execute("SELECT COUNT(*) FROM pages").fetchone()[0]
|
|
assert before == after # erreur AVANT toute écriture
|
|
ok = TemplateService.instantiate(conn, actor, dto["id"], {"sujet": "X"}, {})
|
|
assert ok["created_type"] == "page"
|
|
|
|
|
|
def test_row_template_degraded_property_warning(client, actor):
|
|
with get_conn() as conn:
|
|
ws = _ensure_ws(conn, actor)
|
|
cid = conn.execute(
|
|
"INSERT INTO collections (name, description, icon, schema_json, workspace_id) VALUES (?,?,?,?,?)",
|
|
("Taches", "", "✅", "[]", ws)).lastrowid
|
|
conn.execute(
|
|
"INSERT INTO collection_properties (collection_id, name, prop_type, position) VALUES (?,?,?,?)",
|
|
(cid, "Status", "status", 0))
|
|
conn.commit()
|
|
dto = TemplateService.create_template(conn, actor, {
|
|
"kind": "row", "name": "Bug", "scope": "personal",
|
|
"target_collection_id": cid,
|
|
"manifest": {"property_defaults": {"Status": "À trier", "Priorite": "P1"}},
|
|
"blocks": [],
|
|
})
|
|
conn.commit()
|
|
result = TemplateService.instantiate(conn, actor, dto["id"], {},
|
|
{"collection_id": cid})
|
|
assert result["created_type"] == "row"
|
|
# Propriété renommée → avertissement, pas d'échec total.
|
|
assert any("Priorite" in w for w in result["warnings"])
|
|
row = conn.execute("SELECT * FROM collection_pages WHERE id=?",
|
|
(int(result["created_id"]),)).fetchone()
|
|
assert row is not None
|
|
|
|
|
|
def test_default_exclusive(client, actor):
|
|
with get_conn() as conn:
|
|
ws = _ensure_ws(conn, actor)
|
|
cid = conn.execute(
|
|
"INSERT INTO collections (name, schema_json, workspace_id) VALUES (?,?,?)",
|
|
("B", "[]", ws)).lastrowid
|
|
conn.commit()
|
|
a = TemplateService.create_template(conn, actor, {
|
|
"kind": "row", "name": "A", "scope": "personal",
|
|
"target_collection_id": cid, "blocks": []})
|
|
b = TemplateService.create_template(conn, actor, {
|
|
"kind": "row", "name": "B", "scope": "personal",
|
|
"target_collection_id": cid, "blocks": []})
|
|
conn.commit()
|
|
TemplateService.set_default(conn, actor, a["id"])
|
|
TemplateService.set_default(conn, actor, b["id"])
|
|
conn.commit()
|
|
rows = conn.execute(
|
|
"SELECT id FROM templates WHERE target_collection_id=? AND is_default=1 AND deleted_at IS NULL",
|
|
(cid,)).fetchall()
|
|
assert [r["id"] for r in rows] == [b["id"]]
|
|
|
|
|
|
def test_idempotency_single_object(client, actor):
|
|
with get_conn() as conn:
|
|
dto = TemplateService.create_template(conn, actor, {
|
|
"kind": "page", "name": "Idem", "scope": "personal", "blocks": []})
|
|
conn.commit()
|
|
key = "test-key-123"
|
|
r1 = TemplateService.instantiate(conn, actor, dto["id"], {}, {},
|
|
idempotency_key=key)
|
|
r2 = TemplateService.instantiate(conn, actor, dto["id"], {}, {},
|
|
idempotency_key=key)
|
|
assert r1["created_id"] == r2["created_id"]
|
|
assert r2.get("replayed") is True
|
|
n = conn.execute(
|
|
"SELECT COUNT(*) FROM pages WHERE title=?", ("Idem",)).fetchone()[0]
|
|
assert n == 1
|
|
|
|
|
|
def test_recurrence_dedup(client, actor):
|
|
with get_conn() as conn:
|
|
ws = _ensure_ws(conn, actor)
|
|
cid = conn.execute(
|
|
"INSERT INTO collections (name, schema_json, workspace_id) VALUES (?,?,?)",
|
|
("Rec", "[]", ws)).lastrowid
|
|
conn.commit()
|
|
dto = TemplateService.create_template(conn, actor, {
|
|
"kind": "row", "name": "Quotidien", "scope": "personal",
|
|
"target_collection_id": cid, "blocks": []})
|
|
conn.commit()
|
|
TemplateService.set_recurrence(conn, actor, dto["id"],
|
|
{"rrule": "FREQ=DAILY", "timezone": ""})
|
|
# Prochain passage forcé dans le passé pour le test.
|
|
conn.execute("UPDATE template_recurrences SET next_run_at='2000-01-01T09:00:00' WHERE template_id=?",
|
|
(dto["id"],))
|
|
conn.commit()
|
|
fired1 = TemplateService.run_due_recurrences(conn, now_iso="2000-01-02T10:00:00")
|
|
assert len(fired1) == 1 and fired1[0]["status"] == "ok"
|
|
# Même occurrence rejouée (redémarrage scheduler) → pas de doublon.
|
|
conn.execute("UPDATE template_recurrences SET next_run_at='2000-01-01T09:00:00' WHERE template_id=?",
|
|
(dto["id"],))
|
|
conn.commit()
|
|
fired2 = TemplateService.run_due_recurrences(conn, now_iso="2000-01-02T10:00:00")
|
|
assert fired2 == []
|
|
n = conn.execute("SELECT COUNT(*) FROM collection_pages WHERE collection_id=?",
|
|
(cid,)).fetchone()[0]
|
|
assert n == 1
|
|
|
|
|
|
def test_api_list_instantiate_undo(client):
|
|
# Liste (le bouton « Templates » consomme cette route).
|
|
r = client.get("/api/templates?kind=page")
|
|
assert r.status_code == 200
|
|
assert isinstance(r.json()["templates"], list)
|
|
assert len(r.json()["templates"]) >= 5 # presets système
|
|
tid = r.json()["templates"][0]["id"]
|
|
# Détail + aperçu.
|
|
assert client.get(f"/api/templates/{tid}").status_code == 200
|
|
pv = client.post(f"/api/templates/{tid}/preview", json={})
|
|
assert pv.status_code == 200
|
|
# Instanciation puis annulation (titre fourni pour les variables requises).
|
|
ins = client.post(f"/api/templates/{tid}/instantiate",
|
|
json={"variables": {"title": "Test"}})
|
|
assert ins.status_code == 200, ins.text
|
|
body = ins.json()
|
|
assert body["url"]
|
|
run_id = body["run_id"]
|
|
if run_id:
|
|
undo = client.post(f"/api/templates/runs/{run_id}/undo")
|
|
assert undo.status_code == 200
|
|
# 404 explicite, jamais un no-op silencieux.
|
|
assert client.get("/api/templates/nope-123").status_code == 404
|
|
bad = client.post("/api/templates/nope-123/instantiate", json={})
|
|
assert bad.status_code == 404
|
|
|
|
|
|
def test_manager_page_renders(client):
|
|
r = client.get("/templates")
|
|
assert r.status_code == 200
|
|
assert "Templates" in r.text
|
|
|
|
|
|
_VALID_BLOCK_TYPES = {
|
|
"paragraph", "heading_1", "heading_2", "heading_3", "heading_4",
|
|
"bulleted_list", "numbered_list", "to_do", "toggle", "quote", "callout",
|
|
"divider", "code", "table_of_contents", "math", "columns", "table",
|
|
}
|
|
|
|
|
|
def test_system_presets_catalog_is_complete(client, actor):
|
|
"""Galerie : > 20 presets, tous les genres, types de blocs valides."""
|
|
from app.services.fd_templates import SYSTEM_PRESETS
|
|
|
|
names = {(p["kind"], p["name"]) for p in SYSTEM_PRESETS}
|
|
assert len(SYSTEM_PRESETS) >= 25
|
|
assert len(names) == len(SYSTEM_PRESETS) # pas de doublon
|
|
kinds = {p["kind"] for p in SYSTEM_PRESETS}
|
|
assert {"page", "database", "blocks", "row"} <= kinds
|
|
# Les blocs utilisent les types réels de l'éditeur (pas heading/todo).
|
|
for p in SYSTEM_PRESETS:
|
|
for b in p.get("blocks") or []:
|
|
assert b["type"] in _VALID_BLOCK_TYPES, (p["name"], b["type"])
|
|
with get_conn() as conn:
|
|
TemplateService.ensure_system_presets(conn)
|
|
rows = conn.execute(
|
|
"SELECT kind, COUNT(*) n FROM templates WHERE scope='system' AND deleted_at IS NULL GROUP BY kind"
|
|
).fetchall()
|
|
by_kind = {r["kind"]: r["n"] for r in rows}
|
|
assert sum(by_kind.values()) >= 25
|
|
assert by_kind.get("page", 0) >= 10
|
|
assert by_kind.get("database", 0) >= 8
|
|
assert by_kind.get("blocks", 0) >= 4
|
|
assert by_kind.get("row", 0) >= 4
|
|
|
|
|
|
def test_database_preset_creates_properties(client, actor):
|
|
with get_conn() as conn:
|
|
items = TemplateService.list_templates(conn, actor, kind="database")
|
|
tache = next(t for t in items if t["name"] == "Tâches")
|
|
res = TemplateService.instantiate(conn, actor, tache["id"], {}, {})
|
|
assert res["created_type"] == "collection"
|
|
props = conn.execute(
|
|
"SELECT name FROM collection_properties WHERE collection_id=?",
|
|
(int(res["created_id"]),)).fetchall()
|
|
names = {p["name"] for p in props}
|
|
assert {"Statut", "Priorité", "Échéance", "Assigné à"} <= names
|
|
|
|
|
|
def test_row_preset_resolves_base_by_name(client, actor):
|
|
with get_conn() as conn:
|
|
ws = _ensure_ws(conn, actor)
|
|
cid = conn.execute(
|
|
"INSERT INTO collections (name, schema_json, workspace_id) VALUES (?,?,?)",
|
|
("Tâches", "[]", ws)).lastrowid
|
|
conn.commit()
|
|
items = TemplateService.list_templates(conn, actor, kind="row")
|
|
tache = next(t for t in items if t["name"] == "Tâche")
|
|
# Sans collection_id : la base est résolue par nom (manifest).
|
|
res = TemplateService.instantiate(conn, actor, tache["id"], {}, {})
|
|
assert res["created_type"] == "row"
|
|
row = conn.execute(
|
|
"SELECT collection_id FROM collection_pages WHERE id=?",
|
|
(int(res["created_id"]),)).fetchone()
|
|
assert row["collection_id"] == cid
|
|
|
|
|
|
def test_row_preset_without_base_reports_clearly(client, actor):
|
|
with get_conn() as conn:
|
|
items = TemplateService.list_templates(conn, actor, kind="row")
|
|
tache = next(t for t in items if t["name"] == "Tâche")
|
|
# Aucune base « Tâches » → message explicite, pas d'objet créé.
|
|
with pytest.raises(ValueError, match="base"):
|
|
TemplateService.instantiate(conn, actor, tache["id"], {}, {})
|
|
|
|
|
|
def test_page_template_fills_current_page_and_sets_title(client, actor):
|
|
"""Appliquer un template de page SANS remplir de variable ne doit pas
|
|
échouer (title réservé), et le titre suit le template."""
|
|
with get_conn() as conn:
|
|
ws = _ensure_ws(conn, actor)
|
|
page_id = conn.execute(
|
|
"INSERT INTO pages (workspace, workspace_id, title, content, content_format) "
|
|
"VALUES (?,?,?,?,?)",
|
|
("T", ws, "Untitled", "[]", "blocks")).lastrowid
|
|
conn.commit()
|
|
items = TemplateService.list_templates(conn, actor, kind="page")
|
|
tpl = next(t for t in items if t["name"] == "Compte rendu client")
|
|
res = TemplateService.instantiate(
|
|
conn, actor, tpl["id"], {},
|
|
{"target": {"type": "current_page", "page_id": page_id},
|
|
"parent_id": page_id})
|
|
conn.commit()
|
|
assert res["created_type"] == "page"
|
|
assert res["title"].startswith("CR —")
|
|
page = conn.execute("SELECT title FROM pages WHERE id=?", (page_id,)).fetchone()
|
|
assert page["title"] == res["title"]
|
|
|
|
|
|
def test_page_template_instantiates_without_variables(client):
|
|
"""Un template de page s'instancie sans variables (title réservé)."""
|
|
r = client.get("/api/templates?kind=page")
|
|
for t in r.json()["templates"]:
|
|
resp = client.post(f"/api/templates/{t['id']}/instantiate", json={"variables": {}})
|
|
assert resp.status_code == 200, (t["name"], resp.text)
|
|
|
|
|
|
def test_preview_never_400_on_required_vars(client, actor):
|
|
"""L'aperçu ne bloque jamais (tokens montrés tels quels, §7.1)."""
|
|
with get_conn() as conn:
|
|
dto = TemplateService.create_template(conn, actor, {
|
|
"kind": "page", "name": "Spec", "scope": "personal",
|
|
"title_template": "Spec — {{sujet}}",
|
|
"variables": [{"name": "sujet", "type": "text", "required": True}],
|
|
"blocks": [],
|
|
})
|
|
conn.commit()
|
|
preview = TemplateService.preview(conn, actor, dto["id"], {})
|
|
assert "{{sujet}}" in preview["title"]
|
|
assert any("sujet" in w for w in preview["warnings"])
|
|
|
|
|
|
def test_preview_endpoint_ok_for_system_presets(client):
|
|
r = client.get("/api/templates?kind=page")
|
|
assert r.status_code == 200
|
|
for t in r.json()["templates"]:
|
|
pv = client.post(f"/api/templates/{t['id']}/preview", json={})
|
|
assert pv.status_code == 200, (t["name"], pv.text)
|
|
|
|
|
|
def test_database_template_inline_in_current_page(client, actor):
|
|
"""Un template de base appliqué depuis un document crée une base INLINE
|
|
rattachée à la page (pas de navigation, embed côté client)."""
|
|
with get_conn() as conn:
|
|
ws = _ensure_ws(conn, actor)
|
|
page_id = conn.execute(
|
|
"INSERT INTO pages (workspace, workspace_id, title, content, content_format) VALUES (?,?,?,?,?)",
|
|
("Test", ws, "Doc", "[]", "blocks")).lastrowid
|
|
conn.commit()
|
|
dto = TemplateService.create_template(conn, actor, {
|
|
"kind": "database", "name": "Projets", "scope": "personal",
|
|
"manifest": {"collection": {"name": "Projets", "schema": [
|
|
{"name": "Status", "type": "status"}]}},
|
|
})
|
|
conn.commit()
|
|
result = TemplateService.instantiate(
|
|
conn, actor, dto["id"], {}, {"parent_page_id": page_id})
|
|
assert result["created_type"] == "collection"
|
|
assert result["url"] == f"/db/{result['created_id']}"
|
|
col = conn.execute("SELECT * FROM collections WHERE id=?",
|
|
(int(result["created_id"]),)).fetchone()
|
|
assert col["parent_page_id"] == page_id
|
|
assert col["is_inline"] == 1
|
|
props = conn.execute(
|
|
"SELECT name FROM collection_properties WHERE collection_id=?",
|
|
(col["id"],)).fetchall()
|
|
assert [p["name"] for p in props] == ["Status"]
|