Xây dựng Device Farm Android với ADB + FFmpeg: Tự động hóa kiểm thử giao diện trên 10–50 thiết bị

Khi ứng dụng của bạn phải hoạt động trên một "vườn thú" gồm hàng chục chiếc điện thoại thật, các nhà sản xuất khác nhau, phiên bản Android, độ phân giải màn hình khác nhau, thì việc kiểm thử thủ công nhanh chóng trở thành cơn ác mộng. Dưới đây là một pipeline Python tự động phát hiện tất cả các thiết bị đã kết nối, cài đặt APK song song trên tất cả chúng, chạy các bài kiểm tra có công cụ (instrumented tests), quay video từng lần chạy, và ghép các bản ghi đó thành một báo cáo video duy nhất bằng FFmpeg.

Toàn bộ stack > Python + ADB + FFmpeg < là công cụ QA tiêu chuẩn. Không có phép màu nào, chỉ là tự động hóa những công việc nhàm chán.

Kiến trúc Pipeline

1adb devices  ──►  danh sách serial
2     │
3     ▼
4Cài đặt APK   (song song trên tất cả thiết bị, dùng ThreadPoolExecutor)
5     │
6     ▼
7Với mỗi thiết bị:
8     screenrecord (chạy nền)  →  am instrument (chạy kiểm thử)  →  dừng + kéo video về
9     │
10     ▼
11FFmpeg: chèn serial lên mỗi clip + ghép nối  ──►  test_report.mp4

Những gì bạn cần

• ADB (Android Debug Bridge) từ Android Platform Tools -> điều khiển thiết bị.
• Python 3.10+ -> điều phối (tôi dùng list[str], tuple[...] mà không cần from __future__).
• FFmpeg -> xử lý và ghép video.
• Các thiết bị đã bật USB debugging, kết nối qua USB (hoặc qua Wi-Fi bằng adb tcpip).

Một nguyên tắc xuyên suốt toàn bộ code: tôi truyền đối số cho subprocess dưới dạng list và không dùng shell=True. Cách này an toàn hơn (không bị tấn công chèn qua tên file) và không bị lỗi với khoảng trắng hoặc ký tự đặc biệt trong đường dẫn.

1. Khám phá thiết bị

adb devices cũng liệt kê các thiết bị ở trạng thái unauthorized / offline. Chúng ta chỉ giữ lại những thiết bị thực sự ở trạng thái 'device'.

1import subprocess
2
3def get_devices() -> list[str]:
4    """Trả về serial của tất cả thiết bị ở trạng thái 'device'."""
5    out = subprocess.run(
6        ["adb", "devices"],
7        capture_output=True, text=True, check=True,
8    ).stdout
9
10    serials: list[str] = []
11    for line in out.splitlines()[1:]:      # dòng đầu tiên là tiêu đề "List of devices"
12        line = line.strip()
13        if line.endswith("\tdevice"):      # bỏ qua unauthorized / offline
14            serials.append(line.split("\t")[0])
15    return serials

2. Cài đặt APK song song

Cài đặt trên 50 thiết bị từng cái một rất chậm. Chúng ta phân chia công việc qua một thread pool: mỗi lệnh adb install là một tiến trình riêng, vì vậy threads hoạt động rất tốt ở đây (chúng ta đang chờ I/O, không tốn CPU).

1from concurrent.futures import ThreadPoolExecutor, as_completed
2
3def install_apk(serial: str, apk_path: str) -> tuple[str, bool, str]:
4    r = subprocess.run(
5        ["adb", "-s", serial, "install", "-r", "-g", apk_path],
6        capture_output=True, text=True,
7    )
8    ok = r.returncode == 0 and "Success" in r.stdout
9    return serial, ok, (r.stdout + r.stderr).strip()
10
11def install_on_all(apk_path: str, serials: list[str]) -> None:
12    with ThreadPoolExecutor(max_workers=len(serials) or 1) as pool:
13        futures = [pool.submit(install_apk, s, apk_path) for s in serials]
14        for f in as_completed(futures):
15            serial, ok, log = f.result()
16            print(f"[{'OK' if ok else 'FAIL'}] {serial}")
17            if not ok:
18                print(f"    {log}")

