[kimi] Add JSON validation guard to queue.json writes (#952) (#995)

tests/loop/test_loop_guard_corrupt_queue.py

@@ -0,0 +1,97 @@
+"""Tests for load_queue corrupt JSON handling in loop_guard.py."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+import scripts.loop_guard as lg
+
+
+@pytest.fixture(autouse=True)
+def _isolate(tmp_path, monkeypatch):
+    """Redirect loop_guard paths to tmp_path for isolation."""
+    monkeypatch.setattr(lg, "QUEUE_FILE", tmp_path / "queue.json")
+    monkeypatch.setattr(lg, "IDLE_STATE_FILE", tmp_path / "idle_state.json")
+    monkeypatch.setattr(lg, "CYCLE_RESULT_FILE", tmp_path / "cycle_result.json")
+    monkeypatch.setattr(lg, "GITEA_API", "http://test:3000/api/v1")
+    monkeypatch.setattr(lg, "REPO_SLUG", "owner/repo")
+
+
+def test_load_queue_missing_file(tmp_path):
+    """Missing queue file returns empty list."""
+    result = lg.load_queue()
+    assert result == []
+
+
+def test_load_queue_valid_data(tmp_path):
+    """Valid queue.json returns ready items."""
+    data = [
+        {"issue": 1, "title": "Ready issue", "ready": True},
+        {"issue": 2, "title": "Not ready", "ready": False},
+    ]
+    lg.QUEUE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    lg.QUEUE_FILE.write_text(json.dumps(data, indent=2))
+
+    result = lg.load_queue()
+    assert len(result) == 1
+    assert result[0]["issue"] == 1
+
+
+def test_load_queue_corrupt_json_logs_warning(tmp_path, capsys):
+    """Corrupt queue.json returns empty list and logs warning."""
+    lg.QUEUE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    lg.QUEUE_FILE.write_text("not valid json {{{")
+
+    result = lg.load_queue()
+    assert result == []
+
+    captured = capsys.readouterr()
+    assert "WARNING" in captured.out
+    assert "Corrupt queue.json" in captured.out
+
+
+def test_load_queue_not_a_list(tmp_path):
+    """Queue.json that is not a list returns empty list."""
+    lg.QUEUE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    lg.QUEUE_FILE.write_text(json.dumps({"not": "a list"}))
+
+    result = lg.load_queue()
+    assert result == []
+
+
+def test_load_queue_no_ready_items(tmp_path):
+    """Queue with no ready items returns empty list."""
+    data = [
+        {"issue": 1, "title": "Not ready 1", "ready": False},
+        {"issue": 2, "title": "Not ready 2", "ready": False},
+    ]
+    lg.QUEUE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    lg.QUEUE_FILE.write_text(json.dumps(data, indent=2))
+
+    result = lg.load_queue()
+    assert result == []
+
+
+def test_load_queue_oserror_logs_warning(tmp_path, monkeypatch, capsys):
+    """OSError when reading queue.json returns empty list and logs warning."""
+    lg.QUEUE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    lg.QUEUE_FILE.write_text("[]")
+
+    # Mock Path.read_text to raise OSError
+    original_read_text = Path.read_text
+
+    def mock_read_text(self, *args, **kwargs):
+        if self.name == "queue.json":
+            raise OSError("Permission denied")
+        return original_read_text(self, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "read_text", mock_read_text)
+
+    result = lg.load_queue()
+    assert result == []
+
+    captured = capsys.readouterr()
+    assert "WARNING" in captured.out
+    assert "Cannot read queue.json" in captured.out