Skip to main content

Testing & Troubleshooting

This section provides guidance on testing your integration with the Revsto Distributor API and troubleshooting common issues. Thorough testing is crucial before moving your integration to production to ensure a smooth experience for your users.

Testing Environments​

Revsto provides two environments for your integration:

  1. Sandbox Environment: https://sandbox.revsto.com/api

    • Use for development and testing
    • Test data is periodically reset
    • No real transactions are processed

  2. Production Environment: https://login.revsto.com/api

    • Only use after thorough sandbox testing
    • Real transactions with actual financial impact
    • Requires formal approval from Revsto
caution

Never use production credentials or real customer data in the sandbox environment. Similarly, never use sandbox credentials in production.

Testing Authentication​

Authentication is the foundation of your API integration. Ensure you can successfully obtain an access token before proceeding with other tests.

Authentication URLs​

  • Production: https://35703.tagpay.fr/api/distributor/v1/oauth2/token
  • Sandbox: https://35702.tagpay.fr/api/distributor/v1/oauth2/token
import requests

def test_authentication(environment="sandbox"):
    """
    Test authentication against Revsto API
    
    Args:
        environment: "sandbox" or "production"
    """
    # URLs for different environments
    urls = {
        "sandbox": "https://35702.tagpay.fr/api/distributor/v1/oauth2/token",
        "production": "https://35703.tagpay.fr/api/distributor/v1/oauth2/token"
    }
    
    url = urls.get(environment, urls["sandbox"])
    
    print(f"Testing authentication against {environment} environment")
    print(f"URL: {url}")
    
    # Get credentials from user input
    print("\nEnter your credentials:")
    username = input("Username: ").strip()
    password = input("Password: ").strip()
    client_id = input("Client ID: ").strip()
    client_secret = input("Client Secret: ").strip()
    
    if not all([username, password, client_id, client_secret]):
        print("All credentials are required!")
        return None
    
    payload = {
        "grant_type": "password",
        "username": username,
        "password": password,
        "client_id": client_id,
        "client_secret": client_secret
    }
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    print("\nSending authentication request...")
    
    try:
        response = requests.post(url, data=payload, headers=headers)
        
        print(f"Response Status: {response.status_code}")
        print(f"Response Headers: {dict(response.headers)}")
        
        if response.status_code == 200:
            token_data = response.json()
            print("\n✅ Authentication successful!")
            print(f"Token Type: {token_data.get('token_type')}")
            print(f"Access Token: {token_data['access_token'][:20]}...")
            print(f"Token expires in: {token_data['expires_in']} seconds")
            
            if 'refresh_token' in token_data:
                print(f"Refresh Token: {token_data['refresh_token'][:20]}...")
            
            # Test the token with a simple API call
            print("\nTesting token with API call...")
            test_token_validity(token_data['access_token'], environment)
            
            return token_data['access_token']
            
        else:
            print(f"\n❌ Authentication failed with status code: {response.status_code}")
            print(f"Response: {response.text}")
            
            # Common error explanations
            if response.status_code == 401:
                print("\nTroubleshooting:")
                print("- Verify your username and password")
                print("- Check that your client_id and client_secret are correct")
                print("- Ensure your account has API access enabled")
            elif response.status_code == 403:
                print("\nTroubleshooting:")
                print("- Your account may not have sufficient permissions")
                print("- Contact Revsto support to verify your API access")
            elif response.status_code == 400:
                print("\nTroubleshooting:")
                print("- Check that all required parameters are provided")
                print("- Verify the grant_type is set to 'password'")
            
            return None
            
    except requests.exceptions.RequestException as e:
        print(f"\n❌ Network error during authentication: {str(e)}")
        return None

def test_token_validity(access_token, environment="sandbox"):
    """
    Test if the access token is valid by making a simple API call
    """
    # Base URLs for API calls
    api_urls = {
        "sandbox": "https://sandbox.revsto.com/api",
        "production": "https://login.revsto.com/api"
    }
    
    base_url = api_urls.get(environment, api_urls["sandbox"])
    test_url = f"{base_url}/distributor/v1/products"
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.get(test_url, headers=headers)
        
        if response.status_code == 200:
            print("✅ Token is valid - API call successful")
            products = response.json()
            if isinstance(products, list):
                print(f"Retrieved {len(products)} products")
            elif isinstance(products, dict) and 'items' in products:
                print(f"Retrieved {len(products['items'])} products")
            else:
                print("Products retrieved successfully")
        elif response.status_code == 401:
            print("❌ Token is invalid or expired")
        elif response.status_code == 403:
            print("⚠️  Token is valid but insufficient permissions for this endpoint")
        else:
            print(f"⚠️  Unexpected response: {response.status_code}")
            
    except requests.exceptions.RequestException as e:
        print(f"❌ Error testing token: {str(e)}")

