aa3d22d89c
Some checks failed
Architecture Lint / Linter Tests (pull_request) Successful in 33s
PR Checklist / pr-checklist (pull_request) Failing after 4m27s
Smoke Test / smoke (pull_request) Failing after 20s
Validate Config / YAML Lint (pull_request) Failing after 17s
Validate Config / JSON Validate (pull_request) Successful in 16s
Validate Config / Python Syntax & Import Check (pull_request) Failing after 1m33s
Validate Config / Shell Script Lint (pull_request) Failing after 42s
Validate Config / Cron Syntax Check (pull_request) Successful in 9s
Validate Config / Deploy Script Dry Run (pull_request) Successful in 12s
Validate Config / Playbook Schema Validation (pull_request) Successful in 22s
Architecture Lint / Lint Repository (pull_request) Has been cancelled
Validate Config / Python Test Suite (pull_request) Has been cancelled
- scripts/cron-audit-662.py: Full audit tool that categorizes cron jobs into healthy/transient/systemic based on error age (48h threshold) Flags: --disable (pause systemic), --issues (file Gitea issues), --output (JSON report), --jobs-file (override path) Reads from ~/.hermes/cron/ or falls back to 'hermes cron list' - scripts/cron_audit_662.py: Symlink for importability - tests/test_cron_audit.py: 9 tests covering categorization logic: healthy (ok, never-run, paused, completed), transient (recent error), systemic (old error, boundary cases), mixed audit report - cron/audit-report.json: Initial audit of local jobs.json Result: 7/7 healthy, 0 transient, 0 systemic
110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
"""
|
|
Tests for scripts/cron-audit-662.py — cron fleet audit.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime, timezone, timedelta
|
|
from pathlib import Path
|
|
|
|
# Add scripts to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
|
|
from cron_audit_662 import categorize_job, audit_jobs
|
|
|
|
|
|
class TestCategorizeJob(unittest.TestCase):
|
|
def setUp(self):
|
|
self.now = datetime(2026, 4, 14, 20, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_healthy_ok(self):
|
|
job = {"id": "a1", "name": "Test", "last_status": "ok", "enabled": True, "state": "scheduled"}
|
|
result = categorize_job(job, self.now)
|
|
self.assertEqual(result["category"], "healthy")
|
|
|
|
def test_healthy_never_run(self):
|
|
job = {"id": "a2", "name": "Never", "last_status": None, "last_error": None}
|
|
result = categorize_job(job, self.now)
|
|
self.assertEqual(result["category"], "healthy")
|
|
|
|
def test_healthy_paused(self):
|
|
job = {"id": "a3", "name": "Paused", "state": "paused", "paused_reason": "intentional"}
|
|
result = categorize_job(job, self.now)
|
|
self.assertEqual(result["category"], "healthy")
|
|
|
|
def test_healthy_completed(self):
|
|
job = {"id": "a4", "name": "Done", "state": "completed"}
|
|
result = categorize_job(job, self.now)
|
|
self.assertEqual(result["category"], "healthy")
|
|
|
|
def test_transient_recent_error(self):
|
|
recent = (self.now - timedelta(hours=2)).isoformat()
|
|
job = {
|
|
"id": "t1", "name": "RecentErr",
|
|
"last_status": "error",
|
|
"last_error": "Connection timeout",
|
|
"last_run_at": recent,
|
|
"enabled": True,
|
|
"state": "scheduled",
|
|
}
|
|
result = categorize_job(job, self.now)
|
|
self.assertEqual(result["category"], "transient")
|
|
self.assertIn("transient", result["reason"].lower())
|
|
|
|
def test_systemic_old_error(self):
|
|
old = (self.now - timedelta(hours=72)).isoformat()
|
|
job = {
|
|
"id": "s1", "name": "OldErr",
|
|
"last_status": "error",
|
|
"last_error": "ConfigError: bad config",
|
|
"last_run_at": old,
|
|
"enabled": True,
|
|
"state": "scheduled",
|
|
}
|
|
result = categorize_job(job, self.now)
|
|
self.assertEqual(result["category"], "systemic")
|
|
self.assertEqual(result["action"], "disable")
|
|
|
|
def test_systemic_boundary(self):
|
|
"""48.1 hours should be systemic."""
|
|
boundary = (self.now - timedelta(hours=48, minutes=6)).isoformat()
|
|
job = {
|
|
"id": "s2", "name": "Boundary",
|
|
"last_status": "error",
|
|
"last_error": "fail",
|
|
"last_run_at": boundary,
|
|
"enabled": True,
|
|
"state": "scheduled",
|
|
}
|
|
result = categorize_job(job, self.now)
|
|
self.assertEqual(result["category"], "systemic")
|
|
|
|
|
|
class TestAuditJobs(unittest.TestCase):
|
|
def test_empty(self):
|
|
report = audit_jobs([])
|
|
self.assertEqual(report["total_jobs"], 0)
|
|
self.assertEqual(report["summary"]["healthy"], 0)
|
|
|
|
def test_mixed_report(self):
|
|
now = datetime(2026, 4, 14, 20, 0, 0, tzinfo=timezone.utc)
|
|
old = (now - timedelta(hours=72)).isoformat()
|
|
recent = (now - timedelta(hours=1)).isoformat()
|
|
|
|
jobs = [
|
|
{"id": "h1", "name": "Healthy", "last_status": "ok", "enabled": True, "state": "scheduled"},
|
|
{"id": "t1", "name": "Transient", "last_status": "error", "last_error": "timeout", "last_run_at": recent, "enabled": True, "state": "scheduled"},
|
|
{"id": "s1", "name": "Systemic", "last_status": "error", "last_error": "config bad", "last_run_at": old, "enabled": True, "state": "scheduled"},
|
|
{"id": "p1", "name": "Paused", "state": "paused", "paused_reason": "frozen"},
|
|
]
|
|
report = audit_jobs(jobs)
|
|
self.assertEqual(report["summary"]["healthy"], 2)
|
|
self.assertEqual(report["summary"]["transient_errors"], 1)
|
|
self.assertEqual(report["summary"]["systemic_failures"], 1)
|
|
self.assertEqual(len(report["systemic_jobs"]), 1)
|
|
self.assertEqual(report["systemic_jobs"][0]["name"], "Systemic")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|