test: strengthen assertions across 7 test files (batch 1)
Replaced weak 'is not None' / '> 0' / 'len >= 1' assertions with concrete value checks across the most flagged test files: gateway/test_pairing.py (11 weak → 0): - Code assertions verify isinstance + len == CODE_LENGTH - Approval results verify dict structure + specific user_id/user_name - Added code2 != code1 check in rate_limit_expires test_hermes_state.py (6 weak → 0): - ended_at verified as float timestamp - Search result counts exact (== 2, not >= 1) - Context verified as non-empty list - Export verified as dict, session ID verified test_cli_init.py (4 weak → 0): - max_turns asserts exact value (60) - model asserts string with provider/name format gateway/test_hooks.py (2 zero-assert tests → fixed): - test_no_handlers_for_event: verifies no handler registered - test_handler_error_does_not_propagate: verifies handler count + return gateway/test_platform_base.py (9 weak image tests → fixed): - extract_images tests now verify actual URL and alt_text - truncate_message verifies content preservation after splitting cron/test_scheduler.py (1 weak → 0): - resolve_origin verifies dict equality, not just existence cron/test_jobs.py (2 weak → 0 + 4 new tests): - Schedule parsing verifies ISO timestamp type - Cron expression verifies result is valid datetime string - NEW: 4 tests for update_job() (was completely untested)
This commit is contained in:
parent
e9f05b3524
commit
a44e041acf
7 changed files with 115 additions and 28 deletions
|
|
@ -15,6 +15,7 @@ from cron.jobs import (
|
|||
save_jobs,
|
||||
get_job,
|
||||
list_jobs,
|
||||
update_job,
|
||||
remove_job,
|
||||
mark_job_run,
|
||||
get_due_jobs,
|
||||
|
|
@ -70,8 +71,10 @@ class TestParseSchedule:
|
|||
result = parse_schedule("30m")
|
||||
assert result["kind"] == "once"
|
||||
assert "run_at" in result
|
||||
# run_at should be ~30 minutes from now
|
||||
run_at = datetime.fromisoformat(result["run_at"])
|
||||
# run_at should be a valid ISO timestamp string ~30 minutes from now
|
||||
run_at_str = result["run_at"]
|
||||
assert isinstance(run_at_str, str)
|
||||
run_at = datetime.fromisoformat(run_at_str)
|
||||
assert run_at > datetime.now()
|
||||
assert run_at < datetime.now() + timedelta(minutes=31)
|
||||
|
||||
|
|
@ -140,8 +143,10 @@ class TestComputeNextRun:
|
|||
pytest.importorskip("croniter")
|
||||
schedule = {"kind": "cron", "expr": "* * * * *"} # every minute
|
||||
result = compute_next_run(schedule)
|
||||
assert result is not None
|
||||
assert isinstance(result, str), f"Expected ISO timestamp string, got {type(result)}"
|
||||
assert len(result) > 0
|
||||
next_dt = datetime.fromisoformat(result)
|
||||
assert isinstance(next_dt, datetime)
|
||||
assert next_dt > datetime.now()
|
||||
|
||||
def test_unknown_kind_returns_none(self):
|
||||
|
|
@ -207,6 +212,48 @@ class TestJobCRUD:
|
|||
assert job["deliver"] == "local"
|
||||
|
||||
|
||||
class TestUpdateJob:
|
||||
def test_update_name(self, tmp_cron_dir):
|
||||
job = create_job(prompt="Check server status", schedule="every 1h", name="Old Name")
|
||||
assert job["name"] == "Old Name"
|
||||
updated = update_job(job["id"], {"name": "New Name"})
|
||||
assert updated is not None
|
||||
assert isinstance(updated, dict)
|
||||
assert updated["name"] == "New Name"
|
||||
# Verify other fields are preserved
|
||||
assert updated["prompt"] == "Check server status"
|
||||
assert updated["id"] == job["id"]
|
||||
assert updated["schedule"] == job["schedule"]
|
||||
# Verify persisted to disk
|
||||
fetched = get_job(job["id"])
|
||||
assert fetched["name"] == "New Name"
|
||||
|
||||
def test_update_schedule(self, tmp_cron_dir):
|
||||
job = create_job(prompt="Daily report", schedule="every 1h")
|
||||
assert job["schedule"]["kind"] == "interval"
|
||||
assert job["schedule"]["minutes"] == 60
|
||||
new_schedule = parse_schedule("every 2h")
|
||||
updated = update_job(job["id"], {"schedule": new_schedule})
|
||||
assert updated is not None
|
||||
assert updated["schedule"]["kind"] == "interval"
|
||||
assert updated["schedule"]["minutes"] == 120
|
||||
# Verify persisted to disk
|
||||
fetched = get_job(job["id"])
|
||||
assert fetched["schedule"]["minutes"] == 120
|
||||
|
||||
def test_update_enable_disable(self, tmp_cron_dir):
|
||||
job = create_job(prompt="Toggle me", schedule="every 1h")
|
||||
assert job["enabled"] is True
|
||||
updated = update_job(job["id"], {"enabled": False})
|
||||
assert updated["enabled"] is False
|
||||
fetched = get_job(job["id"])
|
||||
assert fetched["enabled"] is False
|
||||
|
||||
def test_update_nonexistent_returns_none(self, tmp_cron_dir):
|
||||
result = update_job("nonexistent_id", {"name": "X"})
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestMarkJobRun:
|
||||
def test_increments_completed(self, tmp_cron_dir):
|
||||
job = create_job(prompt="Test", schedule="every 1h")
|
||||
|
|
|
|||
|
|
@ -15,9 +15,11 @@ class TestResolveOrigin:
|
|||
}
|
||||
}
|
||||
result = _resolve_origin(job)
|
||||
assert result is not None
|
||||
assert isinstance(result, dict)
|
||||
assert result == job["origin"]
|
||||
assert result["platform"] == "telegram"
|
||||
assert result["chat_id"] == "123456"
|
||||
assert result["chat_name"] == "Test Chat"
|
||||
|
||||
def test_no_origin(self):
|
||||
assert _resolve_origin({}) is None
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue