test_x402_integration.py•8.14 kB
"""
Integration tests for x402 payment flows.
These tests verify payment integration without making actual payments.
"""
import asyncio
import os
import sys
import unittest
from unittest.mock import Mock, patch, AsyncMock
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from src.api.task_manager_client import TaskManager
from src.utils.logging_config import setup_logging
logger = setup_logging(__name__)
class TestX402Integration(unittest.TestCase):
"""Test x402 payment integration flows."""
def setUp(self):
"""Set up test environment."""
# Mock environment variables for testing
self.test_env_vars = {
'PRIVATE_KEY': '0x' + 'a' * 64, # Mock private key
'TASK_MANAGER_API_URL': 'http://localhost:8000',
'PAY_TO_ADDRESS': '0x671cE47E4F38051ba3A990Ba306E2885C2Fe4102',
'X402_NETWORK': 'base-sepolia',
'TASK_CREATION_PRICE': '$1.00'
}
# Apply mock environment
self.env_patcher = patch.dict(os.environ, self.test_env_vars)
self.env_patcher.start()
# Initialize TaskManager client
self.task_manager = TaskManager()
def tearDown(self):
"""Clean up after tests."""
self.env_patcher.stop()
@patch('src.api.task_manager_client.x402HttpxClient')
def test_task_manager_initialization_with_payment(self, mock_x402_client):
"""Test TaskManager initializes correctly with payment configuration."""
task_manager = TaskManager()
# Verify account was created
self.assertIsNotNone(task_manager.account)
self.assertEqual(task_manager.base_url, 'http://localhost:8000')
logger.info("✓ TaskManager initialization with payment config passed")
@patch('src.api.task_manager_client.x402HttpxClient')
async def test_create_task_with_payment_flow(self, mock_x402_client):
"""Test task creation with mocked payment flow."""
# Mock the x402 client response
mock_response = AsyncMock()
mock_response.headers = {
'X-Payment-Response': '{"transaction": "0x123abc", "amount": "$1.00"}',
'content-type': 'application/json'
}
mock_response.aread = AsyncMock(return_value=b'{"task_id": 123}')
mock_client_instance = AsyncMock()
mock_client_instance.request = AsyncMock(return_value=mock_response)
mock_x402_client.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_x402_client.return_value.__aexit__ = AsyncMock(return_value=None)
# Test task creation
task_id = await self.task_manager.create_task(
task_description="Test monitoring task",
frequency_minutes=60,
runtime_minutes=1440,
recipient_email="test@example.com"
)
# Verify payment flow was called
self.assertEqual(task_id, 123)
mock_client_instance.request.assert_called_once()
# Verify the request was made to the correct endpoint
call_args = mock_client_instance.request.call_args
self.assertEqual(call_args[0][0], "POST") # HTTP method
self.assertEqual(call_args[0][1], "/tasks/") # Endpoint
logger.info("✓ Create task with payment flow test passed")
@patch('src.api.task_manager_client.x402HttpxClient')
async def test_free_endpoints_no_payment(self, mock_x402_client):
"""Test that free endpoints work without payment."""
# Mock response for free endpoint
mock_response = AsyncMock()
mock_response.headers = {'content-type': 'application/json'} # No payment response header
mock_response.aread = AsyncMock(return_value=b'{"task_id": 123, "description": "test"}')
mock_client_instance = AsyncMock()
mock_client_instance.request = AsyncMock(return_value=mock_response)
mock_x402_client.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_x402_client.return_value.__aexit__ = AsyncMock(return_value=None)
# Test free endpoint (get task)
task_config = await self.task_manager.get_task(123)
self.assertIsNotNone(task_config)
self.assertEqual(task_config["task_id"], 123)
# Verify no payment was processed (no X-Payment-Response header)
mock_client_instance.request.assert_called_once()
logger.info("✓ Free endpoints test passed")
def test_missing_private_key_handling(self):
"""Test TaskManager handles missing private key gracefully."""
with patch.dict(os.environ, {}, clear=True):
task_manager = TaskManager()
# Should initialize but warn about missing key
self.assertIsNone(task_manager.account)
logger.info("✓ Missing private key handling test passed")
@patch('src.api.task_manager_client.x402HttpxClient')
async def test_payment_failure_handling(self, mock_x402_client):
"""Test handling of payment failures."""
# Mock payment failure
mock_client_instance = AsyncMock()
mock_client_instance.request = AsyncMock(side_effect=Exception("Payment failed"))
mock_x402_client.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_x402_client.return_value.__aexit__ = AsyncMock(return_value=None)
# Test that payment failure is handled properly
with self.assertRaises(Exception) as context:
await self.task_manager.create_task(
task_description="Test task",
frequency_minutes=60,
runtime_minutes=1440,
recipient_email="test@example.com"
)
self.assertIn("Task Manager API error", str(context.exception))
logger.info("✓ Payment failure handling test passed")
def run_x402_tests():
"""Run x402 integration tests."""
print("🔒 X402 PAYMENT INTEGRATION TESTS")
print("=" * 60)
# Run async tests manually with proper setup
async def run_async_tests():
success = True
test_instance = TestX402Integration()
test_instance.setUp()
try:
# Test 1: Create task with payment flow
print("Running: test_create_task_with_payment_flow...")
await test_instance.test_create_task_with_payment_flow()
print("✅ test_create_task_with_payment_flow passed")
# Test 2: Free endpoints
print("Running: test_free_endpoints_no_payment...")
await test_instance.test_free_endpoints_no_payment()
print("✅ test_free_endpoints_no_payment passed")
# Test 3: Payment failure handling
print("Running: test_payment_failure_handling...")
await test_instance.test_payment_failure_handling()
print("✅ test_payment_failure_handling passed")
# Test 4: Sync tests
print("Running: test_task_manager_initialization_with_payment...")
test_instance.test_task_manager_initialization_with_payment()
print("✅ test_task_manager_initialization_with_payment passed")
print("Running: test_missing_private_key_handling...")
test_instance.test_missing_private_key_handling()
print("✅ test_missing_private_key_handling passed")
print("\n✅ All x402 integration tests passed!")
except Exception as e:
print(f"\n❌ X402 tests failed: {e}")
logger.error(f"X402 test error: {e}", exc_info=True)
success = False
finally:
test_instance.tearDown()
return success
# Run async tests
return asyncio.run(run_async_tests())
if __name__ == "__main__":
success = run_x402_tests()
sys.exit(0 if success else 1)