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