Azure HPC/AI MCP Server

import sys import subprocess import json # Provide a lightweight dummy FastMCP so importing server.py works in tests class _DummyFastMCP: def __init__(self, *args, **kwargs): pass def tool(self): def decorator(func): return func return decorator def run(self): pass sys.modules.setdefault("fastmcp", type("mod", (), {"FastMCP": _DummyFastMCP})) import importlib.util from pathlib import Path # Import server.py by file path so pytest can load it regardless of sys.path spec = importlib.util.spec_from_file_location( "server", str(Path(__file__).resolve().parents[1] / "server.py") ) server = importlib.util.module_from_spec(spec) spec.loader.exec_module(server) list_nodes = server.list_nodes get_node_topologies = server.get_node_topologies def _proc_ok(stdout_text: str): class P: returncode = 0 stdout = stdout_text stderr = "" return P() def _proc_err(stderr_text: str): class P: returncode = 1 stdout = "" stderr = stderr_text return P() def test_list_nodes_parses_ok(monkeypatch): sample = { "items": [ { "metadata": {"name": "node-1", "labels": {"accelerator": "gpu"}}, "status": {"conditions": [{"type": "Ready", "status": "True"}]}, } ] } stdout = json.dumps(sample) def fake_run(*args, **kwargs): return _proc_ok(stdout) monkeypatch.setattr(subprocess, "run", fake_run) res = list_nodes() assert isinstance(res, list) assert res[0]["name"] == "node-1" assert res[0]["status"] == "Ready" def test_list_nodes_handles_kubectl_error(monkeypatch): stderr = "kubectl: command not found" def fake_run(*args, **kwargs): return _proc_err(stderr) monkeypatch.setattr(subprocess, "run", fake_run) res = list_nodes() assert isinstance(res, list) assert "error" in res[0] def test_get_node_topologies_parses_ok(monkeypatch): sample = { "items": [ { "metadata": { "name": "node-1", "labels": { "kubernetes.azure.com/agentpool": "gpupool", "ib/pkey": "0x1234", "ib/torset": "42", }, }, }, { "metadata": { "name": "node-2", "labels": { "kubernetes.azure.com/agentpool": "gpupool", # missing pkey/torset is allowed }, }, }, ] } stdout = json.dumps(sample) def fake_run(*args, **kwargs): return _proc_ok(stdout) monkeypatch.setattr(subprocess, "run", fake_run) res = get_node_topologies() assert isinstance(res, list) assert res[0]["name"] == "node-1" assert res[0]["agentpool"] == "gpupool" assert res[0]["pkey"] == "0x1234" assert res[0]["torset"] == "42" assert res[1]["name"] == "node-2" assert res[1]["pkey"] is None assert res[1]["torset"] is None def test_get_node_topologies_handles_kubectl_error(monkeypatch): stderr = "kubectl: command not found" def fake_run(*args, **kwargs): return _proc_err(stderr) monkeypatch.setattr(subprocess, "run", fake_run) res = get_node_topologies() assert isinstance(res, list) assert "error" in res[0]