def get_environment_info():
    """
    Display information about different environments
    """
    print("Environment Information:")
    print("=" * 50)
    print("\n🧪 SANDBOX Environment:")
    print("  Auth URL: https://35702.tagpay.fr/api/distributor/v1/oauth2/token")
    print("  API Base: https://sandbox.revsto.com/api")
    print("  Purpose:  Testing and development")
    print("  Data:     Test data, periodically reset")
    print("  Impact:   No real financial transactions")
    
    print("\n🏢 PRODUCTION Environment:")
    print("  Auth URL: https://35703.tagpay.fr/api/distributor/v1/oauth2/token") 
    print("  API Base: https://login.revsto.com/api")
    print("  Purpose:  Live operations")
    print("  Data:     Real customer data")
    print("  Impact:   Real financial transactions")
    
    print("\n⚠️  Important Notes:")
    print("  - Always test thoroughly in sandbox before production")
    print("  - Never use production credentials in sandbox")
    print("  - Tokens expire after 5 minutes (300 seconds)")
    print("  - Store tokens securely and implement refresh logic")

def main():
    """
    Main function to run authentication tests
    """
    print("=== Revsto API Authentication Test ===")
    
    get_environment_info()
    
    print("\nSelect environment:")
    print("1. Sandbox (recommended for testing)")
    print("2. Production (live environment)")
    
    choice = input("\nEnter choice (1 or 2): ").strip()
    
    if choice == "1":
        environment = "sandbox"
    elif choice == "2":
        environment = "production"
        confirm = input("⚠️  You selected PRODUCTION. Are you sure? (yes/no): ").strip().lower()
        if confirm != "yes":
            print("Switching to sandbox for safety...")
            environment = "sandbox"
    else:
        print("Invalid choice, using sandbox...")
        environment = "sandbox"
    
    # Run the authentication test
    access_token = test_authentication(environment)
    
    if access_token:
        print(f"\n🎉 Success! You can now use this token for API calls:")
        print(f"Authorization: Bearer {access_token}")
        print(f"\nToken will expire in 5 minutes. Store it securely for your session.")
        
        # Option to save token for other tests
        save_token = input("\nSave token for other tests? (y/n): ").strip().lower()
        if save_token == 'y':
            with open('access_token.txt', 'w') as f:
                f.write(access_token)
            print("Token saved to 'access_token.txt'")
    else:
        print("\n❌ Authentication failed. Please check your credentials and try again.")

if __name__ == "__main__":
    test_authentication()

Common Authentication Issues​

IssuePossible CausesSolution
401 UnauthorizedInvalid credentials or expired tokenVerify username, password, client_id, and client_secret
403 ForbiddenInsufficient permissionsContact Revsto to ensure your account has the necessary permissions
Token expirationUsing an expired tokenImplement token refresh logic; tokens expire after 300 seconds (5 minutes)

Testing Webhooks​

Webhooks allow your application to receive real-time notifications. Testing them requires a publicly accessible endpoint.

Webhook Testing Steps​

  1. Set up a public endpoint: Use tools like ngrok or Webhook.site for testing
  2. Subscribe to events: Register your endpoint for specific events
  3. Trigger events: Perform actions that trigger the events (e.g., account creation)
  4. Verify receipt: Confirm your endpoint receives the event payloads
import requests
from flask import Flask, request, jsonify
import threading
import time

# Simple Flask server to receive webhook callbacks
app = Flask(__name__)

# Store received webhooks for testing
received_webhooks = []

@app.route('/webhook', methods=['POST'])
def webhook_handler():
    """
    Handle incoming webhook notifications from Revsto
    """
    try:
        # Get the webhook data
        webhook_data = request.json
        
        # Log the incoming webhook data
        print("\n" + "="*50)
        print("WEBHOOK RECEIVED:")
        print("="*50)
        print(f"Event ID: {webhook_data.get('id')}")
        print(f"Webhook ID: {webhook_data.get('webhookId')}")
        print(f"Type: {webhook_data.get('type')}")
        print(f"Event: {webhook_data.get('event')}")
        print(f"Data: {json.dumps(webhook_data.get('data'), indent=2)}")
        print("="*50)
        
        # Store the webhook for verification
        received_webhooks.append({
            'timestamp': time.time(),
            'data': webhook_data
        })
        
        # Always return a 200 OK response to acknowledge receipt
        return jsonify({"status": "received", "message": "Webhook processed successfully"}), 200
        
    except Exception as e:
        print(f"Error processing webhook: {str(e)}")
        # Still return 200 to avoid retries for malformed data
        return jsonify({"status": "error", "message": str(e)}), 200

def run_webhook_server():
    """
    Run the Flask webhook server in a separate thread
    """
    app.run(debug=False, port=5000, use_reloader=False)

def subscribe_to_webhook(access_token, webhook_url):
    """
    Test webhook subscription by subscribing to test events
    """
    url = "https://sandbox.revsto.com/api/distributor/v1/hooks"
    
    # Subscribe to multiple relevant events for comprehensive testing
    payload = {
        "url": webhook_url,
        "events": [
            "account.status.blocked",
            "account.status.unblocked", 
            "transaction.new",
            "identity.blocked",
            "identity.unblocked",
            "identity.closed"
        ]
    }
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    print(f"Subscribing to webhook at: {webhook_url}")
    print(f"Events: {', '.join(payload['events'])}")
    
    response = requests.post(url, json=payload, headers=headers)
    
    if response.status_code in [200, 201]:
        hook_data = response.json()
        print("\nWebhook subscription successful!")
        print(f"Response: {json.dumps(hook_data, indent=2)}")
        
        # Expected response format:
        expected_format = {
            "id": "hook_id_number",
            "name": "webhook_name", 
            "body": webhook_url,
            "distributorId": "distributor_id",
            "events": payload['events']
        }
        print(f"\nExpected response format: {json.dumps(expected_format, indent=2)}")
        
        return hook_data.get('id')
    else:
        print(f"Webhook subscription failed with status code: {response.status_code}")
        print(f"Response: {response.text}")
        return None

