From d986959927a4738cdb37ca709b88d523ca40086a Mon Sep 17 00:00:00 2001 From: Bruno Charest Date: Sun, 13 Sep 2026 09:51:54 -0400 Subject: [PATCH 1/2] fix(cloudflare): protect agent panel, workspaces & shell scripts from Rocket Loader; trust proxy headers --- Dockerfile | 2 +- app/templates/_header.html | 2 +- app/templates/_page_editor_realtime.html | 2 +- app/templates/agent_panel.html | 4 ++-- app/templates/base.html | 6 +++--- app/templates/workspaces.html | 2 +- 6 files changed, 9 insertions(+), 9 deletions(-) diff --git a/Dockerfile b/Dockerfile index a3f06ff..32eff4f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -41,4 +41,4 @@ EXPOSE 8080 HEALTHCHECK --interval=30s --timeout=5s --start-period=5s --retries=3 \ CMD curl -f http://localhost:8080/api/health || exit 1 -CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"] \ No newline at end of file +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080", "--proxy-headers", "--forwarded-allow-ips", "*"] \ No newline at end of file diff --git a/app/templates/_header.html b/app/templates/_header.html index 24c982f..e10e79e 100644 --- a/app/templates/_header.html +++ b/app/templates/_header.html @@ -142,7 +142,7 @@ - + - +
@@ -1942,7 +1942,7 @@ } }); - + + + +
+
+ + FlowDeck · Import +
+

Importer des données

+

Notes, vaults, bases de données, documents ou signets — choisissez une source ou laissez la détection automatique.

