"""Tests for distribution tools with real blockchain data.
These tests validate that distribution tools return correct data when called through
the MCP interface. NO MOCK DATA - validates actual tool behavior.
Distribution tools are critical for:
- Staking rewards tracking
- Validator commission monitoring
- Delegation management
"""
import pytest
import sys
from pathlib import Path
# Add src to path
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
from mcp_server.tools.distribution_tools import (
get_distribution_params,
get_validator_outstanding_rewards,
get_validator_commission,
get_validator_slashes,
get_delegation_rewards,
get_delegation_total_rewards,
get_delegator_validators,
get_delegator_withdraw_address,
get_community_pool,
)
@pytest.mark.asyncio
@pytest.mark.tools
@pytest.mark.online
class TestDistributionToolsOnline:
"""Test distribution tools with live network connection.
These tests validate critical staking and rewards functionality.
"""
async def test_get_distribution_params_returns_data(self):
"""Test that get_distribution_params returns module parameters.
VALIDATES: Can we query distribution module configuration?
IMPACT: Needed for understanding staking reward mechanics
"""
result = await get_distribution_params()
assert isinstance(result, dict), "Result must be dictionary"
# Should have params or success indicator
if "error" not in result:
print(f"✅ Can query distribution params")
# May have params field or be structured differently
if "params" in result:
params = result["params"]
print(f" Params available: {list(params.keys())[:5]}")
else:
# Some endpoints may not be available
print(f"⚠️ Distribution params: {result.get('error')}")
async def test_get_community_pool_returns_balance(self):
"""Test that community pool balance can be queried.
VALIDATES: Can we see community pool funds?
IMPACT: Needed for governance and treasury analysis
"""
result = await get_community_pool()
assert isinstance(result, dict), "Result must be dictionary"
if "error" not in result:
print(f"✅ Can query community pool")
if "pool" in result:
pool = result["pool"]
print(f" Pool has {len(pool)} token types")
else:
print(f"⚠️ Community pool query: {result.get('error')}")
async def test_get_validator_outstanding_rewards_structure(self):
"""Test validator outstanding rewards query structure.
VALIDATES: Can we query validator rewards?
IMPACT: Needed for validator operators to track earnings
"""
# Use a known validator address (this is a Regen Network validator)
# Note: May not exist or may have been removed
test_validator = "regenvaloper1xhhquwctvwm40qn3nmmqq067pc2gw22ezy9n2h"
result = await get_validator_outstanding_rewards(test_validator)
assert isinstance(result, dict), "Result must be dictionary"
# Validator may or may not exist, both are valid responses
if "error" in result:
# Expected - validator may not exist or be active
print(f"⚠️ Validator rewards: {result.get('error')}")
else:
# If validator exists, should have rewards structure
print(f"✅ Can query validator outstanding rewards")
if "rewards" in result:
print(f" Rewards structure available")
async def test_get_validator_commission_structure(self):
"""Test validator commission query structure.
VALIDATES: Can we query validator commission?
IMPACT: Needed for validator operators to track commission
"""
test_validator = "regenvaloper1xhhquwctvwm40qn3nmmqq067pc2gw22ezy9n2h"
result = await get_validator_commission(test_validator)
assert isinstance(result, dict), "Result must be dictionary"
if "error" in result:
print(f"⚠️ Validator commission: {result.get('error')}")
else:
print(f"✅ Can query validator commission")
if "commission" in result:
print(f" Commission structure available")
async def test_get_validator_slashes_with_pagination(self):
"""Test validator slashes query with pagination.
VALIDATES: Can we query slashing events?
IMPACT: Needed for validator risk assessment
"""
test_validator = "regenvaloper1xhhquwctvwm40qn3nmmqq067pc2gw22ezy9n2h"
result = await get_validator_slashes(
test_validator,
page=1,
limit=10
)
assert isinstance(result, dict), "Result must be dictionary"
# Validator may not have slashes (good!) or may not exist
if "error" in result:
print(f"⚠️ Validator slashes: {result.get('error')}")
else:
print(f"✅ Can query validator slashes")
if "slashes" in result:
slashes = result["slashes"]
print(f" Slashes found: {len(slashes)}")
async def test_get_delegation_total_rewards_for_address(self):
"""Test total delegation rewards for an address.
VALIDATES: Can stakers see their total rewards?
IMPACT: CRITICAL for stakers tracking earnings
"""
# Use a known address (this may or may not be a delegator)
test_address = "regen1xhhquwctvwm40qn3nmmqq067pc2gw22e6q3p3c"
result = await get_delegation_total_rewards(test_address)
assert isinstance(result, dict), "Result must be dictionary"
# Address may or may not be delegating
if "error" in result:
print(f"⚠️ Total delegation rewards: {result.get('error')}")
else:
print(f"✅ Can query total delegation rewards")
if "rewards" in result or "total" in result:
print(f" Rewards structure available")
async def test_get_delegator_validators_for_address(self):
"""Test getting validators for a delegator.
VALIDATES: Can delegators see which validators they're bonded to?
IMPACT: Needed for delegation portfolio management
"""
test_address = "regen1xhhquwctvwm40qn3nmmqq067pc2gw22e6q3p3c"
result = await get_delegator_validators(test_address)
assert isinstance(result, dict), "Result must be dictionary"
if "error" in result:
print(f"⚠️ Delegator validators: {result.get('error')}")
else:
print(f"✅ Can query delegator validators")
if "validators" in result:
validators = result["validators"]
print(f" Bonded to {len(validators)} validators")
async def test_get_delegator_withdraw_address_structure(self):
"""Test getting withdraw address for delegator.
VALIDATES: Can we query where rewards are sent?
IMPACT: Needed for reward management
"""
test_address = "regen1xhhquwctvwm40qn3nmmqq067pc2gw22e6q3p3c"
result = await get_delegator_withdraw_address(test_address)
assert isinstance(result, dict), "Result must be dictionary"
if "error" in result:
print(f"⚠️ Withdraw address: {result.get('error')}")
else:
print(f"✅ Can query withdraw address")
if "withdraw_address" in result:
print(f" Withdraw address: {result['withdraw_address']}")
async def test_get_delegation_rewards_for_pair(self):
"""Test getting rewards for specific delegator-validator pair.
VALIDATES: Can we get rewards for specific delegation?
IMPACT: Needed for detailed reward tracking
"""
test_delegator = "regen1xhhquwctvwm40qn3nmmqq067pc2gw22e6q3p3c"
test_validator = "regenvaloper1xhhquwctvwm40qn3nmmqq067pc2gw22ezy9n2h"
result = await get_delegation_rewards(test_delegator, test_validator)
assert isinstance(result, dict), "Result must be dictionary"
# This specific pair may not exist
if "error" in result:
print(f"⚠️ Delegation rewards for pair: {result.get('error')}")
else:
print(f"✅ Can query delegation rewards for pair")
if "rewards" in result:
print(f" Rewards structure available")
@pytest.mark.asyncio
@pytest.mark.tools
class TestDistributionToolsValidation:
"""Test distribution tools parameter validation and error handling."""
async def test_validator_address_validation_outstanding_rewards(self):
"""Test that invalid validator addresses are rejected."""
# Invalid address format
result = await get_validator_outstanding_rewards("invalid_address")
assert isinstance(result, dict), "Should return dict even on error"
assert "error" in result, "Should indicate error for invalid address"
async def test_validator_address_validation_commission(self):
"""Test that invalid validator addresses are rejected."""
result = await get_validator_commission("")
assert isinstance(result, dict), "Should return dict"
assert "error" in result, "Should indicate error"
async def test_delegator_address_validation_total_rewards(self):
"""Test that invalid delegator addresses are rejected."""
result = await get_delegation_total_rewards("invalid")
assert isinstance(result, dict), "Should return dict"
assert "error" in result, "Should indicate error"
async def test_delegator_address_validation_validators(self):
"""Test that invalid delegator addresses are rejected."""
result = await get_delegator_validators("")
assert isinstance(result, dict), "Should return dict"
assert "error" in result, "Should indicate error"
async def test_delegation_rewards_validates_both_addresses(self):
"""Test that both addresses are validated."""
# Invalid delegator
result = await get_delegation_rewards("invalid", "regenvaloper1xxx")
assert isinstance(result, dict)
assert "error" in result
# Invalid validator
result = await get_delegation_rewards("regen1xxx", "invalid")
assert isinstance(result, dict)
assert "error" in result
@pytest.mark.asyncio
@pytest.mark.tools
@pytest.mark.online
class TestDistributionToolsUserJourneys:
"""Test distribution tools for real user scenarios.
These tests validate complete staking and rewards workflows.
"""
async def test_staker_can_track_total_rewards(self):
"""
USER: Staker/Delegator
GOAL: Track total staking rewards across all validators
VALIDATES: Can staker see their earnings?
"""
test_address = "regen1xhhquwctvwm40qn3nmmqq067pc2gw22e6q3p3c"
# Step 1: Get total rewards
total_rewards = await get_delegation_total_rewards(test_address)
assert isinstance(total_rewards, dict), \
"Staker cannot track rewards - wrong format"
# Step 2: Should have reward information or clear error
if "error" not in total_rewards:
print("✅ STAKING REWARDS: Staker can track total rewards")
print(f" Address: {test_address}")
# Step 3: Should be able to see which validators
validators = await get_delegator_validators(test_address)
if "error" not in validators:
if "validators" in validators:
val_count = len(validators["validators"])
print(f" Bonded to {val_count} validators")
else:
# Address may not be delegating - that's okay
print(f"⚠️ STAKING REWARDS: Address not delegating")
print(f" Error: {total_rewards.get('error')}")
async def test_validator_operator_can_monitor_commission(self):
"""
USER: Validator Operator
GOAL: Monitor validator commission and outstanding rewards
VALIDATES: Can validator operator track earnings?
"""
test_validator = "regenvaloper1xhhquwctvwm40qn3nmmqq067pc2gw22ezy9n2h"
# Step 1: Get outstanding rewards
rewards = await get_validator_outstanding_rewards(test_validator)
assert isinstance(rewards, dict), \
"Validator cannot monitor rewards - wrong format"
# Step 2: Get commission
commission = await get_validator_commission(test_validator)
assert isinstance(commission, dict), \
"Validator cannot check commission - wrong format"
if "error" not in rewards or "error" not in commission:
print("✅ VALIDATOR MONITORING: Operator can track earnings")
print(f" Validator: {test_validator}")
# Step 3: Check for slashing events
slashes = await get_validator_slashes(test_validator)
if "error" not in slashes:
if "slashes" in slashes:
slash_count = len(slashes["slashes"])
print(f" Slashing events: {slash_count}")
else:
print(f"⚠️ VALIDATOR MONITORING: Validator not found/active")
async def test_network_analyst_can_assess_community_pool(self):
"""
USER: Network Analyst
GOAL: Assess community pool size for governance
VALIDATES: Can analyst see treasury funds?
"""
# Step 1: Get community pool
pool = await get_community_pool()
assert isinstance(pool, dict), \
"Analyst cannot assess pool - wrong format"
# Step 2: Get distribution params
params = await get_distribution_params()
assert isinstance(params, dict), \
"Analyst cannot get params - wrong format"
if "error" not in pool:
print("✅ NETWORK ECONOMICS: Analyst can assess community pool")
if "pool" in pool:
token_types = len(pool["pool"])
print(f" Community pool has {token_types} token types")
# Check params available
if "error" not in params:
print(f" Distribution params available")
else:
print(f"⚠️ NETWORK ECONOMICS: Data not available")
print(f" Pool error: {pool.get('error')}")
