如何用 Python 和 HeroSMS API 端到端測試 SMS OTP 流程

Disclosure: I work with HeroSMS, the virtual number provider used in the examples below. Written with AI tools; code and endpoints checked against the HeroSMS API reference.

Your sign-up form sends a one-time code by SMS. The unit tests mock the SMS client, so they pass. Then a tester abroad reports that the code never arrived, or arrived after the form timed out. CI couldn't catch that, because it never sent a real message.

Here's how to test that flow end-to-end in Python. The script rents a real number through an SMS verification API, lets your app send the code, reads it by polling or webhook, submits it back, and cleans up so failed runs get refunded.

When to use real numbers vs. test numbers and mocks

Use the cheapest tool that answers your question:

• Mocks of your SMS client. For unit and most integration tests. Fast and deterministic, but they prove your logic, not delivery.
• Twilio test credentials. Twilio won't charge you, reach real phones or fire status callbacks, and the magic From number +15005550006 passes validation. Useful for checking your request code, though Twilio says test credentials don't work with Verify.
• Firebase Auth test phone numbers. Under "Phone numbers for testing" in the Phone sign-in provider settings, you can add up to 10 fictional numbers, each with a fixed 6-digit code. No SMS is sent, no quota is used, and repeated runs aren't throttled. Good for UI tests.
• Real numbers. These answer "does a real SMS arrive, and does the whole flow accept it?" Use them for a pre-release staging smoke test, after changing SMS providers or templates, or when launching in a new country.

Real-number runs cost money and take minutes, so put them on a schedule, not on every commit.

Setup

You buy an activation, one number reserved for one service in one country. You need a HeroSMS API key and requests:

pip install requests
export HEROSMS_API_KEY="your-key"
export HEROSMS_COUNTRY=16    # country ID; 16 is the United Kingdom
export APP_SIGNUP_URL="<your staging sign-up endpoint>"
export APP_VERIFY_URL="<your staging verify endpoint>"
# optional: HEROSMS_SERVICE (default "ot"), HEROSMS_MAX_PRICE, OTP_TIMEOUT

The REST API lives at https://hero-sms.com/api/v1 and expects an Authorization: ApiKey <token> header. Errors carry title and details. The flow uses five endpoints:

For your own app, use the service code ot ("Any other") and the country you're launching in. The API docs list all codes and country IDs.

The Python script: receive the OTP over the API

# otp_e2e.py - pip install requests
import os
import re
import sys
import time

import requests

API = "https://hero-sms.com/api/v1"
session = requests.Session()
session.headers["Authorization"] = f"ApiKey {os.environ['HEROSMS_API_KEY']}"
CLOSED = {"ACTIVATION_NOT_ACTIVE", "CANCELED", "FINISHED", "REFUNDED"}


class HeroSMSError(RuntimeError):
    def __init__(self, resp):
        try:
            body = resp.json()
        except ValueError:
            body = {}
        if not isinstance(body, dict):
            body = {}
        try:
            self.retry_after = int(resp.headers.get("Retry-After") or 0)
        except ValueError:  # an HTTP-date instead of seconds
            self.retry_after = 0
        self.status = resp.status_code
        self.title = body.get("title", "")
        details = body.get("details", resp.text[:200])
        super().__init__(f"HTTP {self.status} {self.title}: {details}")


def call(method, path, **kwargs):
    r = session.request(method, API + path, timeout=15, **kwargs)
    if r.status_code >= 400:
        raise HeroSMSError(r)
    return r.json() if r.content else None


def check_offer(service, country):
    try:
        resp = call("GET", "/activations/offers/sms",
                    params={"services": service, "countries": country})
    except HeroSMSError as e:
        if e.status != 404:  # 404 = OFFER_NOT_FOUND
            raise
        resp = None
    data = (resp or {}).get("data") or {}
    offer = (data.get(service) or {}).get(str(country))
    if not offer or offer["counts"]["total"] == 0:
        sys.exit(f"No {service} numbers in country {country} right now")
    return offer


def buy_number(service, country, max_price=None):
    body = {"service": service, "country": country,
            "amount": 1, "verificationType": "sms"}
    if max_price is not None:
        body["maxPrice"] = max_price
    return call("POST", "/activations", json=body)["data"][0]


def wait_for_otp(activation_id, timeout=300, interval=5):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            resp = call("GET", f"/activations/{activation_id}/otp/last")
            otp = (resp or {}).get("data") or {}
            if otp.get("smsCode") or otp.get("smsText"):
                return otp
        except HeroSMSError as e:
            if e.status == 429:
                time.sleep(e.retry_after or interval)
                continue
            # 400/404 = no SMS yet, unless the activation is closed
            if e.status not in (400, 404) or e.title in CLOSED:
                raise
        time.sleep(interval)
    return None


def release(activation_id, got_sms):
    try:
        if got_sms:
            call("POST", f"/activations/{activation_id}/finish")
        else:
            call("DELETE", f"/activations/{activation_id}")  # cancel + refund
    except (HeroSMSError, requests.RequestException) as e:
        print(f"Cleanup for {activation_id} failed: {e}", file=sys.stderr)


# Swap in your app's own calls (staging API or Playwright).
def trigger_signup(phone):
    r = requests.post(os.environ["APP_SIGNUP_URL"],
                      json={"phone": phone}, timeout=15)
    r.raise_for_status()