+ +
+
1 · Source
+
+
+
✨ Détection auto
+
Reconnaît le format d'après le fichier.
+
+ +
+
+ +
+
2 · Fichiers (sélection multiple possible)
+
+ Glissez un ou plusieurs fichiers ici ou cliquez pour parcourir
Markdown, .zip, .csv, .tsv, .xlsx, .json, .html, .docx, .pdf, .ics, .opml… +
+ + +
+ +
+
+ + +
+
+
+
+ +
+
Ou · importer une page web (URL)
+
+ + +
+
+
+ +
+
Ou · importer depuis une forge
+
+ + + + + + +
+
+
+ +
+
Aperçu —
+ + + + + +
TitreTypeÉlémentsColonnes
+
+ +
+
+ +
+
Rapport d'import
+
+
Pages créées
+
Mises à jour
+
Collections
+
Lignes
+
Fichiers
+
Ignorés
+
Erreurs
+
+
+ +
+
+ +
+
+
+
+
+ + Retour au workspace +
+
+
+ + + + diff --git a/requirements.txt b/requirements.txt index f2497c2..e5d2a17 100644 --- a/requirements.txt +++ b/requirements.txt @@ -13,3 +13,8 @@ itsdangerous==2.2.* slowapi==0.1.* weasyprint==69.0 xhtml2pdf==0.2.* +openpyxl==3.1.* +beautifulsoup4==4.* +PyYAML==6.* +python-docx>=1.1 +pypdf==6.* diff --git a/tests/test_v56_import.py b/tests/test_v56_import.py new file mode 100644 index 0000000..b5a23d6 --- /dev/null +++ b/tests/test_v56_import.py @@ -0,0 +1,877 @@ +"""FlowDeck — v5.6.0 import framework tests (Phases 0–4). + +Covers the unified pipeline, hierarchy, attachments, dedup, background jobs and +the per-source importers (Obsidian, Notion, Logseq/Roam, Google Keep, HTML +notes, typed CSV/TSV, Excel, JSON, bookmarks, .ics, OPML, Standard Notes, Word, +PDF, forge issues). +""" +from __future__ import annotations + +import io +import json +import os +import secrets +import time +import zipfile + +from app.db import get_conn + + +def _login(client): + from app.auth.session import SessionManager + + with get_conn() as conn: + login = f"importer_{secrets.token_hex(4)}" + conn.execute( + "INSERT INTO users (login, full_name, email, password_hash, is_admin) " + "VALUES (?, 'Importer', ?, ?, 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}) + r = client.get("/api/csrf-token", cookies={"flowdeck_session": session}) + return session, r.json()["csrf_token"], login + + +def _create_workspace(client, cookie, name="Import WS"): + r = client.post("/workspace", json={"name": name}, cookies={"flowdeck_session": cookie}) + assert r.status_code == 200 + ws = r.json() + client.cookies.set("flowdeck_workspace", str(ws["id"])) + return ws["id"] + + +def _run(client, cookie, csrf, filename, data, *, source=None, extra=None): + form = dict(extra or {}) + if source: + form["source"] = source + return client.post( + "/api/import/run", + files={"file": (filename, data, "application/octet-stream")}, + data=form, + headers={"X-CSRF-Token": csrf}, + cookies={"flowdeck_session": cookie}, + ) + + +def _page_blocks(page_id): + with get_conn() as conn: + row = conn.execute("SELECT content, content_format FROM pages WHERE id=?", (page_id,)).fetchone() + return json.loads(row["content"]) if row and row["content_format"] == "blocks" else [] + + +def _collection_by_name(name): + with get_conn() as conn: + return conn.execute("SELECT * FROM collections WHERE name=?", (name,)).fetchone() + + +def _props_of(collection_id): + with get_conn() as conn: + return { + r["name"]: r["prop_type"] + for r in conn.execute( + "SELECT name, prop_type FROM collection_properties WHERE collection_id=?", + (collection_id,), + ).fetchall() + } + + +def _pages_of(collection_id): + with get_conn() as conn: + return conn.execute( + "SELECT * FROM collection_pages WHERE collection_id=? ORDER BY position", + (collection_id,), + ).fetchall() + + +def _props_named(collection_id, page_row): + """Map a row's id-keyed property values back to property names.""" + with get_conn() as conn: + names = { + r["id"]: r["name"] + for r in conn.execute( + "SELECT id, name FROM collection_properties WHERE collection_id=?", + (collection_id,), + ).fetchall() + } + values = json.loads(page_row["property_values_json"]) + return {names.get(int(k), k): v for k, v in values.items()} + + +class TestImportRegistry: + def test_sources_listed(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = client.get("/api/import/sources", cookies={"flowdeck_session": cookie}) + assert r.status_code == 200 + ids = {s["source_id"] for s in r.json()["sources"]} + assert {"obsidian", "notion", "logseq", "roam", "google_keep", "html_notes", + "csv", "excel", "json", "markdown"} <= ids + + def test_preview_is_dry_run(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = client.post( + "/api/import/preview", + files={"file": ("notes.csv", b"Name,Age\nAlice,30", "text/csv")}, + headers={"X-CSRF-Token": csrf}, + cookies={"flowdeck_session": cookie}, + ) + assert r.status_code == 200 + body = r.json() + assert body["dry_run"] is True + assert body["detected_source"] == "csv" + assert body["pages"][0]["type"] == "collection" + assert body["pages"][0]["rows"] == 1 + assert _collection_by_name("notes") is None + + +class TestCsvTypedImport: + def test_types_inferred_and_rows_created(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + csv_data = ( + b"Name,Age,Due,Active,Email,Website,Tags\n" + b"Alice,30,2026-01-02,true,a@b.com,https://x.com,red;blue\n" + b"Bob,42,2026-02-03,false,b@c.com,https://y.com,green\n" + ) + r = _run(client, cookie, csrf, "people.csv", csv_data) + assert r.status_code == 200, r.text + report = r.json() + assert report["collections_created"] == 1 + assert report["rows_created"] == 2 + + coll = _collection_by_name("people") + assert coll is not None + types = _props_of(coll["id"]) + assert types["Age"] == "number" + assert types["Due"] == "date" + assert types["Active"] == "checkbox" + assert types["Email"] == "email" + assert types["Website"] == "url" + assert types["Tags"] == "multi_select" + + pages = _pages_of(coll["id"]) + assert pages[0]["title"] == "Alice" + props = _props_named(coll["id"], pages[0]) + assert props["Age"] == 30 + assert props["Active"] is True + assert props["Tags"] == ["red", "blue"] + + def test_dedup_skips_second_import(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + data = b"Name,Age\nAlice,30" + assert _run(client, cookie, csrf, "d.csv", data).json()["rows_created"] == 1 + second = _run(client, cookie, csrf, "d.csv", data).json() + assert second["rows_created"] == 0 + assert second["skipped"] == 1 + + def test_tsv_delimiter(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + data = b"Name\tScore\nAlice\t9\nBob\t8" + r = _run(client, cookie, csrf, "scores.tsv", data) + assert r.status_code == 200 + coll = _collection_by_name("scores") + assert _props_of(coll["id"])["Score"] == "number" + assert len(_pages_of(coll["id"])) == 2 + + +class TestExcelJsonImport: + def test_excel_multi_sheet(self, client): + from openpyxl import Workbook + + wb = Workbook() + ws1 = wb.active + ws1.title = "Tasks" + ws1.append(["Task", "Priority"]) + ws1.append(["Write", "High"]) + ws2 = wb.create_sheet("Contacts") + ws2.append(["Name", "Email"]) + ws2.append(["Alice", "a@b.com"]) + buf = io.BytesIO() + wb.save(buf) + + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = _run(client, cookie, csrf, "book.xlsx", buf.getvalue()) + assert r.status_code == 200, r.text + assert r.json()["collections_created"] == 2 + assert _collection_by_name("book — Tasks") is not None + assert _collection_by_name("book — Contacts") is not None + + def test_generic_json(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + data = json.dumps([ + {"title": "One", "count": 3}, + {"title": "Two", "count": 5}, + ]).encode() + r = _run(client, cookie, csrf, "items.json", data) + assert r.status_code == 200 + coll = _collection_by_name("items") + assert _props_of(coll["id"])["count"] == "number" + assert len(_pages_of(coll["id"])) == 2 + + +class TestObsidianImport: + def _vault(self) -> bytes: + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr(".obsidian/app.json", "{}") + zf.writestr( + "Projects/Roadmap.md", + "---\ntitle: Feuille de route\ntags: [projet, 2026]\n---\n" + "# Roadmap\n\nVoir [[Ideas]] et l'image ![[diagram.png]]\n", + ) + zf.writestr("Ideas.md", "Quelques idées\n") + zf.writestr("Projects/diagram.png", b"\x89PNG\r\n\x1a\nfake") + return buf.getvalue() + + def test_hierarchy_frontmatter_and_attachment(self, client, tmp_path): + cookie, csrf, _ = _login(client) + ws_id = _create_workspace(client, cookie) + old = os.environ.get("FLOWDECK_DATA_DIR") + os.environ["FLOWDECK_DATA_DIR"] = str(tmp_path) + try: + r = _run(client, cookie, csrf, "vault.zip", self._vault()) + assert r.status_code == 200, r.text + report = r.json() + assert report["detected_source"] == "obsidian" + assert report["attachments"] == 1 + + with get_conn() as conn: + folder = conn.execute("SELECT * FROM pages WHERE title='Projects'").fetchone() + note = conn.execute("SELECT * FROM pages WHERE title='Feuille de route'").fetchone() + assert folder is not None and note is not None + assert note["parent_id"] == folder["id"] + + blocks = _page_blocks(note["id"]) + assert any(b["type"] == "heading_1" and b["content"] == "Roadmap" for b in blocks) + images = [b for b in blocks if b["type"] == "image"] + assert images and f"/api/files/{ws_id}/import/" in images[0]["src"] + assert any(b["type"] == "callout" for b in blocks) + finally: + if old is None: + os.environ.pop("FLOWDECK_DATA_DIR", None) + else: + os.environ["FLOWDECK_DATA_DIR"] = old + + def test_existing_md_import_still_works(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = client.post( + "/board/api/pages/import", + json={"markdown": "# Hello\n\n- a\n- b", "title": "Legacy"}, + headers={"X-CSRF-Token": csrf, "Content-Type": "application/json"}, + cookies={"flowdeck_session": cookie}, + ) + assert r.status_code == 200 + + +class TestNotionImport: + def test_pages_and_database(self, client): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("Home abcdef0123456789abcdef0123456789.md", + "# Home\n\nBienvenue [Child](Child%20abcdef0123456789abcdef0123456789.md)\n") + zf.writestr("Home/Child abcdef0123456789abcdef0123456789.md", "# Child\n\ntexte\n") + zf.writestr("Tasks 11111111111111111111111111111111.csv", + "Name,Status,Done\nT1,Done,true\nT2,Todo,false\n") + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = _run(client, cookie, csrf, "notion.zip", buf.getvalue()) + assert r.status_code == 200, r.text + report = r.json() + assert report["detected_source"] == "notion" + assert report["collections_created"] == 1 + coll = _collection_by_name("Tasks") + assert coll is not None + assert _props_of(coll["id"])["Done"] == "checkbox" + assert len(_pages_of(coll["id"])) == 2 + with get_conn() as conn: + home = conn.execute("SELECT * FROM pages WHERE title='Home'").fetchone() + text = json.dumps(_page_blocks(home["id"])) + assert "abcdef0123456789" not in text + assert "Child" in text + + +class TestOutlineImport: + def test_logseq_properties_and_journal(self, client): + data = ( + "title:: Ma page\n" + "tags:: travail, urgent\n" + "- Première puce\n" + "- Deuxième [[Lien]]\n" + ).encode() + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = _run(client, cookie, csrf, "logseq-page.md", data) + assert r.status_code == 200 + assert r.json()["detected_source"] == "logseq" + with get_conn() as conn: + page = conn.execute("SELECT * FROM pages WHERE title='Ma page'").fetchone() + assert page is not None + blocks = _page_blocks(page["id"]) + assert any(b["type"] == "bulleted_list" for b in blocks) + + def test_roam_todo_marker(self, client): + data = b"- {{[[TODO]]}} finir\n- {{[[DONE]]}} fait\n" + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = _run(client, cookie, csrf, "roam.md", data) + assert r.status_code == 200 + assert r.json()["detected_source"] == "roam" + with get_conn() as conn: + page = conn.execute("SELECT * FROM pages WHERE title='roam'").fetchone() + todos = [b for b in _page_blocks(page["id"]) if b["type"] == "to_do"] + assert todos and todos[0]["checked"] is False + + +class TestHtmlAndKeepImport: + def test_html_note(self, client): + html = b"Ma note

Ma note

Texte

" + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = _run(client, cookie, csrf, "note.html", html) + assert r.status_code == 200 + assert r.json()["detected_source"] == "html_notes" + with get_conn() as conn: + page = conn.execute("SELECT * FROM pages WHERE title='Ma note'").fetchone() + blocks = _page_blocks(page["id"]) + assert any(b["type"] == "heading_1" for b in blocks) + assert sum(1 for b in blocks if b["type"] == "bulleted_list") == 2 + + def test_google_keep_json(self, client): + note = { + "title": "Liste courses", + "listContent": [ + {"text": "Pain", "isChecked": False}, + {"text": "Lait", "isChecked": True}, + ], + "labels": ["perso"], + "isTrashed": False, + "color": "white", + } + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = _run(client, cookie, csrf, "keep.json", json.dumps(note).encode()) + assert r.status_code == 200 + assert r.json()["detected_source"] == "google_keep" + with get_conn() as conn: + page = conn.execute("SELECT * FROM pages WHERE title='Liste courses'").fetchone() + todos = [b for b in _page_blocks(page["id"]) if b["type"] == "to_do"] + assert len(todos) == 2 + + +class TestMappingAndWizard: + def test_column_type_mapping_override(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + data = b"Name,Code\nA,007\nB,042\n" + r = _run( + client, cookie, csrf, "codes.csv", data, + extra={"mapping": json.dumps({"Code": "text"})}, + ) + assert r.status_code == 200 + coll = _collection_by_name("codes") + assert _props_of(coll["id"])["Code"] == "text" + props = _props_named(coll["id"], _pages_of(coll["id"])[0]) + assert props["Code"] == "007" + + def test_wizard_page_requires_auth(self, client): + r = client.get("/import", follow_redirects=False) + assert r.status_code in (302, 307) + + def test_wizard_page_renders(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = client.get("/import", cookies={"flowdeck_session": cookie}) + assert r.status_code == 200 + assert "Importer des données" in r.text + + +class TestAsyncJob: + def test_background_import(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = client.post( + "/api/import/run", + files={"file": ("async.csv", b"Name,Age\nAlice,30", "text/csv")}, + data={"async": "true"}, + headers={"X-CSRF-Token": csrf}, + cookies={"flowdeck_session": cookie}, + ) + assert r.status_code == 200 + job_id = r.json()["job_id"] + status = None + for _ in range(50): + jr = client.get(f"/api/import/jobs/{job_id}", cookies={"flowdeck_session": cookie}) + assert jr.status_code == 200 + status = jr.json()["status"] + if status in ("done", "error"): + break + time.sleep(0.1) + assert status == "done", jr.json() + assert jr.json()["report"]["rows_created"] == 1 + + +# ── Phase 4: bookmarks, calendar, OPML, Standard Notes, forge ──────────── + + +class TestBookmarkImport: + def test_raindrop_csv(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + csv_data = ( + b"id,title,note,excerpt,url,folder,tags,created\n" + b"1,Notion,super outil,,https://notion.so,Tools,\"prod,web\",2026-01-02T10:00:00Z\n" + ) + r = _run(client, cookie, csrf, "raindrop.csv", csv_data) + assert r.status_code == 200, r.text + assert r.json()["detected_source"] == "raindrop" + coll = _collection_by_name("Raindrop") + assert coll is not None + row = _pages_of(coll["id"])[0] + props = _props_named(coll["id"], row) + assert props["URL"] == "https://notion.so" + assert props["Tags"] == ["prod", "web"] + + def test_pocket_csv(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + csv_data = b"title,url,time_added,tags,status\nArticle,https://ex.com/1,1700000000,news,unread\n" + r = _run(client, cookie, csrf, "pocket.csv", csv_data) + assert r.status_code == 200 + assert r.json()["detected_source"] == "pocket" + coll = _collection_by_name("Pocket") + assert _pages_of(coll["id"])[0]["title"] == "Article" + + def test_readwise_csv(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + csv_data = ( + "Highlight,Book Title,Book Author,Note,Document Tags,Highlighted at\n" + "Une idée clé,Livre X,Auteur Y,à retenir,philo,2026-03-01\n" + ).encode() + r = _run(client, cookie, csrf, "readwise.csv", csv_data) + assert r.status_code == 200 + assert r.json()["detected_source"] == "readwise" + coll = _collection_by_name("Readwise") + props = _props_named(coll["id"], _pages_of(coll["id"])[0]) + assert props["Book"] == "Livre X" + assert props["Tags"] == ["philo"] + + def test_shaarli_json(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + data = json.dumps([ + {"url": "https://a.io", "title": "A", "tags": "dev", "created": "2026-01-01"}, + ]).encode() + r = _run(client, cookie, csrf, "shaarli.json", data) + assert r.status_code == 200 + assert r.json()["detected_source"] == "shaarli" + coll = _collection_by_name("Shaarli") + assert _pages_of(coll["id"])[0]["title"] == "A" + + def test_netscape_bookmarks(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + html = ( + b'

' + b'

Dev

' + b'

Python' + b'

' + ) + r = _run(client, cookie, csrf, "bookmarks.html", html) + assert r.status_code == 200 + assert r.json()["detected_source"] == "bookmarks" + coll = _collection_by_name("Bookmarks") + props = _props_named(coll["id"], _pages_of(coll["id"])[0]) + assert props["URL"] == "https://python.org" + assert props["Tags"] == ["Dev"] + + +class TestCalendarOpmlNotes: + def test_ics_events(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + ics = ( + "BEGIN:VCALENDAR\r\nX-WR-CALNAME:Work\r\nBEGIN:VEVENT\r\n" + "SUMMARY:Réunion\r\nDTSTART;VALUE=DATE:20260115\r\n" + "LOCATION:Bureau\r\nEND:VEVENT\r\n" + "BEGIN:VEVENT\r\nSUMMARY:Call\r\nDTSTART:20260116T140000Z\r\n" + "DTEND:20260116T150000Z\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n" + ).encode() + r = _run(client, cookie, csrf, "agenda.ics", ics) + assert r.status_code == 200 + assert r.json()["detected_source"] == "ics" + coll = _collection_by_name("Calendar") + pages = _pages_of(coll["id"]) + assert len(pages) == 2 + props = _props_named(coll["id"], pages[0]) + assert props["Start"] == "2026-01-15" + assert props["All day"] is True + assert props["Calendar"] == "Work" + + def test_opml_feeds(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + opml = ( + b'' + b'' + b'' + ) + r = _run(client, cookie, csrf, "feeds.opml", opml) + assert r.status_code == 200 + assert r.json()["detected_source"] == "opml" + coll = _collection_by_name("OPML feeds") + rows = [_props_named(coll["id"], p) for p in _pages_of(coll["id"])] + feed = next(p for p in rows if p.get("Feed URL")) + assert feed["Feed URL"] == "https://hn/rss" + assert feed["Folder"] == "Tech" + + def test_standard_notes(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + backup = { + "items": [ + {"uuid": "u1", "content_type": "Note", + "content": json.dumps({"title": "Ma note", "text": "Contenu clair"})}, + {"uuid": "u2", "content_type": "Note", + "content": json.dumps({"encrypted": True, "content": "xxx"})}, + {"uuid": "u3", "content_type": "Tag", "content": "ignore"}, + ] + } + r = _run(client, cookie, csrf, "standard-notes.json", json.dumps(backup).encode()) + assert r.status_code == 200 + assert r.json()["detected_source"] == "standard_notes" + with get_conn() as conn: + page = conn.execute("SELECT * FROM pages WHERE title='Ma note'").fetchone() + assert page is not None + assert r.json()["warnings"] + + +class TestDocumentImport: + def test_docx(self, client): + from docx import Document + from docx.shared import Inches + from PIL import Image + + doc = Document() + doc.add_heading("Titre Word", level=1) + doc.add_paragraph("Un paragraphe.") + doc.add_paragraph("Puce un", style="List Bullet") + table = doc.add_table(rows=2, cols=2) + table.cell(0, 0).text = "A" + table.cell(0, 1).text = "B" + table.cell(1, 0).text = "1" + table.cell(1, 1).text = "2" + img = io.BytesIO() + Image.new("RGB", (10, 10), "red").save(img, format="PNG") + img.seek(0) + doc.add_picture(img, width=Inches(1)) + buf = io.BytesIO() + doc.save(buf) + + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = _run(client, cookie, csrf, "doc.docx", buf.getvalue()) + assert r.status_code == 200, r.text + assert r.json()["detected_source"] == "docx" + assert r.json()["attachments"] == 1 + with get_conn() as conn: + page = conn.execute("SELECT * FROM pages WHERE title='doc'").fetchone() + blocks = _page_blocks(page["id"]) + assert any(b["type"] == "heading_1" and b["content"] == "Titre Word" for b in blocks) + assert any(b["type"] == "table" for b in blocks) + assert any(b["type"] == "image" for b in blocks) + + def test_pdf_text(self, client): + from xhtml2pdf import pisa + + buf = io.BytesIO() + pisa.CreatePDF("

Hello PDF

Bonjour le monde

", dest=buf) + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = _run(client, cookie, csrf, "report.pdf", buf.getvalue()) + assert r.status_code == 200, r.text + assert r.json()["detected_source"] == "pdf" + with get_conn() as conn: + page = conn.execute("SELECT * FROM pages WHERE title='report'").fetchone() + text = json.dumps(_page_blocks(page["id"])) + assert "Bonjour le monde" in text + + +class TestForgeImport: + def test_build_issues_result(self): + from app.services.importers.forge import build_issues_result + + issues = [ + {"number": 1, "title": "Bug", "state": "open", "labels": [{"name": "bug"}], + "milestone": {"title": "v1"}, "assignees": [{"login": "bob"}], + "created_at": "2026-01-01T00:00:00Z", "html_url": "https://x/1", "body": "desc"}, + {"number": 2, "title": "PR", "pull_request": {"url": "x"}}, + ] + labels = [{"name": "bug", "color": "red"}] + milestones = [{"title": "v1", "state": "open", "due_on": "2026-02-01"}] + result = build_issues_result(issues, labels=labels, milestones=milestones, + owner="me", repo="proj", provider="gitea") + assert result.source == "forge:gitea" + assert len(result.pages) == 3 + issue_page = result.pages[0] + assert issue_page.collection["name"] == "me/proj issues" + assert len(issue_page.collection["rows"]) == 1 + props = issue_page.collection["rows"][0]["properties"] + assert props["Labels"] == ["bug"] + assert props["Milestone"] == "v1" + assert props["Assignee"] == "bob" + + def test_fetch_forge_issues(self): + import asyncio + + from app.services.importers.forge import fetch_forge_issues + + class FakeAdapter: + async def list_issues(self, owner, repo, state="all"): + return [{"number": 1, "title": "T", "state": "closed", "labels": []}] + + async def list_labels(self, owner, repo): + return [{"name": "x"}] + + async def list_milestones(self, owner, repo, state="all"): + return [{"title": "M", "state": "closed"}] + + result = asyncio.run(fetch_forge_issues( + FakeAdapter(), "o", "r", provider="github", + )) + assert len(result.pages) == 3 + + def test_gitea_adapter_wraps_client(self): + import asyncio + + from app.services.importers.forge import GiteaForgeAdapter + + class FakeGiteaClient: + async def get_issues(self, owner, repo, state="all", page=1, limit=50): + return [{"number": 1, "title": "G", "state": "open"}] if page == 1 else [] + + async def get_labels(self, owner, repo): + return [{"name": "bug"}] + + async def get_milestones(self, owner, repo, state="open"): + return [{"title": "M1", "state": "open"}] + + async def run(): + adapter = GiteaForgeAdapter(FakeGiteaClient()) + issues = await adapter.list_issues("o", "r") + labels = await adapter.list_labels("o", "r") + milestones = await adapter.list_milestones("o", "r") + return issues, labels, milestones + + issues, labels, milestones = asyncio.run(run()) + assert len(issues) == 1 and labels and milestones + + def test_forge_endpoint_requires_connection(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = client.post( + "/api/import/forge", + json={"provider": "gitea", "owner": "o", "repo": "r"}, + headers={"X-CSRF-Token": csrf}, + cookies={"flowdeck_session": cookie}, + ) + assert r.status_code == 400 + + +# ── Phase 5: incremental, batch, URL, forge-repo, relations, reports ────── + + +class TestIncrementalImport: + def test_mode_update_resyncs_rows(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + assert _run(client, cookie, csrf, "m.csv", b"Name,Age\nAlice,30").json()["rows_created"] == 1 + second = _run(client, cookie, csrf, "m.csv", b"Name,Age\nAlice,31\nBob,20", + extra={"mode": "update"}).json() + assert second["rows_updated"] == 1 + assert second["rows_created"] == 1 + coll = _collection_by_name("m") + pages = {p["title"]: _props_named(coll["id"], p) for p in _pages_of(coll["id"])} + assert pages["Alice"]["Age"] == 31 + + def test_mode_duplicate_creates_again(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + _run(client, cookie, csrf, "d.csv", b"Name\nA") + second = _run(client, cookie, csrf, "d.csv", b"Name\nA", + extra={"mode": "duplicate"}).json() + assert second["collections_created"] == 1 + assert second["skipped"] == 0 + + def test_mode_skip_is_default(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + _run(client, cookie, csrf, "s.csv", b"Name\nA") + second = _run(client, cookie, csrf, "s.csv", b"Name\nA").json() + assert second["skipped"] == 1 + + def test_report_shape(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + report = _run(client, cookie, csrf, "r.csv", b"Name\nA").json() + assert report["status"] == "ok" + assert report["errors"] == [] + assert report["mode"] == "skip" + + +class TestBatchImport: + def test_run_batch_multiple_files(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = client.post( + "/api/import/run-batch", + files=[ + ("file", ("a.csv", b"Name,Age\nA,1", "text/csv")), + ("file", ("b.csv", b"Name,Age\nB,2", "text/csv")), + ], + headers={"X-CSRF-Token": csrf}, + cookies={"flowdeck_session": cookie}, + ) + assert r.status_code == 200, r.text + body = r.json() + assert body["summary"]["files"] == 2 + assert len(body["results"]) == 2 + assert body["summary"]["rows_created"] == 2 + + +class TestUrlImport: + def test_fetch_url_result(self, monkeypatch): + import asyncio + + import httpx + + from app.services.importers import url_fetch + + monkeypatch.setattr(url_fetch, "_is_public_host", lambda host: True) + html = ( + "Titre" + "" + "" + "

Salut

Contenu

" + ) + + def handler(request): + return httpx.Response(200, headers={"content-type": "text/html"}, + text=html, request=request) + + result = asyncio.run(url_fetch.fetch_url_result( + "https://example.com/post", transport=httpx.MockTransport(handler), + )) + assert result.source == "url" + assert len(result.pages) == 1 + blocks = result.pages[0].blocks + assert blocks[0]["type"] == "bookmark" + assert blocks[0]["url"] == "https://example.com/post" + + def test_url_endpoint_rejects_private_host(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = client.post( + "/api/import/url", + json={"url": "http://localhost/secret"}, + headers={"X-CSRF-Token": csrf}, + cookies={"flowdeck_session": cookie}, + ) + assert r.status_code == 400 + + +class TestForgeRepoImport: + def test_build_repo_result(self): + from app.services.importers.forge_repo import build_repo_result + + result = build_repo_result( + [("docs/README.md", "# Hi\n\ntext"), ("src/app.py", "print('x')")], + owner="o", repo="r", provider="github", + ) + assert len(result.pages) == 2 + md = next(p for p in result.pages if p.title == "README.md") + assert md.parent_path == "docs" + assert md.markdown == "# Hi\n\ntext" + py = next(p for p in result.pages if p.title == "app.py") + assert "```python" in py.markdown + + def test_fetch_forge_repo(self): + import asyncio + + from app.services.importers.forge_repo import fetch_forge_repo + + class FakeAdapter: + async def list_repo_files(self, owner, repo, path=""): + return [{"path": "a.md", "size": 10}, + {"path": "big.bin", "size": 9999999}, + {"path": "img.png", "size": 10}] + + async def get_file_content(self, owner, repo, path): + return "# A" + + result = asyncio.run(fetch_forge_repo( + FakeAdapter(), "o", "r", provider="github", + )) + assert len(result.pages) == 1 + assert result.pages[0].title == "a.md" + + +class TestRelationsResolve: + def test_text_column_becomes_relation(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + _run(client, cookie, csrf, "members.csv", b"Name,Team\nAlice,Red\nBob,Blue") + _run(client, cookie, csrf, "teams.csv", b"Name\nRed\nBlue") + r = client.post( + "/api/import/relations/resolve", + headers={"X-CSRF-Token": csrf}, + cookies={"flowdeck_session": cookie}, + ) + assert r.status_code == 200 + assert r.json()["relations_resolved"] == 1 + + members = _collection_by_name("members") + with get_conn() as conn: + prop = conn.execute( + "SELECT id, prop_type, related_collection_id FROM collection_properties " + "WHERE collection_id=? AND name='Team'", (members["id"],), + ).fetchone() + assert prop["prop_type"] == "relation" + assert prop["related_collection_id"] is not None + + teams = _collection_by_name("teams") + team_ids = {p["title"]: p["id"] for p in _pages_of(teams["id"])} + row = _pages_of(members["id"])[0] + values = json.loads(row["property_values_json"]) + assert values[str(prop["id"])] == [team_ids["Red"]] + + +class TestJobReport: + def test_download_job_report(self, client): + cookie, csrf, _ = _login(client) + _create_workspace(client, cookie) + r = client.post( + "/api/import/run", + files={"file": ("j.csv", b"Name,Age\nA,1", "text/csv")}, + data={"async": "true"}, + headers={"X-CSRF-Token": csrf}, + cookies={"flowdeck_session": cookie}, + ) + job_id = r.json()["job_id"] + for _ in range(50): + jr = client.get(f"/api/import/jobs/{job_id}", cookies={"flowdeck_session": cookie}) + if jr.json()["status"] in ("done", "error"): + break + time.sleep(0.1) + rep = client.get(f"/api/import/jobs/{job_id}/report", cookies={"flowdeck_session": cookie}) + assert rep.status_code == 200 + assert "attachment" in rep.headers.get("content-disposition", "") + assert rep.json()["rows_created"] == 1