File size: 3,936 Bytes
c922f8b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
"""
Test Function Fix Module
This module provides a fix for the test_memory_implementation function signature mismatch
in the Supabase memory system. It creates a compatibility layer so that calls with or
without the client parameter will work correctly.
"""
import logging
from typing import Any, Dict, Optional, Union, Callable
logger = logging.getLogger("test_function_fix")
def fix_test_memory_implementation():
"""
Fix the test_memory_implementation function signature mismatch by creating
a compatibility wrapper that accepts both calling conventions.
"""
# Import here to avoid circular imports
from src.gaia.memory.supabase_memory_consolidated import (
test_memory_implementation as original_function,
SupabaseMemory,
Client
)
# Get the module where the function is defined
import inspect
module = inspect.getmodule(original_function)
# Create a wrapper function that can handle both signatures
def test_memory_implementation_wrapper(*args, **kwargs):
"""
Compatibility wrapper for test_memory_implementation.
Handles both:
- test_memory_implementation(client, table="gaia_memory")
- test_memory_implementation(memory_or_config=None)
"""
# Check if the first positional arg is a Client
if args and isinstance(args[0], Client):
# The old signature with client parameter
logger.info("Using compatibility mode for test_memory_implementation with client parameter")
# We need to call the implementation that matches this signature
# Since the 'client' parameter isn't supported in the consolidated version,
# we'll need to adapt it
client = args[0]
table = kwargs.get('table', 'gaia_memory')
# Create memory config from client
memory_config = {
"url": client.supabase_url,
"key": client.rest_url.split("?")[1].split("&")[0].split("=")[1],
"table_name": table
}
# Call the function with memory_config
return original_function(memory_config)
elif 'client' in kwargs:
# The old signature with client as keyword arg
logger.info("Using compatibility mode for test_memory_implementation with client keyword parameter")
client = kwargs.pop('client')
table = kwargs.get('table', 'gaia_memory')
# Create memory config from client
memory_config = {
"url": client.supabase_url,
"key": client.rest_url.split("?")[1].split("&")[0].split("=")[1],
"table_name": table
}
# Call the function with memory_config
return original_function(memory_config)
else:
# The new signature without client
return original_function(*args, **kwargs)
# Copy attributes from the original function to the wrapper
for attr in dir(original_function):
if not attr.startswith('__'):
setattr(test_memory_implementation_wrapper, attr, getattr(original_function, attr))
# Replace the original function with our wrapper
setattr(module, 'test_memory_implementation', test_memory_implementation_wrapper)
logger.info("Fixed test_memory_implementation function signature mismatch")
return True
def apply_fixes():
"""Apply all test function fixes."""
fix_test_memory_implementation()
logger.info("All test function fixes applied successfully")
if __name__ == "__main__":
# Configure logging for standalone use
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
# Apply fixes
apply_fixes() |