1import contextlib
2import io
3import json
4from pathlib import Path
5import subprocess
6import tempfile
7import unittest
8
9import vm
10
11
12class VmLifecycleTest(unittest.TestCase):
13 def setUp(self):
14 self.original = {name: getattr(vm, name) for name in (
15 "VM_HOME", "IMAGES", "MEDIA", "INSTANCES", "RUN", "AGENT_ISO", "TARGETS",
16 )}
17 self.temporary = tempfile.TemporaryDirectory(prefix="one-vm-test-")
18 home = Path(self.temporary.name)
19 vm.VM_HOME = home
20 vm.IMAGES = home / "images"
21 vm.MEDIA = home / "media"
22 vm.INSTANCES = home / "instances"
23 vm.RUN = home / "run"
24 vm.AGENT_ISO = vm.MEDIA / "win7-agent.iso"
25 vm.TARGETS = home / "targets.json"
26 vm.IMAGES.mkdir(parents=True)
27 for base in ("win7", "win11"):
28 subprocess.run([
29 vm.qemu("qemu-img"), "create", "-q", "-f", "qcow2",
30 str(vm.build_disk(base)), "64M",
31 ], check=True)
32 with contextlib.redirect_stdout(io.StringIO()):
33 vm.seal(base)
34
35 def tearDown(self):
36 self.temporary.cleanup()
37 for name, value in self.original.items():
38 setattr(vm, name, value)
39
40 def test_clone_has_unique_identity_and_removable_overlay(self):
41 output = io.StringIO()
42 with contextlib.redirect_stdout(output):
43 vm.create_instance("alpha")
44 vm.create_instance("beta")
45 config = json.loads(vm.instance_path("alpha").read_text())
46 other = json.loads(vm.instance_path("beta").read_text())
47 targets = json.loads(vm.TARGETS.read_text())
48 self.assertEqual(config["hostname"], "ONE-ALPHA")
49 self.assertEqual(targets["alpha"]["base"],
50 "http://127.0.0.1:%d" % config["port"])
51 self.assertEqual(targets["alpha"]["token"], config["token"])
52 self.assertNotEqual(config["hostname"], other["hostname"])
53 self.assertNotEqual(config["port"], other["port"])
54 self.assertNotEqual(config["token"], other["token"])
55 self.assertNotIn(config["token"], output.getvalue())
56 self.assertTrue((vm.IMAGES / "instances" / "alpha.qcow2").is_file())
57 self.assertTrue((vm.MEDIA / "instances" / "alpha.iso").is_file())
58 with contextlib.redirect_stdout(io.StringIO()):
59 vm.delete_instance("alpha")
60 vm.delete_instance("beta")
61 self.assertFalse(vm.instance_path("alpha").exists())
62 self.assertNotIn("alpha", json.loads(vm.TARGETS.read_text()))
63
64 def test_reserved_target_name_is_rejected(self):
65 with self.assertRaisesRegex(SystemExit, "reserved"):
66 vm.create_instance("local")
67 with self.assertRaisesRegex(SystemExit, "reserved"):
68 vm.create_instance("win11-build")
69
70 def test_clone_records_its_base(self):
71 with contextlib.redirect_stdout(io.StringIO()):
72 vm.create_instance("arm", base="win11")
73 self.assertEqual(json.loads(vm.instance_path("arm").read_text())["base"], "win11")
74 backing = json.loads(subprocess.check_output([
75 vm.qemu("qemu-img"), "info", "--output=json",
76 str(vm.IMAGES / "instances" / "arm.qcow2"),
77 ]))["backing-filename"]
78 self.assertEqual(backing, str(vm.base_disk("win11")))
79
80
81if __name__ == "__main__":
82 unittest.main()