test_vm_harness.py (7469B)
1 """Engine-free VM command, seed, image-lock and acceptance-report controls.""" 2 import copy 3 import json 4 from pathlib import Path 5 import subprocess 6 import tempfile 7 import tarfile 8 import unittest 9 10 import vm_matrix as vm 11 12 13 class VMTests(unittest.TestCase): 14 def identity(self): 15 return dict(kernel='6.8.0-100-generic', machine='x86_64', 16 os_release='ID=ubuntu\nVERSION_ID="24.04"\n', proc_version='Linux', 17 kernel_sha256='a'*64, kernel_config_sha256='b'*64) 18 19 def test_guest_syntax(self): 20 source = Path(vm.__file__).with_name('vm_guest.py') 21 compile(source.read_text(), str(source), 'exec') 22 23 def test_commands(self): 24 root = Path('/tmp/vm path') 25 command = vm.qemu_command(root/'run.qcow2', root/'seed.iso', root, 'kvm', False) 26 self.assertEqual(command[-2:], ['-nic', 'none']) 27 self.assertIn('q35,accel=kvm', command) 28 self.assertNotIn('-virtfs', command) 29 self.assertNotIn('-fsdev', command) 30 self.assertIn('virtserialport,chardev=report,name=tricu.report', command) 31 online = vm.qemu_command(root/'prepared.qcow2', root/'seed.iso', root, 'tcg', True) 32 self.assertIn('q35,accel=tcg', online) 33 self.assertEqual(online[-1], 'user,model=virtio-net-pci') 34 for path in ('/tmp/a,b', '/tmp/a\nb'): 35 with self.assertRaises(ValueError): 36 vm.safe_path(path) 37 38 def test_real_seed_iso(self): 39 with tempfile.TemporaryDirectory() as directory: 40 root = Path(directory) 41 (root/'kit.tar.gz').write_bytes(b'kit') 42 (root/'consumers.tar.gz').write_bytes(b'consumers') 43 request = dict(mode='prepare', kernel_prefix='6.8.') 44 iso = vm.seed(root, request, root) 45 files = subprocess.check_output(['isoinfo', '-R', '-i', str(iso), '-f'], text=True) 46 for name in ('user-data', 'meta-data', 'request.json', 'vm_guest.py', 'kit.tar.gz', 'consumers.tar.gz'): 47 self.assertIn('/'+name, files) 48 self.assertEqual(json.loads((root/'seed/request.json').read_text()), request) 49 userdata = (root/'seed/user-data').read_text() 50 self.assertTrue(userdata.startswith('#cloud-config\n')) 51 command = json.loads(userdata.split(' - ', 1)[1]) 52 self.assertEqual(command[:2], ['/bin/sh', '-ec']) 53 self.assertIn('/mnt/tricu-seed/vm_guest.py', command[2]) 54 55 def test_consumer_links_and_real_extraction(self): 56 with tempfile.TemporaryDirectory() as directory: 57 root = Path(directory)/'consumers' 58 for name in ('node/bin/node', 'node/lib/node_modules/npm/bin/npm-cli.js', 59 'tricu-js/node_modules/koffi/package.json'): 60 path = root/name 61 path.parent.mkdir(parents=True, exist_ok=True) 62 path.write_text('fixture') 63 (root/'node/bin/nodejs').symlink_to('/usr/local/bin/node') 64 (root/'node/bin/npm').symlink_to('../lib/node_modules/npm/bin/npm-cli.js') 65 for name in ('yarn', 'yarnpkg'): 66 (root/'node/bin'/name).symlink_to('/opt/yarn-v1.22.22/bin/'+name) 67 archive_path = Path(directory)/'consumers.tar.gz' 68 def pack(): 69 with tarfile.open(archive_path, 'w:gz') as archive: 70 for name in ('node', 'tricu-js'): 71 archive.add(root/name, arcname=name) 72 pack() 73 with self.assertRaises(tarfile.AbsoluteLinkError): 74 vm.validate_consumer_archive(archive_path) 75 changes = vm.normalize_consumers(root) 76 self.assertEqual(len(changes), 3) 77 self.assertEqual(str((root/'node/bin/nodejs').readlink()), 'node') 78 self.assertFalse((root/'node/bin/yarn').is_symlink()) 79 self.assertEqual((root/'node/bin/npm').read_text(), 'fixture') 80 pack() 81 vm.validate_consumer_archive(archive_path) 82 bad = root/'node/bin/escape' 83 bad.symlink_to('/etc/passwd') 84 with self.assertRaises(ValueError): 85 vm.normalize_consumers(root) 86 bad.unlink() 87 (root.parent/'outside').write_text('outside') 88 bad.symlink_to('../../../outside') 89 with self.assertRaises(ValueError): 90 vm.normalize_consumers(root) 91 92 def test_prepared_lock(self): 93 with tempfile.TemporaryDirectory() as directory: 94 root = Path(directory) 95 files = ('base.qcow2', 'prepared.qcow2', 'kit.tar.gz', 'consumers.tar.gz') 96 for name in files: 97 (root/name).write_bytes(name.encode()) 98 hashes = {name: vm.digest(root/name) for name in files} 99 lock = dict(schema='tricu.vm-images.v1', files=hashes, identity=self.identity(), 100 request=dict(kernel_prefix='6.8.', kit_sha256=hashes['kit.tar.gz'], 101 consumers_sha256=hashes['consumers.tar.gz'])) 102 path = root/'images.json' 103 vm.write_json(path, lock) 104 self.assertEqual(vm.load_prepared(path), lock) 105 (root/'base.qcow2').write_bytes(b'changed') 106 with self.assertRaises(ValueError): 107 vm.load_prepared(path) 108 109 def test_acceptance_controls(self): 110 identity = self.identity() 111 request = dict(mode='run', node_version='22.22.0') 112 expected = dict(result=42) 113 lock = dict(identity=identity, packages='packages', expected=expected) 114 binding = dict(passed=['python', 'js', 'php'], node='v22.22.0', 115 results={name: expected for name in ('python', 'js', 'php')}) 116 runs = [dict(location=location, nix_store_absent=True, cli_record_verify_mutation=True, 117 bindings=copy.deepcopy(binding)) for location in ('first', 'relocated path with spaces')] 118 report = dict(status='passed', identity=identity, request=request, packages='packages', 119 workload=dict(kernel=identity['kernel'], expectation='pass', runs=runs)) 120 vm.validate_result(report, lock, request) 121 mutations = [dict(status='failed'), dict(packages='changed'), 122 dict(identity=dict(identity, kernel='6.18.42')), 123 dict(workload=dict(report['workload'], runs=runs[:1]))] 124 for change in mutations: 125 with self.assertRaises(ValueError): 126 vm.validate_result(dict(report, **change), lock, request) 127 for mutate in ('node', 'missing-consumer', 'wrong-result', 'missing-isolation'): 128 altered = copy.deepcopy(report) 129 run = altered['workload']['runs'][0] 130 if mutate == 'node': run['bindings']['node'] = 'v20.19.0' 131 if mutate == 'missing-consumer': run['bindings']['passed'] = ['python'] 132 if mutate == 'wrong-result': run['bindings']['results']['js'] = dict(result=0) 133 if mutate == 'missing-isolation': run['nix_store_absent'] = False 134 with self.assertRaises(ValueError): 135 vm.validate_result(altered, lock, request) 136 for change in (dict(kernel='6.18.42'), dict(machine='aarch64'), dict(os_release='ID=debian'), 137 dict(kernel_config_sha256='missing')): 138 with self.assertRaises(ValueError): 139 vm.validate_identity(dict(identity, **change), '6.8.') 140 141 142 if __name__ == '__main__': 143 unittest.main()