Published about 4 hours ago 14 min read

Appium OTP automation với Python: test OTP SMS trên Android emulator và máy thật

Appium OTP automation hay gãy đúng ở một chỗ. Job nightly chạy test đăng ký trên Android emulator, mọi bước đều xanh, bấm "Gửi mã" xong thì test đứng ở màn hình OTP 30 giây rồi đỏ với TimeoutException. Log Appium không có lỗi nào: backend đã gửi SMS, chỉ là không ai nhận. Emulator không có SIM thật, máy cắm ở runner cũng không có SIM, nên test dùng số của một bạn QA và mỗi lần chạy lại phải nhắn hỏi mã. SMS Retriever hay autofill cũng không cứu được, vì tin nhắn không hề đến máy.

Bài này xử lý trọn bước OTP trong kiểm thử tự động với Appium Python: chọn phương án, helper Python đọc mã qua REST API, các bước Appium, pytest fixture luôn dọn phiên, chạy song song nhiều thiết bị, chi phí và chuyện dùng đúng mục đích. Luồng API mình đã viết ở bài nhận SMS OTP tự động qua API bằng Python, phía Firebase ở bài test Firebase Phone Auth trong Flutter; lần này ráp chúng vào Appium.

Appium OTP automation: chọn cách xử lý bước OTP trước khi viết code

Không phải test nào có bước OTP cũng cần số thật:

Cách	Chứng minh được gì	Rủi ro	Chi phí
1. Mock/stub SMS gateway ở backend staging, trả mã cố định	Luồng UI: nhập số, chuyển màn, gõ mã, vào app	Không chứng minh SMS được gửi; mock lệch dần khỏi gateway thật	Công dựng và bảo trì mock
2. Số test của Firebase (số fictional, mã 6 chữ số cố định, tối đa 10 số)	App gọi đúng API xác thực; chạy được trên emulator	Không gửi SMS nên không kiểm tra được khâu giao tin	Không tốn SMS, không trừ quota
3. Đọc OTP từ DB, Redis hoặc log của staging	Backend sinh và lưu mã đúng	SecOps hay chặn CI vào DB; test dính chặt vào schema nội bộ	Công viết lớp kết nối
4. Số thật thuê qua API	SMS thật đi qua gateway và nhà mạng tới một số thật, app nhận đúng mã đó	Mỗi lượt tốn tiền và mất vài phút; phụ thuộc nhà mạng	Giá thuê số + SMS do gateway của bạn gửi

Cách 1–3 chạy cho mỗi pull request vì nhanh và rẻ. Cách 4 chạy theo lịch (nightly), trước release, sau khi đổi nhà cung cấp SMS, đổi template tin nhắn hay mở thị trường mới. Phần còn lại của bài nói về cách 4.

Còn cờ kiểu SKIP_OTP=true trong backend thì đừng thêm: test xanh mà khâu SMS không được kiểm chứng, và cấu hình nhầm một lần là cờ bật trên production. Cần chế độ test thì đặt ở tầng SMS gateway (cách 1).

Vì sao Appium không dùng được SMS Retriever hay tự động nhập mã OTP ở đây

Số thuê không phải SIM của thiết bị. SMS gửi tới số đó được nhận ở phía nhà cung cấp, không bao giờ chạm tới điện thoại hay emulator đang chạy test. Hệ quả:

Android SMS Retriever API, SMS User Consent API, auto-retrieval của Firebase (verificationCompleted, chỉ có trên Android) và gợi ý mã oneTimeCode trên bàn phím iOS đều không kích hoạt.
Test phải tự đọc mã qua API rồi gõ vào ô như một người dùng.

Nếu muốn test chính giao diện tự động nhập mã OTP, Appium Python client có driver.send_sms(phone_number, message): nó gọi mobile: sendSms để giả lập một SMS đến, chỉ trên Android emulator. Gửi tin có mã kèm app hash mà SMS Retriever cần là kiểm tra được luồng autofill. Đó là một test riêng và không chứng minh SMS thật đi qua nhà mạng.

