test_simple.py•2.96 kB
#!/usr/bin/env python3
"""Test the simplified NIX MCP implementation"""
import asyncio
import json
import logging
from src.nix_mcp.simple_client import SimpleNixClient
from src.nix_mcp.abi_fetcher import ABIFetcher
from src.nix_mcp.abi_resolver import ABIResolver
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def test_list_queries():
"""Test listing available queries"""
print("\n=== Testing list_queries ===")
fetcher = ABIFetcher()
try:
actions = fetcher.get_actions("nix.q")
print(f"Found {len(actions)} queries in nix.q:")
# Show first 10
for action in actions[:10]:
print(f" - {action}")
# Filter for global queries
global_queries = [a for a in actions if "global" in a.lower()]
print(f"\nGlobal-related queries ({len(global_queries)}):")
for q in global_queries:
print(f" - {q}")
except Exception as e:
print(f"Error: {e}")
async def test_get_query_abi():
"""Test getting query ABI and examples"""
print("\n=== Testing get_query_abi ===")
client = SimpleNixClient()
fetcher = ABIFetcher()
try:
# Get ABI for globalconfn
schema = fetcher.get_action_schema("nix.q", "globalconfn")
print(f"Schema for 'globalconfn': {json.dumps(schema, indent=2)}")
# Get example
example = client.generate_json_example("globalconfn", "protobuf::native_indexer.NetworkIdentifier")
print(f"\nExample JSON for 'globalconfn': {json.dumps(example, indent=2)}")
except Exception as e:
print(f"Error: {e}")
async def test_query():
"""Test executing a simple query"""
print("\n=== Testing query execution ===")
client = SimpleNixClient()
try:
# Test globalconfs (no parameters)
print("Executing 'globalconfs' query...")
result = await client.query(
contract="nix.q",
action="globalconfs",
params={},
decode_response=True
)
print(f"Result: {json.dumps(result, indent=2)}")
# Test globalconfn with parameters
print("\nExecuting 'globalconfn' query for ETH testnet...")
result = await client.query(
contract="nix.q",
action="globalconfn",
params={
"blockchain": "ETH",
"network": "testnet"
},
decode_response=True
)
print(f"Result: {json.dumps(result, indent=2)}")
except Exception as e:
print(f"Error: {e}")
async def main():
"""Run all tests"""
print("Testing Simplified NIX MCP Implementation")
print("=" * 50)
await test_list_queries()
await test_get_query_abi()
await test_query()
print("\n" + "=" * 50)
print("Tests completed!")
if __name__ == "__main__":
asyncio.run(main())