tests/test_database.py

"""Tests for database.py — uses in-memory SQLite."""

import pytest

from reachy_mini_conversation_app.database import MiniMinderDB


@pytest.fixture
def db() -> MiniMinderDB:
    """Create an in-memory database for testing."""
    return MiniMinderDB(":memory:")


def test_schema_creation(db: MiniMinderDB) -> None:
    """Tables should exist after init."""
    conn = db._get_conn()
    tables = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
    ).fetchall()
    table_names = [t["name"] for t in tables]
    assert "headache_entries" in table_names
    assert "medication_entries" in table_names


def test_insert_headache(db: MiniMinderDB) -> None:
    """Insert and retrieve a headache entry."""
    row_id = db.insert_headache(
        {
            "intensity": 7,
            "location": "left temple",
            "pain_type": "throbbing",
            "onset_time": "2 hours ago",
        }
    )
    assert row_id == 1

    entries = db.get_recent_headaches(days=1)
    assert len(entries) == 1
    assert entries[0]["intensity"] == 7
    assert entries[0]["location"] == "left temple"


def test_insert_medication(db: MiniMinderDB) -> None:
    """Insert and retrieve a medication entry."""
    row_id = db.insert_medication(
        {
            "medication_name": "Ibuprofen",
            "dose": "400mg",
            "taken": 1,
        }
    )
    assert row_id == 1

    entries = db.get_recent_medications(days=1)
    assert len(entries) == 1
    assert entries[0]["medication_name"] == "Ibuprofen"
    assert entries[0]["dose"] == "400mg"


def test_list_to_string_conversion(db: MiniMinderDB) -> None:
    """List fields should be stored as comma-separated strings."""
    db.insert_headache(
        {
            "intensity": 5,
            "associated_symptoms": ["nausea", "light sensitivity"],
            "triggers": ["stress", "lack of sleep"],
        }
    )
    entries = db.get_recent_headaches(days=1)
    assert "nausea" in entries[0]["associated_symptoms"]
    assert "stress" in entries[0]["triggers"]


def test_get_recent_respects_days(db: MiniMinderDB) -> None:
    """Entries older than N days should not be returned (mocked via small cutoff)."""
    db.insert_headache({"intensity": 3})
    # All entries created 'now' should appear within 1 day
    assert len(db.get_recent_headaches(days=1)) == 1
    # But also within 30 days
    assert len(db.get_recent_headaches(days=30)) == 1


def test_export_csv_empty(db: MiniMinderDB) -> None:
    """Export CSV should return empty string when no entries."""
    assert db.export_csv("headache", 30) == ""


def test_export_csv_with_data(db: MiniMinderDB) -> None:
    """Export CSV should contain header and data rows."""
    db.insert_headache({"intensity": 5, "location": "forehead"})
    csv_output = db.export_csv("headache", 30)
    assert "intensity" in csv_output
    assert "forehead" in csv_output


def test_export_html_empty(db: MiniMinderDB) -> None:
    """Export HTML should show 'No entries found' when empty."""
    html = db.export_html("headache", 30)
    assert "No entries found" in html


def test_export_html_with_data(db: MiniMinderDB) -> None:
    """Export HTML should contain a table when data exists."""
    db.insert_medication({"medication_name": "Aspirin", "dose": "325mg"})
    html = db.export_html("medication", 30)
    assert "<table>" in html
    assert "Aspirin" in html


def test_multiple_entries(db: MiniMinderDB) -> None:
    """Insert multiple entries and verify count."""
    for i in range(5):
        db.insert_headache({"intensity": i + 1})
    assert len(db.get_recent_headaches(days=1)) == 5


def test_close(db: MiniMinderDB) -> None:
    """close() should not raise and should clean up connection."""
    db.insert_headache({"intensity": 1})
    assert len(db.get_recent_headaches(days=1)) == 1
    db.close()
    assert db._conn is None


# ---- Token Usage / Caching Tests ----


def test_log_token_usage_with_cached(db: MiniMinderDB) -> None:
    """Token usage should persist with cached token counts."""
    row_id = db.log_token_usage(
        session_id="test-session",
        input_tokens_text=1000,
        input_tokens_audio=500,
        output_tokens_text=200,
        output_tokens_audio=100,
        cached_tokens_text=600,
        cached_tokens_audio=300,
        estimated_cost_usd=0.042,
    )
    assert row_id == 1

    # Verify via raw query
    conn = db._get_conn()
    row = conn.execute("SELECT * FROM token_usage WHERE id = ?", (row_id,)).fetchone()
    assert row["cached_tokens_text"] == 600
    assert row["cached_tokens_audio"] == 300
    assert row["input_tokens_text"] == 1000


