• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
127def command_judge(cmd):
128    ret = False
129    cmd_parts = cmd.split()
130    if 'file send' in cmd and cmd[:9] == 'file send' and len(cmd_parts) == 4:
131        ret = True
132    if 'file recv' in cmd and cmd[:9] == 'file recv' and len(cmd_parts) == 4:
133        ret = True
134    if 'install' in cmd and cmd[:7] == 'install' and len(cmd_parts) == 2:
135        ret = True
136    return ret
137
138
139def command_callback(cmd, head, need_del, res=""):
140    cmd_parts = cmd.split()
141    if 'file send' in cmd and cmd[:9] == 'file send' and len(cmd_parts) == 4:
142        if need_del:
143            assert "FileTransfer finish" in res
144        check_file_send(cmd_parts[2], cmd_parts[3], head, need_del)
145    if 'file recv' in cmd and cmd[:9] == 'file recv' and len(cmd_parts) == 4:
146        if need_del:
147            assert "FileTransfer finish" in res
148        check_file_recv(cmd_parts[2], cmd_parts[3], head, need_del)
149    if 'install' in cmd and cmd[:7] == 'install' and len(cmd_parts) == 2:
150        check_install(head, res)
151    if cmd == 'smode':
152        time.sleep(4)
153        check_root(head)
154    if cmd == 'smode -r':
155        time.sleep(4)
156        check_user(head)
157    if cmd == 'target boot':
158        time.sleep(35)
159
160
161def check_file_send(local_file, remote_file, head, need_del):
162    ret = get_win_file_type(local_file)
163    file_type = ""
164    if 'file' in ret:
165        file_type = '-f'
166    if "dir" in ret:
167        file_type = '-d'
168    res = run_command_stdout("{}{}{}{}{}".format("shell \"[ ", file_type, " ", remote_file,
169                             " ] && echo yes || echo no\""), head)
170    assert 'yes' in res
171    if file_type == '-d' or 'empty.txt' in local_file:
172        rm_send_file(remote_file, head, need_del)
173        return
174    local_md5 = get_win_md5(local_file)
175    remote_md5 = get_md5(remote_file, head)
176    rm_send_file(remote_file, head, need_del)
177    assert local_md5 == remote_md5
178    print("check_file_send success ", res)
179
180
181def rm_send_file(file_path, head, need_del):
182    if need_del:
183        run_command("{}{}{}{}".format("shell \"rm -rf ", file_path, "\"", head))
184
185
186def check_file_recv(remote_file, local_file, head, need_del):
187    if 'dir' in get_win_file_type(local_file) or 'empty.txt' in local_file:
188        rm_recv_file(local_file, need_del)
189        return
190    res = run_command_stdout("{}{}{}".format("attrib ", local_file, ""))
191    local_md5 = get_win_md5(local_file)
192    remote_md5 = get_md5(remote_file, head)
193    assert '-' not in res
194    if local_md5 != remote_md5:
195        print("check_file_recv fail ", remote_file, local_file)
196    assert local_md5 == remote_md5
197    rm_recv_file(local_file, need_del)
198    print("check_file_recv success ", res)
199
200
201def get_win_file_type(file_path):
202    ret = run_command_stdout("{}{}{}".format("if exist ", file_path, " echo yes"), '')
203    assert "yes" in ret
204    res = run_command_stdout("{}{}{}".format("dir/ad ", file_path, " >nul 2>nul && echo dir || echo file"), '')
205    return res
206
207
208def rm_recv_file(file_path, need_del):
209    if need_del:
210        res = get_win_file_type(file_path)
211        if "dir" in res:
212            run_command("{}{}".format("rmdir /s/q ", file_path), "")
213        if "file" in res:
214            run_command("{}{}".format("del ", file_path), "")
215
216
217def check_install(head, res):
218    print("check_install")
219    print(res)
220    if "msg:install bundle successfully." not in res:
221        print("install msg error")
222        assert False
223    res = run_command_stdout("shell \"bm dump -a\"", head)
224    if HAP_ONE['PACKAGE_NAME'] in res:
225        print("check_install success ", HAP_ONE['PACKAGE_NAME'])
226        assert True
227    else:
228        assert False
229
230
231def check_root(head):
232    res = run_command_stdout("shell \"whoami\"", head)
233    print("check_root res: ", res)
234    assert 'root' in res
235
236
237def check_user(head):
238    res = run_command_stdout("shell \"whoami\"", head)
239    print("check_user res: ", res)
240    assert 'shell' in res
241    return 'shell' in res
242
243
244def get_devs(head=NORMAL_HEAD):
245    res = run_command_stdout(BASIC_COMMANDS['component'][0], head)
246    devs = res.split()
247    print(devs)
248    return devs
249
250
251def tmode_to_tcp():
252    run_command(TCP_CONN['tmode'])
253    res = run_command_stdout(TCP_CONN['tconn'])
254    print(res)
255    if "Connect OK" in res:
256        return True
257    return False
258
259
260def remote_server_start(server_head):
261    global IP
262    cmd = "{}{}".format(server_head, "-m")
263    print(cmd)
264    os.popen(cmd)
265
266
267def run_command(cmd, head=NORMAL_HEAD, need_del=True, need_callback=True):
268    command = "{}{}".format(head, cmd)
269    if head != '':
270        print(command)
271
272    subprocess.Popen(command,
273                     shell=True).communicate()
274    if need_callback:
275        command_callback(cmd, head, need_del)
276
277
278def run_command_stdout(cmd, head=NORMAL_HEAD, need_del=True, need_callback=True):
279    command = "{}{}".format(head, cmd)
280    if head != '' and 'echo' not in cmd:
281        print(command)
282    dec = "UTF-8"
283    if head == '':
284        dec = 'gbk'
285    res = subprocess.Popen(command,
286                           shell=True,
287                           stdout=subprocess.PIPE).communicate()
288    res = res[0].decode(dec)
289    if need_callback:
290        command_callback(cmd, head, need_del, res)
291    return res
292
293
294def run_commands(cmds, head=NORMAL_HEAD, need_del=True):
295    for command in cmds:
296        run_command(command, head, need_del)
297
298
299def get_basic_commands():
300    commands = []
301    for tasks in BASIC_COMMANDS.values():
302        commands += tasks
303    return commands
304
305
306def run_global_cmd():
307    global_commands = EXTRA_COMMANDS['global']
308    run_commands(global_commands)
309
310
311def run_split_commands(commands, head=NORMAL_HEAD):
312    for command in commands:
313        if command_judge(command):
314            run_command_stdout(command, head, False)
315        else:
316            run_command(command, head, False)
317
318
319def run_device_cmd(head=NORMAL_HEAD):
320    run_command_stdout("{}{}{}{}".format("file send ", PATH['dir_send']['local'], " ", PATH['dir_recv']['remote']),
321                       head, False)
322    for smd in EXTRA_COMMANDS['smode']:
323        run_command(smd)
324        commands = get_basic_commands() + EXTRA_COMMANDS['boot']
325        if smd == "smode -r":
326            for item in commands:
327                if "{}{}".format(REMOTE_PATH, "log") in item:
328                    print(item)
329                    commands.remove(item)
330        run_split_commands(commands, head)
331
332
333def extract_ip():
334    global IP
335    st = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
336    try:
337        st.connect(('10.255.255.255', 1))
338        IP = st.getsockname()[0]
339    except Exception:
340        IP = ""
341    finally:
342        st.close()
343    return IP
344
345
346def mix_path(path, i):
347    ret = path.find('.')
348    if ret > 0:
349        return "{}{}{}".format(path[:ret], i, path[ret:])
350    return path
351
352
353def file_send(send_path, recv_path, i, wait_time=0):
354    time.sleep(wait_time)
355    res = run_command_stdout("{}{}{}{}".format("file send ", send_path, " ", mix_path(recv_path, str(i))))
356    print(res)
357
358
359def file_recv(remote_path, recv_path, i, wait_time=0):
360    time.sleep(wait_time)
361    res = run_command_stdout("{}{}{}{}".format("file recv ", remote_path, " ", mix_path(recv_path, str(i))))
362    print(res)
363
364
365def get_win_md5(file_path):
366    real_md5 = "win_md5"
367    send_md5 = run_command_stdout("{}{}{}".format("certutil -hashfile ", os.path.abspath(file_path), " MD5"),
368                                  "").split()
369    for x in send_md5:
370        if len(x) == 32:
371            real_md5 = x
372    return real_md5
373
374
375def get_md5(file_path, head=NORMAL_HEAD):
376    md5 = run_command_stdout("{}{}{}".format("shell \"md5sum ", file_path, "\""), head)
377    return md5.split()[0]
378
379
380class TestCommands:
381    def test_file_cmd(self):
382        print("HDC TEST: start test_file_cmd\n")
383        for item in TEST_FILES:
384            send_file = TEST_FILES[item]['send_file']
385            send_file_one = TEST_FILES[item]['send_file_one']
386            send_file_two = TEST_FILES[item]['send_file_two']
387
388            recv_file = TEST_FILES[item]['recv_file']
389            recv_file_one = TEST_FILES[item]['recv_file_one']
390            recv_file_two = TEST_FILES[item]['recv_file_two']
391
392            run_command_stdout("{}{}{}{}".format("file send ", os.path.abspath(PATH['file_empty']['local']), ' ',
393                               PATH['file_empty']['remote']), NORMAL_HEAD, False, False)
394            run_command_stdout("{}{}{}{}".format("file send ", os.path.abspath(send_file), ' ', recv_file),
395                               NORMAL_HEAD, False)
396            for i in range(10):
397                wait_time = random.uniform(0, 1)
398                if i == 0:
399                    wait_time = 0
400                print("{}{}".format("HDC TEST: start test_file_cmd \n", str(i)))
401                send_one = threading.Thread(target=file_send, args=(os.path.abspath(send_file), send_file_one, i))
402                send_two = threading.Thread(target=file_send,
403                                            args=(os.path.abspath(send_file), send_file_two, i, wait_time))
404                recv_one = threading.Thread(target=file_recv, args=(recv_file, os.path.abspath(recv_file_one), i))
405                recv_two = threading.Thread(target=file_recv,
406                                            args=(recv_file, os.path.abspath(recv_file_two), i, wait_time))
407
408                send_one.start()
409                send_two.start()
410                recv_one.start()
411                recv_two.start()
412                send_one.join()
413                send_two.join()
414                recv_one.join()
415                recv_two.join()
416
417    def test_global_server(self):
418        print("HDC TEST: start test_global_server_cmd\n")
419        run_global_cmd()
420
421    def test_device_cmd(self):
422        print("HDC TEST: start test_device_cmd\n")
423        devs = get_devs()
424        if len(devs) == 1:
425            run_device_cmd()
426        if len(devs) > 1:
427            for dev in devs:
428                run_device_cmd("{}{}{}{}".format(NORMAL_HEAD, EXTRA_COMMANDS['choose'], dev, " "))
429
430    def test_tcp_mode(self):
431        print("HDC TEST: start test_tcp_mode\n")
432        extract_ip()
433        global IP
434        if len(IP) == 0 or TCP_CFG['ip'] == '':
435            print("请连接热点 配置TCP_CFG")
436            return
437        if tmode_to_tcp():
438            commands = get_basic_commands()
439            run_commands(commands, NORMAL_HEAD, False)
440
441    def test_remote_server(self):
442        time.sleep(10)
443        print("HDC TEST: start test_remote_server\n")
444        extract_ip()
445        ret = run_command_stdout("kill")
446        if 'finish' not in ret:
447            print('test_remote_server kill server failed')
448            return
449        global IP
450        server_head = "{}{}{}{}{}".format(NORMAL_HEAD, "-s ", IP, ":", "8710 ")
451        thread_start = threading.Thread(target=remote_server_start(server_head))
452        thread_start.start()
453        thread_start.join()
454        devs = get_devs(server_head)
455        for dev in devs:
456            head = "{}{}{}{}".format(server_head, EXTRA_COMMANDS['choose'], dev, " ")
457            run_command_stdout("{}{}{}{}".format("file send ", PATH['dir_send']['local'], " ",
458                                                 PATH['dir_recv']['remote']), head, False)
459            threading.Thread(target=run_split_commands(get_basic_commands(), head)).start()
460        run_command("kill", server_head)
461