49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
from agent.db import connect, ensure_database, run_migrations
|
|
from agent.firefly import FireflyError
|
|
from agent.jobs.heartbeat import run_heartbeat
|
|
|
|
|
|
class FakeFirefly:
|
|
def __init__(self, version: str | None = None, error: Exception | None = None):
|
|
self._version = version
|
|
self._error = error
|
|
|
|
def about(self) -> dict:
|
|
if self._error:
|
|
raise self._error
|
|
return {"version": self._version}
|
|
|
|
|
|
def _last_run(cfg):
|
|
with connect(cfg) as conn:
|
|
return conn.execute(
|
|
"SELECT connector, status, detail, started_at, finished_at"
|
|
" FROM sync_runs ORDER BY id DESC LIMIT 1"
|
|
).fetchone()
|
|
|
|
|
|
def test_heartbeat_records_ok(cfg):
|
|
ensure_database(cfg)
|
|
run_migrations(cfg)
|
|
|
|
status = run_heartbeat(cfg, client=FakeFirefly(version="6.1.1"))
|
|
|
|
assert status == "ok"
|
|
connector, db_status, detail, started_at, finished_at = _last_run(cfg)
|
|
assert connector == "heartbeat"
|
|
assert db_status == "ok"
|
|
assert "6.1.1" in detail
|
|
assert finished_at >= started_at
|
|
|
|
|
|
def test_heartbeat_records_error_and_does_not_raise(cfg):
|
|
ensure_database(cfg)
|
|
run_migrations(cfg)
|
|
|
|
status = run_heartbeat(cfg, client=FakeFirefly(error=FireflyError("GET /api/v1/about -> 401")))
|
|
|
|
assert status == "error"
|
|
connector, db_status, detail, *_ = _last_run(cfg)
|
|
assert db_status == "error"
|
|
assert "401" in detail
|