test_env_switching.py•7.42 kB
#!/usr/bin/env python3
"""Test environment switching functionality"""
import asyncio
import json
import logging
import sys
from pathlib import Path
# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))
from src.nix_mcp.simple_client import SimpleNixClient
from src.nix_mcp.env_config import EnvironmentConfig
from src.nix_mcp.abi_fetcher import ABIFetcher
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def test_environment_info():
"""Test listing available environments"""
print("\n=== Available Environments ===")
envs = EnvironmentConfig.list_environments()
for env_name, env_info in envs.items():
print(f"\n{env_name}:")
print(f" Nodeos: {env_info['nodeos']}")
print(f" Rodeos: {env_info['rodeos']}")
print(f" Has Signature: {env_info.get('signature_provider') is not None}")
async def test_dynamic_endpoints():
"""Test getting endpoints for different environments"""
print("\n=== Testing Dynamic Endpoints ===")
config = EnvironmentConfig()
# Test different environments
test_envs = ["dev", "uat", "prod", "local"]
for env in test_envs:
try:
nodeos, rodeos = config.get_endpoints(env)
print(f"\n{env.upper()} Environment:")
print(f" Nodeos: {nodeos}")
print(f" Rodeos: {rodeos}")
except Exception as e:
print(f" Error: {e}")
async def test_query_different_envs():
"""Test executing queries on different environments"""
print("\n=== Testing Queries on Different Environments ===")
# Query to test - get global configurations
test_action = "globalconfs"
test_params = {}
# Test environments (be careful with prod!)
test_envs = ["dev", "uat"] # Not including prod for safety
for env in test_envs:
print(f"\n--- Testing {env.upper()} environment ---")
try:
# Create client for specific environment
client = SimpleNixClient(environment=env)
print(f"Using endpoints - Nodeos: {client.nodeos_api}")
print(f" Rodeos: {client.rodeos_api}")
# Execute query
result = await client.query(
contract="nix.q",
action=test_action,
params=test_params,
decode_response=True
)
print(f"Query successful! Result type: {type(result)}")
# Show first part of result
if isinstance(result, dict):
# If it's a decoded response, show some info
if "text_response" in result:
text = result["text_response"]
print(f"Response length: {len(text)} chars")
print(f"First 200 chars: {text[:200]}...")
elif "hex_response" in result:
print(f"Hex response length: {len(result['hex_response'])} chars")
else:
# Show the JSON result (truncated if too long)
result_str = json.dumps(result, indent=2)
if len(result_str) > 500:
print(f"Result (truncated):\n{result_str[:500]}...")
else:
print(f"Result:\n{result_str}")
else:
print(f"Result: {result}")
except Exception as e:
print(f"Error querying {env}: {e}")
async def test_specific_network_query():
"""Test querying specific network configuration (e.g., SOL mainnet on prod)"""
print("\n=== Testing Specific Network Query ===")
# Example: Get SOL mainnet configuration from different environments
test_query = "globalconfn"
test_params = {
"blockchain": "SOL",
"network": "mainnet"
}
# Try different environments to see where SOL mainnet config exists
for env in ["dev", "uat", "prod"]:
print(f"\n--- Checking {env.upper()} for SOL mainnet config ---")
try:
client = SimpleNixClient(environment=env)
# First, let's check if the query exists
fetcher = ABIFetcher(nodeos_api=client.nodeos_api, environment=env)
actions = fetcher.get_actions("nix.q")
if test_query not in actions:
print(f"Query '{test_query}' not found in {env}")
continue
# Execute the query
result = await client.query(
contract="nix.q",
action=test_query,
params=test_params,
decode_response=True
)
if isinstance(result, dict):
if "text_response" in result:
print(f"Found SOL mainnet config in {env}!")
print(f"Response: {result['text_response'][:300]}...")
elif "status" in result and result["status"] == "success":
print(f"Query executed but no data returned")
else:
print(f"Result: {json.dumps(result, indent=2)[:300]}...")
else:
print(f"Result: {result}")
except Exception as e:
print(f"Error: {e}")
async def test_client_creation_methods():
"""Test different ways to create clients"""
print("\n=== Testing Client Creation Methods ===")
# Method 1: Environment name
print("\n1. Using environment name:")
client1 = SimpleNixClient(environment="uat")
print(f" Nodeos: {client1.nodeos_api}")
print(f" Rodeos: {client1.rodeos_api}")
# Method 2: Explicit endpoints (overrides environment)
print("\n2. Using explicit endpoints:")
client2 = SimpleNixClient(
nodeos_api="http://custom-nodeos:8888",
rodeos_api="http://custom-rodeos:8880"
)
print(f" Nodeos: {client2.nodeos_api}")
print(f" Rodeos: {client2.rodeos_api}")
# Method 3: Environment with override
print("\n3. Using environment with partial override:")
client3 = SimpleNixClient(
environment="dev",
rodeos_api="http://override-rodeos:8880"
)
print(f" Nodeos: {client3.nodeos_api}")
print(f" Rodeos: {client3.rodeos_api}")
# Method 4: Default (uses NODEOS_ENV or 'dev')
print("\n4. Using defaults:")
client4 = SimpleNixClient()
print(f" Nodeos: {client4.nodeos_api}")
print(f" Rodeos: {client4.rodeos_api}")
async def main():
"""Run all tests"""
print("=" * 60)
print("NIX MCP Environment Switching Test Suite")
print("=" * 60)
await test_environment_info()
await test_dynamic_endpoints()
await test_client_creation_methods()
# Be careful with these tests as they make actual queries
print("\n" + "=" * 60)
print("WARNING: The following tests will make actual blockchain queries")
print("=" * 60)
response = input("\nRun query tests? (y/n): ")
if response.lower() == 'y':
await test_query_different_envs()
await test_specific_network_query()
else:
print("Skipping query tests")
print("\n" + "=" * 60)
print("Test suite completed!")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(main())