import time
from oeqa.oetest import loadTests, runTests
from oeqa.targetcontrol import get_target_controller
+ from oeqa.utils.dump import get_host_dumper
pn = d.getVar("PN", True)
export = oe.utils.conditional("TEST_EXPORT_ONLY", "1", True, False, d)
testslist = get_tests_list(d)
testsrequired = [t for t in d.getVar("TEST_SUITES", True).split() if t != "auto"]
+ # we need the host dumper in test context
+ host_dumper = get_host_dumper(d)
+
# the robot dance
target = get_target_controller(d)
self.testsrequired = testsrequired
self.filesdir = os.path.join(os.path.dirname(os.path.abspath(oeqa.runtime.__file__)),"files")
self.target = target
+ self.host_dumper = host_dumper
self.imagefeatures = d.getVar("IMAGE_FEATURES", True).split()
self.distrofeatures = d.getVar("DISTRO_FEATURES", True).split()
manifest = os.path.join(d.getVar("DEPLOY_DIR_IMAGE", True), d.getVar("IMAGE_LINK_NAME", True) + ".manifest")
import unittest
import inspect
import subprocess
-import datetime
-import commands
import bb
from oeqa.utils.decorators import LogResults
from sys import exc_info, exc_clear
# If a test fails or there is an exception
if not exc_info() == (None, None, None):
exc_clear()
- dump_dir = self.create_dump_dir()
+ self.tc.host_dumper.create_dir(self._testMethodName)
+ self.target.target_dumper.dump_target(
+ self.tc.host_dumper.dump_dir)
+ self.tc.host_dumper.dump_host()
print ("%s dump data from host and target "
- "stored in %s" % (self._testMethodName, dump_dir))
- self.dump_host_logs(dump_dir)
- self.dump_target_logs(dump_dir)
-
- def create_dump_dir(self):
- dump_sub_dir = ("%s_%s" % (
- datetime.datetime.now().strftime('%Y%m%d%H%M'),
- self._testMethodName))
- dump_dir = os.path.join(self.target.dump_dir, dump_sub_dir)
- os.makedirs(dump_dir)
- return dump_dir
-
- def dump_host_logs(self, dump_dir):
- for cmd in self.target.dump_host.split('\n'):
- cmd = cmd.lstrip()
- if not cmd:
- continue
- output = commands.getoutput(cmd)
- filename = "host_%s" % cmd.split()[0]
- with open(os.path.join(dump_dir, filename), 'w') as f:
- f.write(output)
-
- def dump_target_logs(self, dump_dir):
- for cmd in self.target.dump_target.split('\n'):
- cmd = cmd.lstrip()
- if not cmd:
- continue
- # This will ping the host from target
- if cmd == "_ping":
- comm = "ping -c3 %s" % self.target.server_ip
- # This will get all the logs from /var/log/
- elif cmd == "_logs":
- comm = 'find /var/log/ -type f 2>/dev/null '
- comm = '%s-exec echo "%s" \\; ' % (comm, '='*20)
- comm = '%s-exec echo {} \\; ' % comm
- comm = '%s-exec echo "%s" \\; ' % (comm, '='*20)
- comm = '%s-exec cat {} \\; -exec echo "" \\;' % comm
- else:
- comm = cmd
- (status, output) = self.target.run_serial(comm)
- filename = "target_%s" % cmd.split()[0]
- with open(os.path.join(dump_dir, filename), 'w') as f:
- f.write(output)
+ "stored in %s" % (self._testMethodName,
+ self.target.target_dumper.dump_dir))
#TODO: use package_manager.py to install packages on any type of image
def install_packages(self, packagelist):
from oeqa.utils.sshcontrol import SSHControl
from oeqa.utils.qemurunner import QemuRunner
from oeqa.utils.qemutinyrunner import QemuTinyRunner
+from oeqa.utils.dump import TargetDumper
from oeqa.controllers.testtargetloader import TestTargetLoader
from abc import ABCMeta, abstractmethod
self.origrootfs = os.path.join(d.getVar("DEPLOY_DIR_IMAGE", True), d.getVar("IMAGE_LINK_NAME", True) + '.' + self.image_fstype)
self.rootfs = os.path.join(self.testdir, d.getVar("IMAGE_LINK_NAME", True) + '-testimage.' + self.image_fstype)
self.kernel = os.path.join(d.getVar("DEPLOY_DIR_IMAGE", True), d.getVar("KERNEL_IMAGETYPE", False) + '-' + d.getVar('MACHINE', False) + '.bin')
- self.dump_target = d.getVar("testimage_dump_target", True)
- self.dump_host = d.getVar("testimage_dump_host", True)
- self.dump_dir = d.getVar("TESTIMAGE_DUMP_DIR", True)
# Log QemuRunner log output to a file
import oe.path
logfile = self.qemulog,
boottime = int(d.getVar("TEST_QEMUBOOT_TIMEOUT", True)))
+ self.target_dumper = TargetDumper(d, self.runner)
+
def deploy(self):
try:
bb.utils.mkdirhier(self.testdir)
--- /dev/null
+import os
+import sys
+import errno
+import datetime
+import itertools
+from commands import runCmd
+
+def get_host_dumper(d):
+ return HostDumper(d)
+
+
+class BaseDumper(object):
+
+ def __init__(self, d):
+ self.parent_dir = d.getVar("TESTIMAGE_DUMP_DIR", True)
+
+ def create_dir(self, dir_suffix):
+ dump_subdir = ("%s_%s" % (
+ datetime.datetime.now().strftime('%Y%m%d%H%M'),
+ dir_suffix))
+ dump_dir = os.path.join(self.parent_dir, dump_subdir)
+ try:
+ os.makedirs(dump_dir)
+ except OSError as err:
+ if err.errno != errno.EEXIST:
+ raise err
+ self.dump_dir = dump_dir
+
+ def write_dump(self, command, output):
+ if isinstance(self, HostDumper):
+ prefix = "host"
+ elif isinstance(self, TargetDumper):
+ prefix = "target"
+ else:
+ prefix = "unknown"
+ for i in itertools.count():
+ filename = "%s_%02d_%s" % (prefix, i, command)
+ fullname = os.path.join(self.dump_dir, filename)
+ if not os.path.exists(fullname):
+ break
+ with open(fullname, 'w') as dump_file:
+ dump_file.write(output)
+
+
+class HostDumper(BaseDumper):
+
+ def __init__(self, d):
+ super(HostDumper, self).__init__(d)
+ self.host_cmds = d.getVar("testimage_dump_host", True)
+
+ def dump_host(self, dump_dir=""):
+ if dump_dir:
+ self.dump_dir = dump_dir
+ for cmd in self.host_cmds.split('\n'):
+ cmd = cmd.lstrip()
+ if not cmd or cmd[0] == '#':
+ continue
+ result = runCmd(cmd, ignore_status=True)
+ self.write_dump(cmd.split()[0], result.output)
+
+
+class TargetDumper(BaseDumper):
+
+ def __init__(self, d, qemurunner):
+ super(TargetDumper, self).__init__(d)
+ self.target_cmds = d.getVar("testimage_dump_target", True)
+ self.runner = qemurunner
+
+ def dump_target(self, dump_dir=""):
+ if dump_dir:
+ self.dump_dir = dump_dir
+ for cmd in self.target_cmds.split('\n'):
+ cmd = cmd.lstrip()
+ if not cmd or cmd[0] == '#':
+ continue
+ (status, output) = self.runner.run_serial(cmd)
+ self.write_dump(cmd.split()[0], output)
