Source code for cfme.cloud.instance

import attr
from navmazing import NavigateToSibling, NavigateToAttribute, NavigationDestinationNotFound
from widgetastic.exceptions import NoSuchElementException
from widgetastic_patternfly import CheckableBootstrapTreeview, Dropdown, Button
from widgetastic.utils import VersionPick, Version, partial_match
from widgetastic.widget import View

from cfme.base.login import BaseLoggedInPage
from cfme.common.vm import VM, VMCollection
from cfme.common.vm_views import (
    ProvisionView, VMToolbar, VMEntities, VMDetailsEntities, EditView,
    SetOwnershipView, ManagementEngineView, PolicySimulationView, RetirementViewWithOffset)
from cfme.exceptions import (InstanceNotFound, ItemNotFound, DestinationNotFound)
from cfme.utils.appliance.implementations.ui import navigate_to, CFMENavigateStep, navigator
from cfme.utils.log import logger
from cfme.utils.providers import get_crud_by_name
from cfme.utils.wait import wait_for
from widgetastic_manageiq import (ManageIQTree, TimelinesView, Accordion, CompareToolBarActionsView,
    Search)


[docs]class InstanceDetailsToolbar(View): """ The toolbar on the details screen for an instance """ reload = Button(title=VersionPick({Version.lowest(): 'Reload current display', '5.9': 'Refresh this page'})) configuration = Dropdown('Configuration') policy = Dropdown('Policy') lifecycle = Dropdown('Lifecycle') monitoring = Dropdown('Monitoring') power = Dropdown('Instance Power Functions') # title download = Button(title='Download summary in PDF format') access = Dropdown("Access")
[docs]class InstanceAccordion(View): """ The accordion on the instances page """ @View.nested class instances_by_provider(Accordion): # noqa ACCORDION_NAME = 'Instances by Provider' tree = ManageIQTree() @View.nested class images_by_provider(Accordion): # noqa ACCORDION_NAME = 'Images by Provider' tree = ManageIQTree() @View.nested class instances(Accordion): # noqa ACCORDION_NAME = 'Instances' tree = ManageIQTree() @View.nested class images(Accordion): # noqa ACCORDION_NAME = 'Images' tree = ManageIQTree()
[docs]class InstanceCompareAccordion(View): """ The accordion on the instance comparison page """ @View.nested class comparison_sections(Accordion): # noqa ACCORDION_NAME = 'Comparison Sections' tree = CheckableBootstrapTreeview() apply = Button('Apply')
[docs]class CloudInstanceView(BaseLoggedInPage): """Base view for header/nav check, inherit for navigatable views""" @property def in_cloud_instance(self): return ( self.logged_in_as_current_user and self.navigation.currently_selected == ['Compute', 'Clouds', 'Instances'] )
[docs]class InstanceAllView(CloudInstanceView): """ The collection page for instances """ @property def is_displayed(self): return ( self.in_cloud_instance and self.entities.title.text == 'All Instances' and self.sidebar.instances.is_opened) actions = View.nested(CompareToolBarActionsView) toolbar = View.nested(VMToolbar) sidebar = View.nested(InstanceAccordion) search = View.nested(Search) including_entities = View.include(VMEntities, use_parent=True)
[docs]class InstanceProviderAllView(CloudInstanceView): @property def is_displayed(self): return ( self.in_cloud_instance and self.entities.title.text == 'Instances under Provider "{}"' .format(self.context['object'].provider.name) and self.sidebar.instances_by_provider.is_opened) toolbar = View.nested(VMToolbar) sidebar = View.nested(InstanceAccordion) including_entities = View.include(VMEntities, use_parent=True) @View.nested class instances_by_provider(Accordion): # noqa ACCORDION_NAME = 'Instances by Provider' tree = ManageIQTree()
[docs]class InstanceDetailsView(CloudInstanceView): @property def is_displayed(self): expected_name = self.context['object'].name expected_provider = self.context['object'].provider.name try: # Not displayed when the instance is archived relationships = self.entities.summary('Relationships') relationship_provider_name = relationships.get_text_of('Cloud Provider') except (NameError, NoSuchElementException): logger.warning('No "Cloud Provider" Relationship, assume instance view not displayed') return False return ( self.in_cloud_instance and self.entities.title.text == 'Instance "{}"'.format(expected_name) and relationship_provider_name == expected_provider) toolbar = View.nested(InstanceDetailsToolbar) sidebar = View.nested(InstanceAccordion) entities = View.nested(VMDetailsEntities)
[docs]class InstanceTimelinesView(TimelinesView, CloudInstanceView): @property def is_displayed(self): expected_name = self.context['object'].name return ( self.in_cloud_instance and self.title.text == 'Timelines for Instance "{}"'.format(expected_name))
[docs]class InstanceCompareView(CloudInstanceView): """ The comparison page for instances """ @property def is_displayed(self): return self.in_cloud_instance and self.entities.title.text == 'Compare VM or Template' toolbar = View.nested(CompareToolBarActionsView) sidebar = View.nested(InstanceCompareAccordion)
@attr.s
[docs]class Instance(VM): """Represents a generic instance in CFME. This class is used if none of the inherited classes will match. Args: name: Name of the instance provider: :py:class:`cfme.cloud.provider.Provider` object template_name: Name of the template to use for provisioning appliance: :py:class: `utils.appliance.IPAppliance` object Note: This class cannot be instantiated. Use :py:func:`instance_factory` instead. """ ALL_LIST_LOCATION = "clouds_instances" TO_RETIRE = "Retire this Instance" QUADICON_TYPE = "instance" VM_TYPE = "Instance" REMOVE_SINGLE = 'Remove Instance' TO_OPEN_EDIT = "Edit this Instance" DETAILS_VIEW_CLASS = InstanceDetailsView
[docs] def update(self, values, cancel=False, reset=False): """Update cloud instance Args: values: Dictionary of form key/value pairs cancel: Boolean, cancel the form submission reset: Boolean, reset form after fill - returns immediately after reset Note: The edit form contains a 'Reset' button - if this is c """ view = navigate_to(self, 'Edit') # form is the view's parent view.form.fill(values) if reset: view.form.reset_button.click() return else: button = view.form.cancel_button if cancel else view.form.submit_button button.click()
[docs] def on_details(self, force=False): """A function to determine if the browser is already on the proper instance details page. An instance may not be assigned to a provider if archived or orphaned If no provider is listed, default to False since we may be on the details page for an instance on the wrong provider. """ if not force: return self.browser.create_view(InstanceDetailsView).is_displayed else: navigate_to(self, 'Details') return True
[docs] def get_vm_via_rest(self): # Try except block, because instances collection isn't available on 5.4 try: instance = self.appliance.rest_api.collections.instances.get(name=self.name) except AttributeError: raise Exception("Collection instances isn't available") else: return instance
[docs] def get_collection_via_rest(self): return self.appliance.rest_api.collections.instances
[docs] def wait_for_instance_state_change(self, desired_state, timeout=900): """Wait for an instance to come to desired state. This function waits just the needed amount of time thanks to wait_for. Args: desired_state: A string or list of strings indicating desired state timeout: Specify amount of time (in seconds) to wait until TimedOutError is raised """ def _looking_for_state_change(): view = navigate_to(self, 'Details') current_state = view.entities.summary("Power Management").get_text_of("Power State") logger.info('Current Instance state: {}'.format(current_state)) logger.info('Desired Instance state: {}'.format(desired_state)) if isinstance(desired_state, (list, tuple)): return current_state in desired_state else: return current_state == desired_state return wait_for(_looking_for_state_change, num_sec=timeout, delay=15, message='Checking for instance state change', fail_func=self.provider.refresh_provider_relationships, handle_exception=True)
[docs] def find_quadicon(self, **kwargs): """Find and return a quadicon belonging to a specific instance TODO: remove this method and refactor callers to use view entities instead Args: Returns: entity of appropriate type """ view = navigate_to(self.parent, 'All') view.toolbar.view_selector.select('Grid View') try: return view.entities.get_entity(name=self.name, surf_pages=True) except ItemNotFound: raise InstanceNotFound("Instance '{}' not found in UI!".format(self.name))
@property def exists(self): try: navigate_to(self, 'Details') return True except InstanceNotFound: return False
[docs] def power_control_from_cfme(self, *args, **kwargs): """Power controls a VM from within CFME using details or collection Raises: InstanceNotFound: the instance wasn't found when navigating OptionNotAvailable: option param is not visible or enabled """ # TODO push this to common.vm when infra vm classes have widgets if not kwargs.get('option'): raise ValueError('Need to provide option for power_control_from_cfme, no default.') if kwargs.get('from_details', True): view = navigate_to(self, 'Details') else: view = navigate_to(self, 'AllForProvider') view.toolbar.view_selector.select('List View') try: row = view.entities.get_entity(name=self.name) except ItemNotFound: raise InstanceNotFound('Failed to find instance in table: {}'.format(self.name)) row.check() # cancel is the kwarg, when true we want item_select to dismiss the alert, flip the bool view.toolbar.power.item_select(kwargs.get('option'), handle_alert=not kwargs.get('cancel', False))
@property def vm_default_args(self): """Represents dictionary used for Vm/Instance provision with minimum required default args """ provisioning = self.provider.data['provisioning'] inst_args = { 'request': {'email': 'vm_provision@cfmeqe.com'}, 'catalog': { 'vm_name': self.name}, 'environment': { 'availability_zone': provisioning.get('availability_zone'), 'cloud_network': provisioning.get('cloud_network'), 'cloud_subnet': provisioning.get('cloud_subnet'), 'resource_groups': provisioning.get('resource_group') }, 'properties': { 'instance_type': partial_match(provisioning.get('instance_type')), 'guest_keypair': provisioning.get('guest_keypair') } } return inst_args @property def vm_default_args_rest(self): """Represents dictionary used for REST API Instance provision with minimum required default args """ from cfme.cloud.provider.azure import AzureProvider from cfme.cloud.provider.ec2 import EC2Provider if not self.provider.is_refreshed(): self.provider.refresh_provider_relationships() wait_for(self.provider.is_refreshed, func_kwargs=dict(refresh_delta=10), timeout=600) provisioning = self.provider.data['provisioning'] provider_rest = self.appliance.rest_api.collections.providers.get(name=self.provider.name) # find out image guid image_name = provisioning['image']['name'] image = self.appliance.rest_api.collections.templates.get(name=image_name, ems_id=provider_rest.id) # find out flavor if ':' in provisioning['instance_type'] and self.provider.one_of(EC2Provider): instance_type = provisioning['instance_type'].split(':')[0].strip() else: instance_type = provisioning['instance_type'] flavor = self.appliance.rest_api.collections.flavors.get(name=instance_type, ems_id=provider_rest.id) # find out cloud network cloud_network_name = provisioning.get('cloud_network').strip() if self.provider.one_of(EC2Provider, AzureProvider): cloud_network_name = cloud_network_name.split()[0] cloud_network = self.appliance.rest_api.collections.cloud_networks.get( name=cloud_network_name, enabled='true') # find out cloud subnet cloud_subnet = self.appliance.rest_api.collections.cloud_subnets.get( cloud_network_id=cloud_network['id']) # find out availability zone azone_id = None av_zone_name = provisioning.get('availability_zone') if av_zone_name: azone_id = self.appliance.rest_api.collections.availability_zones.get( name=av_zone_name, ems_id=flavor.ems_id).id # find out cloud tenant tenant_name = provisioning.get('cloud_tenant') if tenant_name: try: tenant = self.appliance.rest_api.collections.cloud_tenants.get( name=tenant_name, ems_id=provider_rest.id, enabled='true') except IndexError: raise ItemNotFound("Tenant {} not found on provider {}".format( tenant_name, self.provider.name)) resource_group_id = None if self.provider.one_of(AzureProvider): resource_groups = self.appliance.rest_api.get( '{}?attributes=resource_groups'.format(provider_rest._href))['resource_groups'] resource_group_id = None resource_group_name = provisioning.get('resource_group') for res_group in resource_groups: if (res_group['name'] == resource_group_name and res_group['ems_id'] == provider_rest.id): resource_group_id = res_group['id'] break inst_args = { "version": "1.1", "template_fields": { "guid": image.guid }, "vm_fields": { "placement_auto": False, "vm_name": self.name, "instance_type": flavor['id'], "request_type": "template", "cloud_network": cloud_network['id'], "cloud_subnet": cloud_subnet['id'], }, "requester": { "user_name": "admin", "owner_email": "admin@example.com", "auto_approve": True, }, "tags": { }, "ems_custom_attributes": { }, "miq_custom_attributes": { } } if tenant_name: inst_args['vm_fields']['cloud_tenant'] = tenant['id'] if resource_group_id: inst_args['vm_fields']['resource_group'] = resource_group_id if azone_id: inst_args['vm_fields']['placement_availability_zone'] = azone_id if self.provider.one_of(EC2Provider): inst_args['vm_fields']['monitoring'] = 'basic' return inst_args
@attr.s
[docs]class InstanceCollection(VMCollection): ENTITY = Instance
[docs] def all(self): """Return entities for all items in collection""" # Pretty much same as image, but defining at VMCollection would only work for cloud # provider filter means we're viewing instances through provider details relationships provider = self.filters.get('provider') # None if no filter, need for entity instantiation view = navigate_to(self, 'AllForProvider' if provider else 'All') # iterate pages here instead of use surf_pages=True because data is needed entities = [] for _ in view.entities.paginator.pages(): # auto-resets to first page page_entities = [entity for entity in view.entities.get_all(surf_pages=False)] entities.extend( # when provider filtered view, there's no provider data value [self.instantiate(e.data['name'], provider or get_crud_by_name(e.data['provider'])) for e in page_entities if e.data.get('provider') != ''] # safe provider check, archived shows no provider ) return entities
@navigator.register(InstanceCollection, 'All')
[docs]class All(CFMENavigateStep): VIEW = InstanceAllView prerequisite = NavigateToAttribute('appliance.server', 'LoggedIn')
[docs] def step(self): self.prerequisite_view.navigation.select('Compute', 'Clouds', 'Instances') self.view.sidebar.instances.tree.click_path('All Instances')
[docs] def resetter(self, *args, **kwargs): # If a filter was applied, it will persist through navigation and needs to be cleared self.view.entities.search.remove_search_filters() self.view.toolbar.reload.click()
@navigator.register(InstanceCollection, 'AllForProvider') @navigator.register(Instance, 'AllForProvider')
[docs]class AllForProvider(CFMENavigateStep): VIEW = InstanceProviderAllView
[docs] def prerequisite(self): try: view = navigate_to(self.obj, 'All') except NavigationDestinationNotFound: view = navigate_to(self.obj.parent, 'All') finally: return view
[docs] def step(self, *args, **kwargs): if isinstance(self.obj, InstanceCollection) and self.obj.filters.get('provider'): # the collection is navigation target, use its filter value provider_name = self.obj.filters['provider'].name elif isinstance(self.obj, Instance): provider_name = self.obj.provider.name else: raise DestinationNotFound("Unable to identify a provider for AllForProvider navigation") self.view.sidebar.instances_by_provider.tree.click_path('Instances by Provider', provider_name)
[docs] def resetter(self, *args, **kwargs): # If a filter was applied, it will persist through navigation and needs to be cleared if self.view.adv_search_clear.is_displayed: logger.debug('Clearing advanced search filter') self.view.adv_search_clear.click() self.view.toolbar.reload.click()
@navigator.register(Instance, 'Details')
[docs]class Details(CFMENavigateStep): VIEW = InstanceDetailsView
[docs] def prerequisite(self, *args, **kwargs): return navigate_to(self.obj.parent, 'AllForProvider' if self.obj.parent.filters.get('provider') else 'All')
[docs] def step(self): try: row = self.prerequisite_view.entities.get_entity(name=self.obj.name, surf_pages=True, use_search=True) except ItemNotFound: raise InstanceNotFound('Failed to locate instance with name "{}"'.format(self.obj.name)) row.click()
[docs] def resetter(self, *args, **kwargs): self.view.toolbar.reload.click()
@navigator.register(Instance, 'ArchiveDetails')
[docs]class ArchiveDetails(CFMENavigateStep): VIEW = InstanceDetailsView prerequisite = NavigateToAttribute('parent', 'All')
[docs] def step(self): try: row = self.prerequisite_view.entities.get_entity(name=self.obj.name, surf_pages=True, use_search=True) except ItemNotFound: raise InstanceNotFound('Failed to locate instance with name "{}"'.format(self.obj.name)) row.click()
[docs] def resetter(self, *args, **kwargs): self.view.toolbar.reload.click()
@navigator.register(Instance, 'Edit')
[docs]class Edit(CFMENavigateStep): VIEW = EditView prerequisite = NavigateToSibling('Details')
[docs] def step(self, *args, **kwargs): self.prerequisite_view.toolbar.configuration.item_select('Edit this Instance')
@navigator.register(Instance, 'EditManagementEngineRelationship')
[docs]class EditManagementEngineRelationship(CFMENavigateStep): VIEW = ManagementEngineView prerequisite = NavigateToSibling('Details')
[docs] def step(self, *args, **kwargs): configuration = self.prerequisite_view.toolbar.configuration configuration.item_select('Edit Management Engine Relationship')
@navigator.register(InstanceCollection, 'Provision')
[docs]class Provision(CFMENavigateStep): VIEW = ProvisionView prerequisite = NavigateToSibling('All')
[docs] def step(self, *args, **kwargs): self.prerequisite_view.toolbar.lifecycle.item_select('Provision Instances')
@navigator.register(Instance, 'PolicySimulation')
[docs]class PolicySimulation(CFMENavigateStep): VIEW = PolicySimulationView prerequisite = NavigateToSibling('Details')
[docs] def step(self, *args, **kwargs): self.prerequisite_view.toolbar.policy.item_select('Policy Simulation')
@navigator.register(Instance, 'SetOwnership')
[docs]class SetOwnership(CFMENavigateStep): VIEW = SetOwnershipView prerequisite = NavigateToSibling('Details') # No am_i_here because the page only indicates name and not provider
[docs] def step(self, *args, **kwargs): self.prerequisite_view.toolbar.configuration.item_select('Set Ownership')
@navigator.register(Instance, 'SetRetirement')
[docs]class SetRetirement(CFMENavigateStep): VIEW = RetirementViewWithOffset prerequisite = NavigateToSibling('Details')
[docs] def step(self, *args, **kwargs): self.prerequisite_view.toolbar.lifecycle.item_select('Set Retirement Date')
@navigator.register(Instance, 'Timelines')
[docs]class Timelines(CFMENavigateStep): VIEW = InstanceTimelinesView prerequisite = NavigateToSibling('ArchiveDetails')
[docs] def step(self, *args, **kwargs): self.prerequisite_view.toolbar.monitoring.item_select('Timelines')
@navigator.register(Instance, 'candu')
[docs]class InstanceUtilization(CFMENavigateStep): @property def VIEW(self): # noqa return self.obj.provider.vm_utilization_view prerequisite = NavigateToSibling('Details')
[docs] def step(self): self.prerequisite_view.toolbar.monitoring.item_select('Utilization')