Emulator hay máy thật? App gửi OTP qua backend và SMS gateway của bạn thì emulator chạy tốt: thiết bị chỉ cần mạng để gọi API, còn SMS đi tới số thuê. App dùng Firebase Phone Auth thì số thật chỉ đăng nhập được trên thiết bị thật, và app verification phải cấu hình đủ; chi tiết trong bài về Firebase Phone Auth.

Chuẩn bị Appium Python: server, UiAutomator2, Android emulator hoặc máy thật

npm install -g appium              # Appium 3 cần Node.js ^20.19, ^22.12 hoặc >= 24
appium driver install uiautomator2
appium driver doctor uiautomator2  # báo thiếu ANDROID_HOME, JDK...
appium                             # mặc định cổng 4723, client gọi http://127.0.0.1:4723

pip install Appium-Python-Client pytest pytest-xdist requests  # client 6.x cần Python >= 3.10
adb devices                        # lấy udid: emulator-5554, hoặc serial máy thật bật USB debugging

export HEROSMS_API_KEY=...         # CI secret, không commit
export HEROSMS_COUNTRY=16          # ID quốc gia (16 = Anh), tra danh sách trong tài liệu API
export HEROSMS_SERVICE=ot          # "ot" = "Any other", dùng cho app của chính bạn
# export HEROSMS_MAX_PRICE=<trần giá>  # tùy chọn: trần giá mỗi số
export APP_APK=/path/to/app-staging.apk
export ANDROID_UDIDS=emulator-5554,emulator-5556

Locator lấy bằng Appium Inspector: kết nối tới server với cùng capabilities, bấm vào phần tử để xem resource-id hoặc accessibility id (content-desc) của ô số điện thoại, nút "Gửi mã", ô nhập mã và một phần tử chỉ có ở màn hình chính. Ưu tiên hai loại này thay vì XPath vì chúng ổn định hơn khi layout đổi.

Helper Python nhận OTP qua API: offers → mua số → poll otp/last → finish/cancel

Mình lấy HeroSMS làm ví dụ vì REST API của họ được mô tả rõ trong tài liệu API của HeroSMS; nhà cung cấp khác có API tương tự thì chỉ cần viết lại file otp_client.py.

Base URL là https://hero-sms.com/api/v1, xác thực bằng header Authorization: ApiKey <token>, lỗi trả JSON có title và details. Mỗi lần mua là một activation (phiên thuê số): một số cho một dịch vụ ở một quốc gia. Cần năm endpoint:

Bước	Endpoint
Xem giá và số còn trong kho	`GET /activations/offers/sms?services=ot&countries=<ID>`
Mua số	`POST /activations`
Đọc OTP	`GET /activations/{activationId}/otp/last`
Đã nhận mã, đóng phiên	`POST /activations/{activationId}/finish`
Không có mã, hủy và hoàn tiền	`DELETE /activations/{activationId}`

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

import requests

API = "https://hero-sms.com/api/v1"
CLOSED = {"ACTIVATION_NOT_ACTIVE", "CANCELED", "FINISHED", "REFUNDED"}
_session: requests.Session | None = None


def _http() -> requests.Session:
    # Tạo session ở lần gọi đầu: pytest import file này cả khi test số thật bị skip
    global _session
    if _session is None:
        _session = requests.Session()
        _session.headers["Authorization"] = f"ApiKey {os.environ['HEROSMS_API_KEY']}"
    return _session


class HeroSMSError(RuntimeError):
    def __init__(self, resp: requests.Response):
        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:  # Retry-After dạng ngày giờ HTTP
            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}")


class NoNumbers(Exception):
    pass


def call(method: str, path: str, **kwargs):
    for attempt in range(3):
        r = _http().request(method, API + path, timeout=15, **kwargs)
        if r.status_code == 429 and attempt < 2:  # RATE_LIMIT: chờ rồi thử lại
            time.sleep(HeroSMSError(r).retry_after or 5)
            continue
        if r.status_code >= 400:
            raise HeroSMSError(r)
        return r.json() if r.content else None  # finish/DELETE trả 204 rỗng


