mcp-server/tests/unit/test_result_tool.py

# 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_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"