Android Uiautomator2 Python Wrapper自定义工具类开发：扩展框架功能的方法

【免费下载链接】uiautomator2 Android Uiautomator2 Python Wrapper 项目地址: https://gitcode.com/gh_mirrors/ui/uiautomator2

1. 框架扩展需求与实现路径

Android自动化测试中，原生框架功能往往难以满足复杂业务场景需求。Uiautomator2 Python Wrapper作为轻量级封装库，提供了基础的设备控制能力，但在实际测试开发中仍需针对特定场景构建定制化工具。本文将系统讲解通过自定义工具类扩展框架功能的完整方案，包括功能抽象、接口设计、实现模式及最佳实践。

1.1 扩展必要性分析

原生框架局限	自定义扩展方案	典型应用场景
缺乏业务流程封装	页面元素封装类	登录/支付等通用流程
图像识别能力弱	图像处理工具类	验证码识别/图标匹配
性能数据采集缺失	性能监控工具类	FPS/内存占用统计
测试报告不直观	报告生成工具类	可视化测试结果

1.2 扩展架构设计

2. 核心扩展技术与实现方法

2.1 基础工具类开发规范

自定义工具类应遵循以下设计原则：

• 单一职责：每个工具类专注解决一类问题
• 依赖注入：通过构造函数传入Device实例
• 异常封装：统一异常处理机制
• 可测试性：预留mock接口

基础模板代码：

from uiautomator2 import Device
from uiautomator2.exceptions import UiObjectNotFoundError

class BaseExtension:
    def __init__(self, device: Device):
        self._device = device
        self._logger = logging.getLogger(self.__class__.__name__)
        
    def _get_device(self) -> Device:
        """获取设备实例，提供权限控制"""
        if not hasattr(self, '_device'):
            raise RuntimeError("Device instance not initialized")
        return self._device
        
    def _safe_click(self, selector: dict, timeout: int = 3) -> bool:
        """安全点击封装，包含重试机制"""
        try:
            self._device(**selector).click(timeout=timeout)
            return True
        except UiObjectNotFoundError:
            self._logger.warning(f"Element not found: {selector}")
            return False
        except Exception as e:
            self._logger.error(f"Click failed: {str(e)}")
            return False

2.2 功能扩展实现模式

2.2.1 页面元素操作封装

针对电商应用商品列表的封装示例：

class ShoppingUtils(BaseExtension):
    def __init__(self, device: Device):
        super().__init__(device)
        self._item_selector = {"resourceId": "com.example:id/product_item"}
        self._add_cart_btn = {"resourceId": "com.example:id/add_cart"}
        
    def get_product_list(self) -> list:
        """获取商品列表元素"""
        items = self._device(**self._item_selector).all()
        return [self._parse_item(item) for item in items]
        
    def _parse_item(self, item) -> dict:
        """解析商品元素信息"""
        return {
            "title": item.child(text="").get_text(),
            "price": item.child(resourceId="com.example:id/price").get_text(),
            "stock": int(item.child(resourceId="com.example:id/stock").get_text().replace("库存:", "")),
            "element": item
        }
        
    def add_to_cart_by_title(self, title: str) -> bool:
        """根据标题添加商品到购物车"""
        for item in self.get_product_list():
            if title in item["title"]:
                item["element"].click()
                self._device(**self._add_cart_btn).click(timeout=5)
                return self._check_cart_success()
        return False
        
    def _check_cart_success(self) -> bool:
        """验证添加购物车成功"""
        return self._device(text="添加成功").exists(timeout=3)

2.2.2 图像识别功能扩展

基于OpenCV实现图像匹配工具类：

import cv2
import numpy as np
from PIL import Image

