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 pytest 18from utils import GP, check_hdc_cmd, get_shell_result, run_command_with_timeout, check_shell, get_local_path, \ 19 get_remote_path, rmdir, load_gp 20 21 22class TestTcpByFport: 23 daemon_port = 58710 24 address = "127.0.0.1" 25 fport_args = f"tcp:{daemon_port} tcp:{daemon_port}" 26 base_file_table = [ 27 ("empty", "it_empty"), 28 ("small", "it_small"), 29 ("medium", "it_medium"), 30 ("word_100M.txt", "it_word_100M"), 31 ("problem_dir", "it_problem_dir") 32 ] 33 34 @classmethod 35 def setup_class(self): 36 assert (check_hdc_cmd(f"tmode port {self.daemon_port}")) 37 time.sleep(3) # sleep 3s to wait for the device to connect channel 38 run_command_with_timeout(f"{GP.hdc_head} wait", 3) # wait 3s for the device to connect channel 39 time.sleep(3) # sleep 3s to wait for the device to connect channel 40 run_command_with_timeout(f"{GP.hdc_head} wait", 3) # wait 3s for the device to connect channel 41 assert check_hdc_cmd(f"shell param get persist.hdc.port", f"{self.daemon_port}") 42 assert check_hdc_cmd(f"fport {self.fport_args}", "Forwardport result:OK") 43 assert check_hdc_cmd(f"fport ls ", self.fport_args) 44 assert check_shell(f"tconn {self.address}:{self.daemon_port}", "Connect OK", head=GP.hdc_exe) 45 time.sleep(3) 46 assert check_shell("list targets", f"{self.address}:{self.daemon_port}", head=GP.hdc_exe) 47 48 @classmethod 49 def teardown_class(self): 50 check_hdc_cmd("shell rm -rf data/local/tmp/it_*") 51 for local_path, _ in self.base_file_table: 52 if os.path.exists(get_local_path(f'{local_path}_fport_recv')): 53 rmdir(get_local_path(f'{local_path}_fport_recv')) 54 assert check_shell(f"tconn {self.address}:{self.daemon_port} -remove", head=GP.hdc_exe) 55 assert not check_shell("list targets", f"{self.address}:{self.daemon_port}", head=GP.hdc_exe) 56 assert check_hdc_cmd(f"fport rm {self.fport_args}", "Remove forward ruler success") 57 assert not check_hdc_cmd(f"fport ls ", self.fport_args) 58 59 @pytest.mark.L1 60 @pytest.mark.parametrize("local_path, remote_path", base_file_table) 61 def test_fport_file(self, local_path, remote_path): 62 tcp_head = f"{GP.hdc_exe} -t {self.address}:{self.daemon_port}" 63 assert check_hdc_cmd(f"file send {get_local_path(local_path)} " 64 f"{get_remote_path(f'{remote_path}_fport')}", head=tcp_head) 65 assert check_hdc_cmd(f"file recv {get_remote_path(f'{remote_path}_fport')} " 66 f"{get_local_path(f'{local_path}_fport_recv')}", head=tcp_head)