def check_offer(service: str, country: int) -> dict:
    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:
        raise NoNumbers(f"Hết số {service} ở quốc gia {country}")
    return offer


def buy_number(service: str, country: int, max_price: float | None = None) -> dict:
    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: int, timeout: int = 240, interval: int = 5) -> dict | None:
    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:  # vẫn bị giới hạn: chờ rồi poll tiếp
                time.sleep(e.retry_after or interval)
                continue
            # 400/404 = chưa có SMS, trừ khi phiên đã đóng
            if e.status not in (400, 404) or e.title in CLOSED:
                raise
        time.sleep(interval)
    return None


def extract_code(otp: dict) -> str | None:
    if otp.get("smsCode"):
        return otp["smsCode"]
    match = re.search(r"\b\d{4,8}\b", otp.get("smsText") or "")
    return match.group() if match else None


def release(activation_id: int, got_sms: bool) -> None:
    try:
        if got_sms:
            call("POST", f"/activations/{activation_id}/finish")
        else:
            call("DELETE", f"/activations/{activation_id}")  # hủy + hoàn tiền
    except (HeroSMSError, requests.RequestException) as e:
        print(f"Không đóng được phiên {activation_id}: {e}", file=sys.stderr)

Vài điểm dễ sót:

phone trong phản hồi mua số là số nguyên, không có dấu +, nên phải str() trước khi ghép. Form có ô chọn quốc gia riêng thì bỏ phần countryPhoneCode ở đầu.
smsCode (mã đã tách sẵn) và smsText (nguyên tin) đều có thể null. Tài liệu không mô tả phản hồi trước khi SMS đầu tiên về, nên helper coi 400, 404 và data rỗng là "chưa có"; gặp tiêu đề phiên đã đóng như ACTIVATION_NOT_ACTIVE thì dừng ngay. Hệ quả: activationId sai cũng có thể trông giống timeout.
Gặp 429 RATE_LIMIT (endpoint offers có ghi mã này), call() chờ theo Retry-After rồi thử lại tối đa hai lần; riêng vòng poll thì chờ rồi poll tiếp tới hết timeout.
Nhịp poll 5 giây là mình tự chọn, tài liệu không khuyến nghị con số nào. Timeout chờ để vài phút, thấp hơn nhiều so với expiredAt (phiên tiêu chuẩn kéo dài 20 phút). Không muốn poll thì nhận mã qua webhook sms-incoming: nhà cung cấp POST tin nhắn tới URL HTTPS khai trong cài đặt tài khoản.

pytest fixture cho Appium: driver, số thuê và cleanup trong finally

Đặt otp_client.py, conftest.py và test_signup_otp.py cùng một thư mục. driver mở session Appium cho đúng thiết bị của worker, otp_number mua số và luôn đóng phiên: code sau yield chạy dù test pass hay fail, còn try/finally bắt đầu ngay sau lệnh mua nên lỗi lúc dựng fixture cũng không bỏ quên phiên.

# conftest.py
import os
from dataclasses import dataclass

import pytest
from appium import webdriver
from appium.options.android import UiAutomator2Options

import otp_client

APPIUM_URL = os.getenv("APPIUM_URL", "http://127.0.0.1:4723")
UDIDS = [u.strip() for u in os.getenv("ANDROID_UDIDS", "").split(",") if u.strip()]


def pytest_configure(config):
    config.addinivalue_line("markers", "real_sms: gửi SMS thật, tốn tiền")


def pytest_collection_modifyitems(config, items):
    # Không có key thì bỏ qua test số thật trước khi mở Appium hay mua số
    if os.getenv("HEROSMS_API_KEY"):
        return
    skip = pytest.mark.skip(reason="Không có HEROSMS_API_KEY")
    for item in items:
        if "real_sms" in item.keywords:
            item.add_marker(skip)


