Attachment 573140 Details for Bug 807405 – ovirt-engine-sdk code where this error happens

ovirt-engine-sdk code where this error happens

test_vm.py (text/plain), 9.21 KB, created by Martina Kollarova on 2012-03-27 17:28:19 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Martina Kollarova

Created: 2012-03-27 17:28:19 UTC

Size: 9.21 KB

patch

obsolete

>#!/usr/bin/env python
># -*- coding: utf-8 -*-
>#
># Copyright (c) 2012 Red Hat, Inc.
>#
># Licensed under the Apache License, Version 2.0 (the "License");
># you may not use this file except in compliance with the License.
># You may obtain a copy of the License at
>#
>#           http://www.apache.org/licenses/LICENSE-2.0
>#
># Unless required by applicable law or agreed to in writing, software
># distributed under the License is distributed on an "AS IS" BASIS,
># WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
># See the License for the specific language governing permissions and
># limitations under the License.
>#
>
>''' Test operations on a VM (start, suspend, stop, create snapshot etc).
>
>
>'''
>
>from common import *
>from nose.tools import *
>
>MB = 1024*1024
>GB = 1024*MB
>
># Names of created objects. Should be removed at the end of this test module
># and not used by any other test module.
>ISO_NAME =      'my_iso'
>EXPORT_NAME =   'my_export'
>VM_NAME =       'my_vm'
>SNAPSHOT_NAME = 'my_snapshot'
>TEMPLATE_NAME = 'my_template'
>
>
>logger = logging.getLogger(__name__)
>
>    
>
>
>def setUpModule():
>    ''' Creates VM at the beginning of testing this module (not in each test).
>
>    '''
>
>    logger.info('Creating VM "' + VM_NAME + '"')
>    assert api.vms.get(VM_NAME) == None, \
>        "VM with the name '" + VM_NAME + "' already exists"
>    api.vms.add(params.VM(
>        name=VM_NAME, memory=2*GB, cluster=api.clusters.get(CLUSTER_NAME),
>        template=api.templates.get('Blank')))
>
>    logger.info('Attaching NIC to VM')
>    api.vms.get(VM_NAME).nics.add(params.NIC(name='eth0',
>        network=params.Network(name='ovirtmgmt'), interface='virtio'))
>
>    logger.info('Attaching disk space to VM')
>    param = params.StorageDomains(
>            storage_domain=[api.storagedomains.get(STORAGE_NAME)])
>
>    api.vms.get(VM_NAME).disks.add(params.Disk(
>        storage_domains=param, size=512*MB, type_='system', status=None,
>        interface='virtio', format='cow', sparse=True, bootable=True))
>
>    waitForVMState(VM_NAME, 'down')
>    assert api.vms.get(VM_NAME).status.state == 'down'
>
>
>def tearDownModule():
>    ''' Removes VM at the end of testing this module (only once) '''
>
>    if api.vms.get(VM_NAME).status.state != 'down':
>        logging.warning('VM is not down at the end of the testcase, \
>                        stopping it now')
>        api.vms.get(VM_NAME).stop()
>        waitForVMState(VM_NAME, 'down');
>
>    logger.info('Removing VM "' + VM_NAME + '"')
>    api.vms.get(VM_NAME).delete()
>
>    logger.info('Waiting for VM removal')
>    t = 0
>    while VM_NAME in [vm.name for vm in api.vms.list()] and t <= TIMEOUT:
>        t += 1
>        sleep(1)
>    assert api.vms.get(VM_NAME) == None
>
>
>def test_startSuspendVM():
>    ''' Starts a VM and then suspends it. 
>
>        Expected:
>            VM state at beginning: down
>            VM state at end: suspended
>    '''
>    assert api.vms.get(VM_NAME).status.state != 'up', 'VM already up'
>    logger.info('Starting up VM')
>    api.vms.get(VM_NAME).start()
>    waitForVMState(VM_NAME, 'up');
>    assert api.vms.get(VM_NAME).status.state == 'up'
>
>    logger.info("Suspending VM")
>    t = 0
>    while api.vms.get(VM_NAME).status.state != 'suspended':
>        try:
>            api.vms.get(VM_NAME).suspend()
>            waitForVMState(VM_NAME, 'suspended')
>
>        except errors.RequestError as e:
>            if e.reason == 'Bad Request' and \
>                    'asynchronous running tasks' in e.detail and \
>                    t <= TIMEOUT:
>
>                logger.info("Asynchronous running tasks in VM, \
>                        cannot suspend, trying again")
>                sleep(1)
>                t += 1
>            else:
>                raise
>    assert api.vms.get(VM_NAME).status.state == 'suspended'
>
>
>def test_resumeStopVM():
>    ''' Resumes a suspended VM and stops it'.
>        Depends on 'test_startSuspendVM'.
>
>        Expected:
>            VM state at beginning: suspended
>            VM state at end: down
>
>    '''
>    logger.info("Resuming VM")
>    assert api.vms.get(VM_NAME).status.state == 'suspended', \
>        'VM not suspended, cannot resume'
>    api.vms.get(VM_NAME).start()
>    waitForVMState(VM_NAME, 'up');
>    assert api.vms.get(VM_NAME).status.state == 'up'
>
>    logger.info("Stopping VM")
>    api.vms.get(VM_NAME).stop()
>    waitForVMState(VM_NAME, 'down');
>    assert api.vms.get(VM_NAME).status.state == 'down'
>
>
>def setUpExport():
>    ''' Creates export domain for further use. '''
>    storageParam = params.Storage(type_='nfs',
>                    address=EXPORT_ADDRESS, path=EXPORT_PATH)
>    domainParam = params.StorageDomain(name=EXPORT_NAME,
>                    data_center=api.datacenters.get(DC_NAME),
>                    type_='export', host=api.hosts.get(HOST_NAME),
>                    storage = storageParam)
>    logger.info("Creating export domain")
>    api.storagedomains.add(domainParam)
>
>    dc = api.datacenters.get(DC_NAME)
>    dc.storagedomains.add(api.storagedomains.get(EXPORT_NAME))
>
>    waitForStorageActivation(EXPORT_NAME)
>    assert dc.storagedomains.get(EXPORT_NAME).status.state == "active"
>
>
>def tearDownExport():
>    ''' Removes export domain. '''
>    logger.info("Removing export domain")
>    dc = api.datacenters.get(name=DC_NAME)
>    dc.storagedomains.get(EXPORT_NAME).deactivate()
>    assert dc.storagedomains.get(EXPORT_NAME).status.state == 'maintenance'
>
>    dc.storagedomains.get(EXPORT_NAME).delete()
>    assert api.storagedomains.get(EXPORT_NAME).status.state == 'unattached'
>
>    param = params.StorageDomain(name=EXPORT_NAME,
>                host=params.Host(name=HOST_NAME), format=True)
>    api.storagedomains.get(EXPORT_NAME).delete(param)
>    assert api.storagedomains.get(EXPORT_NAME) == None
>
>
>@with_setup(setUpExport, tearDownExport)
>def test_exportDeleteImportVM():
>    ''' Export VM, remove it and then re-create by importing.
>        
>        Will not change VM if successfull and the VM will be put into the 'down' state. 
>
>        Expected:
>            VM state at beginning: down
>            VM state at end: down
>
>    '''
>    logger.info("Exporting VM")
>    api.vms.get(VM_NAME).export(params.Action(
>        storage_domain=api.storagedomains.get(EXPORT_NAME)))
>    waitForVMState(VM_NAME, 'down')
>    assert api.vms.get(VM_NAME).status.state == 'down'
>
>    logger.info("Deleting VM")
>    api.vms.get(VM_NAME).delete()
>    while api.vms.get(VM_NAME) != None:
>        sleep(1)
>
>    logger.info("Importing VM")
>    param = params.Action(storage_domain=api.storagedomains.get(STORAGE_NAME),
>                cluster=api.clusters.get(name=CLUSTER_NAME))
>    api.storagedomains.get(EXPORT_NAME).vms.get(VM_NAME).import_vm(param)
>    assert api.vms.get(VM_NAME) != None
>    waitForVMState(VM_NAME, 'down')
>    assert api.vms.get(VM_NAME).status.state == 'down'
>    assert False
>
>
>@timed(TIMEOUT)
>def test_createDeleteVMSnapshot():
>    ''' Create and delete a VM snapshot. 
>    
>        Expected:
>            VM state at beginning: down
>            VM state at end: down
>
>    '''
>    VM = api.vms.get(VM_NAME)
>    assert VM.status.state == 'down'
>    param = params.Snapshot(vm=VM, name=SNAPSHOT_NAME,
>            description=SNAPSHOT_NAME)
>    VM.snapshots.add(param)
>    logger.info('Creating a VM snapshot')
>    waitForVMState(VM_NAME, 'down')
>
>    # TODO test if snapshot was created correctly
>    # TODO why can't I just use VM.snapshots.get(SNAPSHOT_NAME)?
>    snapshots = api.vms.get(VM_NAME).snapshots.list()
>    snap = [s for s in snapshots if s.get_description() == SNAPSHOT_NAME]
>    assert len(snap) == 1, "found " + len(snap) + " snapshots with same desc"
>    snap = snap[0]
>
>    logger.info("Removing VM snapshot")
>    snap.delete()
>    waitForVMState(VM_NAME, 'down')
>
>    snapshots = api.vms.get(VM_NAME).snapshots.list()
>    snap = [s for s in snapshots if s.get_description() == SNAPSHOT_NAME]
>    assert len(snap) == 0, "Snapshot not deleted"
>
>
>def test_createTemplateFromVM():
>    ''' Create template from the default VM.
>    
>        Will NOT delete template, has to be deleted somewhere else (it is used
>        in other tests).
>    '''
>    logger.info('Creating a Template from VM')
>    param = params.Template(name=TEMPLATE_NAME,
>            vm=api.vms.get(VM_NAME), cluster=api.clusters.get(CLUSTER_NAME))
>    api.templates.add(param)
>    waitForVMState(VM_NAME, 'down')
>    assert api.templates.get(TEMPLATE_NAME) != None
>
>
>@timed(30)
>def test_createVMFromTemplate():
>    ''' Create new VM from previously created template, delete VM.
>
>        Depends on 'test_createTemplateFromVM'. Will not delete template, not
>        independent.
>    '''
>    NEW_VM_NAME = 'my_vm_from_template'
>    logger.info("Creating new VM '{}' from template".format(NEW_VM_NAME)) 
>    param = params.VM(name=NEW_VM_NAME,
>            cluster=api.clusters.get(CLUSTER_NAME),
>            template=api.templates.get(TEMPLATE_NAME))
>    api.vms.add(param)
>    waitForVMState(NEW_VM_NAME, 'down')
>    
>    assert api.vms.get(NEW_VM_NAME) != None
>
>    logger.info("Removing VM created from template")
>    api.vms.get(NEW_VM_NAME).delete()
>    while api.vms.get(NEW_VM_NAME) != None:
>        sleep(1)
>
>def test_removeTemplate():
>    ''' Delete previously created template. 
>    
>        Depends on 'test_createTemplateFromVM'.
>    '''
>    logger.info("Removing VM")
>    api.templates.get(TEMPLATE_NAME).delete()
>    assert api.templates.get(TEMPLATE_NAME) == None
>

Actions: View

Attachments on bug 807405: 573132 | 573140 | 573153