Các flag: -r -> cài đặt lại trong khi giữ dữ liệu, -g -> cấp tất cả quyền runtime ngay lập tức (tiện lợi để kiểm thử không bị vướng các hộp thoại xin quyền).

3. Chạy kiểm thử có công cụ (Instrumented Tests)

am instrument chạy các bài kiểm thử Espresso/JUnit trên thiết bị. Nó in ra OK khi thành công và FAILURES!!! khi thất bại ra stdout -> đó là cách chúng ta xác định kết quả.

1def run_instrumented_tests(
2    serial: str,
3    package: str,
4    runner: str = "androidx.test.runner.AndroidJUnitRunner",
5) -> tuple[str, bool]:
6    r = subprocess.run(
7        ["adb", "-s", serial, "shell", "am", "instrument", "-w",
8         f"{package}/{runner}"],
9        capture_output=True, text=True,
10    )
11    ok = "FAILURES!!!" not in r.stdout and r.returncode == 0
12    return serial, ok

package là ID gói kiểm thử, thường là com.example.app.test.

4. Quay màn hình trong khi chạy kiểm thử

screenrecord quay video ngay trên thiết bị. Có những giới hạn cần nhớ: giới hạn ~3 phút mỗi file và không có âm thanh. Chúng ta bắt đầu quay ở chế độ nền, chạy kiểm thử, sau đó dừng nó một cách sạch sẽ và kéo file về máy chủ.

Cách đáng tin cậy nhất để dừng quay không phải là gửi tín hiệu đến adb cục bộ, mà là dùng pkill ngay trên thiết bị -> bằng cách đó screenrecord sẽ hoàn thiện container MP4 một cách chính xác.

1import time
2
3def start_recording(serial: str, remote: str = "/sdcard/run.mp4") -> subprocess.Popen:
4    return subprocess.Popen(
5        ["adb", "-s", serial, "shell", "screenrecord", remote]
6    )
7
8def stop_recording(
9    serial: str,
10    proc: subprocess.Popen,
11    remote: str = "/sdcard/run.mp4",
12    local: str = "run.mp4",
13) -> None:
14    # SIGINT trên thiết bị giúp screenrecord đóng file đúng cách
15    subprocess.run(["adb", "-s", serial, "shell", "pkill", "-SIGINT", "screenrecord"])
16    proc.wait(timeout=10)
17    time.sleep(1)  # chờ thiết bị một chút để hoàn thiện container
18    subprocess.run(["adb", "-s", serial, "pull", remote, local], check=True)

5. Lắp ráp báo cáo video với FFmpeg

Các thiết bị khác nhau có độ phân giải màn hình khác nhau, vì vậy bạn không thể chỉ ghép nối chúng bằng -c copy. Chúng ta chuẩn hóa mỗi clip về một định dạng chung (1080×1920) và chèn serial qua drawtext trong quá trình xử lý. Sau đó, tất cả các clip đều giống hệt nhau và việc ghép nối cuối cùng là một concat nhanh mà không cần mã hóa lại.

1def label_clip(src: str, dst: str, label: str) -> None:
2    """Scale về 1080x1920 và chèn nhãn (serial thiết bị)."""
3    vf = (
4        "scale=1080:1920:force_original_aspect_ratio=decrease,"
5        "pad=1080:1920:(ow-iw)/2:(oh-ih)/2,"
6        f"drawtext=text='{label}':x=20:y=20:fontsize=42:"
7        "fontcolor=white:box=1:[email protected]"
8    )
9    subprocess.run(
10        ["ffmpeg", "-y", "-i", src, "-vf", vf,
11         "-an", "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", dst],
12        check=True,
13    )
14
15def concat_report(clips: list[str], out: str = "test_report.mp4") -> None:
16    with open("concat_list.txt", "w") as f:
17        for c in clips:
18            f.write(f"file '{c}'\n")
19    subprocess.run(
20        ["ffmpeg", "-y", "-f", "concat", "-safe", "0",
21         "-i", "concat_list.txt", "-c", "copy", out],
22        check=True,
23    )

