Source code for vmshepherd.iaas.dummy_driver

import logging
from .abstract import AbstractIaasDriver
from .exception import IaasException
from .vm import Vm, VmState
from vmshepherd.utils import next_id, add_async_delay


[docs]class DummyIaasUserExc(IaasException): def __init__(self, message, details): super().__init__(message, details)
[docs]class DummyIaasVmNotFound(DummyIaasUserExc): def __init__(self): super().__init__('VMNOTFOUND', 'Vm Not Found')
[docs]class DummyIaasDriver(AbstractIaasDriver): """ DummyIaasDriver - mainly for test purposes. Implements required API of fake-in-memory-iaas (dict). To mimic communication with external service additional delay has been introduced. """ def __init__(self, config=None): """ Init: create list of vms. """ self._vms = {} self._id_it = next_id() @add_async_delay async def create_vm(self, *, preset_name, image, flavor, security_groups=None, userdata=None, key_name=None, availability_zone=None, subnet=None): """ Dummy create_vm func. """ info = { 'id': next(self._id_it), 'name': preset_name, 'ip': ['127.0.0.1'], 'created': 0, 'state': VmState.RUNNING, 'flavor': flavor, 'image': image, 'metadata': {'test-meta': 'abctest'}, 'timed_shutdown_at': 1522753481, 'tags': ['a-tag', 'b-tag', 'c-tag'] } logging.debug('Prepare vm: %s', info) vm = Vm(self, **info) self._vms[vm.id] = vm logging.debug('Create: %s', vm) return None @add_async_delay async def list_vms(self, preset_name): """Dummy list_vms func""" return list(vm for vm in self._vms.values() if vm.name == preset_name) @add_async_delay async def terminate_vm(self, vm_id): """ Dummy terminate_vm func """ if vm_id not in self._vms: raise DummyIaasVmNotFound() del self._vms[vm_id] return None @add_async_delay async def get_vm(self, vm_id): """ Dummy get_vm func """ if vm_id not in self._vms: raise DummyIaasVmNotFound() return self._vms[vm_id]