# -*- coding: utf-8 -*-
import fauxfactory
from widgetastic.utils import partial_match
from cfme.infrastructure.provider import InfraProvider
from cfme.infrastructure.provider.rhevm import RHEVMProvider
from cfme.infrastructure.provider.virtualcenter import VMwareProvider
from cfme.utils import version
from cfme.utils.log import logger
from cfme.utils.rest import create_resource
from cfme.utils.virtual_machines import deploy_template
from cfme.utils.wait import wait_for
from cfme.fixtures.provider import setup_one_by_class_or_skip
from wrapanapi import VmState
TEMPLATE_TORSO = """{
"AWSTemplateFormatVersion" : "2010-09-09",
"Description" : "AWS CloudFormation Sample Template Rails_Single_Instance.",
"Parameters" : {
"KeyName": {
"Description" : "Name of an existing EC2 KeyPair to enable SSH access to the instances",
"Type": "AWS::EC2::KeyPair::KeyName",
"ConstraintDescription" : "must be the name of an existing EC2 KeyPair."
}
}
}
"""
[docs]def service_catalogs(request, rest_api, num=5):
"""Create service catalogs using REST API."""
scls_data = []
for _ in range(num):
scls_data.append({
'name': 'cat_{}'.format(fauxfactory.gen_alphanumeric()),
'description': 'my catalog',
'service_templates': []
})
return _creating_skeleton(request, rest_api, 'service_catalogs', scls_data, col_action='add')
[docs]def service_catalog_obj(request, appliance):
"""Return service catalog object."""
rest_catalog = service_catalogs(request, appliance.rest_api, num=1)[0]
return appliance.collections.catalogs.instantiate(name=rest_catalog.name,
description=rest_catalog.description)
[docs]def categories(request, rest_api, num=1):
ctg_data = []
for _ in range(num):
uniq = fauxfactory.gen_alphanumeric().lower()
ctg_data.append({
'name': 'test_category_{}'.format(uniq),
'description': 'test_category_{}'.format(uniq)
})
return _creating_skeleton(request, rest_api, 'categories', ctg_data)
[docs]def dialog_rest(request, rest_api):
"""Creates service dialog using REST API."""
uid = fauxfactory.gen_alphanumeric()
data = {
"description": "my dialog {}".format(uid),
"label": uid,
"buttons": "submit,cancel",
"dialog_tabs": [{
"description": "my tab desc {}".format(uid),
"position": 0,
"label": "tab_{}".format(uid),
"display": "edit",
"dialog_groups": [{
"description": "my box desc {}".format(uid),
"label": "box_{}".format(uid),
"display": "edit",
"position": 0,
"dialog_fields": [{
"name": "service_name",
"description": "my ele desc {}".format(uid),
"label": "ele_{}".format(uid),
"data_type": "string",
"display": "edit",
"required": False,
"default_value": uid,
"options": {
"protected": False
},
"position": 0,
"dynamic": False,
"read_only": False,
"visible": True,
"type": "DialogFieldTextBox",
"resource_action": {
"resource_type": "DialogField",
"ae_attributes": {}
}
}]
}]
}]
}
service_dialog = _creating_skeleton(request, rest_api, "service_dialogs", [data])
return service_dialog[0]
[docs]def dialog(request, appliance):
"""Returns service dialog object."""
rest_resource = dialog_rest(request, appliance.rest_api)
service_dialogs = appliance.collections.service_dialogs
service_dialog = service_dialogs.instantiate(
label=rest_resource.label,
description=rest_resource.description)
return service_dialog
[docs]def services(request, appliance, a_provider, service_dialog=None, service_catalog=None):
"""
The attempt to add the service entities via web
"""
service_template = service_templates_ui(
request,
appliance,
service_dialog=service_dialog,
service_catalog=service_catalog,
a_provider=a_provider,
num=1
)
service_template = service_template[0]
service_catalog = appliance.rest_api.get_entity(
'service_catalogs',
service_template.service_template_catalog_id
)
template_subcollection = appliance.rest_api.get_entity(
service_catalog.service_templates,
service_template.id
)
template_subcollection.action.order()
results = appliance.rest_api.response.json()
service_request = appliance.rest_api.get_entity('service_requests', results['id'])
def _order_finished():
service_request.reload()
return service_request.request_state.lower() == 'finished'
wait_for(_order_finished, num_sec=2000, delay=10)
assert 'error' not in service_request.message.lower(), \
'Provisioning failed with the message `{}`'.format(service_request.message)
service_name = str(service_request.options['dialog']['dialog_service_name'])
assert '[{}]'.format(service_name) in service_request.message
provisioned_service = appliance.rest_api.collections.services.get(
service_template_id=service_template.id)
@request.addfinalizer
def _finished():
try:
provisioned_service.action.delete()
except Exception:
# service can be deleted by test
logger.warning('Failed to delete service `{}`.'.format(service_name))
# tests expect iterable
return [provisioned_service]
[docs]def rates(request, rest_api, num=3):
chargeback = rest_api.collections.chargebacks.get(rate_type='Compute')
data = []
for _ in range(num):
req = {'description': 'test_rate_{}'.format(fauxfactory.gen_alphanumeric()),
'source': 'allocated',
'group': 'cpu',
'per_time': 'daily',
'per_unit': 'megahertz',
'chargeback_rate_id': chargeback.id,
'chargeable_field_id': chargeback.id}
data.append(req)
return _creating_skeleton(request, rest_api, 'rates', data)
[docs]def a_provider(request):
return setup_one_by_class_or_skip(request, InfraProvider)
[docs]def vm(request, a_provider, rest_api):
provider_rest = rest_api.collections.providers.get(name=a_provider.name)
vm = deploy_template(
a_provider.key,
'test_rest_vm_{}'.format(fauxfactory.gen_alphanumeric(length=4))
)
vm_name = vm.name
@request.addfinalizer
def _finished():
try:
vm.cleanup()
except Exception:
# vm can be deleted/retired by test
logger.warning("Failed to delete vm %r", vm)
provider_rest.action.refresh()
wait_for(
lambda: rest_api.collections.vms.find_by(name=vm_name) or False,
num_sec=600, delay=5)
return vm_name
[docs]def service_templates_ui(request, appliance, service_dialog=None, service_catalog=None,
a_provider=None, num=4):
if not service_dialog:
service_dialog = dialog(request, appliance)
if not service_catalog:
service_catalog = service_catalog_obj(request, appliance)
cat_items_col = appliance.collections.catalog_items
catalog_item_type = a_provider.catalog_item_type if a_provider else cat_items_col.GENERIC
catalog_items = []
new_names = []
for _ in range(num):
if a_provider:
template, host, datastore, vlan = map(
a_provider.data.get('provisioning').get,
('template', 'host', 'datastore', 'vlan'))
vm_name = 'test_rest_{}'.format(fauxfactory.gen_alphanumeric())
provisioning_data = {
'catalog': {'catalog_name': {'name': template},
'vm_name': vm_name},
'environment': {'host_name': {'name': host},
'datastore_name': {'name': datastore},
},
'network': {},
}
if a_provider.one_of(RHEVMProvider):
provisioning_data['catalog']['provision_type'] = 'Native Clone'
provisioning_data['network']['vlan'] = partial_match(vlan)
elif a_provider.one_of(VMwareProvider):
provisioning_data['catalog']['provision_type'] = 'VMware'
provisioning_data['network']['vlan'] = partial_match(vlan)
new_name = 'item_{}'.format(fauxfactory.gen_alphanumeric())
new_names.append(new_name)
cat_items_col.create(
catalog_item_type,
name=new_name,
description='my catalog',
display_in=True,
catalog=service_catalog,
dialog=service_dialog,
prov_data=provisioning_data
)
for catalog_item in catalog_items:
catalog_item.create()
collection = appliance.rest_api.collections.service_templates
for new_name in new_names:
wait_for(lambda: collection.find_by(name=new_name) or False, num_sec=180, delay=10)
s_tpls = [ent for ent in collection if ent.name in new_names]
@request.addfinalizer
def _finished():
collection.reload()
to_delete = [ent for ent in collection if ent.name in new_names]
if to_delete:
collection.action.delete(*to_delete)
return s_tpls
[docs]def service_templates_rest(request, appliance, service_dialog=None, service_catalog=None, num=4):
if not service_dialog:
service_dialog = dialog(request, appliance)
if not service_catalog:
service_catalog = service_catalog_obj(request, appliance)
catalog_id = appliance.rest_api.collections.service_catalogs.get(name=service_catalog.name).id
dialog_id = appliance.rest_api.collections.service_dialogs.get(label=service_dialog.label).id
data = []
for _ in range(num):
uniq = fauxfactory.gen_alphanumeric(5)
data.append({
"name": 'item_{}'.format(uniq),
"description": "my catalog {}".format(uniq),
"service_type": "atomic",
"prov_type": "generic",
"display": True,
"service_template_catalog_id": catalog_id,
"config_info": {
"provision": {
"dialog_id": dialog_id,
"fqname": "/Service/Provisioning/StateMachines/"
"ServiceProvision_Template/CatalogItemInitialization"
},
"retirement": {
"dialog_id": dialog_id,
"fqname": "/Service/Retirement/StateMachines/ServiceRetirement/Default"
},
}
})
return _creating_skeleton(request, appliance.rest_api, "service_templates", data)
[docs]def service_templates(request, appliance, service_dialog=None, service_catalog=None, num=4):
# TODO: remove, because it copies service_templates_rest for supported versions.
# tmplt = service_templates_ui if appliance.version < '5.8' else service_templates_rest
return service_templates_rest(
request, appliance, service_dialog=service_dialog, service_catalog=service_catalog, num=num)
[docs]def automation_requests_data(vm, requests_collection=False, approve=True, num=4):
# for creating automation request using /api/automation_requests
automation_requests_col = {
"uri_parts": {
"namespace": "System",
"class": "Request",
"instance": "InspectME",
"message": "create",
},
"parameters": {
"vm_name": vm,
},
"requester": {
"auto_approve": approve
}
}
# for creating automation request using /api/requests
requests_col = {
"options": {
"request_type": "automation",
"message": "create",
"namespace": "System",
"class_name": "Request",
"instance_name": "InspectME",
"attrs": {
"vm_name": vm,
"userid": "admin"
}
},
"requester": {
"user_name": "admin"
},
"auto_approve": approve
}
data = requests_col if requests_collection else automation_requests_col
return [data for _ in range(num)]
[docs]def groups(request, rest_api, role, tenant, num=1):
data = []
for _ in range(num):
data.append({
"description": "group_description_{}".format(fauxfactory.gen_alphanumeric()),
"role": {"href": role.href},
"tenant": {"href": tenant.href}
})
groups = _creating_skeleton(request, rest_api, "groups", data)
if num == 1:
return groups.pop()
return groups
[docs]def roles(request, rest_api, num=1):
data = []
for _ in range(num):
data.append({"name": "role_name_{}".format(fauxfactory.gen_alphanumeric())})
roles = _creating_skeleton(request, rest_api, "roles", data)
if num == 1:
return roles.pop()
return roles
[docs]def copy_role(rest_api, orig_name, new_name=None):
orig_role = rest_api.collections.roles.get(name=orig_name)
orig_features = orig_role._data.get('features')
orig_settings = orig_role._data.get('settings')
if not orig_features and hasattr(orig_role, 'features'):
features_subcol = orig_role.features
features_subcol.reload()
orig_features = features_subcol._data.get('resources')
if not orig_features:
raise NotImplementedError('Role copy is not implemented for this version.')
new_role = rest_api.collections.roles.action.create(
name=new_name or 'EvmRole-{}'.format(fauxfactory.gen_alphanumeric()),
features=orig_features,
settings=orig_settings
)
return new_role[0]
[docs]def tenants(request, rest_api, num=1):
parent = rest_api.collections.tenants.get(name='My Company')
data = []
for _ in range(num):
uniq = fauxfactory.gen_alphanumeric()
data.append({
'description': 'test_tenants_{}'.format(uniq),
'name': 'test_tenants_{}'.format(uniq),
'divisible': 'true',
'use_config_for_attributes': 'false',
'parent': {'href': parent.href}
})
tenants = _creating_skeleton(request, rest_api, 'tenants', data)
if num == 1:
return tenants.pop()
return tenants
[docs]def users(request, rest_api, num=1):
data = []
for _ in range(num):
uniq = fauxfactory.gen_alphanumeric(4).lower()
data.append({
"userid": "user_{}".format(uniq),
"name": "name_{}".format(uniq),
"password": fauxfactory.gen_alphanumeric(),
"email": "user@example.com",
"group": {"description": "EvmGroup-user_self_service"}
})
resources = _creating_skeleton(request, rest_api, "users", data)
return resources, data
def _creating_skeleton(request, rest_api, col_name, col_data, col_action='create',
substr_search=False):
entities = create_resource(
rest_api, col_name, col_data, col_action=col_action, substr_search=substr_search)
# make sure the original list of `entities` is preserved for cleanup
original_entities = list(entities)
@request.addfinalizer
def _finished():
collection = getattr(rest_api.collections, col_name)
collection.reload()
ids = [e.id for e in original_entities]
delete_entities = [e for e in collection if e.id in ids]
if delete_entities:
collection.action.delete(*delete_entities)
return entities
[docs]def mark_vm_as_template(rest_api, provider, vm_name):
"""
Function marks vm as template via mgmt and returns template Entity
Usage:
mark_vm_as_template(rest_api, provider, vm_name)
"""
t_vm = rest_api.collections.vms.get(name=vm_name)
t_vm.action.stop()
vm_mgmt = provider.mgmt.get_vm(vm_name)
vm_mgmt.ensure_state(VmState.STOPPED, timeout=1000)
vm_mgmt.mark_as_template()
wait_for(
lambda: rest_api.collections.templates.find_by(name=vm_name).subcount != 0,
num_sec=700, delay=15)
return rest_api.collections.templates.get(name=vm_name)
[docs]def arbitration_settings(request, rest_api, num=2):
data = []
for _ in range(num):
uniq = fauxfactory.gen_alphanumeric(5)
data.append({
'name': 'test_settings_{}'.format(uniq),
'display_name': 'Test Settings {}'.format(uniq)})
return _creating_skeleton(request, rest_api, 'arbitration_settings', data)
[docs]def orchestration_templates(request, rest_api, num=2):
data = []
for _ in range(num):
uniq = fauxfactory.gen_alphanumeric(5)
data.append({
'name': 'test_{}'.format(uniq),
'description': 'Test Template {}'.format(uniq),
'type': version.pick({
version.LOWEST: 'OrchestrationTemplateCfn',
'5.9': 'ManageIQ::Providers::Amazon::CloudManager::OrchestrationTemplate'}),
'orderable': False,
'draft': False,
'content': TEMPLATE_TORSO.replace('CloudFormation', uniq)})
return _creating_skeleton(request, rest_api, 'orchestration_templates', data)
[docs]def arbitration_profiles(request, rest_api, a_provider, num=2):
provider = rest_api.collections.providers.get(name=a_provider.name)
data = []
providers = [{'id': provider.id}, {'href': provider.href}]
for index in range(num):
data.append({
'name': 'test_settings_{}'.format(fauxfactory.gen_alphanumeric(5)),
'provider': providers[index % 2]
})
return _creating_skeleton(request, rest_api, 'arbitration_profiles', data)
[docs]def arbitration_rules(request, rest_api, num=2):
data = []
for _ in range(num):
data.append({
'description': 'test admin rule {}'.format(fauxfactory.gen_alphanumeric(5)),
'operation': 'inject',
'expression': {'EQUAL': {'field': 'User-userid', 'value': 'admin'}}
})
return _creating_skeleton(request, rest_api, 'arbitration_rules', data)
[docs]def blueprints(request, rest_api, num=2):
data = []
for _ in range(num):
uniq = fauxfactory.gen_alphanumeric(5)
data.append({
'name': 'test_blueprint_{}'.format(uniq),
'description': 'Test Blueprint {}'.format(uniq),
'ui_properties': {
'service_catalog': {},
'service_dialog': {},
'automate_entrypoints': {},
'chart_data_model': {}
}
})
return _creating_skeleton(request, rest_api, 'blueprints', data)
[docs]def conditions(request, rest_api, num=2):
data = []
for _ in range(num):
uniq = fauxfactory.gen_alphanumeric(5)
data.append({
'name': 'test_condition_{}'.format(uniq),
'description': 'Test Condition {}'.format(uniq),
'expression': {'=': {'field': 'ContainerImage-architecture', 'value': 'dsa'}},
'towhat': 'ExtManagementSystem',
'modifier': 'allow'
})
return _creating_skeleton(request, rest_api, 'conditions', data)
[docs]def policies(request, rest_api, num=2):
conditions_response = conditions(request, rest_api, num=2)
data = []
for _ in range(num):
uniq = fauxfactory.gen_alphanumeric(5)
data.append({
'name': 'test_policy_{}'.format(uniq),
'description': 'Test Policy {}'.format(uniq),
'mode': 'compliance',
'towhat': 'ManageIQ::Providers::Redhat::InfraManager',
'conditions_ids': [conditions_response[0].id, conditions_response[1].id],
'policy_contents': [{
'event_id': 2,
'actions': [{'action_id': 1, 'opts': {'qualifier': 'failure'}}]
}]
})
return _creating_skeleton(request, rest_api, 'policies', data)