3 # Use of this source code is governed by a BSD-style license that can be
11 from impl.testvm import Arch, VmState
16 Both are a wrapper around `./tools/testvm --arch=x86_64/aarch64`.
26     - up: Start the VM if it is not already running.
27     - stop: Gracefully stop the VM
28     - kill: Send SIGKILL to the VM
29     - clean: Stop the VM and delete all images
30     - logs: Print logs of the VM console
37 def cli_shorthand(arch: Arch):  argument
38     if arch == "x86_64":
40     elif arch == "aarch64":
43         raise Exception(f"Unknown architecture: {arch}")
46 def arch_or_all(arch: Optional[Arch]):  argument
47     return (arch,) if arch else testvm.ARCH_OPTIONS
53 @argh.arg("--arch-list", "--arch", nargs="*", type=str, default=ARCHS, choices=ARCHS)
54 def up(arch_list: Iterable[Arch] = [], reset: bool = False, wait: bool = False, timeout: int = 120):  argument
56     for arch in arch_list:
57         testvm.up(arch, reset, wait, timeout)
60 @argh.arg("--arch", required=True, choices=testvm.ARCH_OPTIONS)
61 def run(arch: Arch = "x86_64", reset: bool = False):  argument
63     if testvm.is_running(arch):
65     testvm.build_if_needed(arch, reset)
67         arch,
68         testvm.rootfs_img_path(arch),
73 @argh.arg("--arch", required=True, choices=testvm.ARCH_OPTIONS)
74 def shell(arch: Arch = "x86_64", timeout: int = 120):  argument
76     testvm.up(arch, wait=True, timeout=timeout)
77     testvm.ssh_exec(arch)
80 @argh.arg("--arch-list", "--arch", nargs="*", type=str, default=ARCHS, choices=ARCHS)
81 def stop(arch_list: Iterable[Arch] = []):  argument
83     for arch in arch_list:
84         if not testvm.is_running(arch):
85             print(f"{arch} VM is not running")
87         console.print(f"Stopping {arch} VM")
88         testvm.ssh_exec(arch, "sudo poweroff")
91 @argh.arg("--arch-list", "--arch", nargs="*", type=str, default=ARCHS, choices=ARCHS)
92 def kill(arch_list: Iterable[Arch] = []):  argument
94     for arch in arch_list:
95         if not testvm.is_running(arch):
96             console.print(f"{arch} VM is not running")
98         console.print(f"Killing {arch} VM process")
99         testvm.kill_vm(arch)
103 @argh.arg("--arch-list", "--arch", nargs="*", type=str, default=ARCHS, choices=ARCHS)
104 def clean(arch_list: Iterable[Arch] = []):  argument
106     for arch in arch_list:
107         if testvm.is_running(arch):
108             kill(arch)
109         if testvm.data_dir(arch).exists():
110             console.print("Cleaning data directory", testvm.data_dir(arch))
111             shutil.rmtree(testvm.data_dir(arch))
115 def vm_status(arch: Arch):  argument
117         return f"[green][bold]{cli_shorthand(arch)} {' '.join(args)}[/bold][/green]"
119     vm = f"{arch} VM"
120     port = f"[blue]{testvm.SSH_PORTS[arch]}[/blue]"
122     state = testvm.state(arch)
132         logs(arch, 10, style="light_slate_grey")
136 @argh.arg("--arch-list", "--arch", nargs="*", type=str, default=ARCHS, choices=ARCHS)
137 def status(arch_list: Iterable[Arch] = []):  argument
138     for arch in arch_list:
139         vm_status(arch)
143 @argh.arg("--arch", required=True, choices=testvm.ARCH_OPTIONS)
144 def logs(arch: Arch = "x86_64", n: int = 0, style: Optional[str] = None):  argument
145     log_lines = testvm.log_path(arch).read_text().splitlines()
147         log_lines = log_lines[-n:]