import unittest import tempfile import os from io import BytesIO from datetime import datetime, timezone import server from server import ApiError, Handler, MAX_BODY_BYTES, compute_next_run_at class ComputeNextRunAtTest(unittest.TestCase): def test_daily_uses_same_day_when_time_is_future(self): after = datetime(2026, 6, 20, 12, 0, tzinfo=timezone.utc) self.assertEqual( compute_next_run_at("daily", "23:30", 0, after), "2026-06-20T23:30:00Z", ) def test_daily_moves_to_next_day_when_time_has_passed(self): after = datetime(2026, 6, 20, 23, 30, tzinfo=timezone.utc) self.assertEqual( compute_next_run_at("daily", "23:30", 0, after), "2026-06-21T23:30:00Z", ) def test_weekly_uses_next_target_weekday(self): after = datetime(2026, 6, 20, 12, 0, tzinfo=timezone.utc) # Saturday self.assertEqual( compute_next_run_at("weekly", "09:00", 0, after, weekday=1), "2026-06-22T09:00:00Z", ) def test_weekly_rolls_forward_a_week_after_same_day_time_passed(self): after = datetime(2026, 6, 22, 10, 0, tzinfo=timezone.utc) # Monday self.assertEqual( compute_next_run_at("weekly", "09:00", 0, after, weekday=1), "2026-06-29T09:00:00Z", ) def test_monthly_uses_next_valid_monthday(self): after = datetime(2026, 1, 30, 12, 0, tzinfo=timezone.utc) self.assertEqual( compute_next_run_at("monthly", "23:00", 0, after, monthday=31), "2026-01-31T23:00:00Z", ) def test_monthly_skips_short_months_for_day_31(self): after = datetime(2026, 2, 1, 0, 0, tzinfo=timezone.utc) self.assertEqual( compute_next_run_at("monthly", "23:00", 0, after, monthday=31), "2026-03-31T23:00:00Z", ) def test_monthly_last_day_uses_current_month_end(self): after = datetime(2026, 2, 1, 0, 0, tzinfo=timezone.utc) self.assertEqual( compute_next_run_at("monthly", "23:00", 0, after, monthday=0), "2026-02-28T23:00:00Z", ) def test_monthly_last_day_rolls_forward_when_current_month_end_passed(self): after = datetime(2026, 2, 28, 23, 0, tzinfo=timezone.utc) self.assertEqual( compute_next_run_at("monthly", "23:00", 0, after, monthday=0), "2026-03-31T23:00:00Z", ) def test_timezone_offset_is_applied_from_local_schedule_time(self): after = datetime(2026, 6, 20, 12, 0, tzinfo=timezone.utc) self.assertEqual( compute_next_run_at("daily", "23:00", 8 * 60, after), "2026-06-20T15:00:00Z", ) def test_invalid_monthday_raises(self): after = datetime(2026, 6, 20, 12, 0, tzinfo=timezone.utc) with self.assertRaises(ValueError): compute_next_run_at("monthly", "23:00", 0, after, monthday=32) class HandlerSecurityTest(unittest.TestCase): def test_allowed_origin_is_accepted(self): handler = type("DummyHandler", (), {"headers": {"Origin": "http://localhost:5580"}})() Handler.require_allowed_origin(handler) def test_disallowed_origin_is_rejected(self): handler = type("DummyHandler", (), {"headers": {"Origin": "http://example.invalid"}})() with self.assertRaises(ApiError) as ctx: Handler.require_allowed_origin(handler) self.assertEqual(ctx.exception.status, 403) def test_request_body_size_is_limited(self): handler = type( "DummyHandler", (), { "headers": {"Content-Length": str(MAX_BODY_BYTES + 1)}, "rfile": BytesIO(), }, )() with self.assertRaises(ApiError) as ctx: Handler.read_json(handler) self.assertEqual(ctx.exception.status, 413) def test_read_json_accepts_small_object(self): body = b'{"ok":true}' handler = type( "DummyHandler", (), { "headers": {"Content-Length": str(len(body))}, "rfile": BytesIO(body), }, )() self.assertEqual(Handler.read_json(handler), {"ok": True}) class OneTimeJobTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.old_db_path = server.DB_PATH self.old_rc_post = server.rc_post server.DB_PATH = os.path.join(self.tmp.name, "jobs.sqlite") self.next_jobid = 100 def fake_rc_post(path, payload, timeout=30, allow_error_body=False): if path.startswith("sync/"): self.next_jobid += 1 return {"jobid": self.next_jobid} if path == "job/status": return {"finished": False, "jobid": payload["jobid"]} if path == "core/stats": return { "bytes": 1, "totalBytes": 2, "speed": 3, "eta": 4, "transfers": 5, "totalTransfers": 6, "errors": 7, "group": payload["group"], } if path == "job/stop": return {} raise AssertionError(path) server.rc_post = fake_rc_post server.init_db() def tearDown(self): server.rc_post = self.old_rc_post server.DB_PATH = self.old_db_path self.tmp.cleanup() def test_create_and_list_one_time_job(self): job = server.create_one_time_job({"action": "copy", "src": "/tmp/a", "dst": "/tmp/b"}) self.assertEqual(job["id"], 1) self.assertEqual(job["jobid"], 101) self.assertEqual(job["status"], "running") self.assertEqual(job["src"], "/tmp/a") jobs = server.list_one_time_jobs() self.assertEqual(len(jobs), 1) self.assertEqual(jobs[0]["id"], 1) self.assertEqual(jobs[0]["status"], "running") self.assertEqual(jobs[0]["statusSnapshot"]["stats"]["bytes"], 1) self.assertEqual(jobs[0]["statusSnapshot"]["stats"]["totalBytes"], 2) self.assertEqual(jobs[0]["statusSnapshot"]["stats"]["speed"], 3) def test_stop_one_time_job_persists_stopped_status(self): job = server.create_one_time_job({"action": "copy", "src": "/tmp/a", "dst": "/tmp/b"}) stopped = server.stop_one_time_job(job["id"]) self.assertEqual(stopped["status"], "stopped") self.assertFalse(stopped["running"]) self.assertIsNotNone(stopped["statusSnapshot"]) def test_run_one_time_job_again_reuses_record_with_new_jobid(self): job = server.create_one_time_job({"action": "copy", "src": "/tmp/a", "dst": "/tmp/b"}) server.stop_one_time_job(job["id"]) restarted = server.manual_run_one_time_job(job["id"]) self.assertEqual(restarted["id"], job["id"]) self.assertEqual(restarted["jobid"], 102) self.assertEqual(restarted["status"], "running") if __name__ == "__main__": unittest.main()