6. Kết hợp tất cả lại

1def main() -> None:
2    apk = "app-debug.apk"
3    package = "com.example.app.test"
4    runner = "androidx.test.runner.AndroidJUnitRunner"
5
6    serials = get_devices()
7    if not serials:
8        print("Không tìm thấy thiết bị nào. Kiểm tra USB và đầu ra của 'adb devices'.")
9        return
10
11    print(f"Đã tìm thấy thiết bị: {len(serials)}")
12    install_on_all(apk, serials)
13
14    labeled: list[str] = []
15    for serial in serials:
16        proc = start_recording(serial)
17        _, ok = run_instrumented_tests(serial, package, runner)
18        stop_recording(serial, proc, local=f"{serial}.mp4")
19        print(f"[{'PASS' if ok else 'FAIL'}] kiểm thử trên {serial}")
20
21        out = f"{serial}_labeled.mp4"
22        label_clip(f"{serial}.mp4", out, serial)
23        labeled.append(out)
24
25    concat_report(labeled, "test_report.mp4")
26    print("Hoàn tất: test_report.mp4")
27
28if __name__ == "__main__":
29    main()

Ở đây, vòng lặp "ghi hình + kiểm thử" chạy tuần tự qua các thiết bị -> cách này dễ đọc hơn. Đối với một farm thực sự, bạn nên bọc khối này trong một ThreadPoolExecutor để tất cả thiết bị được kiểm thử cùng một lúc; logic tương tự như phần cài đặt ở mục 2.

Đừng phát minh lại bánh xe: các công cụ có sẵn

• scrcpy -> phản chiếu và điều khiển thiết bị theo thời gian thực từ PC của bạn. Không thể thiếu khi gỡ lỗi các bài kiểm thử bị lỗi.
• Appium / Espresso / UI Automator -> các framework kiểm thử UI đầy đủ tính năng; am instrument ở trên chính là engine của chúng.
• Gradle Managed Devices -> chạy kiểm thử trên các trình giả lập trực tiếp từ bản build, không cần thao tác ADB thủ công.
• Firebase Test Lab / AWS Device Farm -> một đội tàu thiết bị thật trên đám mây nếu bạn không muốn tự giữ phần cứng của mình.
• GNU parallel -> nếu bạn muốn điều phối từ bash thay vì Python.

Kinh tế học: chi phí là bao nhiêu và tiết kiệm được gì

Tự động hóa kiểm thử không phải là "kiếm tiền từ không khí" -> nó là cắt giảm hai khoản mục đắt đỏ nhất: giờ công lao động và phút sử dụng đám mây. Dưới đây là ước tính cho ba quy mô điển hình. Các con số chỉ mang tính minh họa và phụ thuộc vào khu vực, nhà cung cấp thiết bị và giá của nhà cung cấp dịch vụ, hãy kiểm tra giá hiện tại trước khi mua.

Farm của riêng bạn -> đầu tư một lần

Hạng mục

Cho 10 thiết bị

Cho 30 thiết bị

Cho 50 thiết bị

Điện thoại Android đã qua sử dụng (~$60 mỗi chiếc)

~$600

~$1,800

~$3,000

Hub USB có nguồn

~$100

~$250

~$400

Mini-PC / máy chủ

~$400

~$400

~$500

Cáp, giá đỡ, linh tinh

~$80

~$150

~$250

Tổng một lần

~$1,200

~$2,600

~$4,150

Tiền điện mỗi tháng

vài xu

~$10–20

