Skip to content
This repository has been archived by the owner on Apr 7, 2022. It is now read-only.

[1LP][RFR] modify snapshot testing #5126

Merged
merged 1 commit into from
Aug 23, 2017
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 46 additions & 42 deletions cfme/tests/infrastructure/test_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from cfme.infrastructure.virtual_machines import Vm # For Vm.Snapshot
from utils import testgen
from utils.conf import credentials
from utils.generators import random_vm_name
from utils.log import logger
from utils.path import data_path
from utils.ssh import SSHClient
Expand All @@ -29,94 +30,99 @@
pytest_generate_tests = testgen.generate([InfraProvider], scope="module")


@pytest.fixture(scope="module")
def vm_name():
return "test_snpsht_" + fauxfactory.gen_alphanumeric()


@pytest.fixture(scope="module")
def domain(request):
dom = DomainCollection().create(name=fauxfactory.gen_alpha(), enabled=True)
request.addfinalizer(dom.delete_if_exists)
return dom


@pytest.fixture(scope="module")
def test_vm(setup_provider_modscope, provider, vm_name, request):
"""Fixture to provision appliance to the provider being tested if necessary"""
vm = VM.factory(vm_name, provider, template_name=provider.data['full_template']['name'])
def provision_vm(provider, request, small_template=True):
vm_name = random_vm_name(context="snapshot")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

if small_template:
vm = VM.factory(vm_name, provider, template_name=provider.data['small_template'])
else:
vm = VM.factory(vm_name, provider, template_name=provider.data['full_template']['name'])

if not provider.mgmt.does_vm_exist(vm_name):
vm.create_on_provider(find_in_cfme=True, allow_skip="default")
request.addfinalizer(vm.delete_from_provider)
return vm


def new_snapshot(test_vm, has_name=True):
@pytest.fixture(scope="module")
def small_test_vm(setup_provider_modscope, provider, request):
return provision_vm(provider, request)


@pytest.fixture(scope="module")
def full_test_vm(setup_provider_modscope, provider, request):
return provision_vm(provider, request, small_template=False)


def new_snapshot(test_vm, has_name=True, memory=False):
if has_name:
new_snapshot = Vm.Snapshot(
name="snpshot_" + fauxfactory.gen_alphanumeric(8),
description="snapshot", memory=False, parent_vm=test_vm
name="snpshot_{}".format(fauxfactory.gen_alphanumeric(8)),
description="snapshot", memory=memory, parent_vm=test_vm
)
else:
new_snapshot = Vm.Snapshot(
description="snapshot_" + fauxfactory.gen_alphanumeric(8),
memory=False, parent_vm=test_vm
description="snapshot_{}".format(fauxfactory.gen_alphanumeric(8)),
memory=memory, parent_vm=test_vm
)
return new_snapshot


@pytest.mark.uncollectif(lambda provider:
not provider.one_of(RHEVMProvider, VMwareProvider) or
(provider.one_of(RHEVMProvider) and provider.version < 4))
def test_snapshot_crud(test_vm, provider):
def test_snapshot_crud(small_test_vm, provider):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This obviously won't get a new vm name each time, I assume this is OK?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yep, this works just fine.

"""Tests snapshot crud

Metadata:
test_flag: snapshot, provision
"""
if provider.one_of(RHEVMProvider):
snapshot = new_snapshot(test_vm, has_name=False)
snapshot = new_snapshot(small_test_vm, has_name=False)
else:
snapshot = new_snapshot(test_vm)
snapshot = new_snapshot(small_test_vm)
snapshot.create()
snapshot.delete()


@pytest.mark.uncollectif(lambda provider: not provider.one_of(VMwareProvider))
def test_delete_all_snapshots(test_vm, provider):
def test_delete_all_snapshots(small_test_vm, provider):
"""Tests snapshot removal

Metadata:
test_flag: snapshot, provision
"""
snapshot1 = new_snapshot(test_vm)
snapshot1 = new_snapshot(small_test_vm)
snapshot1.create()
snapshot2 = new_snapshot(test_vm)
snapshot2 = new_snapshot(small_test_vm)
snapshot2.create()
snapshot2.delete_all()


@pytest.mark.uncollectif(lambda provider:
not provider.one_of(RHEVMProvider, VMwareProvider) or
(provider.one_of(RHEVMProvider) and provider.version < 4))
def test_verify_revert_snapshot(test_vm, provider, soft_assert, register_event, request):
def test_verify_revert_snapshot(full_test_vm, provider, soft_assert, register_event, request):
"""Tests revert snapshot

Metadata:
test_flag: snapshot, provision
"""
if provider.one_of(RHEVMProvider):
snapshot1 = new_snapshot(test_vm, has_name=False)
snapshot1 = new_snapshot(full_test_vm, has_name=False)
else:
snapshot1 = new_snapshot(test_vm)

