1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# Copyright (C) 2021 Huawei Device Co., Ltd. 4# Licensed under the Apache License, Version 2.0 (the "License"); 5# you may not use this file except in compliance with the License. 6# You may obtain a copy of the License at 7# 8# http://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, 12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13# See the License for the specific language governing permissions and 14# limitations under the License. 15# 1.运行环境 16# pycharm 17# Python 3.10 18# 测试框架 pytest (pycharm setting -> Tools -> Testing -> pytest) 19# 2.配置测试脚本 20# 需配置hdc环境变量-NORMAL_HEAD为可执行程序hdc 21# 配置TEST_FILE_PATH为测试用文件所在路径,取resource测试文件放入该路径下 22# 测试tcp模式连接设备，需设备，PC连同一网络(手机热点)：配置TCP_CFG['ip']为设备的ip 23import os 24import random 25import subprocess 26import socket 27import time 28import threading 29 30TCP_CFG = { 31 'ip': '', 32 'port': "8710", 33} 34 35NORMAL_HEAD = "hdc " 36REMOTE_PATH = "/data/local/tmp/" 37IP = "" 38TEST_FILE_PATH = "{}{}".format(os.path.abspath("D:/Download/sdk/"), '\\') 39 40HAP_ONE = { 41 'HAP_NAME': "entry-default-signed-debug.hap", 42 'PACKAGE_NAME': "com.hmos.diagnosis", 43} 44 45TCP_CONN = { 46 'bright_screen': "shell \"power-shell setmode 602\"", 47 'tmode': "{}{}".format("tmode port ", TCP_CFG['port']), 48 'tconn': "{}{}{}{}".format("tconn ", TCP_CFG['ip'], ":", TCP_CFG['port']), 49} 50 51PATH = { 52 'file_send': { 53 'local': "{}{}".format(TEST_FILE_PATH, "hdc.log"), 54 'remote': "{}{}".format(REMOTE_PATH, "hdc.log") 55 }, 56 'dir_send': { 57 'local': "{}{}".format(TEST_FILE_PATH, "log"), 58 'remote': "{}{}".format(REMOTE_PATH, "log") 59 }, 60 'file_recv': { 61 'remote': "{}{}".format(REMOTE_PATH, "hdc.log"), 62 'local': "{}{}".format(TEST_FILE_PATH, "dev_data") 63 }, 64 'dir_recv': { 65 'remote': "{}{}".format(REMOTE_PATH, "log"), 66 'local': "{}{}".format(TEST_FILE_PATH, "hdc\\log") 67 }, 68 'file_empty': { 69 'local': "{}{}".format(TEST_FILE_PATH, "empty.txt"), 70 'remote': "{}{}".format(REMOTE_PATH, "empty.txt") 71 } 72} 73 74EXTRA_COMMANDS = { 75 'global': ["kill -r", "kill", "-l5 start", "start -r", "-v", "version", "checkserver"], 76 'smode': ["smode -r", "smode"], 77 'boot': ["target boot"], 78 'choose': "-t " 79} 80 81BASIC_COMMANDS = { 82 'shell': [ 83 "shell \"ls\"" 84 ], 85 'component': [ 86 "list targets", "list targets -v", "target mount" 87 ], 88 'file_task': [ 89 "{}{}{}{}".format("file send ", PATH['file_send']['local'], " ", PATH['file_send']['remote']), 90 "{}{}{}{}".format("file send ", PATH['file_empty']['local'], " ", PATH['file_empty']['remote']), 91 "{}{}{}{}".format("file send ", PATH['dir_send']['local'], " ", PATH['dir_send']['remote']), 92 "{}{}{}{}".format("file recv ", PATH['file_recv']['remote'], " ", PATH['file_recv']['local']), 93 "{}{}{}{}".format("file recv ", PATH['file_empty']['remote'], " ", PATH['file_empty']['local']), 94 "{}{}{}{}".format("file recv ", PATH['dir_recv']['remote'], " ", PATH['dir_recv']['local']) 95 ], 96 'fport_task': [ 97 "fport tcp:1234 tcp:1080", 98 "rport tcp:13608 localabstract:8888BananaBanana", 99 "fport ls", 100 "fport rm tcp:1234 tcp:1080" 101 ], 102 'install_task': [ 103 "{}{}{}".format("install ", TEST_FILE_PATH, HAP_ONE['HAP_NAME']), 104 "{}{}".format("uninstall ", HAP_ONE['PACKAGE_NAME']) 105 ] 106} 107 108TEST_FILES = { 109 'one': { 110 'send_file': "{}{}".format(TEST_FILE_PATH, "100M.txt"), 111 'send_file_one': "{}{}".format(REMOTE_PATH, "a100M.txt"), 112 'send_file_two': "{}{}".format(REMOTE_PATH, "c100M.txt"), 113 'recv_file': "{}{}".format(REMOTE_PATH, "recv100M.txt"), 114 'recv_file_one': "{}{}".format(TEST_FILE_PATH, "recv100M.txt"), 115 'recv_file_two': "{}{}".format(TEST_FILE_PATH, "recv200M.txt"), 116 }, 117 'two': { 118 'send_file': "{}{}".format(TEST_FILE_PATH, "hdc_file.log"), 119 'send_file_one': "{}{}".format(REMOTE_PATH, "send_one.log"), 120 'send_file_two': "{}{}".format(REMOTE_PATH, "send_two.log"), 121 'recv_file': "{}{}".format(REMOTE_PATH, "hdcd.log"), 122 'recv_file_one': "{}{}".format(TEST_FILE_PATH, "recv_one.log"), 123 'recv_file_two': "{}{}".format(TEST_FILE_PATH, "recv_two.log"), 124 } 125} 126 127 128def command_judge(cmd): 129 ret = False 130 cmd_parts = cmd.split() 131 if 'file send' in cmd and cmd[:9] == 'file send' and len(cmd_parts) == 4: 132 ret = True 133 if 'file recv' in cmd and cmd[:9] == 'file recv' and len(cmd_parts) == 4: 134 ret = True 135 if 'install' in cmd and cmd[:7] == 'install' and len(cmd_parts) == 2: 136 ret = True 137 return ret 138 139 140def command_callback(cmd, head, need_del, res=""): 141 cmd_parts = cmd.split() 142 if 'file send' in cmd and cmd[:9] == 'file send' and len(cmd_parts) == 4: 143 if need_del: 144 assert "FileTransfer finish" in res 145 check_file_send(cmd_parts[2], cmd_parts[3], head, need_del) 146 if 'file recv' in cmd and cmd[:9] == 'file recv' and len(cmd_parts) == 4: 147 if need_del: 148 assert "FileTransfer finish" in res 149 check_file_recv(cmd_parts[2], cmd_parts[3], head, need_del) 150 if 'install' in cmd and cmd[:7] == 'install' and len(cmd_parts) == 2: 151 check_install(head, res) 152 if cmd == 'smode': 153 time.sleep(4) 154 check_root(head) 155 if cmd == 'smode -r': 156 time.sleep(4) 157 check_user(head) 158 if cmd == 'target boot': 159 time.sleep(35) 160 161 162def check_file_send(local_file, remote_file, head, need_del): 163 ret = get_win_file_type(local_file) 164 file_type = "" 165 if 'file' in ret: 166 file_type = '-f' 167 if "dir" in ret: 168 file_type = '-d' 169 res = run_command_stdout("{}{}{}{}{}".format("shell \"[ ", file_type, " ", remote_file, 170 " ] && echo yes || echo no\""), head) 171 assert 'yes' in res 172 if file_type == '-d' or 'empty.txt' in local_file: 173 rm_send_file(remote_file, head, need_del) 174 return 175 local_md5 = get_win_md5(local_file) 176 remote_md5 = get_md5(remote_file, head) 177 rm_send_file(remote_file, head, need_del) 178 assert local_md5 == remote_md5 179 print("check_file_send success ", res) 180 181 182def rm_send_file(file_path, head, need_del): 183 if need_del: 184 run_command("{}{}{}{}".format("shell \"rm -rf ", file_path, "\"", head)) 185 186 187def check_file_recv(remote_file, local_file, head, need_del): 188 if 'dir' in get_win_file_type(local_file) or 'empty.txt' in local_file: 189 rm_recv_file(local_file, need_del) 190 return 191 res = run_command_stdout("{}{}{}".format("attrib ", local_file, "")) 192 local_md5 = get_win_md5(local_file) 193 remote_md5 = get_md5(remote_file, head) 194 assert '-' not in res 195 if local_md5 != remote_md5: 196 print("check_file_recv fail ", remote_file, local_file) 197 assert local_md5 == remote_md5 198 rm_recv_file(local_file, need_del) 199 print("check_file_recv success ", res) 200 201 202def get_win_file_type(file_path): 203 ret = run_command_stdout("{}{}{}".format("if exist ", file_path, " echo yes"), '') 204 assert "yes" in ret 205 res = run_command_stdout("{}{}{}".format("dir/ad ", file_path, " >nul 2>nul && echo dir || echo file"), '') 206 return res 207 208 209def rm_recv_file(file_path, need_del): 210 if need_del: 211 res = get_win_file_type(file_path) 212 if "dir" in res: 213 run_command("{}{}".format("rmdir /s/q ", file_path), "") 214 if "file" in res: 215 run_command("{}{}".format("del ", file_path), "") 216 217 218def check_install(head, res): 219 print("check_install") 220 print(res) 221 if "msg:install bundle successfully." not in res: 222 print("install msg error") 223 assert False 224 res = run_command_stdout("shell \"bm dump -a\"", head) 225 if HAP_ONE['PACKAGE_NAME'] in res: 226 print("check_install success ", HAP_ONE['PACKAGE_NAME']) 227 assert True 228 else: 229 assert False 230 231 232def check_root(head): 233 res = run_command_stdout("shell \"whoami\"", head) 234 print("check_root res: ", res) 235 assert 'root' in res 236 237 238def check_user(head): 239 res = run_command_stdout("shell \"whoami\"", head) 240 print("check_user res: ", res) 241 assert 'shell' in res 242 return 'shell' in res 243 244 245def get_devs(head=NORMAL_HEAD): 246 res = run_command_stdout(BASIC_COMMANDS['component'][0], head) 247 devs = res.split() 248 print(devs) 249 return devs 250 251 252def tmode_to_tcp(): 253 run_command(TCP_CONN['tmode']) 254 res = run_command_stdout(TCP_CONN['tconn']) 255 print(res) 256 if "Connect OK" in res: 257 return True 258 return False 259 260 261def remote_server_start(server_head): 262 global IP 263 cmd = "{}{}".format(server_head, "-m") 264 print(cmd) 265 os.popen(cmd) 266 267 268def run_command(cmd, head=NORMAL_HEAD, need_del=True, need_callback=True): 269 command = "{}{}".format(head, cmd) 270 if head != '': 271 print(command) 272 273 subprocess.Popen(command, 274 shell=True).communicate() 275 if need_callback: 276 command_callback(cmd, head, need_del) 277 278 279def run_command_stdout(cmd, head=NORMAL_HEAD, need_del=True, need_callback=True): 280 command = "{}{}".format(head, cmd) 281 if head != '' and 'echo' not in cmd: 282 print(command) 283 dec = "UTF-8" 284 if head == '': 285 dec = 'gbk' 286 res = subprocess.Popen(command, 287 shell=True, 288 stdout=subprocess.PIPE).communicate() 289 res = res[0].decode(dec) 290 if need_callback: 291 command_callback(cmd, head, need_del, res) 292 return res 293 294 295def run_commands(cmds, head=NORMAL_HEAD, need_del=True): 296 for command in cmds: 297 run_command(command, head, need_del) 298 299 300def get_basic_commands(): 301 commands = [] 302 for tasks in BASIC_COMMANDS.values(): 303 commands += tasks 304 return commands 305 306 307def run_global_cmd(): 308 global_commands = EXTRA_COMMANDS['global'] 309 run_commands(global_commands) 310 311 312def run_split_commands(commands, head=NORMAL_HEAD): 313 for command in commands: 314 if command_judge(command): 315 run_command_stdout(command, head, False) 316 else: 317 run_command(command, head, False) 318 319 320def run_device_cmd(head=NORMAL_HEAD): 321 run_command_stdout("{}{}{}{}".format("file send ", PATH['dir_send']['local'], " ", PATH['dir_recv']['remote']), 322 head, False) 323 for smd in EXTRA_COMMANDS['smode']: 324 run_command(smd) 325 commands = get_basic_commands() + EXTRA_COMMANDS['boot'] 326 if smd == "smode -r": 327 for item in commands: 328 if "{}{}".format(REMOTE_PATH, "log") in item: 329 print(item) 330 commands.remove(item) 331 run_split_commands(commands, head) 332 333 334def extract_ip(): 335 global IP 336 st = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 337 try: 338 st.connect(('10.255.255.255', 1)) 339 IP = st.getsockname()[0] 340 except Exception: 341 IP = "" 342 finally: 343 st.close() 344 return IP 345 346 347def mix_path(path, i): 348 ret = path.find('.') 349 if ret > 0: 350 return "{}{}{}".format(path[:ret], i, path[ret:]) 351 return path 352 353 354def file_send(send_path, recv_path, i, wait_time=0): 355 time.sleep(wait_time) 356 res = run_command_stdout("{}{}{}{}".format("file send ", send_path, " ", mix_path(recv_path, str(i)))) 357 print(res) 358 359 360def file_recv(remote_path, recv_path, i, wait_time=0): 361 time.sleep(wait_time) 362 res = run_command_stdout("{}{}{}{}".format("file recv ", remote_path, " ", mix_path(recv_path, str(i)))) 363 print(res) 364 365 366def get_win_md5(file_path): 367 real_md5 = "win_md5" 368 send_md5 = run_command_stdout("{}{}{}".format("certutil -hashfile ", os.path.abspath(file_path), " MD5"), 369 "").split() 370 for x in send_md5: 371 if len(x) == 32: 372 real_md5 = x 373 return real_md5 374 375 376def get_md5(file_path, head=NORMAL_HEAD): 377 md5 = run_command_stdout("{}{}{}".format("shell \"md5sum ", file_path, "\""), head) 378 return md5.split()[0] 379 380 381class TestCommands: 382 def test_file_cmd(self): 383 print("HDC TEST: start test_file_cmd\n") 384 for item in TEST_FILES: 385 send_file = TEST_FILES[item]['send_file'] 386 send_file_one = TEST_FILES[item]['send_file_one'] 387 send_file_two = TEST_FILES[item]['send_file_two'] 388 389 recv_file = TEST_FILES[item]['recv_file'] 390 recv_file_one = TEST_FILES[item]['recv_file_one'] 391 recv_file_two = TEST_FILES[item]['recv_file_two'] 392 393 run_command_stdout("{}{}{}{}".format("file send ", os.path.abspath(PATH['file_empty']['local']), ' ', 394 PATH['file_empty']['remote']), NORMAL_HEAD, False, False) 395 run_command_stdout("{}{}{}{}".format("file send ", os.path.abspath(send_file), ' ', recv_file), 396 NORMAL_HEAD, False) 397 for i in range(10): 398 wait_time = random.uniform(0, 1) 399 if i == 0: 400 wait_time = 0 401 print("{}{}".format("HDC TEST: start test_file_cmd \n", str(i))) 402 send_one = threading.Thread(target=file_send, args=(os.path.abspath(send_file), send_file_one, i)) 403 send_two = threading.Thread(target=file_send, 404 args=(os.path.abspath(send_file), send_file_two, i, wait_time)) 405 recv_one = threading.Thread(target=file_recv, args=(recv_file, os.path.abspath(recv_file_one), i)) 406 recv_two = threading.Thread(target=file_recv, 407 args=(recv_file, os.path.abspath(recv_file_two), i, wait_time)) 408 409 send_one.start() 410 send_two.start() 411 recv_one.start() 412 recv_two.start() 413 send_one.join() 414 send_two.join() 415 recv_one.join() 416 recv_two.join() 417 418 def test_global_server(self): 419 print("HDC TEST: start test_global_server_cmd\n") 420 run_global_cmd() 421 422 def test_device_cmd(self): 423 print("HDC TEST: start test_device_cmd\n") 424 devs = get_devs() 425 if len(devs) == 1: 426 run_device_cmd() 427 if len(devs) > 1: 428 for dev in devs: 429 run_device_cmd("{}{}{}{}".format(NORMAL_HEAD, EXTRA_COMMANDS['choose'], dev, " ")) 430 431 def test_tcp_mode(self): 432 print("HDC TEST: start test_tcp_mode\n") 433 extract_ip() 434 global IP 435 if len(IP) == 0 or TCP_CFG['ip'] == '': 436 print("请连接热点 配置TCP_CFG") 437 return 438 if tmode_to_tcp(): 439 commands = get_basic_commands() 440 run_commands(commands, NORMAL_HEAD, False) 441 442 def test_remote_server(self): 443 time.sleep(10) 444 print("HDC TEST: start test_remote_server\n") 445 extract_ip() 446 ret = run_command_stdout("kill") 447 if 'finish' not in ret: 448 print('test_remote_server kill server failed') 449 return 450 global IP 451 server_head = "{}{}{}{}{}".format(NORMAL_HEAD, "-s ", IP, ":", "8710 ") 452 thread_start = threading.Thread(target=remote_server_start(server_head)) 453 thread_start.start() 454 thread_start.join() 455 devs = get_devs(server_head) 456 for dev in devs: 457 head = "{}{}{}{}".format(server_head, EXTRA_COMMANDS['choose'], dev, " ") 458 run_command_stdout("{}{}{}{}".format("file send ", PATH['dir_send']['local'], " ", 459 PATH['dir_recv']['remote']), head, False) 460 threading.Thread(target=run_split_commands(get_basic_commands(), head)).start() 461 run_command("kill", server_head) 462