From c734823e790611ff406facf3a5f694b760d93b6d Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 24 Jun 2026 14:40:54 +0800 Subject: [PATCH] feat: add list_pending_jobs and get_pending_job --- spark_executor/tools/submit.py | 25 +++++++++++++++++++++++++ tests/unit/test_submit_tool.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 57 insertions(+) diff --git a/spark_executor/tools/submit.py b/spark_executor/tools/submit.py index 36c42bf..8e09d12 100644 --- a/spark_executor/tools/submit.py +++ b/spark_executor/tools/submit.py @@ -124,3 +124,28 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult: application_id=application_id, tracking_url=tracking_url, ) + + +def list_pending_jobs() -> list[dict[str, object]]: + return [p.model_dump() for p in pending_store.list_all()] + + +def get_pending_job(pending_id: str) -> dict[str, object]: + p = pending_store.get(pending_id) + if p is None: + raise KeyError(f"Unknown pending_id: {pending_id}") + return p.model_dump() + + +def cancel_pending_job(pending_id: str) -> dict[str, str]: + p = pending_store.get(pending_id) + if p is None: + raise KeyError(f"Unknown pending_id: {pending_id}") + if p.status in ("SUBMITTED", "FAILED"): + raise ValueError( + f"pending_id {pending_id} is in status {p.status!r} and cannot be cancelled" + ) + p.status = "CANCELLED" + pending_store.save(p) + logger.info(f"cancel_pending_job pending_id={pending_id}") + return {"pending_id": pending_id, "status": "CANCELLED"} diff --git a/tests/unit/test_submit_tool.py b/tests/unit/test_submit_tool.py index 0241bd8..c3ecd77 100644 --- a/tests/unit/test_submit_tool.py +++ b/tests/unit/test_submit_tool.py @@ -130,3 +130,35 @@ def test_confirm_marks_failed_on_spark_submit_error(monkeypatch): p = submit.pending_store.get(pid) assert p.status == "FAILED" assert "boom" in (p.error or "") + + +# --- list_pending_jobs --- + +def test_list_pending_jobs_empty(): + assert submit.list_pending_jobs() == [] + + +def test_list_pending_jobs_returns_all(monkeypatch): + monkeypatch.setattr(submit, "run_spark_submit", lambda cmd: None) + submit.prepare_submit_job(connection="prod", script_path="/tmp/a.py") + submit.prepare_submit_job(connection="prod", script_path="/tmp/b.py") + out = submit.list_pending_jobs() + assert {p["script_path"] for p in out} == {"/tmp/a.py", "/tmp/b.py"} + assert all(p["status"] == "PENDING" for p in out) + + +# --- get_pending_job --- + +def test_get_pending_job_returns_dump(monkeypatch): + monkeypatch.setattr(submit, "run_spark_submit", lambda cmd: None) + submit.prepare_submit_job(connection="prod", script_path="/tmp/j.py") + pid = _last_pending_id() + out = submit.get_pending_job(pid) + assert out["pending_id"] == pid + assert out["connection"] == "prod" + assert out["status"] == "PENDING" + + +def test_get_pending_job_unknown_raises(): + with pytest.raises(KeyError): + submit.get_pending_job("missing")