31 lines
877 B
Python
31 lines
877 B
Python
from datetime import timedelta
|
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
|
from agent.config import AgentConfig
|
|
from agent.scheduler import build_scheduler
|
|
|
|
|
|
def _cfg(minutes: int) -> AgentConfig:
|
|
return AgentConfig(
|
|
db_host="db",
|
|
db_port=5432,
|
|
db_user="u",
|
|
db_password="p",
|
|
agent_db_name="agentdb",
|
|
firefly_url="http://app:8080",
|
|
firefly_token="tok",
|
|
heartbeat_interval_minutes=minutes,
|
|
)
|
|
|
|
|
|
def test_build_scheduler_registers_heartbeat_at_configured_interval():
|
|
scheduler = build_scheduler(_cfg(minutes=15), scheduler_class=BackgroundScheduler)
|
|
scheduler.start(paused=True)
|
|
try:
|
|
job = scheduler.get_job("heartbeat")
|
|
assert job is not None
|
|
assert job.trigger.interval == timedelta(minutes=15)
|
|
finally:
|
|
scheduler.shutdown(wait=False)
|