191 lines
7.0 KiB
Python
191 lines
7.0 KiB
Python
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()
|