1#!/usr/bin/env python3
2import argparse
3import json
4import os
5from pathlib import Path
6import re
7import shutil
8import vms
9import subprocess
10import sys
11import time
12import uuid
13import xml.etree.ElementTree as ET
14
15
16def main():
17 parser = argparse.ArgumentParser()
18 parser.add_argument("socket")
19 parser.add_argument("--user", default="nobody")
20 parser.add_argument("--images", type=Path, required=True)
21 parser.add_argument("--output", type=Path)
22 args = parser.parse_args()
23 client = Path(__file__).with_name("dashboard-host-vm-test.py")
24
25 def call(operation, payload=None):
26 request = {"operation": "vm." + operation}
27 if payload is not None:
28 request["payload"] = payload
29 result = subprocess.run([sys.executable, str(client), args.socket, "--client", "--user", args.user],
30 input=json.dumps(request).encode() + b"\n", capture_output=True)
31 assert result.returncode == 0, result.stderr.decode()
32 return json.loads(result.stdout)
33
34 name = "boundary-" + uuid.uuid4().hex
35 existing = set(subprocess.check_output(["virsh", "list", "--all", "--name"], text=True).split())
36 spec = {"name": name, "description": "isolated VM fixture", "image": "blank", "vcpus": 1,
37 "memory": 2**29, "disk": 2**30, "autostart": False, "start": False,
38 "mode": "iso", "firmware": "bios", "platform": "linux", "owner": "fixture-user", "username": "fixture"}
39 node = call("node")["value"]
40 for field, value in [("vcpus", node["cpus"] + 1), ("memory", node["memory"] + 1), ("disk", 2**64),
41 ("vcpus", True), ("image", "../escape.iso"), ("name", "../escape"), ("xml", "<domain/>")]:
42 assert call("create", {**spec, field: value})["status"] == 400, (field, value)
43 assert call("act", {"name": name, "action": ["start"]})["status"] == 400
44 assert call("update", {"name": name})["status"] == 400
45 assert "installers" in call("library")["value"]
46 images = args.images
47 created_images = not images.exists()
48 images.mkdir(parents=True, exist_ok=True)
49 preset_dirs = [vms.PRESETS / (name + suffix) for suffix in ["-source", "-backed", "-external", "-unsupported"]]
50 for directory in preset_dirs:
51 directory.mkdir(parents=True)
52 (directory / "preset.json").write_text(json.dumps({"id": directory.name, "os": "Fixture", "description": "", "firmware": "bios", "platform": "linux", "createdAt": time.time(), "recommended": {"vcpus": 1, "memory": 2**29, "disk": 2**30}}))
53 source, backing, external, unsupported = [directory / "disk.qcow2" for directory in preset_dirs]
54 installer = images / (name + ".iso")
55 secret = Path("/run/" + name + "-outside-image-directory")
56 secret.write_bytes(b"synthetic private fixture\n" * 4096)
57 secret.chmod(0o600)
58 inactive = re.search(r"^Active:\s+no", subprocess.check_output(["virsh", "net-info", "default"], text=True), re.MULTILINE)
59 try:
60 subprocess.run(["qemu-img", "create", "-f", "qcow2", "-F", "raw", "-b", str(secret), str(backing)], check=True, capture_output=True)
61 rejected = call("create", {**spec, "mode": "preset", "image": backing.parent.name})
62 assert rejected["status"] == 400 and "prepared" in rejected["error"], rejected
63 assert not (Path("/srv/vm") / name).exists()
64 subprocess.run(["qemu-img", "create", "-f", "qcow2", "-o", "data_file=" + str(secret) + ",data_file_raw=on", str(external), "1048576"], check=True, capture_output=True)
65 assert call("create", {**spec, "mode": "preset", "image": external.parent.name})["status"] == 400
66 subprocess.run(["qemu-img", "create", "-f", "vmdk", str(unsupported), "1048576"], check=True, capture_output=True)
67 assert call("create", {**spec, "mode": "preset", "image": unsupported.parent.name})["status"] == 400
68 subprocess.run(["qemu-img", "create", "-f", "qcow2", str(source), str(2**30)], check=True, capture_output=True)
69 assert call("create", {**spec, "mode": "preset", "image": source.parent.name}) == {"value": None}
70 directory = Path("/srv/vm") / name
71 assert directory.is_dir() and (directory / "disk.qcow2").is_file()
72 assert not (directory / "source-image").exists()
73 description = "--config <literal description>"
74 assert call("update", {"name": name, "description": description, "autostart": True}) == {"value": None}
75 vm = next(vm for vm in call("domains")["value"] if vm["name"] == name)
76 assert vm["description"] == description and vm["autostart"] and vm["owner"] == "fixture-user", vm
77 assert call("owner", {"name": name}) == {"value": "fixture-user"}
78 assert call("act", {"name": name, "action": "start"}) == {"value": None}
79 assert name in call("stats")["value"]
80 subprocess.run(["virsh", "suspend", name], check=True, capture_output=True)
81 assert call("act", {"name": name, "action": "resume"}) == {"value": None}
82 assert call("act", {"name": name, "action": "destroy"}) == {"value": None}
83 assert call("remove", {"name": name, "disks": True}) == {"value": None}
84 assert not directory.exists()
85 installer.write_bytes(b"synthetic installer fixture\n" * 1024)
86 interrupted = """import json, sys, time
87sys.path.insert(0, sys.argv[1])
88import vms
89spec = json.load(sys.stdin)
90if sys.argv[2] == 'copy':
91 alarm = vms.signal.alarm
92 vms.signal.alarm = lambda seconds: alarm(1 if seconds else 0)
93 def copy(incoming, outgoing):
94 outgoing.write(incoming.read(1024))
95 outgoing.flush()
96 time.sleep(60)
97 vms.shutil.copyfileobj = copy
98else:
99 original = vms.virsh
100 def virsh(*args):
101 result = original(*args)
102 if args[0] == 'define':
103 raise OSError('fixture connection lost after definition')
104 return result
105 vms.virsh = virsh
106vms.validate('create', spec)
107try:
108 vms.create(spec)
109except OSError as error:
110 print(str(error), file=sys.stderr)
111 sys.exit(1)
112raise AssertionError('interrupted creation reported success')
113"""
114 for phase in ["copy", "define"]:
115 started = time.monotonic()
116 attempt = subprocess.run([sys.executable, "-c", interrupted, str(Path(__file__).parent), phase],
117 input=json.dumps({**spec, "image": installer.name if phase == "copy" else "blank"}),
118 capture_output=True, text=True, timeout=60,
119 env={**os.environ, "STUDIO_VM_IMAGES_ROOT": str(images)})
120 elapsed = time.monotonic() - started
121 assert attempt.returncode == 1, attempt.stderr
122 if phase == "copy":
123 assert "preparation timed out" in attempt.stderr and elapsed < 60, (elapsed, attempt.stderr)
124 assert not directory.exists() and name not in {vm["name"] for vm in call("domains")["value"]}
125 copy_timeout_seconds = round(elapsed, 2)
126 else:
127 assert "connection lost" in attempt.stderr and (directory / "disk.qcow2").is_file(), attempt.stderr
128 assert name in {vm["name"] for vm in call("domains")["value"]}
129 assert call("remove", {"name": name, "disks": True}) == {"value": None}
130 assert not directory.exists()
131 assert installer.read_bytes() == b"synthetic installer fixture\n" * 1024
132 assert call("create", {**spec, "image": installer.name}) == {"value": None}
133 xml = ET.fromstring(subprocess.check_output(["virsh", "dumpxml", name], text=True))
134 cdrom = xml.find("./devices/disk[@device='cdrom']")
135 assert cdrom.find("driver").get("type") == "raw"
136 assert cdrom.find("source").get("file") == str(directory / installer.name)
137 assert call("remove", {"name": name, "disks": True}) == {"value": None}
138 directory.symlink_to(secret.parent, target_is_directory=True)
139 try:
140 assert call("remove", {"name": name, "disks": True})["status"] == 400
141 assert secret.exists()
142 finally:
143 directory.unlink()
144 assert set(subprocess.check_output(["virsh", "list", "--all", "--name"], text=True).split()) == existing
145 result = {"vm_read_queries": "passed", "vm_resource_and_field_bounds": "passed",
146 "vm_create_start_resume_stop_remove": "passed", "literal_description": "passed",
147 "standalone_image_conversion": "passed", "external_backing_file_rejection": "passed",
148 "external_data_file_and_format_rejection": "passed", "pinned_raw_installer": "passed",
149 "symlink_disk_directory_rejection": "passed", "interrupted_copy_cleanup": "passed",
150 "copy_timeout_seconds": copy_timeout_seconds, "uncertain_definition_preserves_disks": "passed",
151 "existing_domains_preserved": "passed"}
152 if args.output:
153 args.output.write_text(json.dumps(result, indent=2) + "\n")
154 print(json.dumps(result))
155 finally:
156 subprocess.run(["virsh", "destroy", name], capture_output=True)
157 subprocess.run(["virsh", "undefine", name], capture_output=True)
158 directory = Path("/srv/vm") / name
159 if directory.is_dir() and not directory.is_symlink():
160 for file in directory.iterdir():
161 file.unlink()
162 directory.rmdir()
163 for preset_dir in preset_dirs:
164 shutil.rmtree(preset_dir)
165 for file in [installer, secret]:
166 file.unlink(missing_ok=True)
167 if created_images and not any(images.iterdir()):
168 images.rmdir()
169 if inactive and not subprocess.check_output(["virsh", "list", "--name"], text=True).strip():
170 subprocess.run(["virsh", "net-destroy", "default"], check=True, capture_output=True)
171
172
173if __name__ == "__main__":
174 main()