test_computing.py (3963B)
1 from contextlib import contextmanager 2 import os 3 from pathlib import Path 4 import pytest 5 import subprocess 6 from typing import Iterator 7 from unittest import mock 8 import uuid 9 10 from aetherscale import computing 11 from aetherscale.services import ServiceManager 12 13 14 @contextmanager 15 def base_image(directory: Path) -> Iterator[Path]: 16 random_name = str(uuid.uuid4()) 17 img_file = directory / f'{random_name}.qcow2' 18 try: 19 subprocess.run([ 20 'qemu-img', 'create', '-f', 'qcow2', str(img_file), '1G']) 21 yield img_file 22 finally: 23 os.unlink(img_file) 24 25 26 def test_create_user_image(tmppath): 27 with mock.patch('aetherscale.config.BASE_IMAGE_FOLDER', tmppath), \ 28 mock.patch('aetherscale.config.USER_IMAGE_FOLDER', tmppath): 29 30 with base_image(tmppath) as img: 31 user_image = computing.create_user_image('my-vm-id', img.stem) 32 user_image.is_file() 33 34 35 def test_vm_lifecycle(tmppath, mock_service_manager: ServiceManager): 36 with mock.patch('aetherscale.config.BASE_IMAGE_FOLDER', tmppath), \ 37 mock.patch('aetherscale.config.USER_IMAGE_FOLDER', tmppath): 38 39 handler = computing.ComputingHandler( 40 radvd=mock.MagicMock(), service_manager=mock_service_manager) 41 42 with base_image(tmppath) as img: 43 results = list(handler.create_vm({'image': img.stem})) 44 list_results = list(handler.list_vms({})) 45 vm_id = results[0]['vm-id'] 46 service_name = computing.systemd_unit_name_for_vm(vm_id) 47 assert results[0]['status'] == 'allocating' 48 assert results[1]['status'] == 'starting' 49 assert mock_service_manager.service_is_running(service_name) 50 assert list_results[0][0]['vm-id'] == vm_id 51 52 vm_info = list(handler.vm_info({'vm-id': vm_id}))[0] 53 assert vm_info['vm-id'] == vm_id 54 assert vm_info['status'] == 'running' 55 56 # TODO: Test graceful stop, needs mock of QemuMonitor 57 results = list(handler.stop_vm({'vm-id': vm_id, 'kill': True})) 58 list_results = list(handler.list_vms({})) 59 assert results[0]['status'] == 'killed' 60 assert mock_service_manager.service_exists(service_name) 61 assert not mock_service_manager.service_is_running(service_name) 62 assert list_results[0][0]['vm-id'] == vm_id 63 64 results = list(handler.start_vm({'vm-id': vm_id})) 65 list_results = list(handler.list_vms({})) 66 assert results[0]['status'] == 'starting' 67 assert mock_service_manager.service_exists(service_name) 68 assert mock_service_manager.service_is_running(service_name) 69 assert list_results[0][0]['vm-id'] == vm_id 70 71 results = list(handler.delete_vm({'vm-id': vm_id})) 72 list_results = list(handler.list_vms({})) 73 assert results[0]['status'] == 'deleted' 74 assert not mock_service_manager.service_exists(service_name) 75 assert not mock_service_manager.service_is_running(service_name) 76 assert len(list_results[0]) == 0 77 78 79 def test_run_missing_base_image(tmppath, mock_service_manager: ServiceManager): 80 with mock.patch('aetherscale.config.BASE_IMAGE_FOLDER', tmppath), \ 81 mock.patch('aetherscale.config.USER_IMAGE_FOLDER', tmppath): 82 83 handler = computing.ComputingHandler( 84 radvd=mock.MagicMock(), service_manager=mock_service_manager) 85 86 # specify invalid base image 87 with pytest.raises(OSError): 88 # make sure to exhaust the iterator 89 list(handler.create_vm({'image': 'some-missing-image'})) 90 91 # do not specify a base image 92 with pytest.raises(ValueError): 93 # make sure to exhaust the iterator 94 list(handler.create_vm({})) 95 96 97 def test_vm_id_systemd_unit(): 98 assert 'myvmid' == computing.vm_id_from_systemd_unit( 99 computing.systemd_unit_name_for_vm('myvmid'))