import asyncio import pytest from nanobot.cron.service import CronService from nanobot.cron.types import CronSchedule def test_add_job_rejects_unknown_timezone(tmp_path) -> None: service = CronService(tmp_path / "cron" / "jobs.json") with pytest.raises(ValueError, match="unknown timezone 'America/Vancovuer'"): service.add_job( name="tz typo", schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="America/Vancovuer"), message="hello", ) assert service.list_jobs(include_disabled=True) == [] def test_add_job_accepts_valid_timezone(tmp_path) -> None: service = CronService(tmp_path / "cron" / "jobs.json") job = service.add_job( name="tz ok", schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="America/Vancouver"), message="hello", ) assert job.schedule.tz == "America/Vancouver" assert job.state.next_run_at_ms is not None @pytest.mark.asyncio async def test_running_service_honors_external_disable(tmp_path) -> None: store_path = tmp_path / "cron" / "jobs.json" called: list[str] = [] async def on_job(job) -> None: called.append(job.id) service = CronService(store_path, on_job=on_job) job = service.add_job( name="external-disable", schedule=CronSchedule(kind="every", every_ms=200), message="hello", ) await service.start() try: # Wait slightly to ensure file mtime is definitively different await asyncio.sleep(0.05) external = CronService(store_path) updated = external.enable_job(job.id, enabled=False) assert updated is not None assert updated.enabled is False await asyncio.sleep(0.35) assert called == [] finally: service.stop()