@pytest.fixture
def driver(worker_id):
    # worker_id của pytest-xdist: "master" khi chạy thường, "gw0", "gw1"... khi có -n
    idx = 0 if worker_id == "master" else int(worker_id.removeprefix("gw"))
    if idx >= len(UDIDS):
        pytest.fail(f"ANDROID_UDIDS có {len(UDIDS)} máy, không đủ cho worker {worker_id}")
    options = UiAutomator2Options()
    options.udid = UDIDS[idx]
    options.app = os.environ["APP_APK"]
    options.system_port = 8200 + idx  # mỗi máy một cổng UiAutomator2 riêng
    options.auto_grant_permissions = True
    options.new_command_timeout = 300
    drv = webdriver.Remote(APPIUM_URL, options=options)
    try:
        yield drv
    finally:
        drv.quit()


@dataclass
class RentedNumber:
    activation_id: int
    phone_e164: str  # +<mã nước><số>
    national: str    # đã bỏ mã nước, cho form có ô chọn quốc gia riêng
    got_sms: bool = False

    def wait_code(self, timeout: int = 240) -> str:
        otp = otp_client.wait_for_otp(self.activation_id, timeout=timeout)
        if otp is None:
            raise TimeoutError(f"Không có SMS cho activation {self.activation_id} sau {timeout}s")
        self.got_sms = True  # SMS đã về: đóng bằng finish, kể cả khi bước sau lỗi
        code = otp_client.extract_code(otp)
        if not code:
            raise AssertionError("SMS không chứa mã, kiểm tra template SMS")
        return code


@pytest.fixture
def otp_number(driver):  # phụ thuộc driver: Appium lỗi thì chưa mua số
    service = os.getenv("HEROSMS_SERVICE", "ot")
    country = int(os.environ["HEROSMS_COUNTRY"])
    max_price = os.getenv("HEROSMS_MAX_PRICE")
    try:
        offer = otp_client.check_offer(service, country)
    except otp_client.NoNumbers as e:
        pytest.skip(str(e))
    print(f"Còn {offer['counts']['total']} số, giá {offer['prices']['default']}")
    act = otp_client.buy_number(service, country, float(max_price) if max_price else None)
    rented = None
    try:
        phone = str(act["phone"])
        rented = RentedNumber(
            activation_id=act["id"],
            phone_e164=f"+{phone}",
            national=phone.removeprefix(str(act["countryPhoneCode"])),
        )
        print(f"Phiên {act['id']}: giá {act.get('price')}, hết hạn {act.get('expiredAt')}")
        yield rented
    finally:
        otp_client.release(act["id"], got_sms=bool(rented and rented.got_sms))

Cần để ý:

otp_number khai phụ thuộc vào driver, nên pytest luôn mở session Appium trước rồi mới mua số: sai udid hay APK hỏng thì không tốn tiền. Phụ thuộc giữa fixture là thứ pytest đảm bảo, còn thứ tự tham số trong hàm test thì không nên dựa vào. Teardown chạy ngược lại: đóng phiên số trước, driver.quit() sau.
Không có HEROSMS_API_KEY thì mọi test real_sms bị skip ngay lúc collect, không mở Appium, không tiêu tiền. Pipeline cho pull request chạy pytest -m "not real_sms".
Hết số ở quốc gia đó thì test được skip chứ không đỏ; muốn job nightly báo đỏ thì đổi pytest.skip thành pytest.fail.
release(): có SMS thì finish (không hoàn tiền), chưa có thì DELETE để hủy và hoàn tiền. Tài liệu ghi số không hủy được trong 2 phút đầu (EARLY_CANCEL_DENIED), nên test fail sau vài giây sẽ không hủy được ngay; release() chỉ ghi log lỗi đó, không che lỗi gốc của test. Phiên không nhận SMS nào thì tiền tự về số dư khi hết hạn.
worker_id là fixture của pytest-xdist, nên phải cài pytest-xdist kể cả khi chưa chạy song song.

