127 lines
4.2 KiB
Python
127 lines
4.2 KiB
Python
from pathlib import Path
|
|
import json
|
|
|
|
from src.storage import HistoryDB
|
|
|
|
|
|
def test_history_db_job_and_events_roundtrip(tmp_path: Path) -> None:
|
|
db = HistoryDB(tmp_path / "screenjob_test.db")
|
|
job_id = "job_test_001"
|
|
db.create_job(
|
|
job_id=job_id,
|
|
objective="Open example.com",
|
|
model="gpt-5.4-mini",
|
|
created_at="2026-05-27T00:00:00Z",
|
|
safety_override=False,
|
|
disabled_tools=["click"],
|
|
)
|
|
db.add_event(
|
|
job_id=job_id,
|
|
ts="2026-05-27T00:00:01Z",
|
|
step=1,
|
|
event_type="tool_called",
|
|
payload={"tool": "see_screen"},
|
|
)
|
|
db.update_job(
|
|
job_id,
|
|
status="completed",
|
|
ended_at="2026-05-27T00:00:02Z",
|
|
result="Done",
|
|
response_json=json.dumps({"return": "Done", "data": {"files": ["a.txt", "b.txt"]}}, ensure_ascii=False),
|
|
steps=2,
|
|
estimated_cost_usd=0.1234,
|
|
)
|
|
|
|
job = db.get_job(job_id)
|
|
assert job is not None
|
|
assert job["status"] == "completed"
|
|
assert job["model"] == "gpt-5.4-mini"
|
|
assert job["disabled_tools"] == ["click"]
|
|
assert job["response"]["return"] == "Done"
|
|
assert job["response"]["data"]["files"] == ["a.txt", "b.txt"]
|
|
assert job["usage"]["estimated_cost_usd"] == 0.1234
|
|
|
|
events = db.get_job_events(job_id, limit=10)
|
|
assert len(events) == 1
|
|
assert events[0]["event_type"] == "tool_called"
|
|
assert events[0]["payload"]["tool"] == "see_screen"
|
|
|
|
jobs = db.list_jobs(limit=10)
|
|
assert len(jobs) == 1
|
|
assert jobs[0]["job_id"] == job_id
|
|
|
|
stats = db.stats()
|
|
assert stats["total_jobs"] == 1
|
|
assert stats["completed_jobs"] == 1
|
|
assert abs(stats["total_estimated_cost"] - 0.1234) < 1e-9
|
|
|
|
|
|
def test_storage_response_fallback_uses_result_when_json_missing(tmp_path: Path) -> None:
|
|
db = HistoryDB(tmp_path / "screenjob_test_fallback.db")
|
|
job_id = "job_test_002"
|
|
db.create_job(
|
|
job_id=job_id,
|
|
objective="Fallback check",
|
|
model="gpt-5.4-mini",
|
|
created_at="2026-05-27T00:00:00Z",
|
|
safety_override=False,
|
|
disabled_tools=[],
|
|
)
|
|
db.update_job(job_id, status="completed", result="Legacy result string")
|
|
job = db.get_job(job_id)
|
|
assert job is not None
|
|
assert job["response"]["return"] == "Legacy result string"
|
|
assert job["response"]["data"] is None
|
|
|
|
|
|
def test_history_db_analytics_groups_by_category_and_day(tmp_path: Path) -> None:
|
|
db = HistoryDB(tmp_path / "screenjob_test_analytics.db")
|
|
|
|
db.create_job(
|
|
job_id="job_browser_ok",
|
|
objective="Open amazon.de and checkout",
|
|
model="gpt-5.4-mini",
|
|
created_at="2026-05-27T00:00:01Z",
|
|
safety_override=False,
|
|
disabled_tools=[],
|
|
)
|
|
db.update_job("job_browser_ok", status="completed", steps=4, estimated_cost_usd=0.12)
|
|
|
|
db.create_job(
|
|
job_id="job_browser_fail",
|
|
objective="Open website and login",
|
|
model="gpt-5.4-mini",
|
|
created_at="2026-05-28T00:00:01Z",
|
|
safety_override=False,
|
|
disabled_tools=[],
|
|
)
|
|
db.update_job("job_browser_fail", status="failed", steps=6, estimated_cost_usd=0.24)
|
|
|
|
db.create_job(
|
|
job_id="job_terminal_ok",
|
|
objective="Run a shell command to inspect files",
|
|
model="gpt-5.4-mini",
|
|
created_at="2026-05-28T00:00:02Z",
|
|
safety_override=False,
|
|
disabled_tools=[],
|
|
)
|
|
db.update_job("job_terminal_ok", status="completed", steps=10, estimated_cost_usd=0.05)
|
|
|
|
analytics = db.analytics()
|
|
assert analytics["total_jobs"] == 3
|
|
assert analytics["finished_jobs"] == 3
|
|
assert analytics["completed_jobs"] == 2
|
|
assert analytics["failed_jobs"] == 1
|
|
assert analytics["success_rate"] == 66.67
|
|
assert analytics["avg_steps"] == 6.67
|
|
assert analytics["avg_cost_usd"] == 0.136667
|
|
|
|
browser = next(row for row in analytics["by_category"] if row["label"] == "Browser / web")
|
|
terminal = next(row for row in analytics["by_category"] if row["label"] == "Files / terminal")
|
|
assert browser["finished_jobs"] == 2
|
|
assert browser["success_rate"] == 50.0
|
|
assert browser["avg_steps"] == 5.0
|
|
assert terminal["success_rate"] == 100.0
|
|
|
|
assert [row["label"] for row in analytics["timeline"]] == ["2026-05-27", "2026-05-28"]
|