def verify_code(phone, code):
    r = requests.post(os.environ["APP_VERIFY_URL"],
                      json={"phone": phone, "code": code}, timeout=15)
    r.raise_for_status()  # add asserts on the response body


def main():
    service = os.getenv("HEROSMS_SERVICE", "ot")  # "ot" = "Any other"
    country = int(os.environ["HEROSMS_COUNTRY"])  # 16 = United Kingdom
    max_price = os.getenv("HEROSMS_MAX_PRICE")

    offer = check_offer(service, country)
    print(f"{offer['counts']['total']} numbers, "
          f"price {offer['prices']['default']}")
    activation = buy_number(service, country,
                            float(max_price) if max_price else None)
    act_id = activation["id"]

    otp = None
    try:
        phone = f"+{activation['phone']}"
        print(f"Activation {act_id}: {phone}, "
              f"expires {activation.get('expiredAt')}")
        trigger_signup(phone)
        timeout = int(os.getenv("OTP_TIMEOUT", "300"))
        otp = wait_for_otp(act_id, timeout=timeout)
        if otp is None:
            raise TimeoutError(f"No SMS for {phone} within {timeout}s")
        match = re.search(r"\b\d{4,8}\b", otp.get("smsText") or "")
        code = otp.get("smsCode") or (match.group() if match else None)
        if not code:
            raise AssertionError("SMS has no code; check your template")
        print(f"SMS from {otp.get('phoneFrom')}, code {code[:1]}***")
        verify_code(phone, code)
    finally:
        release(act_id, got_sms=otp is not None)


if __name__ == "__main__":
    main()

What's going on:

• POST /activations returns a data array (amount goes up to 10 for parallel tests). phone comes without the +; strip the countryPhoneCode prefix if your form has a separate country picker. A standard activation lasts 20 minutes (see expiredAt), so keep OTP_TIMEOUT well under that.
• otp/last returns smsCode (the parsed code) and smsText (the full message); either can be null. The reference doesn't document the response before the first SMS, so the script treats 400, 404 and an empty data as "not yet". It backs off on 429 and stops on a closed-activation title like ACTIVATION_NOT_ACTIVE or any other error. A 404 from a wrong ID will look like a timeout. If smsCode is missing, a regex reads the text; an SMS with no code fails the test instead of timing out.
• verify_code submits the code, which turns a delivery check into a flow test. Assert whatever "signed up" means in your app.
• Cleanup runs in finally. With no SMS, DELETE cancels and refunds. After an OTP, a regular activation can't be cancelled, so the script calls finish (not refunded). The docs say a number can't be cancelled in its first 2 minutes (EARLY_CANCEL_DENIED), so a sign-up call that fails within seconds can't cancel right away. release() logs that, plus network errors, without hiding the original failure; an activation that never gets an SMS is refunded automatically.
• trigger_signup and verify_code use plain requests, not the API session, so your HeroSMS key never reaches your app. If your app rejects a number, POST /activations/{activationId}/replace swaps it, and phoneException in the purchase body excludes prefixes up front.

Receive the OTP by webhook instead of polling

Polling is fine for one nightly test. For many parallel runs, have the provider push messages instead. The sms-incoming webhook POSTs JSON to up to three HTTPS URLs set in your account profile.

# webhook.py - pip install flask
from flask import Flask, request

app = Flask(__name__)
HEROSMS_IPS = {"84.32.223.53", "185.138.88.87"}
latest = {}  # activationId -> newest SMS (use Redis or a DB for real)


@app.post("/hooks/herosms/sms")
def sms_incoming():
    # Called over HTTPS only: run behind your TLS proxy and take the
    # client IP from the proxy's header instead of remote_addr.
    if request.remote_addr not in HEROSMS_IPS:
        return "", 403
    event = request.get_json(silent=True) or {}
    act_id = event.get("activationId")
    if act_id is not None:
        prev = latest.get(act_id) or {}
        # Retries repeat an SMS; resends are newer. Keep the newest.
        if (event.get("receivedAt") or "") >= (prev.get("receivedAt") or ""):
            latest[act_id] = event
    return "", 200  # within 3 seconds

Things to get right:

• Reply 200 within 3 seconds. Anything else counts as a failure and is retried at least 7 times, 20–30 seconds apart, so keep the handler idempotent and return 200 for duplicates too.
• The payload has id, activationId, phoneFrom, service, text, code, country and receivedAt. The names differ from REST, which uses smsCode and smsText.
• The docs publish these source IPs but no request signature, so the allowlist is your check. Your test then reads your own store; cleanup stays the same. Keep polling as a fallback, since a staging box behind a VPN often can't receive webhooks.

Use it on your own apps only

• Test only apps you own or are authorized to test. Using rented numbers to get around other services' sign-up limits, mass-create accounts or dodge verification rules typically violates their terms.
• A rented number is yours only for the activation window. Never tie it to an account you'll need to recover, and delete the test accounts you created with it.
• Treat OTPs in logs like secrets. Anyone with CI log access can read a live code while it's still valid, which is why the script masks it.
• Keep the API key in a CI secret, and set HEROSMS_MAX_PRICE to cap the price per number.

Wrap-up

Keep mocks and test numbers for everyday CI, and run one real-number check on a schedule to catch delivery problems.

The examples run against HeroSMS. The full reference, including the webhook schema, is in the API docs.