#!/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 logging

import multiprocessing

import platform

import pytest

import tempfile



from utils import GP, check_shell, run_command_with_timeout, get_cmd_block_output, get_shell_result, get_end_symbol, load_gp



logger = logging.getLogger(__name__)





class TestSessionLifeCycle:



    def server_run_method(self, cmd):

        get_cmd_block_output(cmd, timeout=30)



    def remove(self, source_line, target_list):

        for item in target_list:

            if source_line.find(item) >= 0:

                target_list.remove(item)

                return (True, target_list)

        return (False, target_list)



    @pytest.mark.L0

    def test_session_lifecycle_by_reboot(self):

        check_shell(f"kill")

        time.sleep(3)



        cmd = f"hdc -m"

        p = multiprocessing.Process(target=self.server_run_method, args=(cmd,))

        p.start()



        run_command_with_timeout(f"{GP.hdc_head} wait", 20)



        time.sleep(3)

        devices = get_shell_result(f"list targets").split(get_end_symbol())

        target_lines = list()

        for item in devices:

            if len(item) > 0:

                connect_key = f"{item[:3]}******{item[-3:]}(L:{len(item)})"

                line = f"connectKey:{connect_key} connType:0 connect state:1 faultInfo:"

                target_lines.append(line)

                line = f"connectKey:{connect_key} connType:0 connect state:0 faultInfo:LIBUSB_TRANSFER_"

                target_lines.append(line)

        time.sleep(3)

        check_shell(f"shell reboot")

        time.sleep(5)

        check_shell(f"kill")

        is_ohos = "Harmony" in platform.system()

        # 检查serverOutput是否包含上述字符串

        if not is_ohos:

            tmp_path = tempfile.gettempdir()

        else:

            tmp_path = os.path.expanduser("~")

        hdc_log_path = f"{tmp_path}{os.sep}hdc.log"

        with open(hdc_log_path, 'r') as file:

            for line in file.readlines():

                (_ret, _target_lines1) = self.remove(line, target_lines)

                target_lines = _target_lines1

        p.join()

        assert len(target_lines) == 0



    @pytest.mark.L0

    def test_session_lifecycle_of_multi_tcp(self):

        check_shell(f"kill")

        time.sleep(3)

        p = multiprocessing.Process(target=self.server_run_method, args=(f"hdc -m",))

        p.start()



        run_command_with_timeout(f"{GP.hdc_head} wait", 20)

        fport_tcp_port = 27772

        fport_tcp_host_port = 20001

        assert check_shell(f"tmode port {fport_tcp_port}", "Set device run mode successful")

        run_command_with_timeout(f"{GP.hdc_head} wait", 20)

        for i in range(4):

            assert check_shell(f"fport tcp:{fport_tcp_host_port + i} tcp:{fport_tcp_port}", "Forwardport result:OK")

        time.sleep(3)

        for i in range(4):

            assert check_shell(f"tconn 127.0.0.1:{fport_tcp_host_port + i}", "Connect OK")

        devices = get_shell_result(f"list targets").split(get_end_symbol())

        target_lines = list()

        for item in devices:

            if len(item) > 0 and item.find(".") >= 0:

                connect_key = f"{item[:3]}******{item[-3:]}(L:{len(item)})"

                line = f"connectKey:{connect_key} connType:1 connect state:1 faultInfo:"

                target_lines.append(line)

                line = f"connectKey:{connect_key} connType:1 connect state:0 faultInfo:end of file"

                if item == f"127.0.0.1:{fport_tcp_host_port + 3}":

                    line = line + " commandCount:3"

                target_lines.append(line)



        time.sleep(1)

        item = f"127.0.0.1:{fport_tcp_host_port + 3}"

        connect_key = f"{item[:3]}******{item[-3:]}(L:{len(item)})"

        ls_result_line = f"connectKey:{connect_key} command flag:1001 command result:1 command take time:"

        for i in range(3):

            target_lines.append(ls_result_line)

            check_shell(f"-t 127.0.0.1:{fport_tcp_host_port + 3} shell ls")



        for i in range(4):

            assert check_shell(f"fport rm tcp:{fport_tcp_host_port + i} tcp:{fport_tcp_port}",

                               "Remove forward ruler success")

        time.sleep(5)

        check_shell(f"kill")

        is_ohos = "Harmony" in platform.system()

        # 检查serverOutput是否包含上述字符串

        if not is_ohos:

            tmp_path = tempfile.gettempdir()

        else:

            tmp_path = os.path.expanduser("~")

        hdc_log_path = f"{tmp_path}{os.sep}hdc.log"

        with open(hdc_log_path, 'r') as file:

            for line in file.readlines():

                (_ret, _target_lines1) = self.remove(line, target_lines)

                target_lines = _target_lines1



        p.join()

        assert len(target_lines) == 0