test: harden MCP container smoke validation

Author: LauraGPT

examples/mcp_server/smoke_test.py

@@ -20,9 +20,13 @@
 
 
 def validate_responses(responses):
-    by_id = {response.get("id"): response for response in responses}
-    if set(by_id) != {1, 2}:
-        raise ValueError(f"expected response IDs 1 and 2, got {sorted(by_id)}")
+    if not all(isinstance(response, dict) for response in responses):
+        raise ValueError("expected each stdout payload to be a JSON object")
+
+    response_ids = [response.get("id") for response in responses]
+    if len(responses) != 2 or set(response_ids) != {1, 2}:
+        raise ValueError(f"expected response IDs 1 and 2 only, got {response_ids}")
+    by_id = {response["id"]: response for response in responses}
 
     server_info = by_id[1].get("result", {}).get("serverInfo", {})
     if server_info.get("name") != "funasr":
@@ -34,6 +38,21 @@ def validate_responses(responses):
         raise ValueError(f"transcribe_audio missing from tools/list: {tool_names}")
 
 
+def parse_responses(stdout):
+    responses = []
+    for line_number, raw_line in enumerate(stdout.splitlines(), start=1):
+        line = raw_line.strip()
+        if not line:
+            continue
+        try:
+            responses.append(json.loads(line))
+        except json.JSONDecodeError as error:
+            raise ValueError(
+                f"non-JSON stdout on line {line_number}: {line[:120]}"
+            ) from error
+    return responses
+
+
 def run_smoke_test(image, timeout):
     payload = "".join(f"{json.dumps(request)}\n" for request in REQUESTS)
     completed = subprocess.run(
@@ -49,9 +68,7 @@ def run_smoke_test(image, timeout):
             f"container exited with {completed.returncode}: {completed.stderr.strip()}"
         )
 
-    responses = [
-        json.loads(line) for line in completed.stdout.splitlines() if line.strip()
-    ]
+    responses = parse_responses(completed.stdout)
     validate_responses(responses)
 
 

examples/mcp_server/test_registry_metadata.py

@@ -15,6 +15,8 @@
 
 def load_smoke_test_module():
     spec = importlib.util.spec_from_file_location("funasr_mcp_smoke_test", SMOKE_TEST)
+    if spec is None or spec.loader is None:
+        raise ImportError(f"Could not load smoke test module from {SMOKE_TEST}")
     module = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(module)
     return module
@@ -135,6 +137,26 @@ def test_validate_responses_rejects_missing_tool(self):
         with self.assertRaisesRegex(ValueError, "transcribe_audio"):
             self.smoke_test.validate_responses(responses)
 
+    def test_validate_responses_rejects_non_object_payload(self):
+        with self.assertRaisesRegex(ValueError, "JSON object"):
+            self.smoke_test.validate_responses([{"id": 1}, ["unexpected"]])
+
+    def test_validate_responses_reports_unexpected_ids_without_sorting(self):
+        responses = [
+            {"jsonrpc": "2.0", "id": 1, "result": {}},
+            {"jsonrpc": "2.0", "id": 2, "result": {"tools": []}},
+            {"jsonrpc": "2.0", "method": "notifications/progress"},
+        ]
+
+        with self.assertRaisesRegex(ValueError, "expected response IDs 1 and 2"):
+            self.smoke_test.validate_responses(responses)
+
+    def test_parse_responses_rejects_non_json_stdout(self):
+        stdout = 'library log on stdout\n{"jsonrpc":"2.0","id":1,"result":{}}\n'
+
+        with self.assertRaisesRegex(ValueError, "non-JSON stdout"):
+            self.smoke_test.parse_responses(stdout)
+
 
 if __name__ == "__main__":
     unittest.main()