diff --git a/test/scripts/testModule/test_hdc_forward_ip.py b/test/scripts/testModule/test_hdc_forward_ip.py new file mode 100644 index 0000000000000000000000000000000000000000..2f4b286809c7eec620288b01c874ea925afd2b8f --- /dev/null +++ b/test/scripts/testModule/test_hdc_forward_ip.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# Copyright (C) 2025 Huawei Device Co., Ltd. +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import os +import time +import multiprocessing +import pytest + +from utils import run_command_with_timeout, check_shell, load_gp + + +class TestForwardIpSetting: + + @staticmethod + def new_process_run(cmd): + # 重定向 stdout 和 stderr 到 /dev/null + with open(os.devnull, 'w') as devnull: + old_stdout = os.dup2(devnull.fileno(), 1) # 重定向 stdout + old_stderr = os.dup2(devnull.fileno(), 2) # 重定向 stderr + try: + # 这里是子进程的代码,不会有任何输出到控制台 + check_shell(f'{cmd}') + finally: + # 恢复原始的 stdout 和 stderr + os.dup2(old_stdout, 1) + os.dup2(old_stderr, 2) + + def netstat(self, pattern): + if os.name == 'nt': + result = run_command_with_timeout(f"netstat -ano | findstr {pattern}", 5) + return result + else: + result = run_command_with_timeout(f"netstat -anp | grep {pattern}", 5) + return result + + @pytest.mark.L1 + @pytest.mark.repeat(1) + def test_ip_format_valid(self): + check_shell(f"kill") + time.sleep(1) + assert check_shell(f"-e 123 -m", f"-e content IP incorrect") + assert check_shell(f"-e abd -m", f"-e content IP incorrect") + assert check_shell(f"-e 测试 -m", f"-e content IP incorrect") + assert check_shell(f"-e 12.22.22.22.1 -m", f"-e content IP incorrect") + assert check_shell(f"-e 123.aba -m", f"-e content IP incorrect") + assert check_shell(f"-e 323.21.21.2 -m", f"-e content IP incorrect") + + @pytest.mark.L1 + @pytest.mark.repeat(1) + def test_ip_valid_1(self): + check_shell(f"kill") + time.sleep(1) + + p = multiprocessing.Process(target=self.new_process_run, args=("-e 0.0.0.0 -m",)) + p.start() + time.sleep(1) + + result = self.netstat("127.0.0.1:8710") + assert len(result) > 0 + + assert check_shell(f"fport tcp:9988 tcp:8899", "Forwardport result:OK") + result = self.netstat("0.0.0.0:9988") + assert len(result) > 0 + + assert check_shell(f"fport rm tcp:9988 tcp:8899", "success") + + check_shell(f"kill") + time.sleep(1) + + @pytest.mark.L1 + @pytest.mark.repeat(1) + def test_ip_valid_2(self): + check_shell(f"kill") + time.sleep(1) + + p = multiprocessing.Process(target=self.new_process_run, args=("-e 127.0.0.1 -m",)) + p.start() + time.sleep(1) + + result = self.netstat("127.0.0.1:8710") + assert len(result) > 0 + + assert check_shell(f"fport tcp:9988 tcp:8899", "Forwardport result:OK") + result = self.netstat("127.0.0.1:9988") + assert len(result) > 0 + + assert check_shell(f"fport rm tcp:9988 tcp:8899", "success") + + check_shell(f"kill") + time.sleep(1) + + @pytest.mark.L1 + @pytest.mark.repeat(1) + def test_ip_valid_3(self): + check_shell(f"kill") + time.sleep(1) + + p = multiprocessing.Process(target=self.new_process_run, args=("-m",)) + p.start() + time.sleep(1) + + result = self.netstat("127.0.0.1:8710") + assert len(result) > 0 + + assert check_shell(f"fport tcp:9988 tcp:8899", "Forwardport result:OK") + result = self.netstat("127.0.0.1:9988") + assert len(result) > 0 + + assert check_shell(f"fport rm tcp:9988 tcp:8899", "success") + + check_shell(f"kill") + time.sleep(1) + + + @pytest.mark.L1 + @pytest.mark.repeat(1) + def test_ip_valid_3(self): + check_shell(f"kill") + time.sleep(1) + + p = multiprocessing.Process(target=self.new_process_run, args=("-e 23.168.32.2 -m",)) + p.start() + time.sleep(1) + + result = self.netstat("127.0.0.1:8710") + assert len(result) > 0 + assert check_shell(f"fport tcp:9988 tcp:8899", "failed") + + check_shell(f"kill") + time.sleep(1)