307 lines
12 KiB
Python
307 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
JSON Completion Test 13 - Tests JSON completion at various cut positions
|
|
Tests a single JSON object (~300 chars) with all JSON structure types.
|
|
Cuts the JSON at every position from character 50 to the end, completes it, and validates.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
import os
|
|
from typing import Dict, Any, List
|
|
|
|
# Add the gateway to path (go up 2 levels from tests/functional/)
|
|
_gateway_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
|
|
if _gateway_path not in sys.path:
|
|
sys.path.insert(0, _gateway_path)
|
|
|
|
# Import JSON continuation module
|
|
from modules.shared.jsonContinuation import getContexts
|
|
|
|
|
|
class JsonCompletionTester13:
|
|
def __init__(self):
|
|
self.testResults = {}
|
|
self.logBuffer = []
|
|
self.logFile = None
|
|
|
|
def createTestJson(self) -> str:
|
|
"""
|
|
Create a single JSON object (~300 chars) containing all JSON structure types:
|
|
- Objects (nested)
|
|
- Arrays (nested)
|
|
- Strings
|
|
- Numbers (integers and floats)
|
|
- Booleans (true, false)
|
|
- null
|
|
"""
|
|
testData = {
|
|
"id": 12345,
|
|
"name": "Test Object",
|
|
"active": True,
|
|
"inactive": False,
|
|
"value": None,
|
|
"price": 99.99,
|
|
"tags": ["tag1", "tag2", "tag3"],
|
|
"metadata": {
|
|
"created": "2025-01-01",
|
|
"updated": "2025-01-02",
|
|
"version": 1
|
|
},
|
|
"items": [
|
|
{"id": 1, "name": "Item A", "count": 10},
|
|
{"id": 2, "name": "Item B", "count": 20}
|
|
],
|
|
"settings": {
|
|
"theme": "dark",
|
|
"notifications": True,
|
|
"features": ["feature1", "feature2"]
|
|
}
|
|
}
|
|
|
|
jsonString = json.dumps(testData, indent=2, ensure_ascii=False)
|
|
|
|
# Ensure it's approximately 300 characters (adjust if needed)
|
|
targetLength = 300
|
|
if len(jsonString) < targetLength:
|
|
# Add padding to metadata
|
|
testData["metadata"]["description"] = "A" * (targetLength - len(jsonString) + 20)
|
|
jsonString = json.dumps(testData, indent=2, ensure_ascii=False)
|
|
|
|
# Trim to approximately 300 chars if too long
|
|
if len(jsonString) > targetLength + 50:
|
|
# Remove some content to get closer to target
|
|
testData["metadata"].pop("description", None)
|
|
jsonString = json.dumps(testData, indent=2, ensure_ascii=False)
|
|
|
|
return jsonString
|
|
|
|
def _log(self, message: str):
|
|
"""Add message to log buffer."""
|
|
self.logBuffer.append(message)
|
|
print(message)
|
|
|
|
async def testJsonCompletionAtCuts(self, jsonString: str, startPos: int = 50, step: int = 5) -> Dict[str, Any]:
|
|
"""
|
|
Test JSON completion at various cut positions.
|
|
|
|
Args:
|
|
jsonString: The full JSON string to test
|
|
startPos: Starting position for cuts (default 50)
|
|
step: Step size between cuts (default 5)
|
|
|
|
Returns:
|
|
Dictionary with test results for each cut position
|
|
"""
|
|
jsonLength = len(jsonString)
|
|
results = {}
|
|
|
|
self._log("")
|
|
self._log("="*80)
|
|
self._log("TESTING JSON COMPLETION AT VARIOUS CUT POSITIONS")
|
|
self._log("="*80)
|
|
self._log(f"JSON length: {jsonLength} characters")
|
|
self._log(f"Testing cuts from position {startPos} to {jsonLength} (step: {step})")
|
|
self._log("")
|
|
|
|
# Test at each cut position
|
|
cutPositions = list(range(startPos, jsonLength, step))
|
|
# Always include the last position
|
|
if cutPositions[-1] != jsonLength - 1:
|
|
cutPositions.append(jsonLength - 1)
|
|
|
|
successCount = 0
|
|
totalCuts = len(cutPositions)
|
|
|
|
for cutPos in cutPositions:
|
|
# Get truncated JSON
|
|
truncatedJson = jsonString[:cutPos]
|
|
|
|
# Generate contexts
|
|
try:
|
|
contexts = getContexts(truncatedJson)
|
|
completePart = contexts.completePart
|
|
overlapContext = contexts.overlapContext
|
|
|
|
# Test if completePart is valid JSON
|
|
isValidJson = False
|
|
jsonError = None
|
|
parsedData = None
|
|
|
|
try:
|
|
parsedData = json.loads(completePart)
|
|
isValidJson = True
|
|
except json.JSONDecodeError as e:
|
|
jsonError = str(e)
|
|
isValidJson = False
|
|
|
|
# Store result
|
|
result = {
|
|
"cutPosition": cutPos,
|
|
"truncatedLength": len(truncatedJson),
|
|
"completePartLength": len(completePart),
|
|
"overlapContextLength": len(overlapContext),
|
|
"isValidJson": isValidJson,
|
|
"jsonError": jsonError,
|
|
"truncatedJson": truncatedJson[-50:] if len(truncatedJson) > 50 else truncatedJson, # Last 50 chars
|
|
"completePart": completePart[-100:] if len(completePart) > 100 else completePart, # Last 100 chars
|
|
"overlapContext": overlapContext[-100:] if len(overlapContext) > 100 else overlapContext # Last 100 chars
|
|
}
|
|
|
|
results[cutPos] = result
|
|
|
|
if isValidJson:
|
|
successCount += 1
|
|
self._log(f"✅ Cut at position {cutPos:4d}: Valid JSON (completePart length: {len(completePart)}, overlap length: {len(overlapContext)})")
|
|
self._log(f" Overlap: {overlapContext[-80:] if len(overlapContext) > 80 else overlapContext}")
|
|
else:
|
|
self._log(f"❌ Cut at position {cutPos:4d}: Invalid JSON - {jsonError}")
|
|
self._log(f" Truncated (last 50): {truncatedJson[-50:]}")
|
|
self._log(f" CompletePart (last 100): {completePart[-100:]}")
|
|
self._log(f" Overlap: {overlapContext[-80:] if len(overlapContext) > 80 else overlapContext}")
|
|
|
|
except Exception as e:
|
|
result = {
|
|
"cutPosition": cutPos,
|
|
"truncatedLength": len(truncatedJson),
|
|
"isValidJson": False,
|
|
"jsonError": f"Exception: {str(e)}",
|
|
"truncatedJson": truncatedJson[-50:] if len(truncatedJson) > 50 else truncatedJson
|
|
}
|
|
results[cutPos] = result
|
|
self._log(f"❌ Cut at position {cutPos:4d}: Exception - {str(e)}")
|
|
|
|
# Summary
|
|
self._log("")
|
|
self._log("="*80)
|
|
self._log("CUT TEST SUMMARY")
|
|
self._log("="*80)
|
|
self._log(f"Total cuts tested: {totalCuts}")
|
|
self._log(f"Successful completions: {successCount}")
|
|
self._log(f"Failed completions: {totalCuts - successCount}")
|
|
self._log(f"Success rate: {successCount/totalCuts*100:.1f}%")
|
|
self._log("")
|
|
|
|
# Detailed results for failed cuts
|
|
failedCuts = [pos for pos, res in results.items() if not res.get("isValidJson", False)]
|
|
if failedCuts:
|
|
self._log("Failed cuts:")
|
|
for pos in failedCuts[:10]: # Show first 10 failures
|
|
res = results[pos]
|
|
self._log(f" Position {pos}: {res.get('jsonError', 'Unknown error')}")
|
|
overlap = res.get('overlapContext', 'N/A')
|
|
if overlap != 'N/A':
|
|
self._log(f" Overlap: {overlap[-80:] if len(overlap) > 80 else overlap}")
|
|
if len(failedCuts) > 10:
|
|
self._log(f" ... ({len(failedCuts) - 10} more failures)")
|
|
|
|
return {
|
|
"totalCuts": totalCuts,
|
|
"successCount": successCount,
|
|
"failedCount": totalCuts - successCount,
|
|
"successRate": successCount / totalCuts * 100 if totalCuts > 0 else 0,
|
|
"results": results,
|
|
"failedCuts": failedCuts
|
|
}
|
|
|
|
def _writeLogFile(self):
|
|
"""Write log buffer to file."""
|
|
logDir = os.path.join(os.path.dirname(__file__), "..", "..", "..", "local", "debug")
|
|
os.makedirs(logDir, exist_ok=True)
|
|
logFilePath = os.path.join(logDir, "test13_json_completion_cuts_results.txt")
|
|
|
|
with open(logFilePath, 'w', encoding='utf-8') as f:
|
|
f.write('\n'.join(self.logBuffer))
|
|
|
|
self.logFile = logFilePath
|
|
print(f"\n📝 Detailed log written to: {logFilePath}")
|
|
|
|
async def runTest(self):
|
|
"""Run the complete test."""
|
|
self._log("="*80)
|
|
self._log("JSON COMPLETION TEST 13")
|
|
self._log("="*80)
|
|
|
|
try:
|
|
# Create test JSON
|
|
jsonString = self.createTestJson()
|
|
|
|
self._log("")
|
|
self._log("="*80)
|
|
self._log("TEST JSON OBJECT")
|
|
self._log("="*80)
|
|
self._log(f"Length: {len(jsonString)} characters")
|
|
self._log("")
|
|
self._log("Full JSON content:")
|
|
self._log("-"*80)
|
|
jsonLines = jsonString.split('\n')
|
|
for line in jsonLines:
|
|
self._log(line)
|
|
|
|
# Test completion at various cuts
|
|
results = await self.testJsonCompletionAtCuts(jsonString, startPos=50, step=5)
|
|
|
|
# Write log file
|
|
self._writeLogFile()
|
|
|
|
# Final summary
|
|
self._log("")
|
|
self._log("="*80)
|
|
self._log("FINAL TEST SUMMARY")
|
|
self._log("="*80)
|
|
self._log(f"Total cuts tested: {results['totalCuts']}")
|
|
self._log(f"✅ Successful: {results['successCount']}")
|
|
self._log(f"❌ Failed: {results['failedCount']}")
|
|
self._log(f"Success rate: {results['successRate']:.1f}%")
|
|
|
|
if results['failedCuts']:
|
|
self._log("")
|
|
self._log("Failed cut positions:")
|
|
for pos in results['failedCuts']:
|
|
res = results['results'][pos]
|
|
self._log(f" Position {pos}: {res.get('jsonError', 'Unknown error')}")
|
|
overlap = res.get('overlapContext', 'N/A')
|
|
if overlap != 'N/A':
|
|
self._log(f" Overlap: {overlap[-80:] if len(overlap) > 80 else overlap}")
|
|
|
|
self.testResults = {
|
|
"success": results['successCount'] == results['totalCuts'],
|
|
"totalCuts": results['totalCuts'],
|
|
"successCount": results['successCount'],
|
|
"failedCount": results['failedCount'],
|
|
"successRate": results['successRate'],
|
|
"failedCuts": results['failedCuts'],
|
|
"results": results['results']
|
|
}
|
|
|
|
return self.testResults
|
|
|
|
except Exception as e:
|
|
import traceback
|
|
print(f"\n❌ Test failed with error: {type(e).__name__}: {str(e)}")
|
|
print(f"Traceback:\n{traceback.format_exc()}")
|
|
self.testResults = {
|
|
"success": False,
|
|
"error": str(e),
|
|
"traceback": traceback.format_exc()
|
|
}
|
|
return self.testResults
|
|
|
|
|
|
async def main():
|
|
"""Run JSON completion test 13."""
|
|
tester = JsonCompletionTester13()
|
|
results = await tester.runTest()
|
|
|
|
# Print final results as JSON for easy parsing
|
|
print("\n" + "="*80)
|
|
print("FINAL RESULTS (JSON)")
|
|
print("="*80)
|
|
print(json.dumps(results, indent=2, default=str))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|