class ImageProcessor(BaseExtension):
    def __init__(self, device: Device):
        super().__init__(device)
        self._screen_width, self._screen_height = device.window_size()
        
    def _get_screenshot(self) -> np.ndarray:
        """获取设备截图并转为OpenCV格式"""
        screenshot = self._device.screenshot(format='opencv')
        return cv2.cvtColor(np.array(screenshot), cv2.COLOR_RGB2BGR)
        
    def match_template(self, template_path: str, threshold: float = 0.8) -> dict:
        """模板匹配，返回匹配位置"""
        # 读取模板图像
        template = cv2.imread(template_path)
        if template is None:
            raise FileNotFoundError(f"Template file not found: {template_path}")
            
        # 获取截图并匹配
        screenshot = self._get_screenshot()
        result = cv2.matchTemplate(screenshot, template, cv2.TM_CCOEFF_NORMED)
        locations = np.where(result >= threshold)
        
        # 处理匹配结果
        h, w = template.shape[:2]
        matches = []
        for pt in zip(*locations[::-1]):
            matches.append({
                "x": int(pt[0] + w/2),
                "y": int(pt[1] + h/2),
                "confidence": float(result[pt[1], pt[0]]),
                "rect": (pt[0], pt[1], w, h)
            })
            
        return {"count": len(matches), "matches": matches}
        
    def click_template_match(self, template_path: str, threshold: float = 0.8) -> bool:
        """点击模板匹配到的第一个位置"""
        matches = self.match_template(template_path, threshold)
        if matches["count"] == 0:
            return False
            
        match = matches["matches"][0]
        self._device.click(match["x"], match["y"])
        return True

2.2.3 性能监控工具实现

利用adb shell命令封装性能数据采集工具：

import time
import threading
from collections import defaultdict

class PerformanceMonitor(BaseExtension):
    def __init__(self, device: Device):
        super().__init__(device)
        self._monitoring = False
        self._data = defaultdict(list)
        self._thread = None
        self._package = None
        
    def start_monitor(self, package_name: str, interval: float = 1.0):
        """开始性能监控"""
        self._package = package_name
        self._monitoring = True
        self._interval = interval
        self._thread = threading.Thread(target=self._collect_data)
        self._thread.daemon = True
        self._thread.start()
        
    def _collect_data(self):
        """后台采集性能数据"""
        while self._monitoring:
            timestamp = time.time()
            self._data["fps"].append({
                "time": timestamp,
                "value": self._get_fps()
            })
            self._data["memory"].append({
                "time": timestamp,
                "value": self._get_memory_usage()
            })
            time.sleep(self._interval)
            
    def _get_fps(self) -> float:
        """获取当前FPS"""
        try:
            result = self._device.shell(f"dumpsys gfxinfo {self._package}")
            # 解析FPS数据（实际实现需复杂解析逻辑）
            return float(result.split("Frames rendered:")[1].split()[0]) / 1000
        except Exception as e:
            self._logger.error(f"FPS collection failed: {e}")
            return 0.0
            
    def _get_memory_usage(self) -> dict:
        """获取内存使用情况"""
        try:
            result = self._device.shell(f"dumpsys meminfo {self._package}")
            # 解析内存数据
            return {
                "total": int(result.split("TOTAL")[1].split()[0]),
                "java": int(result.split("Java Heap")[1].split()[0]),
                "native": int(result.split("Native Heap")[1].split()[0])
            }
        except Exception as e:
            self._logger.error(f"Memory collection failed: {e}")
            return {}
            
    def stop_monitor(self) -> dict:
        """停止监控并返回数据"""
        self._monitoring = False
        if self._thread:
            self._thread.join()
        return dict(self._data)
        
    def generate_report(self, output_path: str):
        """生成性能报告"""
        import matplotlib.pyplot as plt
        
        # 绘制FPS曲线
        plt.figure(figsize=(12, 6))
        fps_data = self._data["fps"]
        plt.plot([d["time"] for d in fps_data], [d["value"] for d in fps_data])
        plt.title("FPS Monitoring")
        plt.savefig(f"{output_path}/fps_chart.png")
        plt.close()
        
        # 保存原始数据
        with open(f"{output_path}/performance_data.json", "w") as f:
            json.dump(self._data, f, indent=2)

