# coding=utf-8 from datetime import datetime from pathlib import Path from unittest.mock import patch import pytest from spark_executor.core import connection_store from spark_executor.core.job_store import JobStore from spark_executor.models import Connection, Job from spark_executor.tools import result from spark_executor.tools import connections @pytest.fixture def fresh_stores(tmp_path: Path): """Wire up connection + job stores rooted in tmp_path. Per-test isolation so file-backed JobStore doesn't leak between cases.""" store = connection_store.ConnectionStore(data_dir=str(tmp_path)) connection_store.store = store connections.store = store result.conn_store = store result.store = JobStore(data_dir=str(tmp_path)) def _fresh_stores(): store = connection_store.ConnectionStore() connection_store.store = store connections.store = store result.conn_store = store result.store = JobStore() def _seed(job_id="abc", app_id="application_1"): _fresh_stores() result.conn_store.save(Connection(name="prod", master="yarn", yarn_rm_url="http://rm:8088")) result.store.put( Job( job_id=job_id, application_id=app_id, script_path="/tmp/j.py", queue="default", submit_time=datetime(2026, 6, 24), connection="prod", yarn_rm_url="http://rm:8088", ) ) def test_result_returns_parsed_fields(): _seed() raw = { "app": { "id": "application_1", "state": "FINISHED", "finalStatus": "SUCCEEDED", "diagnostics": "Application completed successfully", "trackingUrl": "http://nm:8088/proxy/application_1", "startedTime": 1700000000000, "finishedTime": 1700000123000, } } import json with patch( "spark_executor.tools.result.get_application_status", return_value=("FINISHED", json.dumps(raw)), ) as m: out = result.get_job_result("abc") assert out.application_id == "application_1" assert out.state == "FINISHED" assert out.final_status == "SUCCEEDED" assert out.diagnostics == "Application completed successfully" assert out.tracking_url == "http://nm:8088/proxy/application_1" assert out.started_time == 1700000000000 assert out.finished_time == 1700000123000 args = m.call_args.args assert args[0] == "application_1" assert args[1].yarn_rm_url == "http://rm:8088" def test_result_handles_running_job(): _seed(job_id="running", app_id="application_2") raw = { "app": { "id": "application_2", "state": "RUNNING", "finalStatus": "UNDEFINED", "trackingUrl": "http://rm:8088/proxy/application_2", "startedTime": 1700000000000, } } import json with patch( "spark_executor.tools.result.get_application_status", return_value=("RUNNING", json.dumps(raw)), ): out = result.get_job_result("running") assert out.state == "RUNNING" assert out.final_status == "UNDEFINED" assert out.finished_time is None def test_result_handles_missing_optional_fields(): _seed(job_id="accepted", app_id="application_3") raw = {"app": {"id": "application_3", "state": "ACCEPTED"}} import json with patch( "spark_executor.tools.result.get_application_status", return_value=("ACCEPTED", json.dumps(raw)), ): out = result.get_job_result("accepted") assert out.state == "ACCEPTED" assert out.application_id == "application_3" assert out.final_status is None assert out.diagnostics is None assert out.tracking_url is None assert out.started_time is None assert out.finished_time is None def test_result_raises_keyerror_for_unknown_job(): _fresh_stores() with pytest.raises(KeyError) as ei: result.get_job_result("missing") # New error message must still flag "Unknown" so callers / agents can # recognize the failure, AND mention application_id so the agent # knows the other form is also accepted. msg = str(ei.value) assert "job_id" in msg assert "application_id" in msg def test_result_raises_400_for_external_application_id(fresh_stores): """Input that looks like a YARN application_id but is not in the local JobStore must raise ValueError (-> 400) with a hint to use the external tool, NOT a generic KeyError (-> 404).""" with pytest.raises(ValueError, match="get_external_job_result"): result.get_job_result("application_17400000001_0001") def test_result_raises_when_connection_missing(): _fresh_stores() result.store.put( Job( job_id="abc", application_id="application_1", script_path="/tmp/j.py", queue="default", submit_time=datetime(2026, 6, 24), connection="missing", ) ) with pytest.raises(KeyError, match="Connection not found"): result.get_job_result("abc") # --- application_id accepted (regression: "agent passed wrong id" bug) --- def test_get_job_result_accepts_application_id(fresh_stores): result.conn_store.save(Connection(name="prod", master="yarn", yarn_rm_url="http://rm:8088")) result.store.put( Job( job_id="a1b2c3d4e5f6", application_id="application_17400000001_0001", script_path="/tmp/j.py", queue="default", submit_time=datetime(2026, 6, 24), connection="prod", yarn_rm_url="http://rm:8088", ) ) raw = { "app": { "id": "application_17400000001_0001", "state": "SUCCEEDED", "finalStatus": "SUCCEEDED", } } import json with patch( "spark_executor.tools.result.get_application_status", return_value=("FINISHED", json.dumps(raw)), ) as m: out = result.get_job_result("application_17400000001_0001") assert out.application_id == "application_17400000001_0001" assert out.state == "FINISHED" assert m.call_args.args[0] == "application_17400000001_0001"