File size: 2,166 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 | from pathlib import Path
import sqlite3
from codex_dashboard.codex_data import list_logs, list_logs_result
def _make_logs_db(path: Path) -> None:
conn = sqlite3.connect(path)
conn.execute(
"""
CREATE TABLE logs (
id INTEGER PRIMARY KEY,
ts INTEGER NOT NULL,
ts_nanos INTEGER NOT NULL DEFAULT 0,
level TEXT NOT NULL DEFAULT '',
target TEXT NOT NULL DEFAULT '',
thread_id TEXT NOT NULL DEFAULT '',
module_path TEXT NOT NULL DEFAULT '',
feedback_log_body TEXT
)
"""
)
conn.executemany(
"""
INSERT INTO logs (id, ts, ts_nanos, level, target, thread_id, module_path, feedback_log_body)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
[
(1, 100, 123_000_000, "INFO", "codex_core", "thread-a", "core", "first message"),
(2, 101, 500_000_000, "WARN", "codex_ui", "thread-b", "ui", "needle message"),
],
)
conn.commit()
conn.close()
def test_list_logs_reads_codex_logs_fixture_and_searches(tmp_path):
db = tmp_path / "logs_2.sqlite"
_make_logs_db(db)
logs = list_logs(db_path=db, query="needle")
assert [entry.id for entry in logs] == [2]
assert logs[0].ts_ms == 101_500
assert logs[0].level == "WARN"
assert logs[0].thread_id == "thread-b"
assert logs[0].message == "needle message"
def test_list_logs_schema_drift_missing_expected_columns_reports_data_error(tmp_path):
db = tmp_path / "logs_2.sqlite"
conn = sqlite3.connect(db)
conn.execute(
"""
CREATE TABLE logs (
id INTEGER PRIMARY KEY,
ts INTEGER NOT NULL,
level TEXT,
target TEXT,
thread_id TEXT
)
"""
)
conn.execute("INSERT INTO logs (id, ts, level, target, thread_id) VALUES (1, 100, 'INFO', 'target', 't')")
conn.commit()
conn.close()
result = list_logs_result(db_path=db)
assert result.data == []
assert result.error is not None
assert "ts_nanos" in result.error.message or "feedback_log_body" in result.error.message
|