| 1 | import contextlib |
| 2 | import io |
| 3 | import json |
| 4 | from pathlib import Path |
| 5 | import subprocess |
| 6 | import tempfile |
| 7 | import unittest |
| 8 | |
| 9 | import linux_vm |
| 10 | |
| 11 | |
| 12 | class LinuxVmLifecycleTest(unittest.TestCase): |
| 13 | def setUp(self): |
| 14 | names = ("VM_HOME", "LINUX_HOME", "IMAGES", "INSTANCES", "RUN", "BASE", |
| 15 | "BASE_MANIFEST") |
| 16 | self.original = {name: getattr(linux_vm, name) for name in names} |
| 17 | self.temporary = tempfile.TemporaryDirectory(prefix="one-linux-test-") |
| 18 | home = Path(self.temporary.name) |
| 19 | linux_vm.VM_HOME = home |
| 20 | linux_vm.LINUX_HOME = home / "linux" |
| 21 | linux_vm.IMAGES = linux_vm.LINUX_HOME / "images" |
| 22 | linux_vm.INSTANCES = linux_vm.LINUX_HOME / "instances" |
| 23 | linux_vm.RUN = linux_vm.LINUX_HOME / "run" |
| 24 | linux_vm.BASE = linux_vm.IMAGES / "base.qcow2" |
| 25 | linux_vm.BASE_MANIFEST = linux_vm.IMAGES / "base.json" |
| 26 | linux_vm.IMAGES.mkdir(parents=True) |
| 27 | subprocess.run([ |
| 28 | linux_vm.executable("qemu-img"), "create", "-q", "-f", "qcow2", |
| 29 | str(linux_vm.BASE), "64M", |
| 30 | ], check=True) |
| 31 | linux_vm.BASE_MANIFEST.write_text("{}\n") |
| 32 | |
| 33 | def tearDown(self): |
| 34 | self.temporary.cleanup() |
| 35 | for name, value in self.original.items(): |
| 36 | setattr(linux_vm, name, value) |
| 37 | |
| 38 | def test_clone_has_unique_ports_keys_network_and_removable_overlay(self): |
| 39 | with contextlib.redirect_stdout(io.StringIO()): |
| 40 | linux_vm.create_instance("samba") |
| 41 | linux_vm.create_instance("other") |
| 42 | config = json.loads(linux_vm.instance_path("samba").read_text()) |
| 43 | other = json.loads(linux_vm.instance_path("other").read_text()) |
| 44 | self.assertEqual(config["hostname"], "one-samba") |
| 45 | self.assertEqual(config["lab_address"], "192.168.77.1") |
| 46 | self.assertNotEqual(config["ssh_port"], other["ssh_port"]) |
| 47 | self.assertNotEqual(config["samba_port"], other["samba_port"]) |
| 48 | self.assertNotEqual(config["lab_mac"], other["lab_mac"]) |
| 49 | self.assertEqual(linux_vm.key_path("samba").stat().st_mode & 0o777, 0o600) |
| 50 | known_hosts = next( |
| 51 | value for value in linux_vm.ssh_argv("samba") |
| 52 | if value.startswith("UserKnownHostsFile=") |
| 53 | ) |
| 54 | self.assertEqual(known_hosts, 'UserKnownHostsFile="%s"' % |
| 55 | (linux_vm.INSTANCES / "samba.known_hosts")) |
| 56 | self.assertTrue(linux_vm.overlay_path("samba").is_file()) |
| 57 | self.assertTrue(linux_vm.seed_path("samba").is_file()) |
| 58 | with contextlib.redirect_stdout(io.StringIO()): |
| 59 | linux_vm.delete_instance("samba") |
| 60 | linux_vm.delete_instance("other") |
| 61 | self.assertFalse(linux_vm.instance_path("samba").exists()) |
| 62 | |
| 63 | def test_invalid_name_is_rejected(self): |
| 64 | with self.assertRaisesRegex(SystemExit, "lowercase"): |
| 65 | linux_vm.create_instance("Bad Name") |
| 66 | |
| 67 | |
| 68 | if __name__ == "__main__": |
| 69 | unittest.main() |