flowdeck/tests/test_search_migrations.py

"""FlowDeck — v5.2.0 (versioned migrations) & v5.0.0 (search / command palette API).

Covers the ``schema_version`` runner, the FTS5 full-text index (with triggers),
and the ``GET /api/search`` endpoint powering the Ctrl+K command palette.
"""
import os
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"
    os.environ["PUBLIC_API_INSECURE_OK"] = "true"

    # Point the process-wide settings singleton at OUR temp DB (xdist-safe).
    from app.config import settings
    settings.database_url = f"sqlite:///{db_path}"
    settings.rate_limit_enabled = False
    settings.public_api_insecure_ok = True

    from app.db import init_db
    from app.main import app
    init_db()

    yield TestClient(app)

    os.unlink(db_path)


def _create_page(title, content="", content_format="blocks"):
    from app.db import get_conn
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO pages (workspace, title, content, content_format, parent_section) "
            "VALUES ('', ?, ?, ?, 'Private')",
            (title, content, content_format),
        )
        conn.commit()
        return cur.lastrowid


def _create_collection(name, description="", icon="📋"):
    from app.db import get_conn
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO collections (name, description, icon, schema_json) "
            "VALUES (?, ?, ?, '[]')",
            (name, description, icon),
        )
        conn.commit()
        return cur.lastrowid


# ── v5.2.0: versioned migrations ──

def test_schema_version_table_tracks_migrations(client):
    from app.db import get_conn
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT version, name FROM schema_version ORDER BY version"
        ).fetchall()
        versions = [r["version"] for r in rows]
        assert 1 in versions  # baseline
        assert max(versions) >= 2  # versioned migrations applied


def test_missing_indexes_created(client):
    from app.db import get_conn
    with get_conn() as conn:
        idx = {r["name"] for r in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='index'"
        ).fetchall()}
        assert "idx_users_email" in idx
        assert "idx_user_oauth_tokens_user" in idx


def test_fts5_virtual_table_and_triggers(client):
    from app.db import get_conn
    from app.migrations import fts5_available
    if not fts5_available():
        pytest.skip("FTS5 not available in this SQLite build")
    with get_conn() as conn:
        tbl = conn.execute(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='pages_fts'"
        ).fetchone()
        assert tbl is not None
        trigs = {r["name"] for r in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='trigger'"
        ).fetchall()}
        assert {"pages_fts_ai", "pages_fts_ad", "pages_fts_au"} <= trigs


def test_fts_index_stays_in_sync_with_pages(client):
    from app.db import get_conn
    from app.migrations import fts5_available
    if not fts5_available():
        pytest.skip("FTS5 not available in this SQLite build")

    pid = _create_page("SyncCheck", "alpha beta gamma")
    with get_conn() as conn:
        hits = conn.execute(
            "SELECT count(*) AS c FROM pages_fts WHERE pages_fts MATCH 'alpha'"
        ).fetchone()["c"]
        assert hits == 1
        # updating a page refreshes the index
        conn.execute("UPDATE pages SET title='SyncCheck2', content='delta epsilon' WHERE id=?", (pid,))
        conn.commit()
    with get_conn() as conn:
        hits = conn.execute(
            "SELECT count(*) AS c FROM pages_fts WHERE pages_fts MATCH 'delta'"
        ).fetchone()["c"]
        assert hits == 1
        old = conn.execute(
            "SELECT count(*) AS c FROM pages_fts WHERE pages_fts MATCH 'alpha'"
        ).fetchone()["c"]
        assert old == 0
        # deleting a page removes it from the index
        conn.execute("DELETE FROM pages WHERE id=?", (pid,))
        conn.commit()
    with get_conn() as conn:
        gone = conn.execute(
            "SELECT count(*) AS c FROM pages_fts WHERE pages_fts MATCH 'delta'"
        ).fetchone()["c"]
        assert gone == 0


# ── v5.0.0: search API ──

def test_search_returns_pages_and_collections(client):
    _create_page("Roadmap 2026", '{"blocks":[{"type":"paragraph","content":"build the future"}]}')
    _create_collection("Projects Database", "track our roadmap projects", icon="🚀")

    resp = client.get("/api/search", params={"q": "roadmap"})
    assert resp.status_code == 200
    data = resp.json()

    titles = [p["title"] for p in data["pages"]] + [c["title"] for c in data["collections"]]
    assert any("Roadmap 2026" in t for t in titles)
    assert any("Projects Database" in t for t in titles)


def test_search_empty_query_returns_no_results(client):
    _create_page("Anything")
    resp = client.get("/api/search", params={"q": ""})
    assert resp.status_code == 200
    data = resp.json()
    assert data["pages"] == []
    assert data["collections"] == []


def test_search_excludes_deleted_pages(client):
    from app.db import get_conn
    pid = _create_page("SecretDraft", "very private content")
    with get_conn() as conn:
        conn.execute("UPDATE pages SET deleted_at=CURRENT_TIMESTAMP WHERE id=?", (pid,))
        conn.commit()

    resp = client.get("/api/search", params={"q": "SecretDraft"})
    assert resp.status_code == 200
    assert not [p for p in resp.json()["pages"] if p["id"] == pid]


def test_search_no_match_returns_empty(client):
    _create_page("Alpha")
    resp = client.get("/api/search", params={"q": "zzzz_nonexistent_qqqq"})
    assert resp.status_code == 200
    data = resp.json()
    assert data["pages"] == []
    assert data["collections"] == []
    assert data["total"] == 0