feat(v4.1.0): Data Sources & Linked Databases

- Add collection_data_sources table for multi-source collections - API: GET/POST/DELETE /db/{id}/sources/api — data source management - API: POST /db/{id}/linked/api — create linked database (copies views/properties) - API: POST /db/{id}/toggle-inline/api — toggle full-page vs inline - API: POST /db/inline/api — create inline database - Linked DB inherits workspace_id (permissions) from source - Migration: add workspace_id + created_by to collections table - 10 new tests (153 total)
2026-07-21 21:55:45 -04:00
parent 3fcfa8fdbb
commit d7c1f4c03e
3 changed files with 501 additions and 0 deletions
@@ -2002,3 +2002,229 @@ def test_session_expired_redirect(client):
    location = resp.headers.get("location", "")
    assert "login" in location
    assert "expired=1" in location
+
+
+# ── v4.1.0: Data Sources & Linked Databases ──
+
+
+def test_data_sources_list_empty(client):
+    """List data sources for a collection — empty by default."""
+    # Create a collection first
+    resp = client.post("/db/api", json={"name": "Sources Test DB"})
+    coll_id = resp.json()["id"]
+
+    resp = client.get(f"/db/{coll_id}/sources/api")
+    assert resp.status_code == 200
+    assert resp.json()["sources"] == []
+
+
+def test_data_sources_add_and_remove(client):
+    """Add and remove a data source from a collection."""
+    # Create two collections
+    r1 = client.post("/db/api", json={"name": "Source A"})
+    r2 = client.post("/db/api", json={"name": "Source B"})
+    coll_a = r1.json()["id"]
+    coll_b = r2.json()["id"]
+
+    # Add B as a data source of A
+    resp = client.post(f"/db/{coll_a}/sources/api", json={
+        "source_collection_id": coll_b,
+        "source_name": "Linked B",
+    })
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "added"
+    source_id = resp.json()["id"]
+
+    # List — should have 1 source
+    resp = client.get(f"/db/{coll_a}/sources/api")
+    assert resp.status_code == 200
+    assert len(resp.json()["sources"]) == 1
+    assert resp.json()["sources"][0]["source_name"] == "Linked B"
+
+    # Remove the source
+    resp = client.delete(f"/db/{coll_a}/sources/{source_id}/api")
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "removed"
+
+    # List — should be empty again
+    resp = client.get(f"/db/{coll_a}/sources/api")
+    assert len(resp.json()["sources"]) == 0
+
+
+def test_data_sources_duplicate_rejected(client):
+    """Adding the same data source twice returns 409."""
+    r1 = client.post("/db/api", json={"name": "Dup Source A"})
+    r2 = client.post("/db/api", json={"name": "Dup Source B"})
+    coll_a = r1.json()["id"]
+    coll_b = r2.json()["id"]
+
+    # First add — OK
+    resp = client.post(f"/db/{coll_a}/sources/api", json={
+        "source_collection_id": coll_b,
+    })
+    assert resp.status_code == 200
+
+    # Second add — conflict
+    resp = client.post(f"/db/{coll_a}/sources/api", json={
+        "source_collection_id": coll_b,
+    })
+    assert resp.status_code == 409
+
+
+def test_data_sources_not_found(client):
+    """Non-existent collection returns 404."""
+    resp = client.get("/db/99999/sources/api")
+    assert resp.status_code == 404
+
+
+def test_create_linked_database(client):
+    """Create a linked database from a source collection."""
+    # Create source collection
+    resp = client.post("/db/api", json={
+        "name": "CRM Contacts",
+        "description": "Customer contacts",
+        "icon": "👥",
+    })
+    source_id = resp.json()["id"]
+
+    # Add a property to the source
+    client.post(f"/db/{source_id}/properties/api", json={
+        "name": "Email",
+        "prop_type": "email",
+    })
+
+    # Create linked DB
+    resp = client.post(f"/db/{source_id}/linked/api", json={})
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["status"] == "created"
+    assert data["source_collection_id"] == source_id
+    linked_id = data["linked_id"]
+    assert linked_id != source_id
+
+    # Linked DB should have a data source pointing to source
+    resp = client.get(f"/db/{linked_id}/sources/api")
+    sources = resp.json()["sources"]
+    assert len(sources) == 1
+    assert sources[0]["source_collection_id"] == source_id
+    assert sources[0]["is_linked"] == 1
+
+    # Linked DB should have copied the source's properties
+    resp = client.get(f"/db/{linked_id}/properties/api")
+    props = resp.json()["properties"]
+    assert any(p["name"] == "Email" for p in props)
+
+    # Linked DB should have copied the source's views
+    resp = client.get(f"/db/{linked_id}/views/api")
+    views = resp.json()["views"]
+    assert len(views) >= 1
+
+
+def test_create_linked_database_custom_name(client):
+    """Create a linked database with a custom name."""
+    resp = client.post("/db/api", json={"name": "Task DB"})
+    source_id = resp.json()["id"]
+
+    resp = client.post(f"/db/{source_id}/linked/api", json={
+        "name": "My Linked Tasks",
+    })
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["name"] == "My Linked Tasks"
+    assert data["status"] == "created"
+
+
+def test_toggle_inline(client):
+    """Toggle a collection between full-page and inline modes."""
+    resp = client.post("/db/api", json={"name": "Toggle Test"})
+    coll_id = resp.json()["id"]
+
+    # Start as full-page (is_inline=0 by default)
+    resp = client.get(f"/db/{coll_id}/api")
+    assert resp.json()["collection"]["is_inline"] == 0
+
+    # Toggle to inline
+    resp = client.post(f"/db/{coll_id}/toggle-inline/api")
+    assert resp.status_code == 200
+    assert resp.json()["is_inline"] is True
+    assert resp.json()["mode"] == "inline"
+
+    # Verify in DB
+    resp = client.get(f"/db/{coll_id}/api")
+    assert resp.json()["collection"]["is_inline"] == 1
+
+    # Toggle back to full-page
+    resp = client.post(f"/db/{coll_id}/toggle-inline/api")
+    assert resp.status_code == 200
+    assert resp.json()["is_inline"] is False
+    assert resp.json()["mode"] == "full-page"
+
+
+def test_create_inline_database(client):
+    """Create an inline database within a parent page."""
+    # Create a page first (dummy)
+    from app.db import get_conn
+    with get_conn() as conn:
+        conn.execute(
+            "INSERT INTO pages (workspace, title) VALUES ('test', 'Parent Page')"
+        )
+        conn.commit()
+        page_id = conn.execute("SELECT id FROM pages ORDER BY id DESC LIMIT 1").fetchone()["id"]
+
+    resp = client.post("/db/inline/api", json={
+        "name": "Inline Comments DB",
+        "description": "Inline comments database",
+        "parent_page_id": page_id,
+    })
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["status"] == "created"
+    assert data["is_inline"] is True
+    assert data["parent_page_id"] == page_id
+
+    # Verify the collection exists with inline flag
+    resp = client.get(f"/db/{data['id']}/api")
+    assert resp.json()["collection"]["is_inline"] == 1
+
+
+def test_linked_db_inherits_workspace(client):
+    """Linked database should inherit the source's workspace_id."""
+    # Create workspace
+    from app.db import get_conn
+    with get_conn() as conn:
+        conn.execute(
+            "INSERT INTO workspaces (name, owner_id) VALUES ('Test WS', 1)"
+        )
+        conn.commit()
+        ws_id = conn.execute("SELECT id FROM workspaces ORDER BY id DESC LIMIT 1").fetchone()["id"]
+
+        # Create collection with workspace_id
+        conn.execute(
+            "INSERT INTO collections (name, workspace_id) VALUES (?, ?)",
+            ("WS Collection", ws_id),
+        )
+        conn.commit()
+        coll_id = conn.execute(
+            "SELECT id FROM collections WHERE name='WS Collection'"
+        ).fetchone()["id"]
+
+    # Create linked DB
+    resp = client.post(f"/db/{coll_id}/linked/api", json={"name": "Linked WS DB"})
+    assert resp.status_code == 200
+    linked_id = resp.json()["linked_id"]
+
+    # Verify linked DB has same workspace_id
+    with get_conn() as conn:
+        linked = conn.execute(
+            "SELECT workspace_id FROM collections WHERE id=?", (linked_id,)
+        ).fetchone()
+    assert linked["workspace_id"] == ws_id
+
+
+def test_remove_data_source_not_found(client):
+    """Deleting a non-existent data source returns 404."""
+    resp = client.post("/db/api", json={"name": "Remove Test"})
+    coll_id = resp.json()["id"]
+
+    resp = client.delete(f"/db/{coll_id}/sources/99999/api")
+    assert resp.status_code == 404