File size: 5,014 Bytes
981aa5d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | 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
|