import subprocess
import json
import tempfile
import os
def test_partial_json_recovery():
print("đ§ Testing Partial JSON Recovery")
print("=" * 40)
partial_json_tests = [
{
"name": "Valid array with invalid object",
"content": '''[
{"id": 1, "name": "Alice", "valid": true},
{"id": 2, "name": "Bob", "valid": true},
{id: 3, name: "Charlie", valid: false}
]''',
"description": "Array with mix of valid and invalid objects"
},
{
"name": "Valid object with invalid nested object",
"content": '''{
"users": [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"}
],
"metadata": {
"total": 2,
invalid_key: "should fail"
}
}''',
"description": "Object with invalid nested structure"
},
{
"name": "Truncated JSON (missing end)",
"content": '''{
"data": {
"users": [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"}
],
"total": 2''',
"description": "JSON cut off mid-structure"
}
]
print("Testing server's ability to handle partial corruption...\n")
results = []
for test_case in partial_json_tests:
print(f"đ {test_case['name']}")
print(f" {test_case['description']}")
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
f.write(test_case['content'])
test_file = f.name
try:
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "json-read",
"arguments": {
"file_path": test_file.replace('\\', '/'),
"format": "compact"
}
}
}
result = subprocess.run(
['cargo', 'run', '--quiet', '--'],
input=json.dumps(request) + '\n',
capture_output=True,
text=True,
timeout=10
)
if result.returncode == 0:
for line in result.stdout.strip().split('\n'):
if line.strip().startswith('{"jsonrpc"'):
response = json.loads(line.strip())
if 'result' in response:
content = response['result']['content'][0]['text']
print(" đ Server processed without fatal error")
print(f" đ Response preview: {content[:100]}...")
if 'Alice' in content or 'Bob' in content:
print(" â
Successfully extracted some valid data")
results.append(True)
else:
print(" â ī¸ No recognizable data extracted")
results.append(False)
elif 'error' in response:
error_msg = response['error']['message']
print(" â Server returned error (no partial recovery)")
print(f" đ Error: {error_msg}")
results.append(False)
break
else:
print(" â Server process failed")
results.append(False)
finally:
try:
os.unlink(test_file)
except:
pass
print()
print("=" * 50)
print("đ Partial Recovery Test Results")
print("=" * 50)
recovery_count = sum(results)
total_tests = len(results)
print(f"Partial recovery success: {recovery_count}/{total_tests} tests")
if recovery_count > 0:
print("â
Server demonstrates some partial recovery capability")
else:
print("â Server has strict JSON validation (no partial recovery)")
print("\nđĄ Current Behavior Analysis:")
print(" The JSON MCP server uses strict JSON parsing which:")
print(" â
Provides clear error messages for malformed JSON")
print(" â
Ensures data integrity and consistency")
print(" â
Prevents silent data corruption")
print(" âšī¸ Does not attempt partial recovery (by design)")
print("\nđ¯ Recommendation:")
print(" Current strict validation is appropriate for production use")
print(" Partial recovery would add complexity and potential data corruption")
print(" Clear error messages help LLMs understand and fix JSON issues")
return recovery_count, total_tests
if __name__ == "__main__":
test_partial_json_recovery()