3. 高级扩展技术与最佳实践

3.1 事件监听机制实现

通过装饰器模式扩展设备事件监听能力：

class EventListener:
    def __init__(self):
        self._listeners = defaultdict(list)
        
    def register(self, event_type: str, callback):
        """注册事件回调"""
        self._listeners[event_type].append(callback)
        
    def unregister(self, event_type: str, callback):
        """注销事件回调"""
        if event_type in self._listeners:
            self._listeners[event_type].remove(callback)
            
    def trigger(self, event_type: str, data: dict):
        """触发事件"""
        for callback in self._listeners.get(event_type, []):
            try:
                callback(data)
            except Exception as e:
                print(f"Event callback failed: {e}")

# 使用示例
class ExtendedDevice(Device):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.event_listener = EventListener()
        self._patch_click()
        
    def _patch_click(self):
        """Patch click method to trigger events"""
        original_click = self.click
        
        def wrapped_click(x, y):
            result = original_click(x, y)
            self.event_listener.trigger("click", {
                "x": x, "y": y, "success": result, "time": time.time()
            })
            return result
            
        self.click = wrapped_click

# 注册事件示例
def click_logger(event):
    print(f"Clicked at ({event['x']}, {event['y']}) - Success: {event['success']}")

device = ExtendedDevice()
device.event_listener.register("click", click_logger)

3.2 多设备协同工具类

实现多设备同步操作的工具类：

class MultiDeviceManager:
    def __init__(self):
        self.devices = {}
        
    def add_device(self, device_id: str, device: Device):
        """添加设备到管理器"""
        self.devices[device_id] = device
        
    def execute_all(self, func, *args, **kwargs) -> dict:
        """在所有设备上执行指定函数"""
        results = {}
        threads = []
        
        def run_on_device(device_id, device):
            try:
                results[device_id] = {
                    "success": True,
                    "result": func(device, *args, **kwargs)
                }
            except Exception as e:
                results[device_id] = {
                    "success": False,
                    "error": str(e)
                }
                
        # 创建并启动线程
        for device_id, device in self.devices.items():
            thread = threading.Thread(
                target=run_on_device,
                args=(device_id, device)
            )
            threads.append(thread)
            thread.start()
            
        # 等待所有线程完成
        for thread in threads:
            thread.join()
            
        return results
        
    def sync_click(self, x: int, y: int) -> dict:
        """在所有设备同步点击指定坐标"""
        return self.execute_all(lambda d: d.click(x, y))
        
    def broadcast_message(self, message: str):
        """向所有设备发送广播消息"""
        return self.execute_all(
            lambda d: d.make_toast(message)
        )

4. 扩展类集成与测试方法

4.1 集成流程与版本控制

4.2 单元测试示例

import unittest
from unittest.mock import Mock, patch
from uiautomator2 import Device

class TestBusinessUtils(unittest.TestCase):
    def setUp(self):
        # 创建模拟设备
        self.mock_device = Mock(spec=Device)
        self.mock_device.window_size.return_value = (1080, 2340)
        self.mock_device.exists.return_value = True
        
        # 创建工具类实例
        self.utils = BusinessUtils(self.mock_device)
        
    def test_login_success(self):
        # 设置模拟返回值
        self.mock_device().click.return_value = True
        
        # 执行测试
        result = self.utils.login("test", "pass")
        
        # 验证结果
        self.assertTrue(result)
        self.mock_device(text="用户名").set_text.assert_called_with("test")
        self.mock_device(text="密码").set_text.assert_called_with("pass")
        self.mock_device(text="登录").click.assert_called_once()
        
    def test_scroll_to_text(self):
        # 设置模拟返回值
        self.mock_device().exists.side_effect = [False, False, True]
        
        # 执行测试
        result = self.utils.scroll_to_text("目标文本")
        
        # 验证结果
        self.assertTrue(result)
        self.assertEqual(self.mock_device.swipe.call_count, 2)  # 滚动两次