def list_webhooks(access_token):
    """
    List all existing webhook subscriptions
    """
    url = "https://sandbox.revsto.com/api/distributor/v1/hooks"
    headers = {"Authorization": f"Bearer {access_token}"}
    
    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        hooks = response.json()
        print("\nExisting webhook subscriptions:")
        print(json.dumps(hooks, indent=2))
        return hooks
    else:
        print(f"Failed to list webhooks: {response.status_code} - {response.text}")
        return []

def delete_webhook(access_token, hook_id):
    """
    Delete a webhook subscription
    """
    url = f"https://sandbox.revsto.com/api/distributor/v1/hooks/{hook_id}"
    headers = {"Authorization": f"Bearer {access_token}"}
    
    response = requests.delete(url, headers=headers)
    
    if response.status_code in [200, 204]:
        print(f"Webhook {hook_id} deleted successfully")
        if response.content:
            delete_data = response.json()
            print(f"Delete response: {json.dumps(delete_data, indent=2)}")
        return True
    else:
        print(f"Failed to delete webhook: {response.status_code} - {response.text}")
        return False

def trigger_test_events(access_token):
    """
    Trigger some test events that should generate webhooks
    """
    print("\nTriggering test events...")
    
    # You can trigger events by:
    # 1. Creating a user (should trigger transaction.new with USER_CREATION)
    # 2. Opening an account (should trigger transaction.new with OPEN_ACCOUNT)
    # 3. Creating a product account (should trigger transaction.new with ACCOUNT_CREATION)
    
    base_url = "https://sandbox.revsto.com/api/distributor/v1"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    # Create a test user to trigger webhook
    test_user_payload = {
        "legalType": "NATURAL_PERSON",
        "profileId": "280",
        "firstName": "Webhook",
        "lastName": "Test",
        "birthDate": "1990-01-01",
        "externalId": f"webhook-test-{int(time.time())}",
        "email": f"webhook-test-{int(time.time())}@example.com",
        "contactPhone": "+1234567890",
        "address": {
            "line1": "123 Webhook Street",
            "zipcode": "12345",
            "city": "Test City",
            "country": "US"
        },
        "specialAttributes": {
            "residence_country": "US"
        }
    }
    
    print("Creating test user to trigger webhook...")
    create_response = requests.post(f"{base_url}/users", json=test_user_payload, headers=headers)
    
    if create_response.status_code in [200, 201]:
        user_data = create_response.json()
        print(f"Test user created: {user_data.get('id')}")
        print("This should trigger a 'transaction.new' webhook with transactionType: 'USER_CREATION'")
        return user_data.get('id')
    else:
        print(f"Failed to create test user: {create_response.status_code} - {create_response.text}")
        return None

def wait_for_webhooks(timeout=30):
    """
    Wait for webhooks to be received
    """
    print(f"\nWaiting {timeout} seconds for webhooks...")
    start_time = time.time()
    initial_count = len(received_webhooks)
    
    while time.time() - start_time < timeout:
        current_count = len(received_webhooks)
        if current_count > initial_count:
            print(f"Received {current_count - initial_count} new webhook(s)")
            break
        time.sleep(1)
    
    if len(received_webhooks) > initial_count:
        print("\nReceived webhooks summary:")
        for i, webhook in enumerate(received_webhooks[initial_count:], 1):
            data = webhook['data']
            print(f"{i}. Event: {data.get('event')} | Type: {data.get('type')} | ID: {data.get('id')}")
    else:
        print("No webhooks received during the wait period")

