1#
2# Copyright (c) 2024 Huawei Device Co., Ltd.
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14#
15
16import os
17import socket
18import time
19import subprocess
20import sys
21import unittest
22
23
24class JsTestBase(unittest.TestCase):
25    hdctool = None
26    tcp_client_socket = None
27    HOST = '127.0.0.1'
28    PORT = 9999
29    BUFSIZ = 1024
30    ADDRESS = (HOST, PORT)
31
32    def get_hdctool(self):
33        self.hdctool = os.environ.get("HDCTool", "hdc")
34
35    def connect_client_socket(self, dbg_name):
36        self.tcp_client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
37        subprocess.run(
38            [
39                self.hdctool,
40                "fport",
41                "tcp:%i" % self.PORT,
42                "tcp:%i" % self.PORT,
43            ],
44            stdout=subprocess.PIPE
45        )
46        connection = 0
47        while(connection < 5):
48            arkdb_server = False
49            ret = None
50            result = subprocess.run(
51                '%s shell "netstat -anp | grep PandaDebugger"' % os.environ.get("HDCTool", "hdc"),
52                stdout=subprocess.PIPE,
53            )
54            if (result.returncode == 0):
55                ret = result.stdout
56            else:
57                print("Failed to grep PandaDebugger!")
58                sys.exit(0)
59
60            out_str = str(ret).split('\\n')
61            for sub_str in out_str:
62                if (sub_str.find('CONNECTED') != -1):
63                    arkdb_server = True
64            if arkdb_server == True:
65                self.tcp_client_socket.connect(self.ADDRESS)
66                break
67            time.sleep(1)
68            connection = connection + 1
69        if connection == 5:
70            self.fail(self, "Not connect to arkdb server.")
71
72    def close_client_socket(self):
73        self.tcp_client_socket.close()
74
75    def send_command(self, command):
76        if command == '':
77            self.fail('Command is not vaild')
78        sent = self.tcp_client_socket.send(command.encode('utf-8'))
79        if sent == 0:
80            self.fail('Failed to send command: %s' % command)
81        try:
82            self.tcp_client_socket.settimeout(15)
83            os.environ["isSkipped"] = "False"
84            data, addr = self.tcp_client_socket.recvfrom(self.BUFSIZ)
85            print("Recv: ", data.decode('utf-8'))
86        except TimeoutError:
87            self.tcp_client_socket.close()
88            os.environ["isSkipped"] = "True"
89            self.fail("TimeoutError")
90
91    def run_arkdb_server(self, dbg_name):
92        panda_dbg = []
93        real_panda_name = None
94        ret = None
95        result = subprocess.run(
96            '%s shell "netstat -anp | grep PandaDebugger"' % os.environ.get("HDCTool", "hdc"),
97            stdout=subprocess.PIPE,
98        )
99        if (result.returncode == 0):
100            ret = result.stdout
101        else:
102            print("Failed to grep PandaDebugger!")
103            sys.exit(0)
104
105        out_str = str(ret).split('\\n')
106        pos_start = out_str[0].find("@") + 1
107        out_str = out_str[0].split('\\r')[0]
108        panda_dbg.append(out_str[pos_start : ])
109        if len(panda_dbg) == 0:
110            print('Not exist PandaDebugger with name %s' % dbg_name)
111            sys.exit(0)
112        else:
113            panda_name_len = 0
114            for sub_panda in panda_dbg:
115                if panda_name_len == 0:
116                    panda_name_len = len(sub_panda)
117                    real_panda_name = sub_panda
118                elif len(sub_panda) < panda_name_len:
119                    panda_name_len = len(sub_panda)
120                    real_panda_name = sub_panda
121
122        subprocess.Popen(
123            [self.hdctool, 'shell', 'arkdb', real_panda_name, 'server'],
124            stdout=subprocess.PIPE,
125        )
126