| from pathlib import Path |
| import sqlite3 |
|
|
| from codex_dashboard.codex_data import get_stats, list_threads, list_threads_result |
|
|
|
|
| def _make_state_db(path: Path) -> None: |
| conn = sqlite3.connect(path) |
| conn.execute( |
| """ |
| CREATE TABLE threads ( |
| id TEXT PRIMARY KEY, |
| rollout_path TEXT NOT NULL DEFAULT '', |
| created_at INTEGER NOT NULL, |
| updated_at INTEGER NOT NULL, |
| source TEXT NOT NULL, |
| model_provider TEXT NOT NULL, |
| cwd TEXT NOT NULL, |
| title TEXT NOT NULL, |
| sandbox_policy TEXT NOT NULL DEFAULT '', |
| approval_mode TEXT NOT NULL DEFAULT '', |
| tokens_used INTEGER NOT NULL DEFAULT 0, |
| has_user_event INTEGER NOT NULL DEFAULT 0, |
| archived INTEGER NOT NULL DEFAULT 0, |
| archived_at INTEGER, |
| git_sha TEXT, |
| git_branch TEXT, |
| git_origin_url TEXT, |
| cli_version TEXT NOT NULL DEFAULT '', |
| first_user_message TEXT NOT NULL DEFAULT '', |
| agent_nickname TEXT, |
| agent_role TEXT, |
| memory_mode TEXT NOT NULL DEFAULT 'enabled', |
| model TEXT, |
| reasoning_effort TEXT, |
| agent_path TEXT, |
| created_at_ms INTEGER, |
| updated_at_ms INTEGER |
| ) |
| """ |
| ) |
| rows = [ |
| ( |
| "new", |
| 100, |
| 200, |
| "codex", |
| "openai", |
| "/work/new", |
| "Newest task", |
| 1200, |
| 0, |
| "main", |
| "gpt-5.4", |
| "high", |
| 100_000, |
| 200_000, |
| ), |
| ( |
| "old", |
| 50, |
| 60, |
| "codex", |
| "openai", |
| "/work/old", |
| "Old task", |
| 300, |
| 1, |
| "archive", |
| "gpt-5.4", |
| "medium", |
| 50_000, |
| 60_000, |
| ), |
| ] |
| conn.executemany( |
| """ |
| INSERT INTO threads ( |
| id, created_at, updated_at, source, model_provider, cwd, title, |
| tokens_used, archived, git_branch, model, reasoning_effort, |
| created_at_ms, updated_at_ms |
| ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| """, |
| rows, |
| ) |
| conn.commit() |
| conn.close() |
|
|
|
|
| def test_list_threads_hides_archived_by_default(tmp_path): |
| db = tmp_path / "state.sqlite" |
| _make_state_db(db) |
|
|
| threads = list_threads(db_path=db) |
|
|
| assert [thread.id for thread in threads] == ["new"] |
| assert threads[0].tokens_used == 1200 |
| assert threads[0].reasoning_effort == "high" |
|
|
|
|
| def test_list_threads_can_include_archived_and_search(tmp_path): |
| db = tmp_path / "state.sqlite" |
| _make_state_db(db) |
|
|
| threads = list_threads(db_path=db, include_archived=True, query="old") |
|
|
| assert [thread.id for thread in threads] == ["old"] |
|
|
|
|
| def test_get_stats_handles_missing_db(tmp_path): |
| stats = get_stats(db_path=tmp_path / "missing.sqlite") |
|
|
| assert stats.thread_count == 0 |
| assert stats.latest_thread is None |
| assert stats.error is None |
|
|
|
|
| def test_list_threads_schema_drift_missing_timestamp_columns_reports_data_error(tmp_path): |
| db = tmp_path / "state.sqlite" |
| conn = sqlite3.connect(db) |
| conn.execute( |
| """ |
| CREATE TABLE threads ( |
| id TEXT PRIMARY KEY, |
| title TEXT, |
| first_user_message TEXT, |
| cwd TEXT, |
| model TEXT, |
| model_provider TEXT, |
| reasoning_effort TEXT, |
| tokens_used INTEGER, |
| archived INTEGER DEFAULT 0, |
| source TEXT, |
| git_branch TEXT |
| ) |
| """ |
| ) |
| conn.execute("INSERT INTO threads (id, title, archived) VALUES ('t1', 'bad schema', 0)") |
| conn.commit() |
| conn.close() |
|
|
| result = list_threads_result(db_path=db) |
|
|
| assert result.data == [] |
| assert result.error is not None |
| assert "created_at" in result.error.message or "updated_at" in result.error.message |
|
|
|
|
| def test_list_threads_malformed_tokens_are_reported_without_exception(tmp_path): |
| db = tmp_path / "state.sqlite" |
| _make_state_db(db) |
| conn = sqlite3.connect(db) |
| conn.execute("UPDATE threads SET tokens_used = 'not-an-int' WHERE id = 'new'") |
| conn.commit() |
| conn.close() |
|
|
| result = list_threads_result(db_path=db) |
|
|
| assert result.data == [] |
| assert result.error is not None |
| assert "not-an-int" in result.error.message |
|
|
|
|
| def test_get_stats_schema_drift_reports_error_without_exception(tmp_path): |
| db = tmp_path / "state.sqlite" |
| conn = sqlite3.connect(db) |
| conn.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, archived INTEGER DEFAULT 0)") |
| conn.commit() |
| conn.close() |
|
|
| stats = get_stats(db_path=db) |
|
|
| assert stats.thread_count == 0 |
| assert stats.latest_thread is None |
| assert stats.error is not None |
| assert "tokens_used" in stats.error.message or "updated_at" in stats.error.message |
|
|
|
|