def main():
    """
    Main function to run the webhook testing suite
    """
    print("=== Revsto Webhook Testing ===")
    print("\nThis test will:")
    print("1. Start a local webhook server")
    print("2. Subscribe to webhook events")
    print("3. Trigger test events")
    print("4. Verify webhook delivery")
    print("5. Clean up subscriptions")
    
    access_token = input("\nEnter your access token: ").strip()
    
    if not access_token:
        print("Access token is required!")
        return
    
    # For local testing, you'll need to expose your local server
    # Use ngrok or similar tool: ngrok http 5000
    webhook_url = input("Enter your public webhook URL (e.g., https://abc123.ngrok.io/webhook): ").strip()
    
    if not webhook_url:
        print("Using webhook.site for testing...")
        webhook_url = "https://webhook.site/your-unique-url"
        print(f"Please update with your actual webhook.site URL: {webhook_url}")
        return
    
    # Start the webhook server in a separate thread
    print("\nStarting webhook server...")
    server_thread = threading.Thread(target=run_webhook_server, daemon=True)
    server_thread.start()
    time.sleep(2)  # Give server time to start
    
    try:
        # List existing webhooks
        print("\nListing existing webhooks...")
        existing_hooks = list_webhooks(access_token)
        
        # Subscribe to webhooks
        print("\nSubscribing to webhook events...")
        hook_id = subscribe_to_webhook(access_token, webhook_url)
        
        if hook_id:
            # Trigger test events
            user_id = trigger_test_events(access_token)
            
            if user_id:
                # Wait for webhooks
                wait_for_webhooks(30)
                
                # Display webhook examples
                print("\n" + "="*60)
                print("EXPECTED WEBHOOK FORMATS:")
                print("="*60)
                
                print("\n1. User Creation Webhook:")
                user_creation_webhook = {
                    "id": "14940",
                    "webhookId": "0196aa75-6d9d-71ca-b4b6-e86be8d4d148",
                    "type": "transaction",
                    "event": "transaction.new",
                    "data": {
                        "transactionLogId": "14940",
                        "transactionType": "USER_CREATION"
                    }
                }
                print(json.dumps(user_creation_webhook, indent=2))
                
                print("\n2. Account Opening Webhook:")
                account_opening_webhook = {
                    "id": "15013",
                    "webhookId": "0196afb7-268e-730e-917f-c690256e7784",
                    "type": "transaction",
                    "event": "transaction.new",
                    "data": {
                        "transactionLogId": "15013",
                        "transactionType": "OPEN_ACCOUNT"
                    }
                }
                print(json.dumps(account_opening_webhook, indent=2))
                
                print("\n3. Account Status Webhook:")
                account_status_webhook = {
                    "id": "9728",
                    "webhookId": "0196a9bd-a943-72b4-aaf3-a559121beb23",
                    "type": "account",
                    "event": "account.status.opened",
                    "data": {
                        "accountNumber": "10000097286",
                        "ownerIdentityId": 3639
                    }
                }
                print(json.dumps(account_status_webhook, indent=2))
            
            # Clean up - delete the test webhook
            cleanup = input("\nDelete test webhook subscription? (y/n): ").strip().lower()
            if cleanup == 'y':
                delete_webhook(access_token, hook_id)
        
    except KeyboardInterrupt:
        print("\nTest interrupted by user")
    except Exception as e:
        print(f"\nError during testing: {str(e)}")
    
    print("\nWebhook testing completed!")
    print("\nNotes:")
    print("- Webhooks should return HTTP 200 OK quickly")
    print("- Failed webhook deliveries are retried with exponential backoff")
    print("- Always verify webhook signatures in production")
    print("- Use webhook.site or ngrok for local testing")

if __name__ == "__main__":
    import json
    main()

Example Webhook Subscription Response​

When successfully subscribing to an event, you'll receive a response like this:

Webhook Subscription Response
{
  "id": 421,
  "name": "revsto test",
  "body": "https://webhook.site/3befaaa1-4313-494d-ac4c-f58255327701",
  "distributorId": 1016,
  "events": [
      "account.credited"
  ]
}

Webhook Testing Tools​

  • Ngrok: Creates a secure tunnel to your local server
  • Webhook.site: Provides a temporary URL that logs all incoming requests
  • RequestBin: Collects and inspects HTTP requests

Common Webhook Issues​

IssuePossible CausesSolution
No events receivedIncorrect URL or subscriptionVerify subscription status and URL
HTTP errorsServer not responding with 200 OKEnsure your endpoint returns a 200 OK response quickly
Timeout issuesSlow response from your serverOptimize your webhook handler to respond within 5 seconds

Testing Account Creation & KYC​

Account creation involves multiple steps that must be tested thoroughly.

Complete Account Creation Flow​

  1. Create a user (Natural Person or Legal Entity)
  2. Upload KYC documents (if required while user is in PENDING status)
  3. Open user account via the open action
  4. Create product accounts for the approved user

User Creation - Natural Person​

Endpoint: POST https://sandbox.revsto.com/api/distributor/v1/users

Sample body request:

Create Natural Person Request
{
  "legalType": "NATURAL_PERSON",
  "profileId": "280",
  "firstName": "Joanna",
  "lastName": "Christoforidou",
  "birthDate": "1996-09-05",
  "externalId": "1671281015",
  "email": "j.christoforidou@revsto.com",
  "contactPhone": "35722376138",
  "address": {
      "line1": "62 Athalassas Avenue",
      "line2": "Office 102",
      "zipcode": "2023",
      "city": "Nicosia",
      "country": "CY"
  },
  "specialAttributes": {
      "residence_country": "CY"
  }
}

User Creation - Legal Entity​

Sample body request:

Create Legal Entity Request
{
  "login": "JC@2456",
  "legalType": "LEGAL_ENTITY",
  "profileId": "306",
  "email": "j.christoforidou@revsto.com",
  "address": {
      "line1": "88 rue du dôme",
      "line2": "Apt 3",
      "zipcode": "92100",
      "province": "Île-de-France",
      "city": "Boulogne-Billancourt",
      "country": "FR"
  },
  "brandName": "My 100 testing",
  "legalForm": "Inc",
  "legalName": "My Test Business",
  "registrationDate": "2019-08-24",
  "registrationPlace": "Paris",
  "registrationNumber": "RCS123546",
  "registrationCountry": "FR",
  "specialAttributes": {
      "UBO_Name": "Joanna",
      "UBO_LastName": "Christoforidou",
      "residence_country": "CY"
  },
  "boardMember": {
      "legalType": "NATURAL_PERSON",
      "profileId": "280",
      "externalId": "EXT123456789",
      "email": "j.christoforidou@revsto.com",
      "address": {
          "line1": "62 Athalassas Avenue",
          "line2": "Office 102",
          "zipcode": "2023",
          "province": "Strovolos",
          "city": "Nicosia",
          "country": "CY"
      },
      "idNumber": "AZE123456789",
      "nationality": "French",
      "otherNationality": "German",
      "contactPhone": 35722376006,
      "gender": "FEMALE",
      "personTitle": "MS",
      "firstName": "Joanna",
      "lastName": "Christoforidou",
      "birthDate": "2019-08-24",
      "birthCountry": "CY",
      "birthPlace": "Nicosia",
      "jobTitle": "Account manager"
  }
}

