feat(calendar): v5.8.0 calendrier & rappels + v5.7.0 database avancee (pt.2)
FlowDeck CI / lint (push) Successful in 49s
FlowDeck CI / test (push) Successful in 5m26s
FlowDeck CI / docker (push) Successful in 43s

v5.8.0 (release 5.11.7) — Calendrier & Rappels :
- moteur de recurrence RRULE subset (daily/weekly/monthly, interval,
  count, until, byweekday, timezone) — app/services/recurrence.py
- vues calendar Jour / Semaine / Mois avec expansion des occurrences
  cote serveur (GET /db/{id}/calendar/api) et popover evenement
  (Time / Timezone / Repeat / Remind)
- rappels avant echeance (scan 60 s, table reminder_log, in-app +
  email, cible = personnes assignees) — app/services/reminders.py
- fuseaux horaires : users.timezone + reglages in-app, timezone par
  evenement, liste de zones (GET /db/timezones/api)
- notifications d'assignation sur PUT /db/pages/{id}/api et
  preferences etendues (reminders, assignments)
- template Meeting notes enrichi (Agenda, Notes — migration 12)
- migrations 11-12 ; 20 tests dedies ; suite complete 523 verte

v5.7.0 (release 5.11.6, termine avant cette session, reste dans
l'arbre sans commit) — Database Avancée Pt.2 :
- proprietes person + auto-proprietes (created/last-edited time & by)
- groupes de proprietes, vues sauvegardees par utilisateur
- swimlanes, WIP limits, cartes configurables, calendar drag & drop,
  gallery couvertures ; 12 tests dedies
This commit is contained in:
2026-09-13 22:49:14 -04:00
parent 055956a351
commit d4adf89db5
18 changed files with 3301 additions and 366 deletions
+277
View File
@@ -0,0 +1,277 @@
"""FlowDeck — v5.7.0 Database Avancée (Pt. 2).
Covers: person + auto properties (created/last-edited time & by), property
groups, per-user saved views, swimlanes/WIP/card/calendar view config, row
covers and the multi-view editor rendering.
"""
from __future__ import annotations
import json
import os
import secrets
import tempfile
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client():
db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
db_path = db_file.name
db_file.close()
os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
os.environ["APP_SECRET_KEY"] = "test-secret-for-tests"
os.environ["RATE_LIMIT_ENABLED"] = "false"
from app.config import settings
settings.database_url = f"sqlite:///{db_path}"
settings.rate_limit_enabled = False
from app.db import init_db
from app.main import app
init_db()
yield TestClient(app)
os.unlink(db_path)
def _add_collection(client, name="DB"):
return client.post("/db/api", json={"name": name}).json()["id"]
def _add_prop(client, cid, name, ptype, **extra):
body = {"name": name, "prop_type": ptype}
body.update(extra)
r = client.post(f"/db/{cid}/properties/api", json=body)
assert r.status_code == 200, r.text
return r.json()["id"]
def _props(client, cid):
return client.get(f"/db/{cid}/properties/api").json()["properties"]
def _login(client):
from app.auth.session import SessionManager
from app.db import get_conn
with get_conn() as conn:
login = f"v57admin_{secrets.token_hex(4)}"
conn.execute(
"INSERT INTO users (login, full_name, email, password_hash, is_admin) "
"VALUES (?, 'V57 Admin', ?, ?, 1)",
(login, f"{login}@test.com", ""),
)
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
conn.commit()
session = SessionManager.create_session({"id": uid, "login": login, "is_admin": 1})
csrf = client.get("/api/csrf-token", cookies={"flowdeck_session": session}).json()["csrf_token"]
return session, csrf, uid
# ── Schema / migration ──
def test_migration_v57_columns(client):
from app.db import get_conn
with get_conn() as conn:
pcols = {r[1] for r in conn.execute("PRAGMA table_info(collection_properties)").fetchall()}
vcols = {r[1] for r in conn.execute("PRAGMA table_info(collection_views)").fetchall()}
cpcols = {r[1] for r in conn.execute("PRAGMA table_info(collection_pages)").fetchall()}
version = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]
assert "group_name" in pcols
assert "created_by" in vcols
assert "cover_url" in cpcols
assert version >= 10
# ── Person property ──
def test_person_property_value_roundtrip(client):
cid = _add_collection(client)
pid = _add_prop(client, cid, "Assignee", "person")
r = client.post(f"/db/{cid}/pages/api", json={
"title": "Row", "properties": {str(pid): [{"id": 7, "login": "alice"}]}
})
assert r.status_code == 200
page = client.get(f"/db/pages/{r.json()['id']}/api").json()
pv = json.loads(page["property_values_json"])
assert pv[str(pid)] == [{"id": 7, "login": "alice"}]
def test_person_property_rejects_non_list(client):
cid = _add_collection(client)
pid = _add_prop(client, cid, "Owner", "person")
r = client.post(f"/db/{cid}/pages/api", json={
"title": "Row", "properties": {str(pid): "alice"}
})
assert r.status_code == 400
assert "person" in r.json()["detail"].lower()
def test_collection_members_endpoint(client):
from app.db import get_conn
with get_conn() as conn:
conn.execute(
"INSERT INTO users (login, full_name, email, avatar_color) VALUES ('bob', 'Bob', '[email protected]', '#123456')"
)
conn.commit()
cid = _add_collection(client)
r = client.get(f"/db/{cid}/members/api")
assert r.status_code == 200
members = r.json()["members"]
assert any(m["login"] == "bob" for m in members)
# ── Auto properties ──
def test_auto_properties_filled_on_create_and_update(client):
cid = _add_collection(client)
ct = _add_prop(client, cid, "Created", "created_time")
cb = _add_prop(client, cid, "Created by", "created_by")
lt = _add_prop(client, cid, "Edited", "last_edited_time")
lb = _add_prop(client, cid, "Edited by", "last_edited_by")
r = client.post(f"/db/{cid}/pages/api", json={"title": "Task"})
assert r.status_code == 200
page_id = r.json()["id"]
pv = json.loads(client.get(f"/db/pages/{page_id}/api").json()["property_values_json"])
assert pv[str(ct)]
assert pv[str(lt)]
assert pv[str(cb)]["login"] == "admin"
assert pv[str(lb)]["login"] == "admin"
created = pv[str(ct)]
# Partial update keeps created_time and refreshes last_edited_time.
r = client.put(f"/db/pages/{page_id}/api", json={"properties": {}})
assert r.status_code == 200
pv2 = json.loads(client.get(f"/db/pages/{page_id}/api").json()["property_values_json"])
assert pv2[str(ct)] == created
assert pv2[str(lt)] >= created
def test_partial_update_merges_properties(client):
cid = _add_collection(client)
a = _add_prop(client, cid, "A", "text")
b = _add_prop(client, cid, "B", "text")
r = client.post(f"/db/{cid}/pages/api", json={
"title": "Row", "properties": {str(a): "one", str(b): "two"}
})
pid = r.json()["id"]
client.put(f"/db/pages/{pid}/api", json={"properties": {str(a): "changed"}})
pv = json.loads(client.get(f"/db/pages/{pid}/api").json()["property_values_json"])
assert pv[str(a)] == "changed"
assert pv[str(b)] == "two"
# ── Property groups ──
def test_property_group_assignment(client):
cid = _add_collection(client)
p1 = _add_prop(client, cid, "Name2", "text", group_name="Basics")
p2 = _add_prop(client, cid, "Status2", "status")
props = {p["id"]: p for p in _props(client, cid)}
assert props[p1]["group_name"] == "Basics"
r = client.post(f"/db/{cid}/property-groups/api", json={
"groups": [{"name": "Workflow", "property_ids": [p1, p2]}]
})
assert r.status_code == 200
props = {p["id"]: p for p in _props(client, cid)}
assert props[p1]["group_name"] == "Workflow"
assert props[p2]["group_name"] == "Workflow"
# Clearing groups
client.post(f"/db/{cid}/property-groups/api", json={"groups": []})
props = {p["id"]: p for p in _props(client, cid)}
assert props[p1]["group_name"] == ""
# ── Per-user saved views ──
def test_saved_views_are_per_user(client):
cookie, csrf, uid = _login(client)
cid = _add_collection(client)
r = client.post(f"/db/{cid}/views/save-as",
json={"name": "My Board", "view_type": "board", "config": {"group_by": 1}},
cookies={"flowdeck_session": cookie}, headers={"X-CSRF-Token": csrf})
assert r.status_code == 200
view = r.json()
assert view["created_by"] == uid
assert view["view_type"] == "board"
# The owner sees it.
views = client.get(f"/db/{cid}/views/api", cookies={"flowdeck_session": cookie}).json()["views"]
assert any(v["id"] == view["id"] for v in views)
# Duplicate + delete.
dup = client.post(f"/db/views/{view['id']}/duplicate", json={},
cookies={"flowdeck_session": cookie}, headers={"X-CSRF-Token": csrf}).json()
assert dup["id"] != view["id"]
d = client.delete(f"/db/views/{view['id']}/api", cookies={"flowdeck_session": cookie},
headers={"X-CSRF-Token": csrf})
assert d.status_code == 200
views = client.get(f"/db/{cid}/views/api", cookies={"flowdeck_session": cookie}).json()["views"]
assert not any(v["id"] == view["id"] for v in views)
def test_view_config_extended_keys(client):
cid = _add_collection(client)
views = client.get(f"/db/{cid}/views/api").json()["views"]
vid = views[0]["id"]
payload = {
"sub_group_by": 3, "wip_limits": {"In progress": 2}, "card_size": "small",
"cover_mode": "color", "card_properties": [1, 2], "date_property": 4,
}
r = client.put(f"/db/views/{vid}/config", json=payload)
assert r.status_code == 200
cfg = r.json()["config"]
for k, v in payload.items():
assert cfg[k] == v
# ── Row cover ──
def test_row_cover_url(client):
cid = _add_collection(client)
r = client.post(f"/db/{cid}/pages/api", json={"title": "Row", "cover_url": "https://x.test/a.png"})
pid = r.json()["id"]
page = client.get(f"/db/pages/{pid}/api").json()
assert page["cover_url"] == "https://x.test/a.png"
client.put(f"/db/pages/{pid}/api", json={"cover_url": "https://x.test/b.png"})
page = client.get(f"/db/pages/{pid}/api").json()
assert page["cover_url"] == "https://x.test/b.png"
# ── Table-data payload includes views ──
def test_table_data_includes_views(client):
cid = _add_collection(client)
data = client.get(f"/api/collections/{cid}/table-data").json()
assert "views" in data
assert isinstance(data["views"], list)
assert len(data["views"]) >= 1
# ── Full-page editor renders the multi-view component ──
def test_collection_page_renders_multi_view(client):
cookie, csrf, uid = _login(client)
page = client.post(
"/board/api/pages?title=My%20DB&section=Private&parent_id=0",
headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": cookie},
).json()["id"]
r = client.post(f"/api/pages/{page}/convert-to-database", json={"name": "My DB"},
headers={"X-CSRF-Token": csrf}, cookies={"flowdeck_session": cookie})
assert r.status_code == 200
resp = client.get(f"/pages/{page}", cookies={"flowdeck_session": cookie})
assert resp.status_code == 200
assert "db-view-bar" in resp.text
assert "FlowDeckDB" in resp.text
# New view types + settings are available client-side.
assert "db-board" in resp.text
assert "db-cal-grid" in resp.text
assert "db-gallery" in resp.text
+419
View File
@@ -0,0 +1,419 @@
"""FlowDeck — v5.8.0 Calendrier & Rappels.
Covers: the recurrence engine (pure), migrations 11/12, meta-key validation
on the rows API, the server-expanded calendar endpoint, per-user timezones,
the reminder scanner (deterministic, injected clock), assignment
notifications and the enriched Meeting notes template.
"""
from __future__ import annotations
import datetime as dt
import json
import os
import secrets
import tempfile
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client():
db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
db_path = db_file.name
db_file.close()
os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
os.environ["APP_SECRET_KEY"] = "test-secret-for-tests"
os.environ["RATE_LIMIT_ENABLED"] = "false"
from app.config import settings
settings.database_url = f"sqlite:///{db_path}"
settings.rate_limit_enabled = False
from app.db import init_db
from app.main import app
init_db()
yield TestClient(app)
try:
os.unlink(db_path)
except PermissionError:
pass
def _login(client):
from app.auth.session import SessionManager
from app.db import get_conn
with get_conn() as conn:
login = f"v58u_{secrets.token_hex(4)}"
conn.execute(
"INSERT INTO users (login, full_name, email, password_hash, is_admin) "
"VALUES (?, 'V58 User', ?, ?, 1)",
(login, f"{login}@test.com", ""),
)
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
conn.commit()
session = SessionManager.create_session({"id": uid, "login": login, "is_admin": 1})
csrf = client.get("/api/csrf-token", cookies={"flowdeck_session": session}).json()["csrf_token"]
return session, csrf, uid
def _add_collection(client, name="Cal"):
return client.post("/db/api", json={"name": name}).json()["id"]
def _add_prop(client, cid, name, ptype, **extra):
body = {"name": name, "prop_type": ptype}
body.update(extra)
r = client.post(f"/db/{cid}/properties/api", json=body)
assert r.status_code == 200, r.text
return r.json()["id"]
# ── Recurrence engine (pure) ──
def test_expand_rule_daily_weekly_monthly():
from app.services.recurrence import expand_rule
w = (dt.date(2026, 9, 1), dt.date(2026, 9, 30))
assert expand_rule("2026-09-05", {"freq": "daily"}, *w)[:2] == ["2026-09-05", "2026-09-06"]
assert expand_rule("2026-09-07", {"freq": "weekly"}, *w) == \
["2026-09-07", "2026-09-14", "2026-09-21", "2026-09-28"]
assert expand_rule("2026-09-07", {"freq": "weekly", "interval": 2}, *w) == \
["2026-09-07", "2026-09-21"]
# Monthly anchored on the 31st clamps to shorter months.
months = expand_rule("2026-01-31", {"freq": "monthly"}, dt.date(2026, 1, 1), dt.date(2026, 4, 30))
assert months == ["2026-01-31", "2026-02-28", "2026-03-31", "2026-04-30"]
def test_expand_rule_count_until_window():
from app.services.recurrence import expand_rule
assert expand_rule("2026-09-05", {"freq": "daily", "count": 3},
dt.date(2026, 9, 1), dt.date(2026, 9, 30)) == \
["2026-09-05", "2026-09-06", "2026-09-07"]
assert expand_rule("2026-08-01", {"freq": "daily", "count": 10},
dt.date(2026, 9, 1), dt.date(2026, 9, 30)) == []
assert expand_rule("2026-09-05", {"freq": "daily", "until": "2026-09-08"},
dt.date(2026, 9, 1), dt.date(2026, 9, 30)) == \
["2026-09-05", "2026-09-06", "2026-09-07", "2026-09-08"]
def test_validate_rule_errors():
from app.services.recurrence import validate_rule
assert validate_rule({"freq": "yearly"})[0] is False
assert validate_rule({"freq": "daily", "count": 2, "until": "2026-01-01"})[0] is False
assert validate_rule({"freq": "weekly", "byweekday": [9]})[0] is False
assert validate_rule({"freq": "daily", "timezone": "Mars/Olympus"})[0] is False
assert validate_rule({"freq": "weekly", "byweekday": [0, 4], "timezone": "Europe/Paris"})[0] is True
assert validate_rule(None)[0] is True
# ── Schema / migrations ──
def test_migration_v58_schema(client):
from app.db import get_conn
with get_conn() as conn:
tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
ucols = {r[1] for r in conn.execute("PRAGMA table_info(users)").fetchall()}
version = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]
assert "reminder_log" in tables
assert "timezone" in ucols
assert version >= 12
def test_meeting_template_enriched(client):
from app.db import get_conn
with get_conn() as conn:
row = conn.execute(
"SELECT schema_json, description FROM database_templates WHERE name='Meeting notes'"
).fetchone()
schema = json.loads(row["schema_json"])
names = [p["name"] for p in schema]
assert "Agenda" in names and "Notes" in names
assert "agenda" in row["description"].lower()
# ── Meta-key validation on the rows API ──
def test_recurrence_meta_roundtrip_and_rejects(client):
cid = _add_collection(client)
did = _add_prop(client, cid, "When", "date")
rid = client.post(f"/db/{cid}/pages/api", json={
"title": "Standup",
"properties": {
str(did): "2026-09-15",
"__recurrence__": {str(did): {"freq": "daily"}},
},
}).json()["id"]
page = client.get(f"/db/pages/{rid}/api").json()
pv = json.loads(page["property_values_json"])
assert pv["__recurrence__"][str(did)]["freq"] == "daily"
# Bad freq → 400
r = client.put(f"/db/pages/{rid}/api", json={"properties": {
"__recurrence__": {str(did): {"freq": "fortnightly"}},
}})
assert r.status_code == 400
# Reminder on a non-date property → 400
tid = _add_prop(client, cid, "Text", "text")
r = client.put(f"/db/pages/{rid}/api", json={"properties": {
"__reminder__": {str(tid): {"value": 10, "unit": "minutes"}},
}})
assert r.status_code == 400
# Unknown timezone → 400
r = client.put(f"/db/pages/{rid}/api", json={"properties": {
"__timezone__": {str(did): "Mars/Olympus"},
}})
assert r.status_code == 400
def test_reminder_meta_roundtrip(client):
cid = _add_collection(client)
did = _add_prop(client, cid, "When", "date")
rid = client.post(f"/db/{cid}/pages/api", json={
"title": "Row",
"properties": {
str(did): "2026-10-01T09:00",
"__reminder__": {str(did): {"value": 2, "unit": "hours"}},
},
}).json()["id"]
pv = json.loads(client.get(f"/db/pages/{rid}/api").json()["property_values_json"])
assert pv["__reminder__"][str(did)] == {"value": 2, "unit": "hours"}
# ── Calendar API (server expansion) ──
def test_calendar_api_expands_recurrences(client):
cid = _add_collection(client)
did = _add_prop(client, cid, "When", "date")
client.post(f"/db/{cid}/pages/api", json={
"title": "Weekly sync",
"properties": {
str(did): "2026-09-07",
"__recurrence__": {str(did): {"freq": "weekly"}},
},
})
client.post(f"/db/{cid}/pages/api", json={ # plain one-off event
"title": "Party", "properties": {str(did): "2026-09-20T18:30"},
})
r = client.get(f"/db/{cid}/calendar/api",
params={"start": "2026-09-01", "end": "2026-09-30"})
assert r.status_code == 200
data = r.json()
evs = data["events"]
syncs = [e for e in evs if e["title"] == "Weekly sync"]
assert [e["date"] for e in syncs] == ["2026-09-07", "2026-09-14", "2026-09-21", "2026-09-28"]
assert all(e["recurring"] for e in syncs)
party = [e for e in evs if e["title"] == "Party"][0]
assert party["date"] == "2026-09-20" and party["time"] == "18:30" and not party["recurring"]
def test_calendar_api_param_errors(client):
cid = _add_collection(client)
_add_prop(client, cid, "When", "date")
assert client.get(f"/db/{cid}/calendar/api", params={"start": "bad"}).status_code == 400
assert client.get(f"/db/{cid}/calendar/api",
params={"start": "2026-09-01", "end": "2028-01-01"}).status_code == 400
assert client.get("/db/999999/calendar/api",
params={"start": "2026-09-01", "end": "2026-09-30"}).status_code == 404
cid2 = client.post("/db/api", json={"name": "NoDate"}).json()["id"]
r = client.get(f"/db/{cid2}/calendar/api", params={"start": "2026-09-01", "end": "2026-09-30"})
assert r.status_code == 200 and r.json()["events"] == []
def test_calendar_api_row_timezone(client):
cid = _add_collection(client)
did = _add_prop(client, cid, "When", "date")
client.post(f"/db/{cid}/pages/api", json={
"title": "TZ row",
"properties": {
str(did): "2026-09-10",
"__timezone__": {str(did): "Europe/Paris"},
"__recurrence__": {str(did): {"freq": "weekly", "timezone": "Europe/Paris"}},
},
})
evs = client.get(f"/db/{cid}/calendar/api",
params={"start": "2026-09-01", "end": "2026-09-30"}).json()["events"]
assert evs and all(e["timezone"] == "Europe/Paris" for e in evs)
# ── Timezone settings endpoints ──
def test_user_timezone_endpoints(client):
session, csrf, uid = _login(client)
r = client.get("/api/notifications/timezone", cookies={"flowdeck_session": session})
assert r.status_code == 200
data = r.json()
assert data["timezone"] == ""
assert "Europe/Paris" in data["zones"]
r = client.post("/api/notifications/timezone", json={"timezone": "Europe/Paris"},
cookies={"flowdeck_session": session}, headers={"X-CSRF-Token": csrf})
assert r.status_code == 200
assert client.get("/api/notifications/timezone",
cookies={"flowdeck_session": session}).json()["timezone"] == "Europe/Paris"
r = client.post("/api/notifications/timezone", json={"timezone": "Mars/Olympus"},
cookies={"flowdeck_session": session}, headers={"X-CSRF-Token": csrf})
assert r.status_code == 400
def test_notifications_unauth(client):
assert client.get("/api/notifications/timezone").status_code == 401
# ── Reminder scanner (deterministic clock) ──
def _mk_row_with_reminder(client, reminder, base="2026-09-15T09:00", freq=None):
cid = _add_collection(client)
did = _add_prop(client, cid, "When", "date")
props = {str(did): base, "__reminder__": {str(did): reminder}}
if freq:
props["__recurrence__"] = {str(did): freq}
page = client.post(f"/db/{cid}/pages/api", json={"title": "Event", "properties": props})
assert page.status_code == 200, page.text
return cid, page.json()["id"]
def test_reminder_scan_fires_once(client):
from app.db import get_conn
from app.services.reminders import scan_and_fire
_login(client) # an admin user must exist to receive the notification
_mk_row_with_reminder(client, {"value": 30, "unit": "minutes"})
now = dt.datetime(2026, 9, 15, 8, 35, tzinfo=dt.UTC) # event 09:00 - 30min = 08:30
fired = scan_and_fire(now=now)
assert fired >= 1
with get_conn() as conn:
n = conn.execute("SELECT COUNT(*) c FROM notifications WHERE ntype='reminder'").fetchone()["c"]
log = conn.execute("SELECT COUNT(*) c FROM reminder_log").fetchone()["c"]
assert n >= 1 and log >= 1
# Dedup: a second scan for the same occurrence fires nothing.
assert scan_and_fire(now=now + dt.timedelta(minutes=5)) == 0
def test_reminder_scan_not_yet_due(client):
from app.db import get_conn
from app.services.reminders import scan_and_fire
_mk_row_with_reminder(client, {"value": 30, "unit": "minutes"})
now = dt.datetime(2026, 9, 15, 8, 0, tzinfo=dt.UTC) # fire_at = 08:30
assert scan_and_fire(now=now) == 0
with get_conn() as conn:
assert conn.execute("SELECT COUNT(*) c FROM notifications WHERE ntype='reminder'").fetchone()["c"] == 0
def test_reminder_recurring_occurrence(client):
from app.db import get_conn
from app.services.reminders import scan_and_fire
_login(client)
_mk_row_with_reminder(
client, {"value": 1, "unit": "days"},
base="2026-09-01T09:00", freq={"freq": "weekly"},
)
# Weekly anchored on 2026-09-01 (Tuesday): occurrences 09-01, 09-08, 09-15…
# 1-day reminder for the 09-15 occurrence fires on 09-14 at/after 09:00.
now = dt.datetime(2026, 9, 14, 9, 5, tzinfo=dt.UTC)
assert scan_and_fire(now=now) >= 1
with get_conn() as conn:
row = conn.execute(
"SELECT message FROM notifications WHERE ntype='reminder' ORDER BY id DESC LIMIT 1"
).fetchone()
assert "2026-09-15" in row["message"]
def test_reminder_targets_assigned_person(client):
from app.db import get_conn
from app.services.reminders import scan_and_fire
session, csrf, uid = _login(client)
cid = _add_collection(client)
did = _add_prop(client, cid, "When", "date")
pid = _add_prop(client, cid, "Owner", "person")
client.post(f"/db/{cid}/pages/api", json={"title": "Mine", "properties": {
str(did): "2026-09-20T10:00",
str(pid): [{"id": uid, "login": "x"}],
"__reminder__": {str(did): {"value": 15, "unit": "minutes"}},
}})
now = dt.datetime(2026, 9, 20, 9, 46, tzinfo=dt.UTC) # fire_at 09:45
assert scan_and_fire(now=now) >= 1
with get_conn() as conn:
row = conn.execute(
"SELECT user_id FROM notifications WHERE ntype='reminder' AND is_read=0"
).fetchone()
assert row["user_id"] == uid
def test_parse_lead_units():
from app.services.reminders import parse_lead
assert parse_lead({"value": 30, "unit": "minutes"}) == 1800
assert parse_lead({"value": 2, "unit": "hours"}) == 7200
assert parse_lead({"value": 1, "unit": "days"}) == 86400
assert parse_lead({"unit": "none"}) is None
assert parse_lead({"value": 0, "unit": "minutes"}) is None
assert parse_lead(None) is None
# ── Assignment notifications ──
def test_assignment_notification_on_update(client):
from app.db import get_conn
session, csrf, actor = _login(client)
with get_conn() as conn:
login = f"v58b_{secrets.token_hex(4)}"
conn.execute("INSERT INTO users (login, full_name, email) VALUES (?, 'Bob', '[email protected]')",
(login,))
bob = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
conn.commit()
cid = _add_collection(client)
pid = _add_prop(client, cid, "Owner", "person")
rid = client.post(f"/db/{cid}/pages/api", json={"title": "Task", "properties": {}}).json()["id"]
# Assign Bob.
r = client.put(f"/db/pages/{rid}/api", json={"properties": {
str(pid): [{"id": bob, "login": "bob"}],
}}, cookies={"flowdeck_session": session})
assert r.status_code == 200
with get_conn() as conn:
note = conn.execute(
"SELECT * FROM notifications WHERE user_id=? AND ntype='assignment'", (bob,)
).fetchone()
assert note is not None and note["title"] == "Assigned to you" and note["resource_id"] == rid
# Re-assigning the same person does not duplicate.
client.put(f"/db/pages/{rid}/api", json={"properties": {
str(pid): [{"id": bob, "login": "bob"}],
}}, cookies={"flowdeck_session": session})
with get_conn() as conn:
n = conn.execute(
"SELECT COUNT(*) c FROM notifications WHERE user_id=? AND ntype='assignment'", (bob,)
).fetchone()["c"]
assert n == 1
def test_notification_prefs_extended_keys(client):
session, csrf, uid = _login(client)
r = client.get("/api/notifications/prefs", cookies={"flowdeck_session": session})
prefs = r.json()["prefs"]
assert prefs["reminders"] is True and prefs["assignments"] is True
r = client.post("/api/notifications/prefs", json={"reminders": False},
cookies={"flowdeck_session": session}, headers={"X-CSRF-Token": csrf})
assert r.json()["prefs"]["reminders"] is False
assert r.json()["prefs"]["assignments"] is True
# ── Front-end wiring (rendered template markers) ──
def test_calendar_front_end_wired(client):
import pathlib
src = pathlib.Path("app/templates/_database_table_scripts.html").read_text(encoding="utf-8")
assert "calendar/api" in src
assert "data-cal-mode" in src and "refreshCalendar" in src
assert "openEventEditor" in src and "__recurrence__" in src and "__reminder__" in src