emulator.py 4.33 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import socket
import subprocess
import telnetlib

import infra
import infra.basetest

# TODO: Most of the telnet stuff need to be replaced by stdio/pexpect to discuss
# with the qemu machine.
class Emulator(object):

    def __init__(self, builddir, downloaddir, logtofile):
        self.qemu = None
        self.__tn = None
        self.downloaddir = downloaddir
        self.log = ""
17
        self.logfile = infra.open_log_file(builddir, "run", logtofile)
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

    # Start Qemu to boot the system
    #
    # arch: Qemu architecture to use
    #
    # kernel: path to the kernel image, or the special string
    # 'builtin'. 'builtin' means a pre-built kernel image will be
    # downloaded from ARTEFACTS_URL and suitable options are
    # automatically passed to qemu and added to the kernel cmdline. So
    # far only armv5, armv7 and i386 builtin kernels are available.
    # If None, then no kernel is used, and we assume a bootable device
    # will be specified.
    #
    # kernel_cmdline: array of kernel arguments to pass to Qemu -append option
    #
    # options: array of command line options to pass to Qemu
    #
    def boot(self, arch, kernel=None, kernel_cmdline=None, options=None):
        if arch in ["armv7", "armv5"]:
            qemu_arch = "arm"
        else:
            qemu_arch = arch

        qemu_cmd = ["qemu-system-{}".format(qemu_arch),
                    "-serial", "telnet::1234,server",
                    "-display", "none"]

        if options:
            qemu_cmd += options

        if kernel_cmdline is None:
            kernel_cmdline = []

        if kernel:
            if kernel == "builtin":
                if arch in ["armv7", "armv5"]:
                    kernel_cmdline.append("console=ttyAMA0")

                if arch == "armv7":
                    kernel = infra.download(self.downloaddir,
                                            "kernel-vexpress")
                    dtb = infra.download(self.downloaddir,
                                         "vexpress-v2p-ca9.dtb")
                    qemu_cmd += ["-dtb", dtb]
                    qemu_cmd += ["-M", "vexpress-a9"]
                elif arch == "armv5":
                    kernel = infra.download(self.downloaddir,
                                            "kernel-versatile")
                    qemu_cmd += ["-M", "versatilepb"]

            qemu_cmd += ["-kernel", kernel]

        if kernel_cmdline:
            qemu_cmd += ["-append", " ".join(kernel_cmdline)]

73
74
        self.logfile.write("> starting qemu with '%s'\n" % " ".join(qemu_cmd))
        self.qemu = subprocess.Popen(qemu_cmd, stdout=self.logfile, stderr=self.logfile)
75
76
77
78
79
80
81
82
83
84
85
86
87

        # Wait for the telnet port to appear and connect to it.
        while True:
            try:
                self.__tn = telnetlib.Telnet("localhost", 1234)
                if self.__tn:
                    break
            except socket.error:
                continue

    def __read_until(self, waitstr, timeout=5):
        data = self.__tn.read_until(waitstr, timeout)
        self.log += data
88
        self.logfile.write(data)
89
90
91
92
93
94
95
96
97
98
        return data

    def __write(self, wstr):
        self.__tn.write(wstr)

    # Wait for the login prompt to appear, and then login as root with
    # the provided password, or no password if not specified.
    def login(self, password=None):
        self.__read_until("buildroot login:", 10)
        if "buildroot login:" not in self.log:
99
            self.logfile.write("==> System does not boot")
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
            raise SystemError("System does not boot")

        self.__write("root\n")
        if password:
            self.__read_until("Password:")
            self.__write(password + "\n")
        self.__read_until("# ")
        if "# " not in self.log:
            raise SystemError("Cannot login")
        self.run("dmesg -n 1")

    # Run the given 'cmd' on the target
    # return a tuple (output, exit_code)
    def run(self, cmd):
        self.__write(cmd + "\n")
        output = self.__read_until("# ")
        output = output.strip().splitlines()
        output = output[1:len(output)-1]

        self.__write("echo $?\n")
        exit_code = self.__read_until("# ")
        exit_code = exit_code.strip().splitlines()[1]
        exit_code = int(exit_code)

        return output, exit_code

    def stop(self):
        if self.qemu is None:
            return
        self.qemu.terminate()
        self.qemu.kill()