emulator.py 4.38 KB
Newer Older
1
import pexpect
2
3
4
5
6
7
8
9
10
11
12
13
14

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 = ""
15
        self.logfile = infra.open_log_file(builddir, "run", logtofile)
16
17
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

    # 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)]

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
76
77
        self.qemu.expect("waiting for connection")
        telnet_cmd = ["telnet", "localhost", "1234"]
        self.__tn = pexpect.spawn(telnet_cmd[0], telnet_cmd[1:])
78
79

    def __read_until(self, waitstr, timeout=5):
80
81
82
83
        index = self.__tn.expect([waitstr, pexpect.TIMEOUT], timeout=timeout)
        data = self.__tn.before
        if index == 0:
            data += self.__tn.after
84
        self.log += data
85
        self.logfile.write(data)
86
87
88
        return data

    def __write(self, wstr):
89
        self.__tn.send(wstr)
90
91
92
93
94
95

    # 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:
96
            self.logfile.write("==> System does not boot")
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
            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):
124
125
        if self.__tn:
            self.__tn.terminate(force=True)
126
127
        if self.qemu is None:
            return
128
        self.qemu.terminate(force=True)