mcp-vala/test_converters.py

#!/usr/bin/env python3
"""
Test the converters implemented in compat.vala
"""

import subprocess
import json
import sys
import os

def test_inject_content_length():
    """Test the InjectContentLength converter"""
    print("=== Testing InjectContentLength Converter ===")
    
    # Create a simple test program in Vala that uses the converter
    test_code = """
using GLib;
using Mcp.Core.Compat;

public class TestInjectContentLength {
    public static int main (string[] args) {
        // Create the converter
        var converter = new InjectContentLength ();
        
        // Test input: JSON message without Content-Length header
        string input = "{\\"jsonrpc\\": \\"2.0\\", \\"method\\": \\"ping\\", \\"id\\": 1}\\n";
        
        // Output buffer
        uint8[] output = new uint8[1024];
        size_t bytes_read, bytes_written;
        
        // Convert
        var result = converter.convert (input.data, output, ConverterFlags.NONE, 
                                       out bytes_read, out bytes_written);
        
        // Extract the output
        string output_str = (string)output;
        output_str = output_str.substring (0, (int)bytes_written);
        
        // Print the result
        print ("Input: %s", input);
        print ("Output: %s", output_str);
        
        // Check if Content-Length header was added
        if (output_str.has_prefix ("Content-Length:")) {
            print ("✓ Content-Length header injected successfully\\n");
            return 0;
        } else {
            print ("✗ Content-Length header not found\\n");
            return 1;
        }
    }
}
"""
    
    # Write the test code to a file
    with open("test_inject.vala", "w") as f:
        f.write(test_code)
    
    # Compile and run the test
    try:
        # Compile
        compile_cmd = [
            "valac",
            "--pkg", "glib-2.0",
            "--pkg", "gio-2.0",
            "--pkg", "json-glib-1.0",
            "--pkg", "jsonrpc-glib-1.0",
            "--pkg", "gee-0.8",
            "--pkg", "posix",
            "--vapidir", "builddir/src",
            "--pkg", "mcp-vala",
            "test_inject.vala",
            "-o", "test_inject"
        ]
        
        result = subprocess.run(compile_cmd, capture_output=True, text=True, cwd=".")
        
        if result.returncode != 0:
            print(f"Compilation failed: {result.stderr}")
            print(f"Compilation stdout: {result.stdout}")
            return False
            
        # Run
        run_result = subprocess.run(["./test_inject"], capture_output=True, text=True)
        print(run_result.stdout)
        
        if run_result.returncode == 0:
            return True
        else:
            print(f"Test failed: {run_result.stderr}")
            return False
            
    except Exception as e:
        print(f"Error running test: {e}")
        return False
    finally:
        # Clean up
        for f in ["test_inject.vala", "test_inject"]:
            try:
                os.remove(f)
            except:
                pass

def test_strip_content_length():
    """Test the StripContentLength converter"""
    print("=== Testing StripContentLength Converter ===")
    
    # Create a simple test program in Vala that uses the converter
    test_code = """
using GLib;
using Mcp.Core.Compat;

public class TestStripContentLength {
    public static int main (string[] args) {
        // Create the converter
        var converter = new StripContentLength ();
        
        // Test input: JSON message with Content-Length header
        string input = "Content-Length: 45\\r\\n\\r\\n{\\"jsonrpc\\": \\"2.0\\", \\"method\\": \\"ping\\", \\"id\\": 1}";
        
        // Output buffer
        uint8[] output = new uint8[1024];
        size_t bytes_read, bytes_written;
        
        // Convert
        var result = converter.convert (input.data, output, ConverterFlags.NONE, 
                                       out bytes_read, out bytes_written);
        
        // Extract the output
        string output_str = (string)output;
        output_str = output_str.substring (0, (int)bytes_written);
        
        // Print the result
        print ("Input: %s", input);
        print ("Output: %s", output_str);
        
        // Check if Content-Length header was removed and newline was added
        if (!output_str.has_prefix ("Content-Length:") && output_str.has_suffix ("\\n")) {
            print ("✓ Content-Length header stripped and newline added successfully\\n");
            return 0;
        } else {
            print ("✗ Content-Length header not stripped or newline not added\\n");
            return 1;
        }
    }
}
"""
    
    # Write the test code to a file
    with open("test_strip.vala", "w") as f:
        f.write(test_code)
    
    # Compile and run the test
    try:
        # Compile
        compile_cmd = [
            "valac",
            "--pkg", "glib-2.0",
            "--pkg", "gio-2.0",
            "--pkg", "json-glib-1.0",
            "--pkg", "jsonrpc-glib-1.0",
            "--pkg", "gee-0.8",
            "--pkg", "posix",
            "--vapidir", "builddir/src",
            "--pkg", "mcp-vala",
            "test_strip.vala",
            "-o", "test_strip"
        ]
        
        result = subprocess.run(compile_cmd, capture_output=True, text=True, cwd=".")
        
        if result.returncode != 0:
            print(f"Compilation failed: {result.stderr}")
            return False
            
        # Run
        run_result = subprocess.run(["./test_strip"], capture_output=True, text=True)
        print(run_result.stdout)
        
        if run_result.returncode == 0:
            return True
        else:
            print(f"Test failed: {run_result.stderr}")
            return False
            
    except Exception as e:
        print(f"Error running test: {e}")
        return False
    finally:
        # Clean up
        for f in ["test_strip.vala", "test_strip"]:
            try:
                os.remove(f)
            except:
                pass

def main():
    print("Testing MCP Converters\n")
    
    # Test both converters
    inject_success = test_inject_content_length()
    strip_success = test_strip_content_length()
    
    if inject_success and strip_success:
        print("✓ All converter tests passed!")
        return 0
    else:
        print("✗ Some converter tests failed!")
        return 1

if __name__ == "__main__":
    sys.exit(main())