Open User Account​

Endpoint: POST https://sandbox.revsto.com/api/distributor/v1/identities/{identityId}/actions/open

Body request:

Open Account Request
{
  "reason": "my reason"
}

Create Product Account​

Endpoint: POST https://sandbox.revsto.com/api/distributor/v1/users/{userId}/accounts

Sample body request:

Create Product Account Request
{
  "productId": 235
}

Sample response:

Create Product Account Response
{
  "id": "10000097583",
  "externalId": null,
  "userId": 3633,
  "productId": 235,
  "status": "OPENED",
  "label": "REVSTO_TESTING EUR #10000097583",
  "balance": {
      "value": 0,
      "currency": "EUR"
  },
  "availableBalance": {
      "value": 0,
      "currency": "EUR"
  },
  "createdAt": "2025-05-08T14:45:37+03:00",
  "lastUsedAt": "2025-05-08T14:45:37+03:00"
}

Relevant Webhooks​

You will receive relevant webhooks during the account creation process:

Transaction New (Account Opening):

Account Opening Webhook
{
  "id": "15013",
  "webhookId": "0196afb7-268e-730e-917f-c690256e7784",
  "type": "transaction",
  "event": "transaction.new",
  "data": {
      "transactionLogId": "15013",
      "transactionType": "OPEN_ACCOUNT"
  }
}

Transaction New (User Creation):

User Creation Webhook
{
  "id": "14940",
  "webhookId": "0196aa75-6d9d-71ca-b4b6-e86be8d4d148",
  "type": "transaction",
  "event": "transaction.new",
  "data": {
      "transactionLogId": "14940",
      "transactionType": "USER_CREATION"
  }
}
import requests
import time
import json

def test_account_creation_flow(access_token):
    """
    Test the complete account creation flow:
    1. Create a user (Natural Person or Legal Entity)
    2. Upload KYC documents (if required while user is in PENDING status)
    3. Open user account via the open action
    4. Create a product account
    """
    base_url = "https://sandbox.revsto.com/api/distributor/v1"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    # Step 1: Create a test user (Natural Person)
    identity_id = "test-" + str(int(time.time()))  # Use timestamp for unique test ID
    
    print("\nSTEP 1: Creating test user (Natural Person)...")
    create_user_url = f"{base_url}/users"
    create_user_payload = {
        "legalType": "NATURAL_PERSON",
        "profileId": "280",
        "firstName": "Test",
        "lastName": "User",
        "birthDate": "1990-01-01",
        "externalId": identity_id,
        "email": f"test-{identity_id}@example.com",
        "contactPhone": "+1234567890",
        "address": {
            "line1": "123 Test Street",
            "line2": "Apt 1",
            "zipcode": "12345",
            "city": "Test City",
            "country": "US"
        },
        "specialAttributes": {
            "residence_country": "US"
        }
    }
    
    user_response = requests.post(create_user_url, json=create_user_payload, headers=headers)
    
    if user_response.status_code not in [200, 201]:
        print(f"User creation failed: {user_response.status_code} - {user_response.text}")
        return False
    
    user_data = user_response.json()
    user_id = user_data.get('id') or user_data.get('identityId')
    print(f"User created with ID: {user_id}")
    print(f"User status: {user_data.get('status', 'Unknown')}")
    
    # Step 2: Simulate KYC document upload (if required)
    # Note: This step is only needed if your setup requires KYC documents
    print("\nSTEP 2: Checking if KYC documents are needed...")
    
    # For this test, we'll assume KYC is handled externally by the EMD
    # In a real scenario, you might upload documents here if required
    print("Assuming KYC is handled externally by EMD")
    
    # Step 3: Open the user account
    print("\nSTEP 3: Opening user account...")
    open_account_url = f"{base_url}/identities/{user_id}/actions/open"
    open_account_payload = {
        "reason": "Test account opening for integration testing"
    }
    
    open_response = requests.post(open_account_url, json=open_account_payload, headers=headers)
    
    if open_response.status_code not in [200, 201, 204]:
        print(f"Account opening failed: {open_response.status_code} - {open_response.text}")
        # Continue anyway as the account might already be open or this might not be required
    else:
        print("User account opened successfully")
    
    # Step 4: Create a product account
    print("\nSTEP 4: Creating product account...")
    
    # First, let's get available products
    products_url = f"{base_url}/products"
    products_response = requests.get(products_url, headers=headers)
    
    if products_response.status_code == 200:
        products_data = products_response.json()
        if products_data and len(products_data) > 0:
            # Use the first available product
            if isinstance(products_data, dict) and 'items' in products_data:
                product_id = products_data['items'][0]['id']
            else:
                product_id = products_data[0].get('id') or products_data[0].get('productId')
        else:
            print("No products available, using default product ID")
            product_id = "235"  # Default product ID from examples
    else:
        print(f"Could not fetch products: {products_response.status_code}")
        product_id = "235"  # Default product ID from examples
    
    create_account_url = f"{base_url}/users/{user_id}/accounts"
    create_account_payload = {
        "productId": product_id
    }
    
    account_response = requests.post(create_account_url, json=create_account_payload, headers=headers)
    
    if account_response.status_code not in [200, 201]:
        print(f"Product account creation failed: {account_response.status_code} - {account_response.text}")
        return False
    
    account_data = account_response.json()
    account_id = account_data.get('id') or account_data.get('accountId')
    print(f"Product account created with ID: {account_id}")
    print(f"Account status: {account_data.get('status')}")
    print(f"Account label: {account_data.get('label', 'N/A')}")
    
    # Step 5: Verify the account details
    print("\nSTEP 5: Verifying account details...")
    get_account_url = f"{base_url}/accounts/{account_id}"
    
    verify_response = requests.get(get_account_url, headers=headers)
    
    if verify_response.status_code == 200:
        verify_data = verify_response.json()
        print("Account verification successful:")
        print(f"  Account ID: {verify_data.get('id')}")
        print(f"  User ID: {verify_data.get('userId')}")
        print(f"  Status: {verify_data.get('status')}")
        print(f"  Currency: {verify_data.get('balance', {}).get('currency', 'Unknown')}")
        print(f"  Balance: {verify_data.get('balance', {}).get('value', 0)}")
        print(f"  IBAN: {verify_data.get('iban', 'Not assigned')}")
    else:
        print(f"Account verification failed: {verify_response.status_code} - {verify_response.text}")
    
    print("\nAccount creation flow test completed successfully!")
    print("\nExpected webhooks:")
    print("1. transaction.new with transactionType: USER_CREATION")
    print("2. transaction.new with transactionType: OPEN_ACCOUNT")
    
    return True

