提交UI自动化插件
This commit is contained in:
张家训
@@ -0,0 +1,3 @@
|
||||
airtest
|
||||
pocoui
|
||||
pyyaml
|
||||
@@ -0,0 +1,289 @@
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
def emit(event_type, level="INFO", message="", **extra):
|
||||
payload = {"type": event_type, "level": level, "message": message}
|
||||
payload.update(extra)
|
||||
print(json.dumps(payload, ensure_ascii=False), flush=True)
|
||||
|
||||
|
||||
def load_yaml(path):
|
||||
try:
|
||||
import yaml
|
||||
except Exception as exc:
|
||||
emit("log", "ERROR", "缺少 PyYAML，请执行: python -m pip install -U pyyaml")
|
||||
raise exc
|
||||
with open(path, "r", encoding="utf-8") as fh:
|
||||
return yaml.safe_load(fh) or {}
|
||||
|
||||
|
||||
def ensure_airtest():
|
||||
try:
|
||||
from airtest.core.api import (
|
||||
G,
|
||||
Template,
|
||||
auto_setup,
|
||||
connect_device,
|
||||
exists,
|
||||
keyevent,
|
||||
sleep,
|
||||
snapshot,
|
||||
start_app,
|
||||
stop_app,
|
||||
swipe,
|
||||
text,
|
||||
touch,
|
||||
wait,
|
||||
)
|
||||
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
|
||||
return {
|
||||
"G": G,
|
||||
"Template": Template,
|
||||
"auto_setup": auto_setup,
|
||||
"connect_device": connect_device,
|
||||
"exists": exists,
|
||||
"keyevent": keyevent,
|
||||
"sleep": sleep,
|
||||
"snapshot": snapshot,
|
||||
"start_app": start_app,
|
||||
"stop_app": stop_app,
|
||||
"swipe": swipe,
|
||||
"text": text,
|
||||
"touch": touch,
|
||||
"wait": wait,
|
||||
"AndroidUiautomationPoco": AndroidUiautomationPoco,
|
||||
}
|
||||
except Exception as exc:
|
||||
emit("log", "ERROR", "缺少 Airtest/Poco，请执行: python -m pip install -U airtest pocoui")
|
||||
raise exc
|
||||
|
||||
|
||||
def run_python_case(case_path, args):
|
||||
command = [sys.executable, case_path]
|
||||
emit("log", "INFO", "运行 Python 用例: " + " ".join(command))
|
||||
proc = subprocess.run(command, cwd=os.path.dirname(case_path) or ".", text=True)
|
||||
return proc.returncode
|
||||
|
||||
|
||||
def run_air_case(case_path, args):
|
||||
command = [sys.executable, "-m", "airtest", "run", case_path, "--device", args.device]
|
||||
emit("log", "INFO", "运行 Airtest 用例: " + " ".join(command))
|
||||
proc = subprocess.run(command, cwd=os.path.dirname(case_path) or ".", text=True)
|
||||
return proc.returncode
|
||||
|
||||
|
||||
def connect(api, args):
|
||||
os.makedirs(args.report, exist_ok=True)
|
||||
api["auto_setup"](__file__, logdir=args.report, devices=[args.device])
|
||||
poco = None
|
||||
try:
|
||||
poco = api["AndroidUiautomationPoco"](use_airtest_input=True, screenshot_each_action=False)
|
||||
emit("log", "INFO", "Poco 初始化成功")
|
||||
except Exception as exc:
|
||||
emit("log", "WARN", "Poco 初始化失败，将仅使用 Airtest/坐标/图片定位: " + str(exc))
|
||||
return poco
|
||||
|
||||
|
||||
def find_poco(poco, query):
|
||||
if poco is None:
|
||||
raise RuntimeError("Poco 未初始化")
|
||||
query = query or {}
|
||||
if "name" in query and query["name"]:
|
||||
return poco(query["name"])
|
||||
if "text" in query and query["text"]:
|
||||
return poco(text=query["text"])
|
||||
if "type" in query and query["type"]:
|
||||
return poco(type=query["type"])
|
||||
raise RuntimeError("Poco 查询条件为空")
|
||||
|
||||
|
||||
def target_path(case_dir, path):
|
||||
if os.path.isabs(path):
|
||||
return path
|
||||
return os.path.join(case_dir, path)
|
||||
|
||||
|
||||
def tap_target(api, poco, target, case_dir):
|
||||
target = target or {}
|
||||
target_type = target.get("type") or target.get("preferred") or "coord"
|
||||
if target_type == "poco":
|
||||
node = find_poco(poco, target.get("query") or target.get("poco"))
|
||||
node.wait_for_appearance(timeout=target.get("timeout", 10))
|
||||
node.click()
|
||||
return
|
||||
if target_type == "image":
|
||||
template = api["Template"](target_path(case_dir, target["path"]), threshold=target.get("threshold", 0.85))
|
||||
api["touch"](template)
|
||||
return
|
||||
x = int(target.get("x", 0))
|
||||
y = int(target.get("y", 0))
|
||||
screen_width = int(target.get("screenWidth", 0) or 0)
|
||||
screen_height = int(target.get("screenHeight", 0) or 0)
|
||||
try:
|
||||
current_width, current_height = api["G"].DEVICE.get_current_resolution()
|
||||
if screen_width > 0 and screen_height > 0:
|
||||
x = round(x / screen_width * current_width)
|
||||
y = round(y / screen_height * current_height)
|
||||
except Exception:
|
||||
pass
|
||||
api["touch"]((x, y))
|
||||
|
||||
|
||||
def wait_target(api, poco, target, case_dir, timeout):
|
||||
target = target or {}
|
||||
target_type = target.get("type") or target.get("preferred") or "coord"
|
||||
if target_type == "poco":
|
||||
node = find_poco(poco, target.get("query") or target.get("poco"))
|
||||
node.wait_for_appearance(timeout=timeout)
|
||||
return
|
||||
if target_type == "image":
|
||||
api["wait"](api["Template"](target_path(case_dir, target["path"]), threshold=target.get("threshold", 0.85)), timeout=timeout)
|
||||
return
|
||||
time.sleep(min(timeout, 1))
|
||||
|
||||
|
||||
def assert_exists(api, poco, target, case_dir):
|
||||
target = target or {}
|
||||
target_type = target.get("type") or target.get("preferred") or "coord"
|
||||
if target_type == "poco":
|
||||
node = find_poco(poco, target.get("query") or target.get("poco"))
|
||||
if not node.exists():
|
||||
raise AssertionError("Poco 元素不存在: " + json.dumps(target, ensure_ascii=False))
|
||||
return
|
||||
if target_type == "image":
|
||||
if not api["exists"](api["Template"](target_path(case_dir, target["path"]), threshold=target.get("threshold", 0.85))):
|
||||
raise AssertionError("图片不存在: " + target["path"])
|
||||
|
||||
|
||||
def run_yaml_case(case_path, args):
|
||||
api = ensure_airtest()
|
||||
case = load_yaml(case_path)
|
||||
case_dir = os.path.dirname(case_path)
|
||||
poco = connect(api, args)
|
||||
package_name = args.package or (case.get("app") or {}).get("package")
|
||||
activity_name = args.activity or (case.get("app") or {}).get("activity")
|
||||
settings = case.get("settings") or {}
|
||||
default_timeout = int(args.timeout or settings.get("timeout") or 20)
|
||||
steps = case.get("steps") or []
|
||||
result = {
|
||||
"caseName": case.get("name") or os.path.basename(case_path),
|
||||
"device": args.device,
|
||||
"package": package_name,
|
||||
"success": False,
|
||||
"startTime": datetime.now().isoformat(),
|
||||
"steps": [],
|
||||
}
|
||||
|
||||
if package_name:
|
||||
emit("log", "INFO", "启动 APP: " + package_name)
|
||||
if activity_name:
|
||||
api["start_app"](package_name, activity=activity_name)
|
||||
else:
|
||||
api["start_app"](package_name)
|
||||
|
||||
for idx, step in enumerate(steps, start=1):
|
||||
name = step.get("name") or f"step-{idx}"
|
||||
action = step.get("action")
|
||||
started = time.time()
|
||||
emit("step_start", "INFO", f"{idx}. {name}", index=idx, name=name)
|
||||
record = {"index": idx, "name": name, "action": action, "status": "PASS"}
|
||||
try:
|
||||
if action in ("tap", "click"):
|
||||
tap_target(api, poco, step.get("target"), case_dir)
|
||||
elif action in ("input", "input_text"):
|
||||
target = step.get("target")
|
||||
if target:
|
||||
tap_target(api, poco, target, case_dir)
|
||||
api["text"](str(step.get("text", "")))
|
||||
elif action == "wait":
|
||||
wait_target(api, poco, step.get("target"), case_dir, int(step.get("timeout") or default_timeout))
|
||||
elif action == "sleep":
|
||||
api["sleep"](float(step.get("seconds") or step.get("duration") or 1))
|
||||
elif action == "keyevent":
|
||||
api["keyevent"](step.get("key") or step.get("value") or "BACK")
|
||||
elif action == "swipe":
|
||||
start = step.get("start") or {}
|
||||
end = step.get("end") or {}
|
||||
api["swipe"]((int(start.get("x", 0)), int(start.get("y", 0))), (int(end.get("x", 0)), int(end.get("y", 0))),
|
||||
duration=float(step.get("durationMs", 500)) / 1000)
|
||||
elif action == "screenshot":
|
||||
api["snapshot"](filename=os.path.join(args.report, step.get("path") or f"step_{idx}.png"))
|
||||
elif action == "assert_exists":
|
||||
assert_exists(api, poco, step.get("target"), case_dir)
|
||||
elif action == "start_app":
|
||||
api["start_app"](step.get("package") or package_name)
|
||||
elif action == "stop_app":
|
||||
api["stop_app"](step.get("package") or package_name)
|
||||
else:
|
||||
raise RuntimeError("不支持的 action: " + str(action))
|
||||
|
||||
wait_ms = int(step.get("waitAfterMs") or 0)
|
||||
if wait_ms > 0:
|
||||
time.sleep(wait_ms / 1000)
|
||||
record["elapsedMs"] = round((time.time() - started) * 1000)
|
||||
emit("step_pass", "INFO", f"{idx}. {name} 通过", index=idx, elapsedMs=record["elapsedMs"])
|
||||
except Exception as exc:
|
||||
record["status"] = "FAIL"
|
||||
record["message"] = str(exc)
|
||||
if args.screenshot_on_fail:
|
||||
screenshot = os.path.join(args.report, f"failed_step_{idx}.png")
|
||||
try:
|
||||
api["snapshot"](filename=screenshot)
|
||||
record["screenshot"] = screenshot
|
||||
emit("screenshot", "INFO", screenshot, path=screenshot)
|
||||
except Exception:
|
||||
pass
|
||||
result["steps"].append(record)
|
||||
result["endTime"] = datetime.now().isoformat()
|
||||
result["success"] = False
|
||||
write_result(args.report, result)
|
||||
emit("step_fail", "ERROR", f"{idx}. {name} 失败: {exc}", index=idx)
|
||||
return 1
|
||||
result["steps"].append(record)
|
||||
|
||||
result["success"] = True
|
||||
result["endTime"] = datetime.now().isoformat()
|
||||
write_result(args.report, result)
|
||||
emit("done", "INFO", "用例运行成功", reportPath=os.path.join(args.report, "result.json"))
|
||||
return 0
|
||||
|
||||
|
||||
def write_result(report_dir, result):
|
||||
os.makedirs(report_dir, exist_ok=True)
|
||||
with open(os.path.join(report_dir, "result.json"), "w", encoding="utf-8") as fh:
|
||||
json.dump(result, fh, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--device", default="Android:///")
|
||||
parser.add_argument("--package", default="")
|
||||
parser.add_argument("--activity", default="")
|
||||
parser.add_argument("--case", required=True)
|
||||
parser.add_argument("--report", required=True)
|
||||
parser.add_argument("--timeout", type=int, default=20)
|
||||
parser.add_argument("--screenshot-on-fail", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
case_path = os.path.abspath(args.case)
|
||||
try:
|
||||
if case_path.lower().endswith(".py"):
|
||||
code = run_python_case(case_path, args)
|
||||
elif case_path.lower().endswith(".air"):
|
||||
code = run_air_case(case_path, args)
|
||||
else:
|
||||
code = run_yaml_case(case_path, args)
|
||||
return code
|
||||
except Exception as exc:
|
||||
emit("done", "ERROR", "运行异常: " + str(exc))
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,24 @@
|
||||
name: sample-login-flow
|
||||
app:
|
||||
package: com.example.app
|
||||
activity:
|
||||
device:
|
||||
connect: Android:///
|
||||
settings:
|
||||
timeout: 20
|
||||
screenshotOnFail: true
|
||||
steps:
|
||||
- name: wait-a-moment
|
||||
action: sleep
|
||||
seconds: 1
|
||||
|
||||
- name: tap-example-coordinate
|
||||
action: tap
|
||||
target:
|
||||
type: coord
|
||||
preferred: coord
|
||||
x: 540
|
||||
y: 1600
|
||||
screenWidth: 1080
|
||||
screenHeight: 2400
|
||||
waitAfterMs: 500
|
||||
Reference in New Issue
Block a user