feat: add JSON-backed PendingStore

tests/unit/test_pending_store.py

@@ -0,0 +1,78 @@
+# coding=utf-8
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+
+from spark_executor.core import pending_store
+from spark_executor.models import PendingSubmission
+
+
+def _pending(pid: str = "p_1") -> PendingSubmission:
+    return PendingSubmission(
+        pending_id=pid,
+        connection="prod",
+        master="yarn",
+        deploy_mode="cluster",
+        script_path="/tmp/j.py",
+        queue="default",
+        executor_memory="4G",
+        executor_cores=2,
+        num_executors=2,
+        spark_conf={},
+        created_at=datetime(2026, 6, 24),
+    )
+
+
+@pytest.fixture(autouse=True)
+def _fresh(tmp_path: Path, monkeypatch):
+    monkeypatch.setattr(pending_store, "DEFAULT_DATA_DIR", str(tmp_path))
+    monkeypatch.setattr(pending_store, "store", pending_store.PendingStore())
+
+
+def test_save_then_get_roundtrip():
+    pending_store.store.save(_pending())
+    out = pending_store.store.get("p_1")
+    assert out is not None
+    assert out.connection == "prod"
+    assert out.master == "yarn"
+
+
+def test_save_persists_to_json(tmp_path: Path):
+    pending_store.store.save(_pending())
+    path = tmp_path / "pending_jobs.json"
+    assert path.is_file()
+
+
+def test_get_missing_returns_none():
+    assert pending_store.store.get("missing") is None
+
+
+def test_list_all_empty_when_file_absent():
+    assert pending_store.store.list_all() == []
+
+
+def test_list_all_returns_saved():
+    pending_store.store.save(_pending("a"))
+    pending_store.store.save(_pending("b"))
+    assert {p.pending_id for p in pending_store.store.list_all()} == {"a", "b"}
+
+
+def test_save_upserts_existing():
+    pending_store.store.save(_pending("a"))
+    updated = _pending("a")
+    updated.status = "SUBMITTED"
+    updated.job_id = "j_1"
+    pending_store.store.save(updated)
+    assert pending_store.store.get("a").status == "SUBMITTED"
+    assert pending_store.store.get("a").job_id == "j_1"
+
+
+def test_delete_returns_true_when_present():
+    pending_store.store.save(_pending())
+    assert pending_store.store.delete("p_1") is True
+    assert pending_store.store.get("p_1") is None
+
+
+def test_delete_returns_false_when_absent():
+    assert pending_store.store.delete("nope") is False