def test_legal_entity_creation(access_token):
    """
    Test creating a Legal Entity user
    """
    base_url = "https://sandbox.revsto.com/api/distributor/v1"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    print("\nTesting Legal Entity Creation...")
    
    entity_id = "entity-" + str(int(time.time()))
    
    create_entity_url = f"{base_url}/users"
    create_entity_payload = {
        "login": f"testentity{int(time.time())}",
        "legalType": "LEGAL_ENTITY",
        "profileId": "306",
        "email": f"entity-{entity_id}@example.com",
        "address": {
            "line1": "123 Business Avenue",
            "line2": "Suite 100",
            "zipcode": "10001",
            "province": "NY",
            "city": "New York",
            "country": "US"
        },
        "brandName": "Test Business Inc",
        "legalForm": "Inc",
        "legalName": "Test Business Incorporated",
        "registrationDate": "2020-01-01",
        "registrationPlace": "New York",
        "registrationNumber": f"TEST{int(time.time())}",
        "registrationCountry": "US",
        "specialAttributes": {
            "UBO_Name": "John",
            "UBO_LastName": "Doe",
            "residence_country": "US"
        },
        "boardMember": {
            "legalType": "NATURAL_PERSON",
            "profileId": "280",
            "externalId": f"BOARD{int(time.time())}",
            "email": f"board-{entity_id}@example.com",
            "address": {
                "line1": "456 Board Street",
                "line2": "Floor 5",
                "zipcode": "10002",
                "province": "NY",
                "city": "New York",
                "country": "US"
            },
            "idNumber": f"ID{int(time.time())}",
            "nationality": "American",
            "contactPhone": 15551234567,
            "gender": "MALE",
            "personTitle": "MR",
            "firstName": "John",
            "lastName": "Doe",
            "birthDate": "1980-01-01",
            "birthCountry": "US",
            "birthPlace": "New York",
            "jobTitle": "CEO"
        }
    }
    
    entity_response = requests.post(create_entity_url, json=create_entity_payload, headers=headers)
    
    if entity_response.status_code in [200, 201]:
        entity_data = entity_response.json()
        print(f"Legal entity created successfully with ID: {entity_data.get('id')}")
        return True
    else:
        print(f"Legal entity creation failed: {entity_response.status_code} - {entity_response.text}")
        return False

if __name__ == "__main__":
    print("=== Revsto Account Creation Test ===")
    print("Make sure you have a valid access token before running this test")
    
    access_token = input("Enter your access token: ").strip()
    
    if not access_token:
        print("Access token is required. Please run the authentication test first.")
        exit(1)
    
    print("\nRunning Natural Person account creation test...")
    success = test_account_creation_flow(access_token)
    
    if success:
        print("\nRunning Legal Entity creation test...")
        test_legal_entity_creation(access_token)
    
    print("\nTest completed!")

Account Testing Checklist​

  • ✅ User creation returns a valid identity ID
  • ✅ KYC document uploads are accepted (if required)
  • ✅ User account opening succeeds
  • ✅ Product account creation succeeds after user is opened
  • ✅ Account retrieval returns expected details
  • ✅ Relevant webhooks are received

Common Account Issues​