Kiểm thử tự động với Appium Python: nhập số, chờ ô mã OTP, gõ mã

# test_signup_otp.py
import pytest
from appium.webdriver.common.appiumby import AppiumBy
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait

# Locator mẫu: thay bằng resource-id / accessibility id lấy từ Appium Inspector
PHONE_INPUT = (AppiumBy.ID, "com.example.app:id/phone_input")
SEND_CODE = (AppiumBy.ACCESSIBILITY_ID, "send_code")
CODE_INPUT = (AppiumBy.ID, "com.example.app:id/otp_input")
HOME = (AppiumBy.ACCESSIBILITY_ID, "home_screen")


@pytest.mark.real_sms
def test_signup_with_real_otp(driver, otp_number):
    wait = WebDriverWait(driver, 30)
    wait.until(EC.element_to_be_clickable(PHONE_INPUT)).send_keys(otp_number.national)
    driver.find_element(*SEND_CODE).click()

    # Ô mã phải hiện ra trước khi chờ SMS: app báo lỗi thì fail ngay
    code_box = wait.until(EC.visibility_of_element_located(CODE_INPUT))
    code = otp_number.wait_code(timeout=240)
    code_box.send_keys(code)

    WebDriverWait(driver, 60).until(EC.visibility_of_element_located(HOME))

Vài điểm dễ sót:

Chờ ô nhập mã hiện ra trước khi poll SMS. App từ chối số hoặc báo vượt giới hạn thì test fail sau 30 giây thay vì ngồi chờ 4 phút, và phiên chưa có SMS không bị tính tiền: tiền về khi phiên được hủy hoặc hết hạn (xem release() ở trên).
national là số đã bỏ mã nước. Form đòi số nội địa có 0 ở đầu (hay gặp ở app Việt Nam) thì thêm "0" phía trước; ô nhận số đầy đủ thì dùng otp_number.phone_e164.

Ô OTP gồm 6 ô riêng thì gõ mỗi ô một chữ số, nhưng chỉ tìm EditText bên trong khung chứa mã (OTP_CONTAINER, cũng lấy từ Appium Inspector), nếu không chữ số đầu có thể rơi vào ô số điện thoại. Widget tự nhảy sang ô kế tiếp thì chỉ cần focus ô đầu và gõ cả mã.

# trong test, thay cho code_box.send_keys(code)
OTP_CONTAINER = (AppiumBy.ID, "com.example.app:id/otp_container")
boxes = driver.find_element(*OTP_CONTAINER).find_elements(
    AppiumBy.CLASS_NAME, "android.widget.EditText")
assert len(boxes) == len(code), f"{len(boxes)} ô cho mã {len(code)} chữ số"
for box, digit in zip(boxes, code):
    box.send_keys(digit)

Bàn phím che nút "Xác nhận" thì gọi driver.hide_keyboard() trước khi bấm.
SMS quốc tế có thể về chậm, đừng đặt timeout ngắn cho bước chờ mã.
Không in nguyên mã ra log; cần debug thì in code[:1] + "***". Chạy xong xóa user test (qua admin API của staging) để lần sau đăng ký lại được.

Chạy Appium song song trên nhiều thiết bị Android với pytest-xdist

export ANDROID_UDIDS=emulator-5554,emulator-5556
pytest -m real_sms -n 2

Mỗi worker có udid riêng, systemPort riêng (8200 + chỉ số) để hai session UiAutomator2 không giành cổng, và activation riêng. Mã không thể lẫn giữa các máy vì otp/last đọc theo activationId. -n lớn hơn số máy trong ANDROID_UDIDS thì fixture driver fail với thông báo rõ ràng, không treo.
Một Appium server chạy được nhiều session cùng lúc. Muốn mỗi thiết bị một server thì map worker sang APPIUM_URL giống cách map udid.
Kho số phải đủ cho số worker: xem counts.total trong offers. Nhiều worker gọi offers cùng lúc có thể gặp 429 RATE_LIMIT; call() tự chờ theo Retry-After rồi thử lại, nhưng vẫn nên giữ -n vừa phải.
POST /activations nhận amount tới 10 nếu muốn mua cả lô trong một fixture dùng chung, nhưng mua theo từng test thì dọn dẹp đơn giản hơn: test nào mua, test đó đóng phiên.

