1  """Tests for hermes_cli.cron command handling."""
  2  
  3  from argparse import Namespace
  4  
  5  import pytest
  6  
  7  from cron.jobs import create_job, get_job, list_jobs
  8  from hermes_cli.cron import cron_command
  9  
 10  
 11  @pytest.fixture()
 12  def tmp_cron_dir(tmp_path, monkeypatch):
 13      monkeypatch.setattr("cron.jobs.CRON_DIR", tmp_path / "cron")
 14      monkeypatch.setattr("cron.jobs.JOBS_FILE", tmp_path / "cron" / "jobs.json")
 15      monkeypatch.setattr("cron.jobs.OUTPUT_DIR", tmp_path / "cron" / "output")
 16      return tmp_path
 17  
 18  
 19  class TestCronCommandLifecycle:
 20      def test_pause_resume_run(self, tmp_cron_dir, capsys):
 21          job = create_job(prompt="Check server status", schedule="every 1h")
 22  
 23          cron_command(Namespace(cron_command="pause", job_id=job["id"]))
 24          paused = get_job(job["id"])
 25          assert paused["state"] == "paused"
 26  
 27          cron_command(Namespace(cron_command="resume", job_id=job["id"]))
 28          resumed = get_job(job["id"])
 29          assert resumed["state"] == "scheduled"
 30  
 31          cron_command(Namespace(cron_command="run", job_id=job["id"]))
 32          triggered = get_job(job["id"])
 33          assert triggered["state"] == "scheduled"
 34  
 35          out = capsys.readouterr().out
 36          assert "Paused job" in out
 37          assert "Resumed job" in out
 38          assert "Triggered job" in out
 39  
 40      def test_edit_can_replace_and_clear_skills(self, tmp_cron_dir, capsys):
 41          job = create_job(
 42              prompt="Combine skill outputs",
 43              schedule="every 1h",
 44              skill="blogwatcher",
 45          )
 46  
 47          cron_command(
 48              Namespace(
 49                  cron_command="edit",
 50                  job_id=job["id"],
 51                  schedule="every 2h",
 52                  prompt="Revised prompt",
 53                  name="Edited Job",
 54                  deliver=None,
 55                  repeat=None,
 56                  skill=None,
 57                  skills=["maps", "blogwatcher"],
 58                  clear_skills=False,
 59              )
 60          )
 61          updated = get_job(job["id"])
 62          assert updated["skills"] == ["maps", "blogwatcher"]
 63          assert updated["name"] == "Edited Job"
 64          assert updated["prompt"] == "Revised prompt"
 65          assert updated["schedule_display"] == "every 120m"
 66  
 67          cron_command(
 68              Namespace(
 69                  cron_command="edit",
 70                  job_id=job["id"],
 71                  schedule=None,
 72                  prompt=None,
 73                  name=None,
 74                  deliver=None,
 75                  repeat=None,
 76                  skill=None,
 77                  skills=None,
 78                  clear_skills=True,
 79              )
 80          )
 81          cleared = get_job(job["id"])
 82          assert cleared["skills"] == []
 83          assert cleared["skill"] is None
 84  
 85          out = capsys.readouterr().out
 86          assert "Updated job" in out
 87  
 88      def test_create_with_multiple_skills(self, tmp_cron_dir, capsys):
 89          cron_command(
 90              Namespace(
 91                  cron_command="create",
 92                  schedule="every 1h",
 93                  prompt="Use both skills",
 94                  name="Skill combo",
 95                  deliver=None,
 96                  repeat=None,
 97                  skill=None,
 98                  skills=["blogwatcher", "maps"],
 99              )
100          )
101          out = capsys.readouterr().out
102          assert "Created job" in out
103  
104          jobs = list_jobs()
105          assert len(jobs) == 1
106          assert jobs[0]["skills"] == ["blogwatcher", "maps"]
107          assert jobs[0]["name"] == "Skill combo"