395 lines
18 KiB
Python
395 lines
18 KiB
Python
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from hydra.profiles import Profile, save_profile
|
|
from hydra.runtime import Runtime, RuntimeError as VMError, build_qemu_argv
|
|
from test_qmp import QMPServer, greeting, send
|
|
|
|
|
|
# External protocol fixtures isolate lifecycle tests from user VMs and real disks.
|
|
FAKE_QEMU = r'''#!/usr/bin/env python3
|
|
import json, os, socket, sys
|
|
args = sys.argv[1:]
|
|
def val(flag): return args[args.index(flag)+1]
|
|
if os.environ.get('HYDRA_TEST_FAIL'):
|
|
print('intentional boot failure', file=sys.stderr)
|
|
sys.exit(1)
|
|
path = val('-chardev').split('path=', 1)[1].split(',server=', 1)[0].replace(',,', ',')
|
|
pid = os.fork()
|
|
if pid:
|
|
sys.exit(0)
|
|
os.setsid()
|
|
with open(val('-pidfile'), 'w') as f: f.write(str(os.getpid()))
|
|
s = socket.socket(socket.AF_UNIX)
|
|
s.bind(path)
|
|
s.listen(8)
|
|
alive = True
|
|
while alive:
|
|
c, _ = s.accept()
|
|
try:
|
|
with c:
|
|
c.sendall(b'{"QMP":{"version":{"qemu":{"major":10,"minor":0,"micro":0},"package":""},"capabilities":[]}}\n')
|
|
with c.makefile('rb') as stream:
|
|
for line in stream:
|
|
cmd = json.loads(line)
|
|
name = cmd['execute']
|
|
result = {'query-name': {'name': val('-name')}, 'query-uuid': {'UUID': val('-uuid')},
|
|
'query-status': {'status': 'running', 'running': True, 'singlestep': False}}.get(name, {})
|
|
c.sendall(json.dumps({'return': result, 'id': cmd['id']}).encode()+b'\n')
|
|
if name == 'quit':
|
|
alive = False
|
|
break
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
pass
|
|
s.close()
|
|
os.unlink(path)
|
|
'''
|
|
|
|
FAKE_IMG = r'''#!/usr/bin/env python3
|
|
import sys
|
|
with open(sys.argv[-2], 'xb') as f: f.write(b'new qcow2 fixture')
|
|
'''
|
|
|
|
|
|
class RuntimeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory(prefix='hr-')
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.runtime = Runtime(self.root / 'config space,comma', self.root / 'state space,comma',
|
|
self.root / 'run,comma', boot_timeout=1)
|
|
self.iso = self.root / 'installer space,comma.iso'
|
|
self.iso.touch()
|
|
self.profile = Profile('guest', self.iso, graphics='software', disk_size='8M')
|
|
self.runtime.add(self.profile)
|
|
self.qemu = self.root / 'fake qemu'
|
|
self.qemu.write_text(FAKE_QEMU)
|
|
self.qemu.chmod(0o700)
|
|
self.img = self.root / 'fake img'
|
|
self.img.write_text(FAKE_IMG)
|
|
self.img.chmod(0o700)
|
|
env = patch.dict(os.environ, {'HYDRA_QEMU': str(self.qemu), 'HYDRA_QEMU_IMG': str(self.img)})
|
|
env.start()
|
|
self.addCleanup(env.stop)
|
|
self.addCleanup(self.cleanup_guest)
|
|
|
|
def cleanup_guest(self):
|
|
try:
|
|
if self.runtime.status('guest')['status'] == 'running':
|
|
self.runtime.stop('guest', force=True)
|
|
except VMError:
|
|
pass
|
|
|
|
def test_mac_persisted_distinct_and_stable_across_restarts(self):
|
|
self.runtime.add(Profile('other', network='bridge'))
|
|
other = self.runtime.profile('other')
|
|
guest = self.runtime.profile('guest')
|
|
self.assertIsNotNone(guest.mac)
|
|
self.assertNotEqual(guest.mac, other.mac)
|
|
octet = int(guest.mac.split(':')[0], 16)
|
|
self.assertEqual(octet & 3, 2) # locally administered, unicast
|
|
def nic(profile, runtime):
|
|
return next(arg for arg in build_qemu_argv(profile, runtime.paths(profile.name))
|
|
if arg.startswith('virtio-net-pci,'))
|
|
expected = nic(guest, self.runtime)
|
|
self.runtime.start('guest')
|
|
self.runtime.stop('guest', force=True)
|
|
self.runtime.start('guest')
|
|
self.runtime.stop('guest', force=True)
|
|
self.assertEqual(nic(self.runtime.profile('guest'), self.runtime), expected)
|
|
changed_runtime = Runtime(self.runtime.config, self.runtime.state, self.root / 'different-run')
|
|
self.assertEqual(nic(changed_runtime.profile('guest'), changed_runtime), expected)
|
|
self.runtime.media('guest', None)
|
|
self.assertEqual(self.runtime.profile('guest').mac, guest.mac)
|
|
|
|
def test_legacy_profiles_gain_stable_owner_specific_argv_mac(self):
|
|
file = self.runtime.config / 'legacy.toml'
|
|
save_profile(file, Profile('legacy', network='bridge'))
|
|
original = file.read_bytes()
|
|
legacy = self.runtime.profile('legacy')
|
|
self.assertIsNone(legacy.mac)
|
|
paths = self.runtime.paths('legacy')
|
|
def mac():
|
|
return next(arg for arg in build_qemu_argv(legacy, paths) if arg.startswith('virtio-net-pci,'))
|
|
with patch('os.getuid', return_value=1000):
|
|
first = mac()
|
|
self.assertEqual(mac(), first)
|
|
self.assertNotEqual(first, next(arg for arg in build_qemu_argv(
|
|
legacy, self.runtime.paths('other')) if arg.startswith('virtio-net-pci,')))
|
|
with patch('os.getuid', return_value=1001):
|
|
self.assertNotEqual(mac(), first)
|
|
self.assertEqual(file.read_bytes(), original)
|
|
|
|
def test_disk_boot_omits_media_and_requests_display(self):
|
|
profile = Profile('guest', width=1920, height=1080, refresh_millihz=144000,
|
|
graphics='software')
|
|
argv = build_qemu_argv(profile, self.runtime.paths('guest'))
|
|
self.assertIn('virtio-vga,xres=1920,yres=1080,refresh_rate=144000', argv)
|
|
self.assertNotIn('ide-cd,drive=iso', argv)
|
|
self.assertIn('order=c', argv)
|
|
devices = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
|
|
self.assertEqual(len(devices), 2)
|
|
stock = build_qemu_argv(Profile('guest', graphics='software'), self.runtime.paths('guest'))
|
|
self.assertFalse(any('refresh_rate=' in arg for arg in stock))
|
|
|
|
def test_missing_iso_is_error_at_start_without_creating_disk(self):
|
|
self.iso.unlink()
|
|
with self.assertRaisesRegex(VMError, 'ISO'):
|
|
self.runtime.start('guest')
|
|
self.assertFalse(self.runtime.paths('guest').disk.exists())
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
|
|
|
|
def test_media_changes_only_while_stopped_and_preserves_disk(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.start('guest')
|
|
disk = paths.disk.read_bytes()
|
|
config = (self.runtime.config / 'guest.toml').read_bytes()
|
|
for iso in (None, self.iso):
|
|
with self.assertRaises(VMError):
|
|
self.runtime.media('guest', iso)
|
|
self.assertEqual((self.runtime.config / 'guest.toml').read_bytes(), config)
|
|
self.iso.unlink()
|
|
self.assertEqual(self.runtime.profile('guest').iso, self.iso)
|
|
self.runtime.stop('guest', force=True)
|
|
self.runtime.media('guest', None)
|
|
self.assertIsNone(self.runtime.profile('guest').iso)
|
|
self.runtime.start('guest')
|
|
self.runtime.stop('guest', force=True)
|
|
self.assertEqual(paths.disk.read_bytes(), disk)
|
|
replacement = self.root / 'replacement.iso'
|
|
replacement.touch()
|
|
self.runtime.media('guest', replacement)
|
|
self.assertEqual(self.runtime.profile('guest').iso, replacement)
|
|
with self.assertRaisesRegex(VMError, 'ISO'):
|
|
self.runtime.media('guest', self.iso)
|
|
self.assertEqual(self.runtime.profile('guest').iso, replacement)
|
|
|
|
def test_argv_is_private_and_comma_safe(self):
|
|
paths = self.runtime.paths('guest')
|
|
argv = build_qemu_argv(self.profile, paths)
|
|
self.assertEqual(argv[0], str(self.qemu))
|
|
blockdevs = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
|
|
self.assertIn(str(paths.disk), [b.get('filename') for b in blockdevs])
|
|
self.assertIn(str(self.iso), [b.get('filename') for b in blockdevs])
|
|
spice = argv[argv.index('-spice')+1]
|
|
self.assertIn('unix=on', spice)
|
|
self.assertIn('disable-copy-paste=on', spice)
|
|
self.assertIn('disable-agent-file-xfer=on', spice)
|
|
self.assertIn('run,,comma', spice)
|
|
self.assertIn('virtio-blk-pci,drive=disk', argv)
|
|
self.assertTrue(any(arg.startswith('virtio-net-pci,netdev=net,mac=') for arg in argv))
|
|
self.assertNotIn('-vnc', argv)
|
|
self.assertNotIn('-virtfs', argv)
|
|
|
|
def test_socket_removed_during_stat_is_safely_not_listening(self):
|
|
paths = self.runtime.paths('guest')
|
|
paths.qmp.parent.mkdir(parents=True)
|
|
with socket.socket(socket.AF_UNIX) as fixture:
|
|
fixture.bind(str(paths.qmp))
|
|
original_lstat = Path.lstat
|
|
def disappearing_socket(path):
|
|
if path == paths.qmp:
|
|
path.unlink(missing_ok=True)
|
|
return original_lstat(path)
|
|
with patch.object(Path, 'lstat', disappearing_socket):
|
|
self.assertFalse(self.runtime._listening(paths.qmp))
|
|
|
|
def test_argv_provides_audio_and_manual_usb_channels(self):
|
|
argv = build_qemu_argv(self.profile, self.runtime.paths('guest'))
|
|
self.assertIn('spice,id=audio', argv)
|
|
self.assertIn('hda-duplex,audiodev=audio', argv)
|
|
for channel in range(3):
|
|
self.assertIn(f'spicevmc,id=usbredir{channel},name=usbredir', argv)
|
|
self.assertIn(f'usb-redir,chardev=usbredir{channel}', argv)
|
|
self.assertFalse(any('usb-host' in argument for argument in argv))
|
|
|
|
def test_metadata_symlink_does_not_overwrite_user_data(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.prepare(paths)
|
|
valuable = self.root / 'valuable data'
|
|
valuable.write_text('preserve me')
|
|
paths.metadata.symlink_to(valuable)
|
|
self.runtime.start('guest')
|
|
self.assertEqual(valuable.read_text(), 'preserve me')
|
|
self.assertFalse(paths.metadata.is_symlink())
|
|
|
|
def test_disk_symlink_never_changes_its_target(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.prepare(paths)
|
|
valuable = self.root / 'valuable disk'
|
|
valuable.write_text('preserve me')
|
|
paths.disk.symlink_to(valuable)
|
|
with self.assertRaises(VMError):
|
|
self.runtime.start('guest')
|
|
self.assertEqual(valuable.read_text(), 'preserve me')
|
|
|
|
def test_malformed_identity_record_is_safe_unknown(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.prepare(paths)
|
|
paths.metadata.write_text('[]')
|
|
server = QMPServer(paths.qmp, greeting)
|
|
self.addCleanup(server.close)
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
|
|
|
|
def test_foreign_qmp_name_blocks_control_even_when_uuid_matches(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.prepare(paths)
|
|
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
|
|
commands = []
|
|
def handler(conn):
|
|
greeting(conn)
|
|
for line in conn.makefile('rb'):
|
|
req = json.loads(line)
|
|
commands.append(req['execute'])
|
|
result = {'query-name': {'name': 'other'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
|
|
send(conn, {'return': result, 'id': req['id']})
|
|
server = QMPServer(paths.qmp, handler)
|
|
self.addCleanup(server.close)
|
|
with self.assertRaises(VMError):
|
|
self.runtime.stop('guest', force=True)
|
|
self.assertNotIn('quit', commands)
|
|
|
|
def test_reused_pid_without_qmp_never_controls_a_process(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.prepare(paths)
|
|
paths.pid.write_text(str(os.getpid()))
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
|
|
with patch('os.kill', side_effect=AssertionError('must not signal any PID')):
|
|
with self.assertRaises(VMError):
|
|
self.runtime.stop('guest', force=True)
|
|
|
|
def test_malformed_query_result_is_safe_unknown(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.prepare(paths)
|
|
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
|
|
def handler(conn):
|
|
greeting(conn)
|
|
for line in conn.makefile('rb'):
|
|
req = json.loads(line)
|
|
send(conn, {'return': [], 'id': req['id']})
|
|
server = QMPServer(paths.qmp, handler)
|
|
self.addCleanup(server.close)
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
|
|
|
|
def test_cli_qmp_failure_reports_error_without_traceback(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.prepare(paths)
|
|
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
|
|
def handler(conn):
|
|
greeting(conn)
|
|
for line in conn.makefile('rb'):
|
|
req = json.loads(line)
|
|
result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
|
|
if req['execute'] == 'system_powerdown':
|
|
send(conn, {'error': {'class': 'GenericError', 'desc': 'cannot power down'}, 'id': req['id']})
|
|
else:
|
|
send(conn, {'return': result, 'id': req['id']})
|
|
server = QMPServer(paths.qmp, handler)
|
|
self.addCleanup(server.close)
|
|
run = subprocess.run([sys.executable, 'bin/hydra', '--config', str(self.runtime.config),
|
|
'--state', str(self.runtime.state), '--runtime', str(self.runtime.runtime),
|
|
'shutdown', 'guest'], capture_output=True, text=True, timeout=4)
|
|
self.assertEqual(run.returncode, 1)
|
|
self.assertIn('cannot power down', run.stderr)
|
|
self.assertNotIn('Traceback', run.stderr)
|
|
|
|
def test_disk_and_profile_survive_shutdown_and_force(self):
|
|
paths = self.runtime.paths('guest')
|
|
paths.disk.parent.mkdir(parents=True, exist_ok=True)
|
|
paths.disk.write_bytes(b'existing valuable disk')
|
|
self.runtime.start('guest')
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'running')
|
|
self.runtime.shutdown('guest')
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'running')
|
|
with self.assertRaises(VMError):
|
|
self.runtime.stop('guest', force=False)
|
|
self.runtime.stop('guest', force=True)
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
|
|
self.assertEqual(paths.disk.read_bytes(), b'existing valuable disk')
|
|
|
|
def test_simultaneous_start_only_launches_one_vm(self):
|
|
barrier = threading.Barrier(2)
|
|
results = []
|
|
def start():
|
|
barrier.wait()
|
|
try:
|
|
results.append(self.runtime.start('guest')['status'])
|
|
except VMError as exc:
|
|
results.append(str(exc))
|
|
threads = [threading.Thread(target=start) for _ in range(2)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(4)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(results.count('running'), 1)
|
|
self.assertEqual(len(results), 2)
|
|
self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')
|
|
|
|
def test_stale_socket_is_replaced_only_when_no_listener(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.prepare(paths)
|
|
stale = socket.socket(socket.AF_UNIX)
|
|
stale.bind(str(paths.qmp))
|
|
stale.close()
|
|
self.runtime.start('guest')
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'running')
|
|
|
|
def test_failed_boot_is_reported_and_disk_preserved(self):
|
|
with patch.dict(os.environ, {'HYDRA_TEST_FAIL': '1'}), self.assertRaises(VMError):
|
|
self.runtime.start('guest')
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
|
|
self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')
|
|
self.assertIn('intentional boot failure', self.runtime.paths('guest').log.read_text())
|
|
|
|
def test_foreign_qmp_uuid_blocks_all_controls_and_restart(self):
|
|
paths = self.runtime.paths('guest')
|
|
self.runtime.prepare(paths)
|
|
paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
|
|
paths.pid.write_text(str(os.getpid()))
|
|
commands = []
|
|
def handler(conn):
|
|
greeting(conn)
|
|
for line in conn.makefile('rb'):
|
|
req = json.loads(line)
|
|
commands.append(req['execute'])
|
|
result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': 'foreign-uuid'}}.get(req['execute'], {})
|
|
send(conn, {'return': result, 'id': req['id']})
|
|
server = QMPServer(paths.qmp, handler)
|
|
self.addCleanup(server.close)
|
|
self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
|
|
for control in (lambda: self.runtime.shutdown('guest'), lambda: self.runtime.stop('guest', force=True),
|
|
lambda: self.runtime.start('guest')):
|
|
with self.assertRaises(VMError):
|
|
control()
|
|
self.assertNotIn('quit', commands)
|
|
self.assertNotIn('system_powerdown', commands)
|
|
self.assertFalse(paths.disk.exists())
|
|
|
|
def test_private_runtime_and_socket_length(self):
|
|
self.runtime.prepare(self.runtime.paths('guest'))
|
|
self.assertEqual(self.runtime.runtime.stat().st_mode & 0o777, 0o700)
|
|
with self.assertRaises(VMError):
|
|
Runtime(self.root / 'c', self.root / 's', self.root / ('r' * 100)).paths('guest')
|
|
|
|
def test_cli_lifecycle_and_global_paths(self):
|
|
base = [sys.executable, 'bin/hydra', '--config', str(self.runtime.config), '--state', str(self.runtime.state),
|
|
'--runtime', str(self.runtime.runtime)]
|
|
for args, expected in [(['list'], 'guest'), (['start', 'guest'], 'running'),
|
|
(['status', 'guest'], 'running'), (['stop', 'guest', '--force'], 'stopped')]:
|
|
run = subprocess.run(base + args, text=True, capture_output=True, timeout=4)
|
|
self.assertEqual(run.returncode, 0, run.stderr)
|
|
self.assertIn(expected, run.stdout)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|