~$20–40

Đây là chi phí vốn: trả một lần, và sau đó farm hoạt động trong nhiều năm với chi phí gần như bằng không.

Đám mây -> bạn trả tiền theo phút

Firebase Test Lab, AWS Device Farm, BrowserStack, và các dịch vụ tương tự tính phí theo phút thiết bị -> khoảng $0.05–0.20 mỗi phút thiết bị. Một lần chạy regression trên 30 thiết bị, mỗi thiết bị 5 phút, là 150 phút thiết bị, tức là ~$7.5–30 mỗi lần chạy.

Bây giờ nhân với tần suất CI:

Tần suất chạy

Số lần chạy mỗi tháng

Chi phí (với ~$15/lần chạy)

2 lần/ngày

~44

~$660/tháng

10 lần/ngày

~220

~$3,300/tháng

Mỗi lần push (nhóm tích cực)

500+

$7,500+/tháng

Điểm hòa vốn: một farm 30 thiết bị (~$2,600) tự trả tiền so với đám mây trong khoảng 4 tháng với tần suất khiêm tốn 2 lần chạy mỗi ngày; với CI tích cực, trong vòng chưa đầy một tháng. Sau đó, hóa đơn đám mây vẫn tiếp tục chảy mỗi tháng, còn farm thì không.

Lao động thủ công -> những gì được giải phóng

Một lần chạy thủ công một kịch bản regression duy nhất trên 30 thiết bị tương đương với một ngày làm việc của một kỹ sư QA. Chạy regression hai lần một tuần cộng lại thành khoảng 8 ngày công mỗi tháng. Với chi phí QA khoảng $1,600–2,700/tháng, đó là một phần đáng kể của mức lương mà pipeline giải phóng cho công việc có ý nghĩa thay vì công việc "kết nối–cài đặt–chạm–ghi hình" ×30.

Điều này chuyển đổi thành tiền như thế nào

Không có thu nhập trực tiếp "từ script" ở đây -> có ba cơ chế gián tiếp nhưng rất thực tế:

• Phát hành nhanh hơn. Regression trong vài phút thay vì một ngày → bạn phát hành tính năng thường xuyên hơn → bạn phản ứng với thị trường nhanh hơn. Đối với một sản phẩm đăng ký, điều này liên quan trực tiếp đến tỷ lệ giữ chân và doanh thu.
• Ít lỗi hơn trong sản xuất. Phát hiện một vụ crash trên một chiếc Samsung cụ thể trước khi phát hành chỉ tốn vài xu; cùng một vụ crash đến tay người dùng có nghĩa là điểm đánh giá trên cửa hàng giảm, tỷ lệ rời bỏ và yêu cầu hoàn tiền. Mỗi lỗi được phát hiện sớm là một số ít đánh giá tiêu cực không bao giờ được viết.
• Nó được bán như một dịch vụ. Thiết lập một device farm và kiểm thử CI là một vai trò có nhu cầu cao trong lĩnh vực freelancing và gia công. Pipeline trên là một lõi có sẵn cho một dịch vụ như vậy.

Sự khác biệt chính so với "các kế hoạch tương tác kiếm tiền": ở đó, tiền đến từ việc đánh lừa thuật toán và kết thúc bằng lệnh cấm. Ở đây, nó đến từ số giờ tiết kiệm được và các khoản lỗ được ngăn chặn. Cái đầu tiên sụp đổ; cái thứ hai là một trường hợp kinh doanh bền vững mà bạn không ngại trình bày với khách hàng.

Tổng kết

Kết quả là một pipeline có thể tái tạo: một lệnh duy nhất, và ứng dụng được kiểm tra trên toàn bộ đội tàu thiết bị, với một báo cáo video ở cuối cho thấy hành vi trên từng thiết bị cụ thể.

Đánh dấu và theo dõi để không bỏ lỡ các bài viết hữu ích giúp bạn tiết kiệm tiền 📝