Skip to main content

​
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_id = os.getenv(f"GT_API_KEY_ID_{env.value.upper()}")
        self.api_secret = os.getenv(f"GT_API_SECRET_{env.value.upper()}")
    
    def _get_base_url(self) -> str:
        urls = {
            Environment.DEVELOPMENT: "https://apis-sbx.hopnow.io",
            Environment.TESTING: "https://apis-sbx.hopnow.io",
            Environment.STAGING: "https://apis-sbx.hopnow.io",
            Environment.PRODUCTION: "https://apis.hopnow.io"
        }
        return urls[self.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.api_secret = "test_secret_key_123"
    
    def test_post_request_signature(self):
        """Test HMAC signature for POST request"""
        method = "POST"
        url = "https://apis.hopnow.io/v1/test"
        timestamp = "1640995200"
        body = '{"test":true}'
        
        expected_payload = f"{method}{url}{timestamp}{body}"
        signature = self._create_signature(method, url, timestamp, body)
        
        # Verify signature is 64 characters (SHA256 hex)
        self.assertEqual(len(signature), 64)
        self.assertTrue(all(c in '0123456789abcdef' for c in signature))
    
    def test_get_request_signature(self):
        """Test HMAC signature for GET request (empty body)"""
        method = "GET"
        url = "https://apis.hopnow.io/v1/customers/cus_123/accounts"
        timestamp = "1640995200"
        body = ""
        
        signature = self._create_signature(method, url, timestamp, body)
        self.assertEqual(len(signature), 64)
    
    def test_signature_consistency(self):
        """Test that same inputs produce same signature"""
        method = "POST"
        url = "https://apis.hopnow.io/v1/test"
        timestamp = "1640995200"
        body = '{"test":true}'
        
        signature1 = self._create_signature(method, url, timestamp, body)
        signature2 = self._create_signature(method, url, timestamp, body)
        
        self.assertEqual(signature1, signature2)
    
    def _create_signature(self, method, url, timestamp, body):
        """Helper to create HMAC signature"""
        payload = f"{method.upper()}{url}{timestamp}{body}"
        return hmac.new(
            self.api_secret.encode('utf-8'),
            payload.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

if __name__ == '__main__':
    unittest.main()

​
2. Request Construction

Test that your API client constructs requests correctly: 
class TestAPIClient(unittest.TestCase):
    
    def setUp(self):
        self.client = GlobalTransferClient("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)

​
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 = GlobalTransferClient(
            os.getenv("GT_API_KEY_ID_TESTING"),
            os.getenv("GT_API_SECRET_TESTING"),
            base_url="https://apis-sbx.hopnow.io"
        )
    
    def test_authentication_success(self):
        """Test successful authentication"""
        response = self.client.get("/v1/auth/test")
        
        self.assertIn("customer_id", response)
        self.assertIn("api_key_id", response)
        self.assertEqual(response["message"], "Authentication successful")
    
    def test_authentication_with_invalid_key(self):
        """Test authentication with invalid API key"""
        client = GlobalTransferClient(
            "invalid_key",
            "invalid_secret",
            base_url="https://apis-sbx.hopnow.io"
        )
        
        with self.assertRaises(AuthenticationError):
            client.get("/v1/auth/test")

​
2. End-to-End Workflow Testing

Test complete workflows like account creation: 
class TestAccountWorkflow(unittest.TestCase):
    
    def setUp(self):
        self.client = GlobalTransferClient(
            os.getenv("GT_API_KEY_ID_TESTING"),
            os.getenv("GT_API_SECRET_TESTING"),
            base_url="https://apis-sbx.hopnow.io"
        )
        self.customer_id = os.getenv("GT_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}"
        )

​
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 = GlobalTransferClient(
            os.getenv("GT_API_KEY_ID_TESTING"),
            os.getenv("GT_API_SECRET_TESTING"),
            base_url="https://apis-sbx.hopnow.io"
        )
    
    def test_rate_limit_handling(self):
        """Test rate limit response and retry behavior"""
        def make_request():
            try:
                return self.client.get("/v1/auth/test")
            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")

​
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 = GlobalTransferClient("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)

​
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 = GlobalTransferClient(...)
        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...

​
Monitoring and Alerts

​
1. Production Health Checks

Implement health checks that monitor your API integration: 
class APIHealthCheck:
    
    def __init__(self, client):
        self.client = client
    
    def check_authentication(self):
        """Check if authentication is working"""
        try:
            response = self.client.get("/v1/auth/test")
            return {
                "status": "healthy",
                "response_time": response.get("response_time"),
                "message": "Authentication successful"
            }
        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("/v1/auth/test")
            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:
        GT_API_KEY_ID_TESTING: ${{ secrets.GT_API_KEY_ID_TESTING }}
        GT_API_SECRET_TESTING: ${{ secrets.GT_API_SECRET_TESTING }}
        GT_TEST_CUSTOMER_ID: ${{ secrets.GT_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.