|
""" |
|
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. |
|
""" |
|
|
|
from src.gaia.memory.supabase_memory_consolidated import ( |
|
test_memory_implementation as original_function, |
|
SupabaseMemory, |
|
Client |
|
) |
|
|
|
|
|
import inspect |
|
module = inspect.getmodule(original_function) |
|
|
|
|
|
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) |
|
""" |
|
|
|
if args and isinstance(args[0], Client): |
|
|
|
logger.info("Using compatibility mode for test_memory_implementation with client parameter") |
|
|
|
|
|
|
|
|
|
client = args[0] |
|
table = kwargs.get('table', 'gaia_memory') |
|
|
|
|
|
memory_config = { |
|
"url": client.supabase_url, |
|
"key": client.rest_url.split("?")[1].split("&")[0].split("=")[1], |
|
"table_name": table |
|
} |
|
|
|
|
|
return original_function(memory_config) |
|
|
|
elif 'client' in kwargs: |
|
|
|
logger.info("Using compatibility mode for test_memory_implementation with client keyword parameter") |
|
client = kwargs.pop('client') |
|
table = kwargs.get('table', 'gaia_memory') |
|
|
|
|
|
memory_config = { |
|
"url": client.supabase_url, |
|
"key": client.rest_url.split("?")[1].split("&")[0].split("=")[1], |
|
"table_name": table |
|
} |
|
|
|
|
|
return original_function(memory_config) |
|
else: |
|
|
|
return original_function(*args, **kwargs) |
|
|
|
|
|
for attr in dir(original_function): |
|
if not attr.startswith('__'): |
|
setattr(test_memory_implementation_wrapper, attr, getattr(original_function, attr)) |
|
|
|
|
|
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__": |
|
|
|
logging.basicConfig( |
|
level=logging.INFO, |
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" |
|
) |
|
|
|
|
|
apply_fixes() |