support/testing/infra/emulator.py

   1 import pexpect
   2
   3 import infra
   4 import infra.basetest
   5
   6 # TODO: Most of the telnet stuff need to be replaced by stdio/pexpect to discuss
   7 # with the qemu machine.
   8 class Emulator(object):
   9
  10     def __init__(self, builddir, downloaddir, logtofile):
  11         self.qemu = None
  12         self.__tn = None
  13         self.downloaddir = downloaddir
  14         self.log = ""
  15         self.logfile = infra.open_log_file(builddir, "run", logtofile)
  16
  17     # Start Qemu to boot the system
  18     #
  19     # arch: Qemu architecture to use
  20     #
  21     # kernel: path to the kernel image, or the special string
  22     # 'builtin'. 'builtin' means a pre-built kernel image will be
  23     # downloaded from ARTEFACTS_URL and suitable options are
  24     # automatically passed to qemu and added to the kernel cmdline. So
  25     # far only armv5, armv7 and i386 builtin kernels are available.
  26     # If None, then no kernel is used, and we assume a bootable device
  27     # will be specified.
  28     #
  29     # kernel_cmdline: array of kernel arguments to pass to Qemu -append option
  30     #
  31     # options: array of command line options to pass to Qemu
  32     #
  33     def boot(self, arch, kernel=None, kernel_cmdline=None, options=None):
  34         if arch in ["armv7", "armv5"]:
  35             qemu_arch = "arm"
  36         else:
  37             qemu_arch = arch
  38
  39         qemu_cmd = ["qemu-system-{}".format(qemu_arch),
  40                     "-serial", "telnet::1234,server",
  41                     "-display", "none"]
  42
  43         if options:
  44             qemu_cmd += options
  45
  46         if kernel_cmdline is None:
  47             kernel_cmdline = []
  48
  49         if kernel:
  50             if kernel == "builtin":
  51                 if arch in ["armv7", "armv5"]:
  52                     kernel_cmdline.append("console=ttyAMA0")
  53
  54                 if arch == "armv7":
  55                     kernel = infra.download(self.downloaddir,
  56                                             "kernel-vexpress")
  57                     dtb = infra.download(self.downloaddir,
  58                                          "vexpress-v2p-ca9.dtb")
  59                     qemu_cmd += ["-dtb", dtb]
  60                     qemu_cmd += ["-M", "vexpress-a9"]
  61                 elif arch == "armv5":
  62                     kernel = infra.download(self.downloaddir,
  63                                             "kernel-versatile")
  64                     qemu_cmd += ["-M", "versatilepb"]
  65
  66             qemu_cmd += ["-kernel", kernel]
  67
  68         if kernel_cmdline:
  69             qemu_cmd += ["-append", " ".join(kernel_cmdline)]
  70
  71         self.logfile.write("> starting qemu with '%s'\n" % " ".join(qemu_cmd))
  72         self.qemu = pexpect.spawn(qemu_cmd[0], qemu_cmd[1:])
  73
  74         # Wait for the telnet port to appear and connect to it.
  75         self.qemu.expect("waiting for connection")
  76         telnet_cmd = ["telnet", "localhost", "1234"]
  77         self.__tn = pexpect.spawn(telnet_cmd[0], telnet_cmd[1:])
  78
  79     def __read_until(self, waitstr, timeout=5):
  80         index = self.__tn.expect([waitstr, pexpect.TIMEOUT], timeout=timeout)
  81         data = self.__tn.before
  82         if index == 0:
  83             data += self.__tn.after
  84         self.log += data
  85         self.logfile.write(data)
  86         return data
  87
  88     def __write(self, wstr):
  89         self.__tn.send(wstr)
  90
  91     # Wait for the login prompt to appear, and then login as root with
  92     # the provided password, or no password if not specified.
  93     def login(self, password=None):
  94         self.__read_until("buildroot login:", 10)
  95         if "buildroot login:" not in self.log:
  96             self.logfile.write("==> System does not boot")
  97             raise SystemError("System does not boot")
  98
  99         self.__write("root\n")
 100         if password:
 101             self.__read_until("Password:")
 102             self.__write(password + "\n")
 103         self.__read_until("# ")
 104         if "# " not in self.log:
 105             raise SystemError("Cannot login")
 106         self.run("dmesg -n 1")
 107
 108     # Run the given 'cmd' on the target
 109     # return a tuple (output, exit_code)
 110     def run(self, cmd):
 111         self.__write(cmd + "\n")
 112         output = self.__read_until("# ")
 113         output = output.strip().splitlines()
 114         output = output[1:len(output)-1]
 115
 116         self.__write("echo $?\n")
 117         exit_code = self.__read_until("# ")
 118         exit_code = exit_code.strip().splitlines()[1]
 119         exit_code = int(exit_code)
 120
 121         return output, exit_code
 122
 123     def stop(self):
 124         if self.__tn:
 125             self.__tn.terminate(force=True)
 126         if self.qemu is None:
 127             return
 128         self.qemu.terminate(force=True)