diff --git a/scripts/test_reset_pipeline_state.py b/scripts/test_reset_pipeline_state.py new file mode 100644 index 00000000..83567b9a --- /dev/null +++ b/scripts/test_reset_pipeline_state.py @@ -0,0 +1,190 @@ +#!/usr/bin/env python3 +"""Tests for scripts/reset_pipeline_state.py — 10 tests.""" + +import json +import os +import sys +import tempfile +from datetime import datetime, timezone, timedelta + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from reset_pipeline_state import reset_pipeline_state, classify_stale, parse_timestamp + + +def test_no_state_file(): + """Reset on missing file returns empty.""" + state, removed = reset_pipeline_state("/nonexistent/pipeline_state.json") + assert state == {} + assert removed == [] + print("PASS: test_no_state_file") + + +def test_empty_state(): + """Empty JSON object is untouched.""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump({}, f) + path = f.name + try: + state, removed = reset_pipeline_state(path) + assert state == {} + assert removed == [] + finally: + os.unlink(path) + print("PASS: test_empty_state") + + +def test_fresh_complete_kept(): + """Recent complete entry is kept.""" + now = datetime.now(timezone.utc) + entry = {"state": "complete", "updated": now.isoformat()} + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump({"my-pipeline": entry}, f) + path = f.name + try: + state, removed = reset_pipeline_state(path) + assert "my-pipeline" in state + assert removed == [] + finally: + os.unlink(path) + print("PASS: test_fresh_complete_kept") + + +def test_old_complete_removed(): + """Complete entry older than 24h is removed.""" + old = (datetime.now(timezone.utc) - timedelta(hours=30)).isoformat() + entry = {"state": "complete", "updated": old} + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump({"old-pipeline": entry}, f) + path = f.name + try: + state, removed = reset_pipeline_state(path) + assert "old-pipeline" not in state + assert len(removed) == 1 + assert "old-pipeline" in removed[0] + finally: + os.unlink(path) + print("PASS: test_old_complete_removed") + + +def test_stuck_running_removed(): + """Running entry older than 6h is treated as stuck and removed.""" + old = (datetime.now(timezone.utc) - timedelta(hours=10)).isoformat() + entry = {"state": "running", "updated": old} + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump({"stuck-pipeline": entry}, f) + path = f.name + try: + state, removed = reset_pipeline_state(path) + assert "stuck-pipeline" not in state + assert len(removed) == 1 + finally: + os.unlink(path) + print("PASS: test_stuck_running_removed") + + +def test_old_failed_removed(): + """Failed entry older than 24h is removed.""" + old = (datetime.now(timezone.utc) - timedelta(hours=48)).isoformat() + entry = {"state": "failed", "updated": old} + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump({"failed-pipeline": entry}, f) + path = f.name + try: + state, removed = reset_pipeline_state(path) + assert "failed-pipeline" not in state + finally: + os.unlink(path) + print("PASS: test_old_failed_removed") + + +def test_running_kept_if_fresh(): + """Fresh running entry is kept.""" + now = datetime.now(timezone.utc) + entry = {"state": "running", "updated": now.isoformat()} + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump({"active-pipeline": entry}, f) + path = f.name + try: + state, removed = reset_pipeline_state(path) + assert "active-pipeline" in state + assert removed == [] + finally: + os.unlink(path) + print("PASS: test_running_kept_if_fresh") + + +def test_dry_run_does_not_modify(): + """Dry run reports removals but doesn't change the file.""" + old = (datetime.now(timezone.utc) - timedelta(hours=30)).isoformat() + content = json.dumps({"old-pipeline": {"state": "complete", "updated": old}}) + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + f.write(content) + path = f.name + try: + state, removed = reset_pipeline_state(path, dry_run=True) + assert "old-pipeline" not in state + assert len(removed) == 1 + # File should be unchanged + with open(path) as f: + file_state = json.load(f) + assert "old-pipeline" in file_state + finally: + os.unlink(path) + print("PASS: test_dry_run_does_not_modify") + + +def test_mixed_entries(): + """Mix of fresh and stale entries — only stale removed.""" + now = datetime.now(timezone.utc) + old = (now - timedelta(hours=30)).isoformat() + state_data = { + "fresh-complete": {"state": "complete", "updated": now.isoformat()}, + "stale-complete": {"state": "complete", "updated": old}, + "fresh-running": {"state": "running", "updated": now.isoformat()}, + "stuck-running": {"state": "running", "updated": (now - timedelta(hours=10)).isoformat()}, + } + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump(state_data, f) + path = f.name + try: + state, removed = reset_pipeline_state(path) + assert "fresh-complete" in state + assert "fresh-running" in state + assert "stale-complete" not in state + assert "stuck-running" not in state + assert len(removed) == 2 + finally: + os.unlink(path) + print("PASS: test_mixed_entries") + + +def test_corrupted_entry_removed(): + """Non-dict entries are removed.""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump({"broken": "not_a_dict", "also-bad": 42}, f) + path = f.name + try: + state, removed = reset_pipeline_state(path) + assert "broken" not in state + assert "also-bad" not in state + finally: + os.unlink(path) + print("PASS: test_corrupted_entry_removed") + + +def run_all(): + test_no_state_file() + test_empty_state() + test_fresh_complete_kept() + test_old_complete_removed() + test_stuck_running_removed() + test_old_failed_removed() + test_running_kept_if_fresh() + test_dry_run_does_not_modify() + test_mixed_entries() + test_corrupted_entry_removed() + print("\nAll 10 tests passed!") + + +if __name__ == "__main__": + run_all()