Overview
This guide provides comprehensive testing strategies for integrating with the HopNow API, including unit tests, integration tests, and production monitoring.Testing Environment Setup
1. API Credentials
Create separate API keys for different environments:- Development: Use sandbox API keys for local development
- Testing: Dedicated test environment keys for CI/CD
- Staging: Pre-production keys that mirror production setup
- Production: Live keys with appropriate restrictions
2. Base Configuration
import os
from enum import Enum
class Environment(Enum):
DEVELOPMENT = "development"
TESTING = "testing"
STAGING = "staging"
PRODUCTION = "production"
class APIConfig:
def __init__(self, env: Environment):
self.environment = env
self.base_url = self._get_base_url()
self.api_key = os.getenv(f"HOP_API_KEY_{env.value.upper()}")
self.secret = os.getenv(f"HOP_API_SECRET_{env.value.upper()}")
def _get_base_url(self) -> str:
urls = {
Environment.DEVELOPMENT: "https://api.sbx.hopnow.io",
Environment.TESTING: "https://api.sbx.hopnow.io",
Environment.STAGING: "https://api.sbx.hopnow.io",
Environment.PRODUCTION: "https://api.hopnow.io"
}
return urls[self.environment]
class APIConfig {
constructor(environment) {
this.environment = environment;
this.baseUrl = this._getBaseUrl();
this.apiKey = process.env[`HOP_API_KEY_${environment.toUpperCase()}`];
this.secret = process.env[`HOP_API_SECRET_${environment.toUpperCase()}`];
}
_getBaseUrl() {
const urls = {
'development': 'https://api.sbx.hopnow.io',
'testing': 'https://api.sbx.hopnow.io',
'staging': 'https://api.sbx.hopnow.io',
'production': 'https://api.hopnow.io'
};
return urls[this.environment];
}
}
Unit Testing
1. HMAC Signature Generation
Test your signature generation logic with known test vectors:import unittest
import hmac
import hashlib
class TestHMACSignature(unittest.TestCase):
def setUp(self):
self.secret = "test_secret_key_123"
def test_post_request_signature(self):
"""Test HMAC signature for POST request"""
method = "POST"
path_and_query = "/v1/test"
timestamp = "1640995200"
nonce = "a1b2c3d4e5f678901234567890abcdef"
body = '{"test":true}'
expected_payload = f"{method}{path_and_query}{timestamp}{nonce}{body}"
signature = self._create_signature(method, path_and_query, timestamp, nonce, body)
self.assertEqual(
expected_payload,
'POST/v1/test1640995200a1b2c3d4e5f678901234567890abcdef{"test":true}'
)
self.assertEqual(
signature,
"91d30335d4bfdc5e0801f31de86e34e5ecc557aca551650f42141dd06eeb9953"
)
def test_get_request_signature(self):
"""Test HMAC signature for GET request with query string"""
method = "GET"
path_and_query = "/v1/customers/cus_123/accounts?page=1&size=10"
timestamp = "1640995200"
nonce = "00112233445566778899aabbccddeeff"
body = ""
signature = self._create_signature(method, path_and_query, timestamp, nonce, body)
self.assertEqual(
signature,
"15decbbdf4fa8f2a26998e95a4c95f75bfa19cab74c8f2c2f08ff6ef191800b4"
)
def test_signature_consistency(self):
"""Test that same inputs produce same signature"""
method = "POST"
path_and_query = "/v1/test"
timestamp = "1640995200"
nonce = "a1b2c3d4e5f678901234567890abcdef"
body = '{"test":true}'
signature1 = self._create_signature(method, path_and_query, timestamp, nonce, body)
signature2 = self._create_signature(method, path_and_query, timestamp, nonce, body)
self.assertEqual(signature1, signature2)
def _create_signature(self, method, path_and_query, timestamp, nonce, body):
"""Helper to create HMAC signature"""
payload = f"{method.upper()}{path_and_query}{timestamp}{nonce}{body}"
return hmac.new(
self.secret.encode('utf-8'),
payload.encode('utf-8'),
hashlib.sha256
).hexdigest()
if __name__ == '__main__':
unittest.main()
const crypto = require('crypto');
describe('HMAC Signature Generation', () => {
const secret = 'test_secret_key_123';
const createSignature = (method, pathAndQuery, timestamp, nonce, body) => {
const payload = `${method.toUpperCase()}${pathAndQuery}${timestamp}${nonce}${body}`;
return crypto.createHmac('sha256', secret)
.update(payload)
.digest('hex');
};
test('should generate signature for POST request', () => {
const method = 'POST';
const pathAndQuery = '/v1/test';
const timestamp = '1640995200';
const nonce = 'a1b2c3d4e5f678901234567890abcdef';
const body = '{"test":true}';
const signature = createSignature(method, pathAndQuery, timestamp, nonce, body);
expect(signature).toBe('91d30335d4bfdc5e0801f31de86e34e5ecc557aca551650f42141dd06eeb9953');
});
test('should generate signature for GET request with query string', () => {
const method = 'GET';
const pathAndQuery = '/v1/customers/cus_123/accounts?page=1&size=10';
const timestamp = '1640995200';
const nonce = '00112233445566778899aabbccddeeff';
const body = '';
const signature = createSignature(method, pathAndQuery, timestamp, nonce, body);
expect(signature).toBe('15decbbdf4fa8f2a26998e95a4c95f75bfa19cab74c8f2c2f08ff6ef191800b4');
});
test('should produce consistent signatures', () => {
const method = 'POST';
const pathAndQuery = '/v1/test';
const timestamp = '1640995200';
const nonce = 'a1b2c3d4e5f678901234567890abcdef';
const body = '{"test":true}';
const signature1 = createSignature(method, pathAndQuery, timestamp, nonce, body);
const signature2 = createSignature(method, pathAndQuery, timestamp, nonce, body);
expect(signature1).toBe(signature2);
});
});
2. Request Construction
Test that your API client constructs requests correctly:class TestAPIClient(unittest.TestCase):
def setUp(self):
self.client = HopClient("test_key", "test_secret")
def test_request_headers(self):
"""Test that all required headers are included"""
with patch('requests.request') as mock_request:
self.client._make_request("POST", "/v1/test", {"data": "test"})
call_args = mock_request.call_args
headers = call_args[1]['headers']
# Check all required headers are present
self.assertIn('X-API-Key', headers)
self.assertIn('X-Signature', headers)
self.assertIn('X-Timestamp', headers)
self.assertIn('Content-Type', headers)
# Verify header values
self.assertEqual(headers['Content-Type'], 'application/json')
self.assertEqual(headers['X-API-Key'], 'test_key')
def test_timestamp_freshness(self):
"""Test that timestamp is current"""
with patch('requests.request') as mock_request:
before = int(time.time())
self.client._make_request("GET", "/v1/test")
after = int(time.time())
call_args = mock_request.call_args
headers = call_args[1]['headers']
timestamp = int(headers['X-Timestamp'])
self.assertGreaterEqual(timestamp, before)
self.assertLessEqual(timestamp, after)
const nock = require('nock');
describe('API Client Request Construction', () => {
let client;
beforeEach(() => {
client = new HopClient('test_key', 'test_secret');
});
test('should include all required headers', async () => {
const scope = nock('https://api.hopnow.io')
.post('/v1/test')
.reply(function(uri, requestBody) {
const headers = this.req.headers;
expect(headers['x-api-key']).toBeDefined();
expect(headers['x-signature']).toBeDefined();
expect(headers['x-timestamp']).toBeDefined();
expect(headers['content-type']).toBe('application/json');
return [200, { success: true }];
});
await client.makeRequest('POST', '/v1/test', { data: 'test' });
expect(scope.isDone()).toBe(true);
});
test('should use fresh timestamp', async () => {
const before = Math.floor(Date.now() / 1000);
const scope = nock('https://api.hopnow.io')
.get('/v1/test')
.reply(function() {
const timestamp = parseInt(this.req.headers['x-timestamp'][0]);
const after = Math.floor(Date.now() / 1000);
expect(timestamp).toBeGreaterThanOrEqual(before);
expect(timestamp).toBeLessThanOrEqual(after);
return [200, { success: true }];
});
await client.makeRequest('GET', '/v1/test');
expect(scope.isDone()).toBe(true);
});
});
Integration Testing
1. Authentication Testing
Test that your authentication works against the actual API:class TestAPIAuthentication(unittest.TestCase):
def setUp(self):
# Use sandbox credentials
self.client = HopClient(
os.getenv("HOP_API_KEY_TESTING"),
os.getenv("HOP_API_SECRET_TESTING"),
base_url="https://api.sbx.hopnow.io"
)
self.customer_id = os.getenv("HOP_TEST_CUSTOMER_ID")
def test_authentication_success(self):
"""Test successful authentication"""
response = self.client.get(f"/v1/customers/{self.customer_id}/accounts?size=1")
self.assertIn("items", response)
self.assertIn("page", response)
def test_authentication_with_invalid_key(self):
"""Test authentication with invalid API key"""
client = HopClient(
"invalid_key",
"invalid_secret",
base_url="https://api.sbx.hopnow.io"
)
with self.assertRaises(AuthenticationError):
client.get(f"/v1/customers/{self.customer_id}/accounts?size=1")
describe('API Authentication Integration', () => {
let client;
beforeEach(() => {
client = new HopClient(
process.env.HOP_API_KEY_TESTING,
process.env.HOP_API_SECRET_TESTING,
'https://api.sbx.hopnow.io'
);
});
test('should authenticate successfully', async () => {
const customerId = process.env.HOP_TEST_CUSTOMER_ID;
const response = await client.get(`/v1/customers/${customerId}/accounts?size=1`);
expect(response.items).toBeDefined();
expect(response.page).toBeDefined();
});
test('should fail with invalid credentials', async () => {
const invalidClient = new HopClient(
'invalid_key',
'invalid_secret',
'https://api.sbx.hopnow.io'
);
const customerId = process.env.HOP_TEST_CUSTOMER_ID;
await expect(invalidClient.get(`/v1/customers/${customerId}/accounts?size=1`)).rejects.toThrow();
});
});
2. End-to-End Workflow Testing
Test complete workflows like account creation:class TestAccountWorkflow(unittest.TestCase):
def setUp(self):
self.client = HopClient(
os.getenv("HOP_API_KEY_TESTING"),
os.getenv("HOP_API_SECRET_TESTING"),
base_url="https://api.sbx.hopnow.io"
)
self.customer_id = os.getenv("HOP_TEST_CUSTOMER_ID")
def test_account_creation_and_retrieval(self):
"""Test creating and retrieving an account"""
# Create account
account_data = {"name": f"Test Account {int(time.time())}"}
created_account = self.client.post(
f"/v1/customers/{self.customer_id}/accounts",
account_data
)
# Verify creation response
self.assertIn("id", created_account)
self.assertEqual(created_account["name"], account_data["name"])
self.assertEqual(created_account["status"], "active")
account_id = created_account["id"]
# Retrieve account
retrieved_account = self.client.get(
f"/v1/customers/{self.customer_id}/accounts/{account_id}"
)
# Verify retrieval
self.assertEqual(retrieved_account["id"], account_id)
self.assertEqual(retrieved_account["name"], account_data["name"])
# Clean up - deactivate account
self.client.delete(
f"/v1/customers/{self.customer_id}/accounts/{account_id}"
)
describe('Account Workflow Integration', () => {
let client;
let customerId;
beforeEach(() => {
client = new HopClient(
process.env.HOP_API_KEY_TESTING,
process.env.HOP_API_SECRET_TESTING,
'https://api.sbx.hopnow.io'
);
customerId = process.env.HOP_TEST_CUSTOMER_ID;
});
test('should create and retrieve account', async () => {
// Create account
const accountData = { name: `Test Account ${Date.now()}` };
const createdAccount = await client.post(
`/v1/customers/${customerId}/accounts`,
accountData
);
// Verify creation
expect(createdAccount.id).toBeDefined();
expect(createdAccount.name).toBe(accountData.name);
expect(createdAccount.status).toBe('active');
const accountId = createdAccount.id;
// Retrieve account
const retrievedAccount = await client.get(
`/v1/customers/${customerId}/accounts/${accountId}`
);
// Verify retrieval
expect(retrievedAccount.id).toBe(accountId);
expect(retrievedAccount.name).toBe(accountData.name);
// Clean up
await client.delete(
`/v1/customers/${customerId}/accounts/${accountId}`
);
});
});
Load Testing
1. Rate Limit Testing
Test your application’s behavior under rate limits:import concurrent.futures
import time
class TestRateLimits(unittest.TestCase):
def setUp(self):
self.client = HopClient(
os.getenv("HOP_API_KEY_TESTING"),
os.getenv("HOP_API_SECRET_TESTING"),
base_url="https://api.sbx.hopnow.io"
)
self.customer_id = os.getenv("HOP_TEST_CUSTOMER_ID")
def test_rate_limit_handling(self):
"""Test rate limit response and retry behavior"""
def make_request():
try:
return self.client.get(f"/v1/customers/{self.customer_id}/accounts?size=1")
except RateLimitError as e:
return e
# Make many concurrent requests
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
futures = [executor.submit(make_request) for _ in range(100)]
results = [future.result() for future in futures]
# Check that some requests succeeded and rate limits were handled
successes = [r for r in results if not isinstance(r, Exception)]
rate_limits = [r for r in results if isinstance(r, RateLimitError)]
self.assertGreater(len(successes), 0, "Some requests should succeed")
self.assertGreater(len(rate_limits), 0, "Some requests should hit rate limits")
describe('Rate Limit Testing', () => {
test('should handle rate limits gracefully', async () => {
const client = new HopClient(
process.env.HOP_API_KEY_TESTING,
process.env.HOP_API_SECRET_TESTING,
'https://api.sbx.hopnow.io'
);
const customerId = process.env.HOP_TEST_CUSTOMER_ID;
const makeRequest = async () => {
try {
return await client.get(`/v1/customers/${customerId}/accounts?size=1`);
} catch (error) {
if (error.status === 429) {
return { rateLimited: true };
}
throw error;
}
};
// Make many concurrent requests
const promises = Array(50).fill().map(() => makeRequest());
const results = await Promise.all(promises);
const successes = results.filter(r => !r.rateLimited);
const rateLimited = results.filter(r => r.rateLimited);
expect(successes.length).toBeGreaterThan(0);
expect(rateLimited.length).toBeGreaterThan(0);
});
});
Error Handling Tests
1. Network Error Simulation
Test how your application handles network issues:from unittest.mock import patch
import requests
class TestErrorHandling(unittest.TestCase):
def setUp(self):
self.client = HopClient("test_key", "test_secret")
@patch('requests.request')
def test_network_timeout_handling(self, mock_request):
"""Test handling of network timeouts"""
mock_request.side_effect = requests.exceptions.Timeout()
with self.assertRaises(NetworkError):
self.client.get("/v1/test")
@patch('requests.request')
def test_server_error_retry(self, mock_request):
"""Test retry behavior for server errors"""
# First call fails, second succeeds
mock_request.side_effect = [
requests.exceptions.HTTPError(response=Mock(status_code=500)),
Mock(json=lambda: {"success": True}, status_code=200)
]
result = self.client.get("/v1/test")
self.assertEqual(result["success"], True)
self.assertEqual(mock_request.call_count, 2)
const nock = require('nock');
describe('Error Handling', () => {
test('should retry on server errors', async () => {
const client = new HopClient('test_key', 'test_secret');
// First request fails with 500, second succeeds
nock('https://api.hopnow.io')
.get('/v1/test')
.reply(500, { error: 'Internal Server Error' })
.get('/v1/test')
.reply(200, { success: true });
const result = await client.get('/v1/test');
expect(result.success).toBe(true);
});
test('should handle network timeouts', async () => {
const client = new HopClient('test_key', 'test_secret');
nock('https://api.hopnow.io')
.get('/v1/test')
.delayConnection(5000)
.reply(200, { success: true });
await expect(client.get('/v1/test')).rejects.toThrow('timeout');
});
});
Test Data Management
1. Test Account Creation
Create isolated test accounts for consistent testing:class TestDataManager:
def __init__(self, client, customer_id):
self.client = client
self.customer_id = customer_id
self.created_accounts = []
def create_test_account(self, name_suffix=""):
"""Create a test account and track for cleanup"""
account_data = {
"name": f"Test Account {int(time.time())}{name_suffix}"
}
account = self.client.post(
f"/v1/customers/{self.customer_id}/accounts",
account_data
)
self.created_accounts.append(account["id"])
return account
def cleanup(self):
"""Clean up all created test accounts"""
for account_id in self.created_accounts:
try:
self.client.delete(
f"/v1/customers/{self.customer_id}/accounts/{account_id}"
)
except Exception as e:
print(f"Failed to clean up account {account_id}: {e}")
self.created_accounts.clear()
# Usage in tests
class TestWithCleanup(unittest.TestCase):
def setUp(self):
self.client = HopClient(...)
self.test_data = TestDataManager(self.client, "cus_test_123")
def tearDown(self):
self.test_data.cleanup()
def test_some_functionality(self):
account = self.test_data.create_test_account("_for_balance_test")
# Test logic here...
class TestDataManager {
constructor(client, customerId) {
this.client = client;
this.customerId = customerId;
this.createdAccounts = [];
}
async createTestAccount(nameSuffix = '') {
const accountData = {
name: `Test Account ${Date.now()}${nameSuffix}`
};
const account = await this.client.post(
`/v1/customers/${this.customerId}/accounts`,
accountData
);
this.createdAccounts.push(account.id);
return account;
}
async cleanup() {
for (const accountId of this.createdAccounts) {
try {
await this.client.delete(
`/v1/customers/${this.customerId}/accounts/${accountId}`
);
} catch (error) {
console.error(`Failed to clean up account ${accountId}:`, error);
}
}
this.createdAccounts = [];
}
}
// Usage in tests
describe('Account Tests', () => {
let client;
let testData;
beforeEach(() => {
client = new HopClient(...);
testData = new TestDataManager(client, 'cus_test_123');
});
afterEach(async () => {
await testData.cleanup();
});
test('should handle account operations', async () => {
const account = await testData.createTestAccount('_for_balance_test');
// Test logic here...
});
});
Monitoring and Alerts
1. Production Health Checks
Implement health checks that monitor your API integration:class APIHealthCheck:
def __init__(self, client, customer_id):
self.client = client
self.customer_id = customer_id
def check_authentication(self):
"""Check if authentication is working"""
try:
response = self.client.get(f"/v1/customers/{self.customer_id}/accounts?size=1")
return {
"status": "healthy",
"items_returned": len(response.get("items", [])),
"message": "Authentication succeeded"
}
except Exception as e:
return {
"status": "unhealthy",
"error": str(e),
"message": "Authentication failed"
}
def check_api_availability(self):
"""Check if API is responding"""
import time
start_time = time.time()
try:
self.client.get(f"/v1/customers/{self.customer_id}/accounts?size=1")
response_time = (time.time() - start_time) * 1000
return {
"status": "healthy" if response_time < 5000 else "degraded",
"response_time_ms": response_time,
"message": f"API responding in {response_time:.0f}ms"
}
except Exception as e:
return {
"status": "unhealthy",
"response_time_ms": (time.time() - start_time) * 1000,
"error": str(e),
"message": "API not responding"
}
def full_health_check(self):
"""Complete health check"""
return {
"authentication": self.check_authentication(),
"availability": self.check_api_availability(),
"timestamp": time.time()
}
2. Error Monitoring
Set up monitoring for authentication and API errors:import logging
from datetime import datetime, timedelta
class APIErrorMonitor:
def __init__(self):
self.logger = logging.getLogger(__name__)
self.error_counts = {}
def log_authentication_error(self, error):
"""Log authentication errors"""
self.logger.error(
"API authentication failed",
extra={
"error_type": "authentication_error",
"error_code": getattr(error, 'code', 'unknown'),
"timestamp": datetime.utcnow().isoformat()
}
)
self._increment_error_count("authentication_error")
def log_rate_limit_error(self, error):
"""Log rate limit errors"""
self.logger.warning(
"API rate limit exceeded",
extra={
"error_type": "rate_limit_error",
"retry_after": getattr(error, 'retry_after', None),
"timestamp": datetime.utcnow().isoformat()
}
)
self._increment_error_count("rate_limit_error")
def _increment_error_count(self, error_type):
"""Track error counts for alerting"""
now = datetime.utcnow()
hour_key = now.replace(minute=0, second=0, microsecond=0)
if hour_key not in self.error_counts:
self.error_counts[hour_key] = {}
if error_type not in self.error_counts[hour_key]:
self.error_counts[hour_key][error_type] = 0
self.error_counts[hour_key][error_type] += 1
# Alert if too many errors in the last hour
if self.error_counts[hour_key][error_type] > 10:
self._send_alert(error_type, self.error_counts[hour_key][error_type])
def _send_alert(self, error_type, count):
"""Send alert for high error rates"""
self.logger.critical(
f"High error rate detected: {count} {error_type} errors in the last hour"
)
# Integrate with your alerting system (PagerDuty, Slack, etc.)
Continuous Integration
1. CI/CD Pipeline Testing
Example GitHub Actions workflow for API testing:name: API Integration Tests
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: 3.9
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install -r requirements-test.txt
- name: Run unit tests
run: pytest tests/unit/ -v
- name: Run integration tests
env:
HOP_API_KEY_TESTING: ${{ secrets.HOP_API_KEY_TESTING }}
HOP_API_SECRET_TESTING: ${{ secrets.HOP_API_SECRET_TESTING }}
HOP_TEST_CUSTOMER_ID: ${{ secrets.HOP_TEST_CUSTOMER_ID }}
run: pytest tests/integration/ -v
- name: Run load tests
run: pytest tests/load/ -v --maxfail=1
- name: Generate coverage report
run: |
coverage run -m pytest
coverage xml
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v1
Best Practices Summary
1. Test Organization
- Separate unit, integration, and load tests
- Use consistent test data management
- Implement proper cleanup in teardown methods
2. Environment Management
- Use separate API keys for each environment
- Never commit credentials to version control
- Test against sandbox environments first
3. Error Testing
- Test all error scenarios (network, authentication, rate limits)
- Verify retry logic and backoff strategies
- Monitor error rates in production
4. Performance Testing
- Test rate limit handling
- Measure and monitor response times
- Load test critical workflows
5. Monitoring
- Implement health checks for production
- Set up alerting for high error rates
- Log structured data for analysis
Need help setting up testing for your specific use case? Contact our support team for assistance.