Chi phí mỗi lần chạy test OTP SMS

Mình không đưa con số cụ thể vì giá thay đổi theo quốc gia, nhà mạng và thời điểm. Mỗi lượt gồm:

Giá activation: price trong phản hồi POST /activations, prices.default trong offers. Fixture in cả hai ra log (chạy pytest -rP để thấy; -s chỉ hiện khi không dùng -n).
SMS do hệ thống của bạn gửi: gateway tính theo tin; Firebase tính theo từng SMS và theo quốc gia nhận (bảng giá có trong bài Firebase Phone Auth). Mỗi lần bấm "Gửi lại mã" là thêm một SMS.
Thời gian thiết bị, emulator và runner CI.

Để chi phí trong tầm kiểm soát: đặt HEROSMS_MAX_PRICE (fixture truyền thành maxPrice), chạy theo lịch thay vì theo từng commit, mở Appium session trước khi mua số, hủy phiên khi không có SMS (fixture đã làm) và giới hạn -n, vì mỗi worker là một số thuê.

Dùng đúng mục đích: chỉ test app của bạn, coi OTP như secret

Chỉ test app của bạn hoặc app bạn được phép test, bằng tài khoản test. Dùng số thuê để lách giới hạn đăng ký hay xác minh của dịch vụ khác thường vi phạm điều khoản của họ.
Số thuê chỉ thuộc về bạn trong thời gian activation. Đừng gắn nó vào tài khoản cần khôi phục về sau, và xóa user test đã tạo.
OTP và API key là secret: key nằm trong CI secrets, không commit, không echo; không in nguyên nội dung SMS. Log của Appium server có thể ghi lại giá trị gõ vào ô, nên đừng đăng log thô lên chỗ công khai.
Chạy trên staging (backend hoặc project test riêng), không chạy trên production.

FAQ

Appium có tự đọc được OTP SMS trên emulator không?

Không. Emulator chỉ nhận SMS giả lập do chính test gửi bằng driver.send_sms() (khi đó bạn đã biết mã). SMS thật gửi tới số thuê không đến máy, nên phải đọc qua API rồi gõ vào ô.

Test OTP trên Android emulator hay máy thật?

Backend của bạn tự gửi SMS qua gateway: cả hai đều được. Firebase Phone Auth với số thật: chỉ máy thật.

Firebase test phone numbers có đủ cho automation test không?

Đủ cho test UI ở mỗi pull request. Chúng không gửi SMS, nên không kiểm tra được khâu giao tin.

Chạy song song có bị nhận nhầm mã không?

Không. Mỗi test có activation riêng, và otp/last đọc theo activationId.

Đặt timeout chờ OTP bao nhiêu?

Vài phút, thấp hơn nhiều so với expiredAt của phiên (phiên tiêu chuẩn 20 phút). Bài này dùng 240 giây.

Kết luận

Appium OTP automation nên chia hai tầng: số test của Firebase hoặc mock cho mỗi pull request, số thật qua API theo lịch hoặc trước release. Checklist:

Locator lấy từ Appium Inspector, ưu tiên accessibility id và resource-id.
otp_number phụ thuộc driver: Appium lỗi thì chưa mua số.
Đóng phiên trong finally: có SMS thì finish, không có thì hủy.
Mỗi worker một udid và một systemPort.
API key trong secrets, maxPrice đã đặt, OTP không nằm trong log.

Bạn gặp ca OTP lạ với nhà mạng Việt Nam hay Đông Nam Á khi chạy Appium thì comment nhé, mình sẽ bổ sung vào bài.