claire/tests/test_db.py

from __future__ import annotations

from claire.db import migrate, open_db


def test_migrate_is_idempotent() -> None:
    conn = open_db(":memory:")
    first = migrate(conn)
    second = migrate(conn)
    assert first == [
        "0001_initial", "0002_chat", "0003_pm", "0004_fleet",
        "0003_pm_alter", "0005_session_liveness", "0006_project_org",
        "0007_usage", "0008_task_blocked_by", "0009_decisions",
        "0010_role_clare_to_claire", "0011_host_telemetry",
        "0012_decision_pinned",
    ]
    assert second == []  # already applied


def test_schema_has_expected_tables() -> None:
    conn = open_db(":memory:")
    migrate(conn)
    rows = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
    ).fetchall()
    names = {r[0] for r in rows}
    expected = {
        "schema_migrations",
        "events",
        "projects",
        "tasks",
        "assignments",
        "sessions",
        "groups",
        "updates",
        "chat_messages",
        "task_splits",
        "orgs",
        "people",
        "epics",
        "tags",
        "categories",
        "domains",
        "task_tags",
        "task_state_history",
        "agent_status",
    }
    assert expected <= names


def test_pm_alter_adds_task_columns() -> None:
    """Migration 0003_pm_alter adds the new task columns idempotently."""
    conn = open_db(":memory:")
    migrate(conn)
    cols = {row[1] for row in conn.execute("PRAGMA table_info(tasks)")}
    expected = {
        "epic_id", "owner_person_id", "task_type",
        "category_id", "domain_id", "color", "emoji",
        "apply_color_rule",
    }
    assert expected <= cols
    # Idempotent — a second migrate is a no-op.
    again = migrate(conn)
    assert again == []