38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
from pathlib import Path
|
|
|
|
from llm_trade_bot.database import Database
|
|
|
|
|
|
async def test_notes_and_admin_deduplication(tmp_path: Path):
|
|
database = Database(tmp_path / "state.db")
|
|
await database.open()
|
|
try:
|
|
await database.add_note("first")
|
|
await database.add_note("second")
|
|
assert await database.recent_notes(10, 100) == ["first", "second"]
|
|
first_id, first_queued = await database.add_admin_request("Need a new data feed")
|
|
second_id, second_queued = await database.add_admin_request("Need a new data feed")
|
|
assert first_queued is True
|
|
assert second_queued is False
|
|
assert second_id == first_id
|
|
outbox = await database.pending_outbox()
|
|
assert len(outbox) == 1
|
|
assert outbox[0].kind == "ask_admin"
|
|
finally:
|
|
await database.close()
|
|
|
|
|
|
async def test_run_lifecycle(tmp_path: Path):
|
|
database = Database(tmp_path / "state.db")
|
|
await database.open()
|
|
try:
|
|
run_id = await database.start_run("test")
|
|
await database.add_event("tool_started", {"name": "note"}, run_id)
|
|
await database.finish_run(run_id, status="completed", summary="idle")
|
|
last = await database.last_run()
|
|
assert last is not None
|
|
assert last["status"] == "completed"
|
|
assert last["summary"] == "idle"
|
|
finally:
|
|
await database.close()
|