• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 pytest
16from utils import GP, check_shell_any_device, check_hdc_cmd, get_shell_result, load_gp
17
18
19class TestXportCommand:
20    @pytest.mark.L0
21    def test_fport_cmd_output(self):
22        local_port = 18070
23        remote_port = 11080
24        fport_arg = f"tcp:{local_port} tcp:{remote_port}"
25        assert check_hdc_cmd(f"fport {fport_arg}", "Forwardport result:OK")
26        assert check_shell_any_device(f"netstat -ano", "LISTENING", False)[0]
27        assert check_shell_any_device(f"netstat -ano", f"{local_port}", False)[0]
28        assert check_hdc_cmd(f"fport ls", fport_arg)
29        assert check_hdc_cmd(f"fport rm {fport_arg}", "success")
30
31    @pytest.mark.L0
32    def test_rport_cmd_output(self):
33        local_port = 17090
34        remote_port = 11080
35        rport_arg = f"tcp:{local_port} tcp:{remote_port}"
36        assert check_hdc_cmd(f"rport {rport_arg}", "Forwardport result:OK")
37        netstat_line = get_shell_result(f'shell "netstat -anp | grep {local_port}"')
38        assert "LISTEN" in netstat_line
39        assert "hdcd" in netstat_line
40        fport_list = get_shell_result(f"fport ls")
41        assert "Reverse" in fport_list
42        assert rport_arg in fport_list
43        assert check_hdc_cmd(f"fport rm {rport_arg}", "success")
44
45    @pytest.mark.L0
46    def test_fport_cmd(self):
47        fport_list = []
48        rport_list = []
49        start_port = 10000
50        end_port = 10020
51        for i in range(start_port, end_port):
52            fport = f"tcp:{i+100} tcp:{i+200}"
53            rport = f"tcp:{i+300} tcp:{i+400}"
54            localabs = f"tcp:{i+500} localabstract:{f'helloworld.com.app.{i+600}'}"
55            fport_list.append(fport)
56            rport_list.append(rport)
57            fport_list.append(localabs)
58
59        for fport in fport_list:
60            assert check_hdc_cmd(f"fport {fport}", "Forwardport result:OK")
61            assert check_hdc_cmd(f"fport {fport}", "TCP Port listen failed at")
62            assert check_hdc_cmd("fport ls", fport)
63
64        for fport in fport_list:
65            assert check_hdc_cmd(f"fport rm {fport}", "success")
66            assert not check_hdc_cmd("fport ls", fport)
67
68        for rport in rport_list:
69            assert check_hdc_cmd(f"rport {rport}", "Forwardport result:OK")
70            assert check_hdc_cmd(f"rport {rport}", "TCP Port listen failed at")
71            assert check_hdc_cmd("rport ls", rport) or check_hdc_cmd("fport ls", rport)
72
73        for rport in rport_list:
74            assert check_hdc_cmd(f"fport rm {rport}", "success")
75            assert not check_hdc_cmd("rport ls", fport) and not check_hdc_cmd("fport ls", fport)
76
77        task_str1 = "tcp:33333 tcp:33333"
78        assert check_hdc_cmd(f"fport {task_str1}", "Forwardport result:OK")
79        assert check_hdc_cmd(f"fport rm {task_str1}", "success")
80        assert check_hdc_cmd(f"fport {task_str1}", "Forwardport result:OK")
81        assert check_hdc_cmd(f"fport rm {task_str1}", "success")
82
83        task_str2 = "tcp:44444 tcp:44444"
84        assert check_hdc_cmd(f"rport {task_str2}", "Forwardport result:OK")
85        assert check_hdc_cmd(f"fport rm {task_str2}", "success")
86        assert check_hdc_cmd(f"rport {task_str2}", "Forwardport result:OK")
87        assert check_hdc_cmd(f"fport rm {task_str2}", "success")
88