IssuePossible CausesSolution
User creation failedInvalid data or duplicate userCheck input data and ensure the user doesn't already exist
KYC rejectionDocument issues or data mismatchVerify document quality and ensure data consistency
Account creation failureUser not in correct status or invalid product IDConfirm user is OPEN or PENDING and use a valid product ID

Testing Payment Flows​

Payment flows involve multiple API calls that must be executed in the correct sequence.

import requests
import time
import json

def test_payment_flow(access_token, source_account, destination_account):
    """
    Test the payment transaction flow:
    1. Initiate a customer instruction
    2. Add a payment transaction
    3. Submit the instruction
    4. Check the transaction status
    """
    base_url = "https://sandbox.revsto.com/api"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    # Step 1: Initiate a customer instruction
    print("\nSTEP 1: Initiating customer instruction...")
    init_url = f"{base_url}/service-domain/v1/customer-instructions/credit-transfers"
    init_payload = {
        "customerInstructionInformation": {
            "paymentInstrument": "CreditTransfer",
            "requestedExecutionDate": time.strftime("%Y-%m-%d"),
            "batchBooking": True,
            "customerInstructionTypeInformation": {
                "serviceLevel": "BACI"
            }
        },
        "customerInstructionOrderingParties": {
            "debtor": {
                "name": "Test Client",
                "accountId": {
                    "value": source_account,
                    "type": "IBAN"
                }
            }
        },
        "externalData": {
            "transferId": f"test-{int(time.time())}"
        }
    }
    
    init_response = requests.post(init_url, json=init_payload, headers=headers)
    if init_response.status_code not in [200, 201, 202]:
        print(f"Instruction initiation failed: {init_response.status_code} - {init_response.text}")
        return False
    
    instruction_data = init_response.json()
    instruction_id = instruction_data.get('customerInstructionId')
    print(f"Customer instruction initiated with ID: {instruction_id}")
    
    # Step 2: Add a payment transaction to the instruction
    print("\nSTEP 2: Adding payment transaction...")
    transaction_url = f"{base_url}/service-domain/v1/customer-instructions/credit-transfers/{instruction_id}/payment-transactions"
    transaction_payload = {
        "paymentTransactionDedicatedInformations": {
            "remittanceInformation": {
                "value": "Test Transaction",
                "type": "UNSTRUCTURED"
            }
        },
        "paymentTransactionAmountInformation": {
            "instructedAmount": {
                "currency": "EUR",
                "value": "1.00"  # Small amount for testing
            }
        },
        "paymentTransactionParties": {
            "creditor": {
                "name": "Test Recipient",
                "accountId": {
                    "value": destination_account,
                    "type": "IBAN"
                }
            }
        }
    }
    
    transaction_response = requests.post(transaction_url, json=transaction_payload, headers=headers)
    if transaction_response.status_code not in [200, 201, 202]:
        print(f"Transaction addition failed: {transaction_response.status_code} - {transaction_response.text}")
        return False
    
    transaction_data = transaction_response.json()
    transaction_id = transaction_data.get('paymentTransactionId')
    print(f"Payment transaction added with ID: {transaction_id}")
    
    # Step 3: Submit the instruction for processing
    print("\nSTEP 3: Submitting customer instruction...")
    submit_url = f"{base_url}/service-domain/v1/customer-instructions/credit-transfers/{instruction_id}/submit"
    submit_response = requests.post(submit_url, headers=headers)
    if submit_response.status_code not in [200, 201, 202]:
        print(f"Instruction submission failed: {submit_response.status_code} - {submit_response.text}")
        return False
    
    submit_data = submit_response.json()
    print(f"Instruction submitted successfully with status: {submit_data.get('status')}")
    
    # Step 4: Check transaction status
    print("\nSTEP 4: Checking transaction status...")
    # Wait briefly for processing
    time.sleep(2)
    
    status_url = f"{base_url}/distributor/v1/transactions/{transaction_id}"
    status_response = requests.get(status_url, headers=headers)
    if status_response.status_code == 200:
        status_data = status_response.json()
        print(f"Transaction status: {status_data.get('status')}")
        print(json.dumps(status_data, indent=2))
    else:
        print(f"Status check failed: {status_response.status_code} - {status_response.text}")
    
    print("\nPayment flow test completed!")
    return True

if __name__ == "__main__":
    access_token = "your_access_token_here"
    source_account = "DK9889000021345847"  # Replace with test source account
    destination_account = "NL24RABO8589312569"  # Replace with test destination account
    test_payment_flow(access_token, source_account, destination_account)

Transaction Testing Checklist​

  • ✅ Customer instruction creation succeeds
  • ✅ Payment transaction addition succeeds
  • ✅ Instruction submission completes
  • ✅ Transaction status updates correctly
  • ✅ Funds are properly debited/credited

Common Transaction Issues​

IssuePossible CausesSolution
Insufficient fundsSource account doesn't have enough balanceAdd funds to the source account
Invalid account detailsIncorrect IBAN or account formatVerify all account details before submission
Pending statusProcessing delay or verification requiredWait for processing to complete or provide additional verification

Testing Internal Posting​

Internal posting involves direct fund transfers between accounts.

import requests
import time
import json

