1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# Copyright (C) 2025 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. 15import os 16import time 17import platform 18import pytest 19from utils import GP, check_hdc_cmd, get_shell_result, run_command_with_timeout, check_shell, get_local_path, \ 20 get_remote_path, rmdir, load_gp, check_cmd_block 21 22 23class TestTcpByFport: 24 daemon_port = 58710 25 address = "127.0.0.1" 26 fport_args = f"tcp:{daemon_port} tcp:{daemon_port}" 27 base_file_table = [ 28 ("empty", "it_empty"), 29 ("small", "it_small"), 30 ("medium", "it_medium"), 31 ("word_100M.txt", "it_word_100M"), 32 ("problem_dir", "it_problem_dir") 33 ] 34 35 @classmethod 36 def setup_class(self): 37 assert (check_hdc_cmd(f"tmode port {self.daemon_port}")) 38 time.sleep(3) # sleep 3s to wait for the device to connect channel 39 run_command_with_timeout(f"{GP.hdc_head} wait", 3) # wait 3s for the device to connect channel 40 time.sleep(3) # sleep 3s to wait for the device to connect channel 41 run_command_with_timeout(f"{GP.hdc_head} wait", 3) # wait 3s for the device to connect channel 42 assert check_hdc_cmd(f"shell param get persist.hdc.port", f"{self.daemon_port}") 43 assert check_hdc_cmd(f"fport {self.fport_args}", "Forwardport result:OK") 44 assert check_hdc_cmd(f"fport ls ", self.fport_args) 45 assert check_shell(f"tconn {self.address}:{self.daemon_port}", "Connect OK", head=GP.hdc_exe) 46 time.sleep(3) 47 assert check_shell("list targets", f"{self.address}:{self.daemon_port}", head=GP.hdc_exe) 48 49 @classmethod 50 def teardown_class(self): 51 check_hdc_cmd("shell rm -rf data/local/tmp/it_*") 52 for local_path, _ in self.base_file_table: 53 if os.path.exists(get_local_path(f'{local_path}_fport_recv')): 54 rmdir(get_local_path(f'{local_path}_fport_recv')) 55 assert check_shell(f"tconn {self.address}:{self.daemon_port} -remove", head=GP.hdc_exe) 56 assert not check_shell("list targets", f"{self.address}:{self.daemon_port}", head=GP.hdc_exe) 57 assert check_hdc_cmd(f"fport rm {self.fport_args}", "Remove forward ruler success") 58 assert not check_hdc_cmd(f"fport ls ", self.fport_args) 59 60 @pytest.mark.L1 61 @pytest.mark.parametrize("local_path, remote_path", base_file_table) 62 def test_fport_file(self, local_path, remote_path): 63 tcp_head = f"{GP.hdc_exe} -t {self.address}:{self.daemon_port}" 64 assert check_hdc_cmd(f"file send {get_local_path(local_path)} " 65 f"{get_remote_path(f'{remote_path}_fport')}", head=tcp_head) 66 assert check_hdc_cmd(f"file recv {get_remote_path(f'{remote_path}_fport')} " 67 f"{get_local_path(f'{local_path}_fport_recv')}", head=tcp_head) 68 69 @pytest.mark.L1 70 def test_no_install_with_multi_device(self): 71 package_hap = "AACommand07.hap" 72 app = os.path.join(GP.local_path, package_hap) 73 install_cmd = f"install {app}" 74 patten = f"please confirm a device by help info" 75 assert check_cmd_block(f"{GP.hdc_exe} {install_cmd}", f"{patten}", timeout=3) 76 77 @pytest.mark.L1 78 def test_remote_tcp_connect(self): 79 is_ohos = "Harmony" in platform.system() 80 if not is_ohos: 81 assert True 82 return 83 pid = os.getpid() 84 is_hishell = check_shell(f'ps -efZ | grep {pid}', 'u:r:hishell_hap:s0') 85 check_hdc_cmd("kill") 86 cmd = "-s 0.0.0.0:8710 -m" 87 if is_hishell: 88 assert check_hdc_cmd(cmd, "Program running. Ver:") 89 else: 90 assert check_hdc_cmd(cmd, "[E001105] Unsupport option [s], please try command in HiShell.") 91