5. 实战案例：电商应用测试扩展包

5.1 完整项目结构

custom_extensions/
├── __init__.py
├── base.py           # 基础扩展类
├── business_utils.py # 业务工具类
├── image_utils.py    # 图像处理工具类
├── perf_utils.py     # 性能监控工具类
├── report_utils.py   # 报告生成工具类
└── tests/            # 单元测试目录
    ├── __init__.py
    ├── test_business.py
    ├── test_image.py
    └── test_perf.py

5.2 初始化与使用示例

import uiautomator2 as u2
from custom_extensions.business_utils import BusinessUtils
from custom_extensions.perf_utils import PerformanceMonitor

# 初始化设备与工具类
device = u2.connect()
business = BusinessUtils(device)
perf_monitor = PerformanceMonitor(device)

# 执行测试流程
business.login("test_user", "test_pass")
business.navigate_to("商品列表")

# 启动性能监控
perf_monitor.start_monitor("com.example.shop")

# 执行核心业务操作
for item in ["商品A", "商品B", "商品C"]:
    business.search(item)
    business.add_to_cart()
    business.back()

# 停止监控并生成报告
perf_data = perf_monitor.stop_monitor()
perf_monitor.generate_report("./performance_report")

# 完成购买流程
business.go_to_cart()
business.checkout()
business.pay()

# 验证结果
assert business.is_order_success(), "订单提交失败"

6. 扩展开发最佳实践与常见问题

6.1 性能优化技巧

1.** 元素缓存机制 **：避免重复查找相同元素

def _get_cart_button(self):
    if not hasattr(self, '_cart_btn'):
        self._cart_btn = self._device(resourceId="com.example:id/cart")
    return self._cart_btn

2.** 异步操作处理 **：使用线程池处理耗时任务

from concurrent.futures import ThreadPoolExecutor

class AsyncImageProcessor(ImageProcessor):
    def __init__(self, device):
        super().__init__(device)
        self.executor = ThreadPoolExecutor(max_workers=2)
        
    def async_match_template(self, template_path, callback):
        """异步执行模板匹配"""
        future = self.executor.submit(
            self.match_template, template_path
        )
        future.add_done_callback(lambda f: callback(f.result()))

6.2 常见问题解决方案

问题场景	解决方案	代码示例
元素定位不稳定	实现重试装饰器	@retry(max_attempts=3, delay=1)
不同分辨率适配	百分比坐标转换	x = int(x_percent * screen_width)
网络请求超时	超时控制与重试	try/except + time.sleep()
测试数据管理	配置文件读取	config = yaml.safe_load(open("config.yaml"))

6.3 版本兼容处理

def _get_android_version(self):
    """获取Android系统版本"""
    return int(self._device.info.get("android_version", "0").split(".")[0])

def swipe(self, direction):
    """根据系统版本选择不同滑动实现"""
    if self._get_android_version() >= 10:
        return self._swipe_android10(direction)
    else:
        return self._swipe_legacy(direction)

7. 总结与扩展方向

自定义工具类开发是提升Uiautomator2框架实用性的关键手段，通过合理的抽象设计与实现模式，可以显著扩展框架能力以适应复杂测试场景。本文介绍的基础封装、图像识别、性能监控等扩展方向，为Android自动化测试提供了更强大的技术支撑。

未来扩展方向： 1.** AI辅助测试 ：集成图像识别与自然语言处理 2. 云测试集成 ：开发云端设备管理扩展 3. 测试用例管理 ：实现测试用例的自动生成与优化 4. 跨平台支持 **：扩展至iOS平台的自动化测试

通过持续优化工具类设计与实现，结合实际项目需求不断迭代，可以构建一套完整的自动化测试基础设施，大幅提升测试效率与覆盖率。

【免费下载链接】uiautomator2 Android Uiautomator2 Python Wrapper 项目地址: https://gitcode.com/gh_mirrors/ui/uiautomator2