test_expected_flow.py•5.95 kB
#!/usr/bin/env python3
"""Test the expected flow from doc/expected_flow.md"""
import asyncio
import json
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.abi_fetcher import ABIFetcher
async def test_expected_flow():
"""Test the complete flow as described in expected_flow.md"""
print("=" * 60)
print("Testing Expected Flow for Raw Transaction Query")
print("=" * 60)
# Use cdev environment as specified
client = SimpleNixClient(environment="cdev")
fetcher = ABIFetcher(environment="cdev")
print(f"\nUsing cdev environment:")
print(f" Nodeos: {client.nodeos_api}")
print(f" Rodeos: {client.rodeos_api}")
# Step 1: List queries to find appropriate one
print("\n1. Finding appropriate query for raw transactions...")
try:
actions = fetcher.get_actions("nix.q")
# Filter for raw transaction queries
raw_queries = [a for a in actions if "raw" in a.lower() and "trx" in a.lower()]
print(f"Found {len(raw_queries)} raw transaction queries:")
for q in raw_queries:
print(f" - {q}")
# Should find rawtrxspb
if "rawtrxspb" in actions:
print("\n✓ Found 'rawtrxspb' query as expected")
else:
print("\n⚠ 'rawtrxspb' not found, available queries with 'raw':")
for a in actions:
if 'raw' in a.lower():
print(f" - {a}")
except Exception as e:
print(f"Error listing queries: {e}")
return
# Step 2: Get ABI for rawtrxspb
print("\n2. Getting ABI for rawtrxspb query...")
try:
schema = fetcher.get_action_schema("nix.q", "rawtrxspb")
if schema:
print("✓ Retrieved ABI schema successfully")
print(f" Input type: {schema.get('type', 'unknown')}")
# Show example structure
example = schema.get('example', {})
if example:
print(f" Example structure:")
print(json.dumps(example, indent=4)[:500] + "...")
else:
print("⚠ Could not get schema for rawtrxspb")
except Exception as e:
print(f"Error getting ABI: {e}")
# Step 3: Execute the query
print("\n3. Executing rawtrxspb query for ETH mainnet transaction...")
# Transaction from the example
tx_hash = "0abe152ba84b35026451d68a55310ec58450a167a82a55fae2ff691ebc7236bf"
# Method 1: Using inline JSON
print("\nMethod 1: Using inline JSON parameters")
params = {
"network": {
"blockchain": "ETH",
"network": "mainnet"
},
"transaction_identifier": {
"hash": tx_hash
}
}
try:
print(f"Query params: {json.dumps(params, indent=2)}")
result = await client.query(
contract="nix.q",
action="rawtrxspb",
params=params,
decode_response=True
)
print("\n✓ Query executed successfully!")
# Show result (truncated if too long)
if isinstance(result, dict):
result_str = json.dumps(result, indent=2)
if len(result_str) > 1000:
print(f"Result (truncated):\n{result_str[:1000]}...")
else:
print(f"Result:\n{result_str}")
# Check if we got expected fields
if "hex_response" in result:
print(f"\nReceived hex response of length: {len(result['hex_response'])}")
elif "text_response" in result:
print(f"\nReceived text response of length: {len(result['text_response'])}")
else:
print(f"Result: {result}")
except Exception as e:
print(f"Error executing query: {e}")
# Method 2: Using JSON file
print("\n\nMethod 2: Using JSON file parameter")
json_file = "examples/raw_transaction_query.json"
if Path(json_file).exists():
try:
print(f"Using JSON file: {json_file}")
result = await client.query(
contract="nix.q",
action="rawtrxspb",
params=json_file,
decode_response=True
)
print("✓ Query with JSON file executed successfully!")
# Just show result type/size
if isinstance(result, dict):
if "hex_response" in result:
print(f"Received hex response of length: {len(result['hex_response'])}")
elif "text_response" in result:
print(f"Received text response of length: {len(result['text_response'])}")
else:
print(f"Result type: {type(result)}")
except Exception as e:
print(f"Error with JSON file query: {e}")
else:
print(f"JSON file {json_file} not found")
print("\n" + "=" * 60)
print("Expected Flow Test Completed")
print("=" * 60)
async def check_abi_cache():
"""Check if ABI cache was initialized"""
print("\n4. Checking ABI cache initialization...")
cache_dir = Path(".abi_cache")
if cache_dir.exists():
cache_files = list(cache_dir.glob("*.json"))
if cache_files:
print(f"✓ ABI cache directory exists with {len(cache_files)} cached ABIs:")
for f in cache_files:
print(f" - {f.name}")
else:
print("⚠ ABI cache directory exists but is empty")
else:
print("⚠ ABI cache directory not found")
if __name__ == "__main__":
print("Testing NIX MCP Expected Flow")
print("==============================\n")
asyncio.run(test_expected_flow())
asyncio.run(check_abi_cache())