Airtest自动化测试第四篇：APP自动化测试实战——从项目搭建到完整用例落地
·
前三篇我们分别掌握了环境搭建、图像识别API和Poco控件识别。本篇将所有知识串联起来，构建一个真正的APP自动化测试项目。从目录结构、分层设计、用例编写到报告生成，一步到位。
一、实战项目概述
1.1 被测APP说明
本篇以一款典型的电商APP为被测对象，涵盖以下核心功能模块：
| 模块 | 测试场景 | 技术要点 |
|---|---|---|
| 登录 | 正常登录/异常登录/退出登录 | Poco输入 + 断言验证 |
| 首页 | 轮播图/Banner/弹窗处理 | 图像识别 + 弹窗处理 |
| 搜索 | 关键词搜索/搜索结果验证 | Poco定位 + 列表遍历 |
| 商品详情 | 查看详情/加入购物车 | 混合定位 + 属性断言 |
| 购物车 | 修改数量/删除商品/结算 | 列表操作 + 数据驱动 |
| 个人中心 | 查看订单/设置 | 页面跳转 + 信息验证 |
1.2 学习目标
完成本篇后你将掌握：
- 规范的Airtest项目目录结构
- Page Object分层设计模式
- 数据驱动测试（CSV/JSON）
- 公共方法的封装与复用
- 失败重试与异常处理
- 测试报告的生成与定制
- 命令行运行与参数化
二、项目目录结构
2.1 推荐的目录结构
2.2 配置文件详解
config/settings.py —— 全局配置：
python
# config/settings.py
import os
# 项目根目录
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# 测试数据目录
TESTDATA_DIR = os.path.join(BASE_DIR, "testdata")
# 报告目录
REPORT_DIR = os.path.join(BASE_DIR, "reports")
# 日志目录
LOG_DIR = os.path.join(BASE_DIR, "logs")
# 截图目录
SCREENSHOT_DIR = os.path.join(BASE_DIR, "screenshots")
# 全局超时时间（秒）
TIMEOUT = 30
SHORT_TIMEOUT = 10
LONG_TIMEOUT = 60
# 操作间隔（秒）
CLICK_INTERVAL = 0.5
PAGE_LOAD_WAIT = 2.0
# 失败重试次数
RETRY_COUNT = 3
# 是否生成报告
GENERATE_REPORT = True
# 报告类型：simple / detailed
REPORT_TYPE = "detailed"
config/devices.yaml —— 设备配置：
yaml
# config/devices.yaml
devices:
# 默认设备
default:
platform: Android
device: "emulator-5554" # adb devices 中看到的设备名
app_package: "com.example.shopping"
app_activity: ".MainActivity"
# 真机1
phone_01:
platform: Android
device: "ABC123456789"
app_package: "com.example.shopping"
app_activity: ".MainActivity"
# 模拟器
emulator:
platform: Android
device: "emulator-5554"
app_package: "com.example.shopping"
app_activity: ".MainActivity"
三、公共模块封装
3.1 页面基类（base_page.py）
这是整个框架的核心，封装了所有页面共用的操作：
python
# common/base_page.py
from airtest.core.api import *
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
import time
class BasePage:
"""页面基类：封装通用操作"""
def __init__(self, poco_instance=None):
"""
初始化基类
:param poco_instance: 外部传入的Poco实例（多页面共享）
"""
if poco_instance is None:
self.poco = AndroidUiautomationPoco()
else:
self.poco = poco_instance
# ========== Poco 封装方法 ==========
def find_element(self, selector, timeout=30):
"""
查找元素（智能等待）
:param selector: poco选择器对象，如 poco(text="登录")
:param timeout: 超时时间
:return: 找到的元素 或 None
"""
try:
selector.wait_for_appearance(timeout=timeout)
return selector
except Exception:
snapshot(msg=f"未找到元素")
return None
def click_element(self, selector, timeout=30, msg=""):
"""
点击元素（带等待和重试）
:param selector: poco选择器
:param timeout: 等待超时
:param msg: 日志描述
:return: True/False
"""
element = self.find_element(selector, timeout)
if element:
element.click()
time.sleep(0.5)
if msg:
print(f"[点击] {msg}")
return True
else:
print(f"[失败] 未找到可点击元素: {msg}")
return False
def input_text(self, selector, text, clear_first=True, timeout=30):
"""
输入文本
:param selector: poco选择器
:param text: 要输入的文本
:param clear_first: 是否先清空
:param timeout: 超时
:return: True/False
"""
element = self.find_element(selector, timeout)
if element:
element.click()
time.sleep(0.3)
if clear_first:
element.set_text("")
time.sleep(0.2)
element.set_text(str(text))
return True
return False
def get_text(self, selector, timeout=30):
"""获取元素文本"""
element = self.find_element(selector, timeout)
if element:
return element.get_text()
return None
def is_element_exists(self, selector, timeout=5):
"""判断元素是否存在（短超时，不截图）"""
return selector.exists()
def wait_for_element(self, selector, timeout=30):
"""等待元素出现（不操作，纯等待）"""
try:
selector.wait_for_appearance(timeout=timeout)
return True
except Exception:
return False
def swipe_screen(self, direction="up", duration=1.0):
"""
全局滑动
:param direction: up/down/left/right
:param duration: 滑动持续时间
"""
if direction == "up":
swipe([0.5, 0.7], [0.5, 0.3], duration=duration)
elif direction == "down":
swipe([0.5, 0.3], [0.5, 0.7], duration=duration)
elif direction == "left":
swipe([0.8, 0.5], [0.2, 0.5], duration=duration)
elif direction == "right":
swipe([0.2, 0.5], [0.8, 0.5], duration=duration)
time.sleep(0.5)
def take_screenshot(self, msg=""):
"""截图"""
snapshot(msg=msg)
# ========== 通用断言方法 ==========
def assert_element_exists(self, selector, timeout=30, msg=""):
"""断言元素存在"""
assert self.wait_for_element(selector, timeout), \
f"断言失败：{msg or '元素不存在'}"
print(f"[断言通过] {msg or '元素存在'}")
def assert_text_equals(self, selector, expected_text, timeout=30):
"""断言文本内容"""
actual = self.get_text(selector, timeout)
assert actual == expected_text, \
f"文本断言失败：期望'{expected_text}'，实际'{actual}'"
print(f"[断言通过] 文本匹配: {actual}")
def assert_element_not_exists(self, selector, msg=""):
"""断言元素不存在"""
assert not self.is_element_exists(selector, timeout=3), \
f"断言失败：{msg or '元素不应该存在'}"
print(f"[断言通过] {msg or '元素不存在'}")
3.2 测试用例基类（base_test.py）
python
# common/base_test.py
from airtest.core.api import *
from airtest.report.report import simple_report
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
import os
import sys
import time
# 将项目根目录加入路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config.settings import *
class BaseTest:
"""测试用例基类"""
def __init__(self, device_config=None):
"""
初始化测试
:param device_config: 设备配置字典
"""
self.device_config = device_config or {}
self.poco = None
self.test_results = []
def setup(self):
"""测试前置：连接设备、启动APP"""
print("=" * 50)
print("开始测试前置操作...")
# 1. 连接设备
device = self.device_config.get("device", "")
if device:
connect_device(f"Android:///{device}")
print(f"已连接设备: {device}")
# 2. 初始化Poco
self.poco = AndroidUiautomationPoco()
# 3. 启动APP
package = self.device_config.get("app_package", "")
if package:
start_app(package)
print(f"已启动APP: {package}")
# 4. 等待APP加载
time.sleep(PAGE_LOAD_WAIT)
print("前置操作完成")
print("=" * 50)
def teardown(self):
"""测试后置：停止APP、生成报告"""
print("=" * 50)
print("开始测试后置操作...")
# 1. 停止APP
package = self.device_config.get("app_package", "")
if package:
stop_app(package)
print(f"已停止APP: {package}")
# 2. 生成报告
if GENERATE_REPORT:
self._generate_report()
# 3. 打印结果汇总
self._print_summary()
print("后置操作完成")
print("=" * 50)
def _generate_report(self):
"""生成测试报告"""
try:
simple_report(
__file__,
logpath=True,
output=os.path.join(REPORT_DIR, f"report_{time.strftime('%Y%m%d_%H%M%S')}.html")
)
print("测试报告已生成")
except Exception as e:
print(f"报告生成失败: {e}")
def _print_summary(self):
"""打印测试结果汇总"""
total = len(self.test_results)
passed = sum(1 for r in self.test_results if r["status"] == "passed")
failed = sum(1 for r in self.test_results if r["status"] == "failed")
print("\n" + "=" * 40)
print(" 测试结果汇总")
print("=" * 40)
print(f" 总计: {total}")
print(f" 通过: {passed} ✅")
print(f" 失败: {failed} ❌")
print(f" 通过率: {passed/total*100:.1f}%" if total > 0 else " 无结果")
print("=" * 40)
def record_result(self, case_name, status, error_msg=""):
"""记录测试结果"""
self.test_results.append({
"case": case_name,
"status": status,
"error": error_msg,
"time": time.strftime("%H:%M:%S")
})
def run_test_case(self, case_func, case_name):
"""运行单个测试用例（带异常捕获）"""
print(f"\n>>> 执行用例: {case_name}")
try:
case_func()
self.record_result(case_name, "passed")
print(f"<<< 用例通过: {case_name}")
except Exception as e:
self.record_result(case_name, "failed", str(e))
print(f"<<< 用例失败: {case_name}")
print(f" 错误信息: {e}")
snapshot(msg=f"失败截图_{case_name}")
3.3 弹窗处理模块（popup_handler.py）
python
# common/popup_handler.py
import time
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
class PopupHandler:
"""通用弹窗处理器"""
def __init__(self, poco_instance=None):
self.poco = poco_instance or AndroidUiautomationPoco()
def handle_all_popups(self, max_attempts=3):
"""
处理所有可能的弹窗
:param max_attempts: 最大尝试次数
:return: 处理的弹窗数量
"""
handled_count = 0
# 按优先级排列的弹窗处理策略
strategies = [
# 1. 更新提示弹窗
{
"name": "更新提示",
"detect": lambda: self.poco(textMatches=".*(更新|升级|新版本).*").exists(),
"close": lambda: self._close_popup_by_buttons(["稍后再说", "暂不更新", "取消", "关闭"])
},
# 2. 权限请求弹窗
{
"name": "权限请求",
"detect": lambda: self.poco(textMatches=".*(允许|授权|权限).*").exists(),
"close": lambda: self._close_popup_by_buttons(["允许", "始终允许", "确定"])
},
# 3. 活动/广告弹窗
{
"name": "活动弹窗",
"detect": lambda: self.poco(textMatches=".*(限时|优惠|福利|红包).*").exists(),
"close": lambda: self._close_popup_by_buttons(["关闭", "×", "取消"])
},
# 4. 协议/公告弹窗
{
"name": "协议公告",
"detect": lambda: self.poco(textMatches=".*(协议|隐私|公告|须知).*").exists(),
"close": lambda: self._close_popup_by_buttons(["同意", "我知道了", "关闭"])
},
# 5. 引导提示
{
"name": "引导提示",
"detect": lambda: self.poco(textMatches=".*(引导|提示|试试).*").exists(),
"close": lambda: self._close_popup_by_buttons(["知道了", "跳过", "关闭"])
},
]
for attempt in range(max_attempts):
found_any = False
for strategy in strategies:
try:
if strategy["detect"]():
print(f"[弹窗处理] 发现{strategy['name']}")
strategy["close"]()
handled_count += 1
found_any = True
time.sleep(1)
except Exception as e:
print(f"[弹窗处理] {strategy['name']}处理异常: {e}")
if not found_any:
break # 没有更多弹窗，退出
print(f"[弹窗处理] 共处理 {handled_count} 个弹窗")
return handled_count
def _close_popup_by_buttons(self, button_texts):
"""尝试用多个可能的按钮关闭弹窗"""
for text in button_texts:
btn = self.poco(text=text)
if btn.exists():
btn.click()
time.sleep(0.5)
return True
# 兜底：用图像识别关闭
from airtest.core.api import touch, Template
try:
touch(Template(r"tpl_close.png"))
return True
except Exception:
pass
return False
3.4 工具函数（utils.py）
python
# common/utils.py
import csv
import json
import yaml
import os
import time
def read_csv(file_path):
"""
读取CSV测试数据
返回：列表，每个元素是字典
"""
data = []
with open(file_path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
data.append(row)
return data
def read_json(file_path):
"""读取JSON测试数据"""
with open(file_path, "r", encoding="utf-8") as f:
return json.load(f)
def read_yaml(file_path):
"""读取YAML测试数据"""
with open(file_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
def get_timestamp():
"""获取当前时间戳字符串"""
return time.strftime("%Y%m%d_%H%M%S")
def ensure_dir(dir_path):
"""确保目录存在"""
if not os.path.exists(dir_path):
os.makedirs(dir_path)
def retry(times=3, delay=1):
"""
重试装饰器
:param times: 最大重试次数
:param delay: 重试间隔（秒）
"""
def decorator(func):
def wrapper(*args, **kwargs):
last_exception = None
for i in range(times):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
print(f"[重试] 第{i+1}次失败: {e}")
if i < times - 1:
time.sleep(delay)
raise last_exception
return wrapper
return decorator
四、Page Object 页面层实现
4.1 登录页（login_page.py）
python
# pages/login_page.py
from common.base_page import BasePage
from config.settings import TIMEOUT
class LoginPage(BasePage):
"""登录页面"""
# ---------- 页面元素定位器 ----------
def username_input(self):
return self.poco("android.widget.EditText")[0]
def password_input(self):
return self.poco("android.widget.EditText")[1]
def login_button(self):
return self.poco(text="登录")
def login_with_phone_button(self):
return self.poco(text="手机号登录")
def register_link(self):
return self.poco(text="注册账号")
def forget_password_link(self):
return self.poco(text="忘记密码")
def error_message(self):
return self.poco(textMatches=".*(错误|失败|不正确).*")
def success_message(self):
return self.poco(textMatches=".*(成功|欢迎).*")
def agree_checkbox(self):
return self.poco("android.widget.CheckBox")
# ---------- 页面操作 ----------
def login(self, username, password):
"""
执行登录操作
:param username: 用户名
:param password: 密码
:return: True/False
"""
print(f"[登录] 开始登录：用户名={username}")
# 等待登录页加载
if not self.wait_for_element(self.login_button(), timeout=TIMEOUT):
print("[登录] 登录页面未加载")
return False
# 勾选同意协议
if self.is_element_exists(self.agree_checkbox()):
checkbox = self.agree_checkbox()
if not checkbox.attr("checked"):
checkbox.click()
print("[登录] 已勾选同意协议")
# 输入用户名
self.input_text(self.username_input(), username, msg="用户名")
# 输入密码
self.input_text(self.password_input(), password, msg="密码")
# 点击登录
self.click_element(self.login_button(), msg="登录按钮")
return True
def login_with_data(self, data_dict):
"""
使用数据字典登录
:param data_dict: {"username": "xxx", "password": "xxx"}
"""
return self.login(
username=data_dict.get("username", ""),
password=data_dict.get("password", "")
)
def is_login_success(self):
"""判断是否登录成功"""
return self.is_element_exists(self.success_message(), timeout=5)
def is_login_failed(self):
"""判断是否登录失败"""
return self.is_element_exists(self.error_message(), timeout=3)
def get_error_text(self):
"""获取错误提示文本"""
element = self.find_element(self.error_message(), timeout=3)
return element.get_text() if element else ""
def go_to_register(self):
"""跳转到注册页面"""
self.click_element(self.register_link(), msg="注册账号")
def wait_for_page_load(self):
"""等待登录页加载完成"""
return self.wait_for_element(self.login_button(), timeout=TIMEOUT)
4.2 首页（home_page.py）
python
# pages/home_page.py
from common.base_page import BasePage
from common.popup_handler import PopupHandler
from config.settings import TIMEOUT
class HomePage(BasePage):
"""首页"""
def __init__(self, poco_instance=None):
super().__init__(poco_instance)
self.popup_handler = PopupHandler(poco_instance)
# ---------- 元素定位器 ----------
def search_box(self):
return self.poco("com.example.app:id/search_box")
def banner(self):
return self.poco("com.example.app:id/banner")
def category_tab(self, name):
return self.poco(text=name)
def product_card(self, index=0):
return self.poco("com.example.app:id/product_card")[index]
def cart_icon(self):
return self.poco(desc="购物车")
def profile_icon(self):
return self.poco(desc="我的")
def home_tab(self):
return self.poco(text="首页")
# ---------- 页面操作 ----------
def wait_for_home_load(self, handle_popups=True):
"""
等待首页加载完成
:param handle_popups: 是否自动处理弹窗
"""
result = self.wait_for_element(self.home_tab(), timeout=TIMEOUT)
if result and handle_popups:
import time
time.sleep(2) # 等待弹窗可能出现
self.popup_handler.handle_all_popups()
return result
def click_search(self):
"""点击搜索框，进入搜索页"""
return self.click_element(self.search_box(), msg="搜索框")
def click_product(self, index=0):
"""点击商品进入详情"""
return self.click_element(self.product_card(index), msg=f"第{index+1}个商品")
def swipe_banner(self):
"""滑动Banner"""
banner = self.find_element(self.banner())
if banner:
self.poco.swipe([0.8, 0.5], [0.2, 0.5])
def go_to_cart(self):
"""进入购物车"""
return self.click_element(self.cart_icon(), msg="购物车")
def go_to_profile(self):
"""进入个人中心"""
return self.click_element(self.profile_icon(), msg="个人中心")
def select_category(self, category_name):
"""选择分类Tab"""
return self.click_element(self.category_tab(category_name), msg=f"分类-{category_name}")
def scroll_to_bottom(self):
"""滚动到底部"""
for _ in range(3):
self.swipe_screen("up")
def get_product_count(self):
"""获取当前可见商品数量"""
try:
cards = self.poco("com.example.app:id/product_card")
return len(cards)
except Exception:
return 0
4.3 搜索页（search_page.py）
python
# pages/search_page.py
from common.base_page import BasePage
from config.settings import TIMEOUT
class SearchPage(BasePage):
"""搜索页面"""
# ---------- 元素定位器 ----------
def search_input(self):
return self.poco("com.example.app:id/search_input")
def search_button(self):
return self.poco(text="搜索")
def search_result_list(self):
return self.poco("com.example.app:id/search_result_list")
def result_item(self, index=0):
return self.poco("com.example.app:id/result_item")[index]
def result_item_name(self, index=0):
return self.result_item(index).child("com.example.app:id/item_name")
def result_item_price(self, index=0):
return self.result_item(index).child("com.example.app:id/item_price")
def no_result_text(self):
return self.poco(textMatches=".*(没有|无|找不到).*")
def result_count_text(self):
return self.poco(textMatches=".*个结果.*")
def sort_button(self):
return self.poco(text="排序")
def filter_button(self):
return self.poco(text="筛选")
def back_button(self):
return self.poco(desc="返回")
# ---------- 页面操作 ----------
def search(self, keyword):
"""
执行搜索
:param keyword: 搜索关键词
:return: True/False
"""
print(f"[搜索] 搜索关键词: {keyword}")
# 等待搜索页加载
self.wait_for_element(self.search_input(), timeout=TIMEOUT)
# 输入关键词
self.input_text(self.search_input(), keyword, msg="搜索关键词")
# 点击搜索
self.click_element(self.search_button(), msg="搜索按钮")
# 等待结果加载
return self.wait_for_element(self.search_result_list(), timeout=TIMEOUT)
def get_result_count(self):
"""获取搜索结果数量"""
element = self.find_element(self.result_count_text(), timeout=5)
if element:
text = element.get_text()
import re
match = re.search(r'\d+', text)
return int(match.group()) if match else 0
return 0
def get_all_result_names(self, max_items=20):
"""
获取所有搜索结果的名称
:param max_items: 最多获取条数
:return: 名称列表
"""
names = []
try:
items = self.poco("com.example.app:id/result_item")
for i in range(min(len(items), max_items)):
try:
name = items[i].child("com.example.app:id/item_name").get_text()
names.append(name)
except Exception:
break
except Exception:
pass
return names
def verify_search_results_contain(self, keyword):
"""
验证搜索结果是否包含关键词
:param keyword: 期望包含的关键词
:return: (是否全部包含, 不包含的项)
"""
names = self.get_all_result_names()
failed_items = [n for n in names if keyword.lower() not in n.lower()]
if failed_items:
print(f"[验证] {len(failed_items)}/{len(names)} 个结果不包含'{keyword}':")
for item in failed_items[:5]:
print(f" - {item}")
return len(failed_items) == 0, failed_items
def click_result(self, index=0):
"""点击第N个搜索结果"""
return self.click_element(self.result_item(index), msg=f"第{index+1}个搜索结果")
def has_results(self):
"""是否有搜索结果"""
return not self.is_element_exists(self.no_result_text(), timeout=3)
def go_back(self):
"""返回上一页"""
return self.click_element(self.back_button(), msg="返回")
4.4 商品详情页（product_detail_page.py）
python
# pages/product_detail_page.py
from common.base_page import BasePage
from config.settings import TIMEOUT
class ProductDetailPage(BasePage):
"""商品详情页"""
# ---------- 元素定位器 ----------
def product_name(self):
return self.poco("com.example.app:id/product_name")
def product_price(self):
return self.poco("com.example.app:id/product_price")
def add_to_cart_button(self):
return self.poco(text="加入购物车")
def buy_now_button(self):
return self.poco(text="立即购买")
def back_button(self):
return self.poco(desc="返回")
def share_button(self):
return self.poco(desc="分享")
def favorite_button(self):
return self.poco(desc="收藏")
def product_images(self):
return self.poco("com.example.app:id/product_image")
def spec_selection(self):
return self.poco("com.example.app:id/spec_container")
# ---------- 页面操作 ----------
def wait_for_page_load(self):
"""等待商品详情页加载"""
return self.wait_for_element(self.product_name(), timeout=TIMEOUT)
def get_product_name(self):
"""获取商品名称"""
return self.get_text(self.product_name())
def get_product_price(self):
"""获取商品价格"""
return self.get_text(self.product_price())
def add_to_cart(self):
"""
加入购物车
:return: True/False
"""
# 先滚动到加入购物车按钮
self.swipe_screen("up", duration=0.5)
result = self.click_element(self.add_to_cart_button(), msg="加入购物车")
# 验证加入成功
if result:
import time
time.sleep(1)
# 检查是否有"已加入"提示
if self.poco(textMatches=".*(已加入|成功).*").exists():
print("[购物车] 商品已成功加入购物车")
return result
def buy_now(self):
"""立即购买"""
return self.click_element(self.buy_now_button(), msg="立即购买")
def swipe_product_images(self):
"""滑动商品图片"""
images = self.find_element(self.product_images())
if images:
self.poco.swipe([0.8, 0.5], [0.2, 0.5])
def click_favorite(self):
"""点击收藏"""
return self.click_element(self.favorite_button(), msg="收藏")
def go_back(self):
"""返回"""
return self.click_element(self.back_button(), msg="返回")
4.5 购物车页（cart_page.py）
python
# pages/cart_page.py
from common.base_page import BasePage
from config.settings import TIMEOUT
class CartPage(BasePage):
"""购物车页面"""
# ---------- 元素定位器 ----------
def cart_title(self):
return self.poco(text="购物车")
def cart_items(self):
return self.poco("com.example.app:id/cart_item")
def item_checkbox(self, index=0):
return self.cart_items()[index].child("android.widget.CheckBox")
def item_name(self, index=0):
return self.cart_items()[index].child("com.example.app:id/item_name")
def item_price(self, index=0):
return self.cart_items()[index].child("com.example.app:id/item_price")
def item_quantity(self, index=0):
return self.cart_items()[index].child("com.example.app:id/item_quantity")
def item_increase_btn(self, index=0):
return self.cart_items()[index].child(text="+")
def item_decrease_btn(self, index=0):
return self.cart_items()[index].child(text="-")
def item_delete_btn(self, index=0):
return self.cart_items()[index].child(text="删除")
def select_all_checkbox(self):
return self.poco("com.example.app:id/select_all")
def total_price(self):
return self.poco("com.example.app:id/total_price")
def checkout_button(self):
return self.poco(text="去结算")
def empty_cart_text(self):
return self.poco(textMatches=".*(空空|没有商品).*")
def delete_confirm_button(self):
return self.poco(text="确定")
# ---------- 页面操作 ----------
def wait_for_page_load(self):
"""等待购物车页加载"""
return self.wait_for_element(self.cart_title(), timeout=TIMEOUT)
def get_item_count(self):
"""获取购物车商品数量"""
try:
return len(self.cart_items())
except Exception:
return 0
def is_cart_empty(self):
"""判断购物车是否为空"""
return self.is_element_exists(self.empty_cart_text(), timeout=3)
def get_item_names(self):
"""获取所有商品名称"""
count = self.get_item_count()
names = []
for i in range(count):
try:
name = self.get_text(self.item_name(i))
names.append(name)
except Exception:
break
return names
def select_item(self, index=0):
"""勾选第N个商品"""
return self.click_element(self.item_checkbox(index), msg=f"勾选商品{index+1}")
def select_all(self):
"""全选"""
return self.click_element(self.select_all_checkbox(), msg="全选")
def change_quantity(self, index=0, increase=True):
"""修改商品数量"""
if increase:
return self.click_element(self.item_increase_btn(index), msg="增加数量")
else:
return self.click_element(self.item_decrease_btn(index), msg="减少数量")
def get_quantity(self, index=0):
"""获取商品数量"""
return self.get_text(self.item_quantity(index))
def delete_item(self, index=0):
"""删除商品"""
self.click_element(self.item_delete_btn(index), msg=f"删除商品{index+1}")
import time
time.sleep(0.5)
# 确认删除
if self.is_element_exists(self.delete_confirm_button()):
self.click_element(self.delete_confirm_button(), msg="确认删除")
return True
def get_total_price(self):
"""获取总价"""
return self.get_text(self.total_price())
def checkout(self):
"""去结算"""
return self.click_element(self.checkout_button(), msg="去结算")
五、测试数据文件
5.1 testdata/login_data.csv
csv
case_id,case_name,username,password,expected_result,expected_message
TC_LOGIN_001,正常登录-正确账号密码,admin,123456,success,登录成功
TC_LOGIN_002,异常登录-错误密码,admin,wrongpwd,fail,密码错误
TC_LOGIN_003,异常登录-空用户名,,123456,fail,请输入用户名
TC_LOGIN_004,异常登录-空密码,admin,,fail,请输入密码
TC_LOGIN_005,异常登录-不存在的用户,nonexist@test.com,123456,fail,用户不存在
5.2 testdata/search_data.json
json
{
"test_cases": [
{
"case_id": "TC_SEARCH_001",
"case_name": "搜索-正常关键词",
"keyword": "手机",
"expected_min_results": 1,
"verify_contains": true
},
{
"case_id": "TC_SEARCH_002",
"case_name": "搜索-精确匹配",
"keyword": "iPhone 15",
"expected_min_results": 1,
"verify_contains": true
},
{
"case_id": "TC_SEARCH_003",
"case_name": "搜索-无结果关键词",
"keyword": "xyz123不存在",
"expected_min_results": 0,
"verify_contains": false
},
{
"case_id": "TC_SEARCH_004",
"case_name": "搜索-特殊字符",
"keyword": "@#$%",
"expected_min_results": 0,
"verify_contains": false
}
]
}
六、测试用例编写
6.1 登录测试（test_login.py）
python
# testcases/test_login.py
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from common.base_test import BaseTest
from common.utils import read_csv
from pages.login_page import LoginPage
from pages.home_page import HomePage
from config.settings import TESTDATA_DIR
import yaml
class TestLogin(BaseTest):
"""登录模块测试"""
def __init__(self):
# 读取设备配置
with open(os.path.join(os.path.dirname(__file__), "..", "config", "devices.yaml"), "r") as f:
config = yaml.safe_load(f)
super().__init__(device_config=config["devices"]["default"])
def run(self):
"""运行所有登录测试用例"""
self.setup()
try:
login_page = LoginPage(self.poco)
home_page = HomePage(self.poco)
# 用例1: 正常登录
self._test_normal_login(login_page, home_page)
# 用例2: 异常登录-错误密码
self._test_wrong_password(login_page)
# 用例3: 异常登录-空用户名
self._test_empty_username(login_page)
# 用例4: 数据驱动测试
self._test_data_driven_login(login_page, home_page)
finally:
self.teardown()
def _test_normal_login(self, login_page, home_page):
"""TC_LOGIN_001: 正常登录"""
case_name = "TC_LOGIN_001-正常登录"
def test():
# 执行登录
assert login_page.login("admin", "123456"), "登录操作失败"
# 验证登录成功
assert login_page.is_login_success(), "未看到登录成功提示"
# 等待首页加载
assert home_page.wait_for_home_load(), "首页加载失败"
# 验证首页显示
home_page.assert_element_exists(
home_page.home_tab(),
msg="首页Tab应该可见"
)
self.run_test_case(test, case_name)
def _test_wrong_password(self, login_page):
"""TC_LOGIN_002: 错误密码"""
case_name = "TC_LOGIN_002-错误密码"
def test():
login_page.login("admin", "wrongpwd")
assert login_page.is_login_failed(), "应该显示登录失败"
error_text = login_page.get_error_text()
assert error_text, "应该有错误提示信息"
print(f"错误提示: {error_text}")
self.run_test_case(test, case_name)
def _test_empty_username(self, login_page):
"""TC_LOGIN_003: 空用户名"""
case_name = "TC_LOGIN_003-空用户名"
def test():
login_page.login("", "123456")
assert login_page.is_login_failed(), "空用户名应该登录失败"
self.run_test_case(test, case_name)
def _test_data_driven_login(self, login_page, home_page):
"""数据驱动登录测试"""
# 读取CSV数据
csv_path = os.path.join(TESTDATA_DIR, "login_data.csv")
test_data = read_csv(csv_path)
for row in test_data:
case_name = f"{row['case_id']}-{row['case_name']}"
expected = row['expected_result']
def test(data=row):
login_page.login(data['username'], data['password'])
if data['expected_result'] == 'success':
assert login_page.is_login_success(), f"期望成功但失败了"
# 退出登录以便下一个用例
home_page = HomePage(self.poco)
home_page.go_to_profile()
# 这里应该调用退出登录
else:
assert login_page.is_login_failed(), f"期望失败但成功了"
self.run_test_case(test, case_name)
# ========== 直接运行 ==========
if __name__ == "__main__":
test = TestLogin()
test.run()
6.2 搜索测试（test_search.py）
python
# testcases/test_search.py
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from common.base_test import BaseTest
from common.utils import read_json
from pages.home_page import HomePage
from pages.search_page import SearchPage
from pages.product_detail_page import ProductDetailPage
from config.settings import TESTDATA_DIR
import yaml
class TestSearch(BaseTest):
"""搜索模块测试"""
def __init__(self):
with open(os.path.join(os.path.dirname(__file__), "..", "config", "devices.yaml"), "r") as f:
config = yaml.safe_load(f)
super().__init__(device_config=config["devices"]["default"])
def run(self):
"""运行所有搜索测试用例"""
self.setup()
try:
home_page = HomePage(self.poco)
search_page = SearchPage(self.poco)
detail_page = ProductDetailPage(self.poco)
# 等待首页加载
assert home_page.wait_for_home_load(), "首页加载失败"
# 用例1: 正常搜索
self._test_normal_search(home_page, search_page)
# 用例2: 搜索无结果
self._test_no_result_search(home_page, search_page)
# 用例3: 搜索结果验证
self._test_search_result_verify(home_page, search_page)
# 用例4: 点击搜索结果进入详情
self._test_click_search_result(home_page, search_page, detail_page)
# 用例5: 数据驱动搜索
self._test_data_driven_search(home_page, search_page)
finally:
self.teardown()
def _test_normal_search(self, home_page, search_page):
"""TC_SEARCH_001: 正常搜索"""
case_name = "TC_SEARCH_001-正常搜索"
def test():
# 点击搜索框
home_page.click_search()
# 执行搜索
assert search_page.search("手机"), "搜索操作失败"
# 验证有结果
assert search_page.has_results(), "应该有搜索结果"
# 验证结果数量
count = search_page.get_result_count()
assert count > 0, f"搜索结果数量应大于0，实际为{count}"
print(f"找到 {count} 个结果")
self.run_test_case(test, case_name)
def _test_no_result_search(self, home_page, search_page):
"""TC_SEARCH_002: 搜索无结果"""
case_name = "TC_SEARCH_002-搜索无结果"
def test():
home_page.click_search()
search_page.search("xyz不存在的商品123")
# 验证无结果提示
search_page.assert_element_exists(
search_page.no_result_text(),
msg="应显示无结果提示"
)
self.run_test_case(test, case_name)
def _test_search_result_verify(self, home_page, search_page):
"""TC_SEARCH_003: 验证搜索结果相关性"""
case_name = "TC_SEARCH_003-搜索结果相关性验证"
def test():
keyword = "手机"
home_page.click_search()
search_page.search(keyword)
# 验证结果包含关键词
all_contain, failed = search_page.verify_search_results_contain(keyword)
assert all_contain, f"有{len(failed)}个结果不包含'{keyword}'"
self.run_test_case(test, case_name)
def _test_click_search_result(self, home_page, search_page, detail_page):
"""TC_SEARCH_004: 点击搜索结果进入详情"""
case_name = "TC_SEARCH_004-点击搜索结果进入详情"
def test():
home_page.click_search()
search_page.search("手机")
# 获取第一个结果的名称
first_name = search_page.get_text(search_page.result_item_name(0))
print(f"点击商品: {first_name}")
# 点击进入详情
search_page.click_result(0)
# 验证详情页加载
assert detail_page.wait_for_page_load(), "商品详情页加载失败"
# 验证商品名称匹配
detail_name = detail_page.get_product_name()
assert first_name in detail_name, \
f"商品名称不匹配: '{first_name}' vs '{detail_name}'"
self.run_test_case(test, case_name)
def _test_data_driven_search(self, home_page, search_page):
"""数据驱动搜索测试"""
json_path = os.path.join(TESTDATA_DIR, "search_data.json")
test_data = read_json(json_path)
for case in test_data["test_cases"]:
case_name = f"{case['case_id']}-{case['case_name']}"
def test(data=case):
home_page.click_search()
search_page.search(data['keyword'])
# 验证最少结果数
count = search_page.get_result_count()
assert count >= data['expected_min_results'], \
f"结果数 {count} < 期望最小值 {data['expected_min_results']}"
# 验证结果包含关键词
if data.get('verify_contains') and count > 0:
all_contain, failed = search_page.verify_search_results_contain(data['keyword'])
assert all_contain, f"部分结果不包含'{data['keyword']}'"
self.run_test_case(test, case_name)
if __name__ == "__main__":
test = TestSearch()
test.run()
七、运行入口与命令行执行
7.1 主运行入口（run.py）
python
# run.py
"""
Airtest APP自动化测试 - 主运行入口
用法：
python run.py # 运行所有测试
python run.py --module login # 只运行登录测试
python run.py --module search # 只运行搜索测试
python run.py --device phone_01 # 指定设备
python run.py --no-report # 不生成报告
"""
import sys
import os
import argparse
import yaml
# 将项目根目录加入路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from config.settings import GENERATE_REPORT
def parse_args():
"""解析命令行参数"""
parser = argparse.ArgumentParser(description="Airtest APP自动化测试")
parser.add_argument("--module", "-m", choices=["login", "search", "cart", "smoke", "all"],
default="all", help="测试模块")
parser.add_argument("--device", "-d", default="default", help="设备名称")
parser.add_argument("--no-report", action="store_true", help="不生成报告")
return parser.parse_args()
def load_device_config(device_name):
"""加载设备配置"""
config_path = os.path.join(os.path.dirname(__file__), "config", "devices.yaml")
with open(config_path, "r", encoding="utf-8") as f:
config = yaml.safe_load(f)
if device_name not in config["devices"]:
print(f"设备 '{device_name}' 未找到，使用默认设备")
return config["devices"]["default"]
return config["devices"][device_name]
def run_test_module(module_name, device_config):
"""运行指定模块的测试"""
if module_name == "login":
from testcases.test_login import TestLogin
test = TestLogin()
test.run()
elif module_name == "search":
from testcases.test_search import TestSearch
test = TestSearch()
test.run()
elif module_name == "cart":
# from testcases.test_cart import TestCart
# test = TestCart()
# test.run()
print("购物车测试模块待实现")
elif module_name == "smoke":
# 冒烟测试：运行所有模块的核心用例
run_smoke_tests(device_config)
elif module_name == "all":
# 运行所有模块
from testcases.test_login import TestLogin
from testcases.test_search import TestSearch
print("\n" + "=" * 60)
print(" 1/2 运行登录测试")
print("=" * 60)
login_test = TestLogin()
login_test.run()
print("\n" + "=" * 60)
print(" 2/2 运行搜索测试")
print("=" * 60)
search_test = TestSearch()
search_test.run()
def run_smoke_tests(device_config):
"""运行冒烟测试"""
from testcases.test_login import TestLogin
from testcases.test_search import TestSearch
print("\n>>> 运行冒烟测试套件 <<<")
# 只运行核心用例
login_test = TestLogin()
login_test.setup()
try:
from pages.login_page import LoginPage
from pages.home_page import HomePage
login_page = LoginPage(login_test.poco)
home_page = HomePage(login_test.poco)
login_test._test_normal_login(login_page, home_page)
finally:
login_test.teardown()
search_test = TestSearch()
search_test.setup()
try:
from pages.home_page import HomePage
from pages.search_page import SearchPage
home_page = HomePage(search_test.poco)
search_page = SearchPage(search_test.poco)
search_test._test_normal_search(home_page, search_page)
finally:
search_test.teardown()
if __name__ == "__main__":
args = parse_args()
# 如果不生成报告
if args.no_report:
import config.settings as settings
settings.GENERATE_REPORT = False
# 加载设备配置
device_config = load_device_config(args.device)
print(f"设备: {args.device}")
print(f"模块: {args.module}")
print(f"报告: {'开启' if GENERATE_REPORT else '关闭'}")
print()
# 运行测试
run_test_module(args.module, device_config)
7.2 命令行运行示例
bash
# 运行所有测试
python run.py
# 只运行登录模块
python run.py --module login
# 指定设备 + 不生成报告
python run.py --device phone_01 --module search --no-report
# 运行冒烟测试
python run.py --module smoke
# 在Airtest IDE命令行中运行
airtest run testcases/test_login.py --device Android:///emulator-5554
# 生成报告
airtest report testcases/test_login.py
八、测试报告
8.1 报告生成方式
方式一：代码生成（推荐）
python
from airtest.report.report import simple_report, LogToHtml
# 简单报告
simple_report(
__file__, # 脚本路径
logpath=True, # 使用默认日志路径
output="reports/my_report.html" # 输出路径
)
# 详细报告
h = LogToHtml(
script_root=__file__,
log_root="logs/",
export_dir="reports/",
static_root=None, # 静态资源路径
lang="zh", # 中文
plugins=["poco"] # 启用Poco插件
)
h.report(
template_name="log_template.html",
output_file="reports/detailed_report.html"
)
方式二：命令行生成
bash
airtest report testcases/test_login.py \
--log_root logs/ \
--export reports/ \
--lang zh \
--static_root /path/to/static/
8.2 报告内容解读
生成的HTML报告包含：
| 模块 | 内容 |
|---|---|
| 基本信息 | 脚本名称、运行时间、设备信息 |
| 执行概览 | 总步骤数、成功/失败数、通过率 |
| 步骤详情 | 每一步的截图、操作类型、耗时 |
| Poco面板 | Poco操作的UI树截图（需开启插件） |
| 错误信息 | 失败步骤的错误详情和截图 |
| 日志面板 | 完整的运行日志 |
九、持续集成（Jenkins）快速接入
9.1 Jenkins Pipeline 配置
groovy
// Jenkinsfile
pipeline {
agent any
environment {
ANDROID_HOME = '/opt/android-sdk'
PYTHON_PATH = '/usr/local/bin/python3'
}
parameters {
choice(name: 'TEST_MODULE', choices: ['all', 'login', 'search', 'smoke'], description: '测试模块')
choice(name: 'DEVICE', choices: ['emulator', 'phone_01'], description: '测试设备')
}
stages {
stage('Checkout') {
steps {
checkout scm
}
}
stage('Install Dependencies') {
steps {
sh 'pip install -r requirements.txt'
}
}
stage('Connect Device') {
steps {
sh 'adb devices'
sh 'adb -s emulator-5554 wait-for-device'
}
}
stage('Run Tests') {
steps {
sh """
python run.py \
--module ${params.TEST_MODULE} \
--device ${params.DEVICE}
"""
}
}
stage('Publish Report') {
steps {
publishHTML([
reportDir: 'reports',
reportFiles: 'report_*.html',
reportName: 'Airtest测试报告'
])
}
}
}
post {
failure {
emailext(
subject: "自动化测试失败: ${env.JOB_NAME}",
body: "测试失败，请查看报告: ${env.BUILD_URL}",
to: 'test-team@company.com'
)
}
}
}
9.2 定时执行
bash
# crontab 定时任务示例
# 每天早上8点运行冒烟测试
0 8 * * * cd /path/to/ShoppingAppTest && python run.py --module smoke >> logs/cron.log 2>&1
# 每周一早上6点运行全量测试
0 6 * * 1 cd /path/to/ShoppingAppTest && python run.py --module all >> logs/cron_full.log 2>&1
十、常见问题与排坑
10.1 项目组织常见问题
| 问题 | 原因 | 解决 |
|---|---|---|
导入模块报错 ModuleNotFoundError | 路径未加入 sys.path | 在入口文件开头加 sys.path.insert(0, ...) |
.air 目录和 .py 混用 | 用了IDE录制后又手动写 | 统一用 .py 脚本，截图放 tpl/ 目录 |
| 配置硬编码在脚本里 | 没有抽离配置文件 | 用 yaml/json 文件管理配置 |
| 多个用例间状态污染 | 前一个用例影响了后一个 | 每个用例结束后恢复初始状态 |
10.2 运行时常见问题
| 问题 | 解决 |
|---|---|
| Poco初始化失败 | 检查设备是否连接、APP是否支持Poco |
| 元素定位超时 | 增加timeout、检查元素是否被遮挡、尝试相对定位 |
| 报告生成空白 | 检查logpath是否正确、静态资源是否完整 |
| 多设备并行冲突 | 每个进程指定不同的设备串号 |
十一、总结
项目框架速查
技能自检清单
- 能搭建规范的Airtest项目目录结构
- 能用Page Object模式封装页面元素和操作
- 能封装公共基类（BasePage/BaseTest）
- 能处理各类弹窗（更新/权限/活动）
- 能实现数据驱动测试（CSV/JSON/YAML）
- 能编写完整的测试用例（登录/搜索/购物车）
- 能通过命令行运行测试并生成报告
- 能接入Jenkins做持续集成
- 能排查常见的定位失败和运行报错
记忆口诀
项目结构五层分：入口、用例、页面、公共、配置
页面对象三要素：定位器、操作法、断言器
基类封装四件事：找元素、点元素、输文本、等元素
数据驱动三步走：读文件、遍历行、执行用例
测试运行一条龙：连设备、启APP、跑用例、出报告
下一篇将进入跨平台测试——小程序与Web端自动化测试，教你用Airtest搞定微信小程序和H5页面的自动化，敬请期待！
北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台
更多推荐
9
0- 0
已为社区贡献3条内容
所有评论(0)