Skip to content
16 changes: 16 additions & 0 deletions matrix/scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,15 @@ def __init__(self) -> None:
def jobs(self) -> list[Job]:
return cast(list[Job], self.scheduler.get_jobs())

def list_jobs(self) -> list[str]:
"""Return a list of names of all registered jobs."""
return [job.name for job in self.jobs]

def list_jobs(self) -> list[str]:
"""Return a list of names of all registered jobs."""
return [job.name for job in self.jobs]


def _parse_cron(self, cron: str) -> dict:
"""
Parse a cron string into a dictionary suitable for CronTrigger.
Expand Down Expand Up @@ -48,6 +57,13 @@ async def morning_update() -> None:
cron_trigger = CronTrigger(**self._parse_cron(cron))
self.scheduler.add_job(func, trigger=cron_trigger, name=func.__name__)

def unschedule(self, name: str) -> None:
"""Remove a registered job by its name."""
for job in self.jobs:
if job.name == name:
self.scheduler.remove_job(job.id)
break

def start(self) -> None:
"""Start the scheduler."""
self.scheduler.start()
21 changes: 21 additions & 0 deletions tests/test_scheduler.py

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removing the test is not an appropriate way to fix the issue. Now the feature isn't fully tested.

Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
import pytest
from matrixpy.scheduler import Scheduler


def sample_task() -> None:
pass

Comment thread
PenguinBoi12 marked this conversation as resolved.

def test_scheduler_job_management() -> None:
scheduler = Scheduler()

# 1. Verify list_jobs returns an empty list on initialization
assert scheduler.list_jobs() == []

# 2. Verify list_jobs returns the correct name after scheduling
scheduler.schedule("*/5 * * * *", sample_task)
assert scheduler.list_jobs() == ["sample_task"]

# 3. Verify unschedule cleanly removes the job from the system
scheduler.unschedule("sample_task")
assert scheduler.list_jobs() == []