import json
import os
import tempfile
import unittest
import mcp_pipe
class TestMcpPipe(unittest.TestCase):
def test_build_server_command_from_repo_config(self):
cmd, env = mcp_pipe.build_server_command("mqtt-light-controller")
self.assertIsInstance(cmd, list)
self.assertGreaterEqual(len(cmd), 1)
self.assertEqual(cmd[:3], ["python", "-m", "mqtt_light_controller"])
self.assertIsInstance(env, dict)
def test_build_server_command_fallback_script_path(self):
cmd, env = mcp_pipe.build_server_command("mqtt_light_controller.py")
self.assertEqual(cmd[0], os.sys.executable)
self.assertTrue(cmd[1].endswith("mqtt_light_controller.py"))
self.assertIsInstance(env, dict)
def test_load_config_missing_file_returns_empty_dict(self):
with tempfile.TemporaryDirectory() as td:
missing = os.path.join(td, "missing_config.json")
old = os.environ.get("MCP_CONFIG")
try:
os.environ["MCP_CONFIG"] = missing
cfg = mcp_pipe.load_config()
self.assertEqual(cfg, {})
finally:
if old is None:
os.environ.pop("MCP_CONFIG", None)
else:
os.environ["MCP_CONFIG"] = old
def test_load_config_reads_valid_json(self):
with tempfile.TemporaryDirectory() as td:
path = os.path.join(td, "cfg.json")
with open(path, "w", encoding="utf-8") as f:
json.dump({"mcpServers": {"x": {"type": "stdio", "command": "python", "args": ["-V"]}}}, f)
old = os.environ.get("MCP_CONFIG")
try:
os.environ["MCP_CONFIG"] = path
cfg = mcp_pipe.load_config()
self.assertIn("mcpServers", cfg)
self.assertIn("x", cfg["mcpServers"])
finally:
if old is None:
os.environ.pop("MCP_CONFIG", None)
else:
os.environ["MCP_CONFIG"] = old
if __name__ == "__main__":
unittest.main()