snapshot1 = new_snapshot(full_test_vm)
ssh_kwargs = {
'hostname': snapshot1.vm.provider.mgmt.get_ip_address(snapshot1.vm.name),
'username': credentials[provider.data['full_template']['creds']]['username'],
'password': credentials[provider.data['full_template']['creds']]['password']
}

ssh_client = SSHClient(**ssh_kwargs)
# We need to wait for ssh to become available on the vm, it can take a while. Without
# this wait, the ssh command would fail with 'port 22 not available' error.
Expand All @@ -127,35 +133,33 @@ def test_verify_revert_snapshot(test_vm, provider, soft_assert, register_event,
wait_for(lambda: ssh_client.run_command('touch snapshot1.txt').rc == 0, num_sec=300,
delay=20, handle_exception=True, fail_func=ssh_client.close())
snapshot1.create()
register_event(target_type='VmOrTemplate', target_name=test_vm.name,
register_event(target_type='VmOrTemplate', target_name=full_test_vm.name,
event_type='vm_snapshot_complete')
register_event(target_type='VmOrTemplate', target_name=test_vm.name,
register_event(target_type='VmOrTemplate', target_name=full_test_vm.name,
event_type='vm_snapshot')
ssh_client.run_command('touch snapshot2.txt')

if provider.one_of(RHEVMProvider):
snapshot2 = new_snapshot(test_vm, has_name=False)
snapshot2 = new_snapshot(full_test_vm, has_name=False)
else:
snapshot2 = new_snapshot(test_vm)
snapshot2 = new_snapshot(full_test_vm)
snapshot2.create()

if provider.one_of(RHEVMProvider):
test_vm.power_control_from_cfme(option=test_vm.POWER_OFF, cancel=False)
test_vm.wait_for_vm_state_change(
desired_state=test_vm.STATE_OFF, timeout=900)
full_test_vm.power_control_from_cfme(option=full_test_vm.POWER_OFF, cancel=False)
full_test_vm.wait_for_vm_state_change(
desired_state=full_test_vm.STATE_OFF, timeout=900)

snapshot1.revert_to()

# Wait for the snapshot to become active
logger.info('Waiting for vm %s to become active', snapshot1.name)
wait_for(lambda: snapshot1.active, num_sec=300, delay=20, fail_func=sel.refresh)
test_vm.wait_for_vm_state_change(desired_state=test_vm.STATE_OFF, timeout=720)
test_vm.power_control_from_cfme(option=test_vm.POWER_ON, cancel=False)
test_vm.wait_for_vm_state_change(desired_state=test_vm.STATE_ON, timeout=900)
current_state = test_vm.find_quadicon().state
full_test_vm.wait_for_vm_state_change(desired_state=full_test_vm.STATE_OFF, timeout=720)
full_test_vm.power_control_from_cfme(option=full_test_vm.POWER_ON, cancel=False)
full_test_vm.wait_for_vm_state_change(desired_state=full_test_vm.STATE_ON, timeout=900)
current_state = full_test_vm.find_quadicon().state
soft_assert(current_state.startswith('currentstate-on'),
"Quadicon state is {}".format(current_state))
soft_assert(test_vm.provider.mgmt.is_vm_running(test_vm.name), "vm not running")
soft_assert(full_test_vm.provider.mgmt.is_vm_running(full_test_vm.name), "vm not running")
wait_for(lambda: ssh_client.run_command('test -e snapshot1.txt').rc == 0,
num_sec=400, delay=20, handle_exception=True, fail_func=ssh_client.close())
try:
Expand All @@ -170,7 +174,7 @@ def test_verify_revert_snapshot(test_vm, provider, soft_assert, register_event,


@pytest.mark.uncollectif(lambda provider: not provider.one_of(VMwareProvider))
def test_create_snapshot_via_ae(request, domain, test_vm):
def test_create_snapshot_via_ae(request, domain, small_test_vm):
"""This test checks whether the vm.create_snapshot works in AE.

Prerequisities:
Expand Down Expand Up @@ -206,12 +210,12 @@ def test_create_snapshot_via_ae(request, domain, test_vm):

# SIMULATE
snap_name = fauxfactory.gen_alpha()
snapshot = Vm.Snapshot(name=snap_name, parent_vm=test_vm)
snapshot = Vm.Snapshot(name=snap_name, parent_vm=small_test_vm)
simulate(
instance="Request",
request="snapshot",
target_type='VM and Instance',
target_object=test_vm.name,
target_object=small_test_vm.name,
execute_methods=True,
attributes_values={"snap_name": snap_name})

Expand Down