1# 2# SPDX-License-Identifier: MIT 3# 4 5import os 6import sys 7import errno 8import datetime 9import itertools 10from .commands import runCmd 11 12class BaseDumper(object): 13 """ Base class to dump commands from host/target """ 14 15 def __init__(self, cmds, parent_dir): 16 self.cmds = [] 17 # Some testing doesn't inherit testimage, so it is needed 18 # to set some defaults. 19 self.parent_dir = parent_dir 20 dft_cmds = """ top -bn1 21 iostat -x -z -N -d -p ALL 20 2 22 ps -ef 23 free 24 df 25 memstat 26 dmesg 27 ip -s link 28 netstat -an""" 29 if not cmds: 30 cmds = dft_cmds 31 for cmd in cmds.split('\n'): 32 cmd = cmd.lstrip() 33 if not cmd or cmd[0] == '#': 34 continue 35 self.cmds.append(cmd) 36 37 def create_dir(self, dir_suffix): 38 dump_subdir = ("%s_%s" % ( 39 datetime.datetime.now().strftime('%Y%m%d%H%M'), 40 dir_suffix)) 41 dump_dir = os.path.join(self.parent_dir, dump_subdir) 42 try: 43 os.makedirs(dump_dir) 44 except OSError as err: 45 if err.errno != errno.EEXIST: 46 raise err 47 self.dump_dir = dump_dir 48 49 def _write_dump(self, command, output): 50 if isinstance(self, HostDumper): 51 prefix = "host" 52 elif isinstance(self, TargetDumper): 53 prefix = "target" 54 else: 55 prefix = "unknown" 56 for i in itertools.count(): 57 filename = "%s_%02d_%s" % (prefix, i, command) 58 fullname = os.path.join(self.dump_dir, filename) 59 if not os.path.exists(fullname): 60 break 61 with open(fullname, 'w') as dump_file: 62 dump_file.write(output) 63 64 65class HostDumper(BaseDumper): 66 """ Class to get dumps from the host running the tests """ 67 68 def __init__(self, cmds, parent_dir): 69 super(HostDumper, self).__init__(cmds, parent_dir) 70 71 def dump_host(self, dump_dir=""): 72 if dump_dir: 73 self.dump_dir = dump_dir 74 env = os.environ.copy() 75 env['PATH'] = '/usr/sbin:/sbin:/usr/bin:/bin' 76 env['COLUMNS'] = '9999' 77 for cmd in self.cmds: 78 result = runCmd(cmd, ignore_status=True, env=env) 79 self._write_dump(cmd.split()[0], result.output) 80 81class TargetDumper(BaseDumper): 82 """ Class to get dumps from target, it only works with QemuRunner """ 83 84 def __init__(self, cmds, parent_dir, runner): 85 super(TargetDumper, self).__init__(cmds, parent_dir) 86 self.runner = runner 87 88 def dump_target(self, dump_dir=""): 89 if dump_dir: 90 self.dump_dir = dump_dir 91 for cmd in self.cmds: 92 # We can continue with the testing if serial commands fail 93 try: 94 (status, output) = self.runner.run_serial(cmd) 95 self._write_dump(cmd.split()[0], output) 96 except: 97 print("Tried to dump info from target but " 98 "serial console failed") 99