def test_session_cost_summary_includes_cache_rate(db: MiniMinderDB) -> None:
    """Cost summary should return cached totals and cache_hit_rate."""
    db.log_token_usage(
        session_id="s1",
        input_tokens_text=1000,
        input_tokens_audio=0,
        cached_tokens_text=400,
        estimated_cost_usd=0.01,
    )
    db.log_token_usage(
        session_id="s1",
        input_tokens_text=2000,
        input_tokens_audio=0,
        cached_tokens_text=1600,
        estimated_cost_usd=0.02,
    )

    summary = db.get_session_cost_summary("s1")
    assert summary["cached_text"] == 2000  # 400 + 1600
    assert summary["cached_audio"] == 0
    assert summary["input_text"] == 3000
    assert summary["response_count"] == 2
    # cache_hit_rate = 2000 / 3000 = 0.6667
    assert abs(summary["cache_hit_rate"] - 0.6667) < 0.001


def test_session_cost_summary_zero_input(db: MiniMinderDB) -> None:
    """Cache hit rate should be 0 when there are no input tokens."""
    summary = db.get_session_cost_summary("nonexistent")
    assert summary["cache_hit_rate"] == 0.0
    assert summary["cached_text"] == 0


def test_migrate_cached_token_columns(db: MiniMinderDB) -> None:
    """Migration should be idempotent — running it again should not error."""
    conn = db._get_conn()
    # Run migration again (already ran during init)
    db._migrate_add_cached_token_columns(conn)
    # Should still have the columns
    cursor = conn.execute("PRAGMA table_info(token_usage)")
    columns = [row[1] for row in cursor.fetchall()]
    assert "cached_tokens_text" in columns
    assert "cached_tokens_audio" in columns


# ---- Cross-Session Summary Tests ----


def test_prior_session_summary_empty(db: MiniMinderDB) -> None:
    """Should return empty string when no prior sessions exist."""
    from reachy_mini_conversation_app.database import ConversationLogger

    logger = ConversationLogger(db)
    result = logger.get_prior_session_summary(current_session_id="current")
    assert result == ""


def test_prior_session_summary_with_data(db: MiniMinderDB) -> None:
    """Should build a formatted summary from prior session turns."""
    from reachy_mini_conversation_app.database import ConversationLogger

    logger = ConversationLogger(db)

    # Log turns for a prior session
    logger.log_turn("prior-1", "user", "I took my morning medication")
    logger.log_turn("prior-1", "assistant", "Great, I've noted that.")
    logger.log_turn("prior-1", "user", "I have a headache today")

    # Request summary excluding current session
    result = logger.get_prior_session_summary(
        current_session_id="current-session",
        max_sessions=3,
    )
    assert "## Prior Session Context" in result
    assert "morning medication" in result
    assert "headache" in result


def test_prior_session_summary_excludes_current(db: MiniMinderDB) -> None:
    """Should exclude the current session from the summary."""
    from reachy_mini_conversation_app.database import ConversationLogger

    logger = ConversationLogger(db)

    # Log turns for current and prior sessions
    logger.log_turn("current", "user", "Hello today")
    logger.log_turn("prior", "user", "Yesterday's chat")

    result = logger.get_prior_session_summary(
        current_session_id="current",
        max_sessions=3,
    )
    assert "Yesterday" in result
    assert "Hello today" not in result


# ---- Prompt Assembly Tests ----


def test_build_session_instructions_with_graph_context() -> None:
    """Prompt should include graph context and prior sessions when provided."""
    from reachy_mini_conversation_app.prompts import (
        SessionState,
        build_session_instructions,
    )

    state = SessionState(
        onboarding_completed=True,
        graph_context="## Knowledge Graph Context\n- Takes Topiramate 50mg",
        prior_session_summary="## Prior Sessions\n- Mentioned headache",
    )
    prompt = build_session_instructions(state)
    assert "Knowledge Graph Context" in prompt
    assert "Topiramate" in prompt
    assert "Prior Sessions" in prompt
    assert "headache" in prompt


def test_build_session_instructions_without_context() -> None:
    """Prompt should work fine without graph or session context."""
    from reachy_mini_conversation_app.prompts import (
        SessionState,
        build_session_instructions,
    )

    state = SessionState(onboarding_completed=True)
    prompt = build_session_instructions(state)
    assert "Knowledge Graph Context" not in prompt
    assert "Prior Session" not in prompt
    # Should still have core sections
    assert len(prompt) > 100