def test_internal_posting(access_token, source_account, destination_account):
    """
    Test the internal posting functionality:
    1. Perform a single internal posting transaction
    2. Check the transaction status
    """
    base_url = "https://sandbox.revsto.com/api/distributor/v1"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    # Step 1: Perform internal posting
    print("\nSTEP 1: Performing internal posting...")
    posting_url = f"{base_url}/transactions/distributorPosting"
    posting_payload = {
        "accounting": [
            {
                "srcAccount": source_account,
                "dstAccount": destination_account,
                "amount": {"value": 1, "currency": "EUR"},
                "type": "TRANSFER"
            }
        ],
        "mode": "TRANSACTION"
    }
    
    posting_response = requests.post(posting_url, json=posting_payload, headers=headers)
    if posting_response.status_code not in [200, 201, 202]:
        print(f"Internal posting failed: {posting_response.status_code} - {posting_response.text}")
        return False
    
    posting_data = posting_response.json()
    transaction_id = posting_data.get('transactionId')
    print(f"Internal posting completed with transaction ID: {transaction_id}")
    print(f"Status: {posting_data.get('status')}")
    
    # Step 2: Check the transaction status
    print("\nSTEP 2: Checking transaction status...")
    # Wait briefly for processing
    time.sleep(2)
    
    status_url = f"{base_url}/transactions/distributorPosting/{transaction_id}"
    status_response = requests.get(status_url, headers=headers)
    if status_response.status_code == 200:
        status_data = status_response.json()
        print(f"Transaction status: {status_data.get('status')}")
        print(json.dumps(status_data, indent=2))
    else:
        print(f"Status check failed: {status_response.status_code} - {status_response.text}")
    
    print("\nInternal posting test completed!")
    return True

if __name__ == "__main__":
    access_token = "your_access_token_here"
    source_account = "10000056878"  # Replace with test source account
    destination_account = "10000056910"  # Replace with test destination account
    test_internal_posting(access_token, source_account, destination_account)

Internal Posting Testing Checklist​

  • ✅ Single transaction posting succeeds
  • ✅ Batch transaction posting succeeds (if supported)
  • ✅ Transaction status can be verified
  • ✅ Funds are correctly transferred

Common Internal Posting Issues​

IssuePossible CausesSolution
Transaction failureInvalid account numbers or insufficient fundsVerify account numbers and balances
Status inconsistencySystem delays or transaction errorsCheck transaction status after a brief delay
Batch partial failureIssues with specific transactions in the batchReview individual transaction status in the batch response

Logging and Monitoring​

Effective logging and monitoring are crucial for troubleshooting API issues.

Best Practices​

  1. Log All API Calls: Record request/response details for each API call
  2. Include Request IDs: Store unique identifiers for each transaction
  3. Monitor Response Times: Track API performance to identify slowdowns
  4. Set Up Alerts: Configure notifications for API errors or anomalies
  5. Audit Webhooks: Log all received webhook events for reconciliation

Example Logging Format​

Logging Format Example
{
  "timestamp": "2023-09-15T14:30:45Z",
  "endpoint": "https://sandbox.revsto.com/api/distributor/v1/transactions",
  "method": "POST",
  "requestId": "req-123456",
  "request": { /* Request payload */ },
  "responseStatus": 200,
  "responseTime": 287,
  "response": { /* Response payload */ }
}

Common HTTP Status Codes​

Status CodeDescriptionTroubleshooting
200 OKRequest succeededNormal operation
201 CreatedResource created successfullyNormal operation for creation requests
400 Bad RequestInvalid request format or dataCheck request payload format and data
401 UnauthorizedAuthentication failureVerify credentials and token validity
403 ForbiddenInsufficient permissionsCheck required permissions for the operation
404 Not FoundResource not foundVerify resource IDs and endpoint paths
409 ConflictResource conflict (e.g., duplicate)Check for existing resources with the same ID
422 Unprocessable EntityValidation errorCheck input data against API requirements
429 Too Many RequestsRate limit exceededImplement backoff strategy and reduce request frequency
500 Server ErrorInternal server errorContact Revsto support if persistent

Support and Escalation​

If you encounter issues that you cannot resolve through troubleshooting:

  1. Documentation: Refer to relevant API documentation sections
  2. Self-Help: Check this troubleshooting guide for common issues
  3. Technical Support: Contact Revsto technical support at api-support@revsto.com
  4. Escalation: For urgent issues, contact your account manager

When contacting support, always include:

  • API endpoint and request method
  • Request and response payloads (with sensitive data redacted)
  • Timestamp of the issue
  • Any error messages or codes received
  • Steps to reproduce the issue

Test Data Reset​

In the sandbox environment, you may periodically need to reset test data:

  1. User accounts: Delete test users via the API
  2. Transactions: Cannot be deleted, but are periodically purged
  3. Webhooks: Unsubscribe from test webhooks when no longer needed
tip

Create unique test identifiers (e.g., with timestamps) to avoid conflicts and make test data easier to identify.

Moving to Production​

Before moving to production, ensure you have:

  1. Completed thorough testing of all API flows
  2. Implemented proper error handling and recovery
  3. Set up monitoring and alerting
  4. Established proper security controls
  5. Received formal approval from Revsto

Production deployments should follow a phased approach:

  1. Limited pilot with controlled user group
  2. Gradual rollout with monitoring
  3. Full production deployment