Source code for cfme.infrastructure.config_management

# -*- coding: utf-8 -*-
from navmazing import NavigateToSibling, NavigateToAttribute
from widgetastic.exceptions import NoSuchElementException
from widgetastic.widget import Checkbox, TextInput, Text, View
from widgetastic_patternfly import BootstrapSelect, Dropdown, Tab

from cfme.base.credential import Credential as BaseCredential
from cfme.base.login import BaseLoggedInPage
from cfme.common import Taggable, TagPageView
from cfme.utils import conf, ParamClassName
from cfme.utils.appliance import Navigatable
from cfme.utils.appliance.implementations.ui import navigator, CFMENavigateStep, navigate_to
from cfme.utils.log import logger
from cfme.utils.pretty import Pretty
from cfme.utils.update import Updateable
from cfme.utils.version import LATEST, LOWEST, VersionPicker
from cfme.utils.wait import wait_for
from widgetastic_manageiq import (
    Accordion, BaseEntitiesView, Button, ItemsToolBarViewSelector, ManageIQTree, SummaryTable,
    Version, VersionPick, Table, Search)


[docs]class ConfigManagementToolbar(View): """Toolbar""" refresh = Button(title=VersionPick({Version.lowest(): 'Reload current display', '5.9': 'Refresh this page'})) configuration = Dropdown('Configuration') lifecycle = Dropdown('Lifecycle') policy = Dropdown('Policy') download = Dropdown(title='Download') view_selector = View.nested(ItemsToolBarViewSelector)
[docs]class ConfigManagementDetailsToolbar(View): """Toolbar on the details page""" history = Dropdown(title='History') refresh = Button(title=VersionPick({Version.lowest(): 'Reload current display', '5.9': 'Refresh this page'})) lifecycle = Dropdown('Lifecycle') policy = Dropdown('Policy') download = Button(title='Download summary in PDF format') view_selector = View.nested(ItemsToolBarViewSelector)
[docs]class ConfigManagementSideBar(View): """Side bar""" @View.nested class providers(Accordion): # noqa ACCORDION_NAME = 'Providers' tree = ManageIQTree() @View.nested class configured_systems(Accordion): # noqa ACCORDION_NAME = 'Configured Systems' tree = ManageIQTree() @View.nested class job_templates(Accordion): # noqa ACCORDION_NAME = 'Job Templates' tree = ManageIQTree()
[docs]class ConfigManagementEntities(BaseEntitiesView): """The entities on the page"""
[docs]class ConfigManagementProfileEntities(BaseEntitiesView): """Entities view for the detail page""" @View.nested class summary(Tab): # noqa TAB_NAME = 'Summary' properties = SummaryTable(title='Properties') environment = SummaryTable(title='Environment') operating_system = SummaryTable(title='Operating System') tenancy = SummaryTable(title='Tenancy') smart_management = SummaryTable(title='Smart Management') @View.nested class configured_systems(Tab): # noqa TAB_NAME = 'Configured Systems' elements = Table('//div[@id="main_div"]//div[@id="list_grid" or @id="gtl_div"]//table')
[docs]class ConfigManagementAddForm(View): """Form to add a provider""" name = TextInput('name') provider_type = BootstrapSelect('provider_type') zone = TextInput('zone') url = TextInput('url') ssl = Checkbox('verify_ssl') username = VersionPick({Version.lowest(): TextInput('log_userid'), '5.9': TextInput('default_userid')}) password = VersionPick({Version.lowest(): TextInput('log_password'), '5.9': TextInput('default_password')}) confirm_password = TextInput('log_verify') validate = Button('Validate')
[docs]class ConfigManagementEditForm(View): """Form to add a provider""" name = TextInput('name') provider_type = BootstrapSelect('provider_type') zone = TextInput('zone') url = TextInput('url') ssl = Checkbox('verify_ssl') username = TextInput('log_userid') password = TextInput('log_password') validate = Button('Validate')
[docs]class ConfigManagementAddEntities(View): """The entities on the add page""" title = Text('//div[@id="main-content"]//h1') form = View.nested(ConfigManagementAddForm) add = Button('Add') cancel = Button('Cancel')
[docs]class ConfigManagementEditEntities(View): """The entities on the edit page""" title = Text('//div[@id="main-content"]//h1') form = View.nested(ConfigManagementEditForm) save = Button('Save') reset = Button('Reset') cancel = Button('Cancel')
[docs]class ConfigManagementView(BaseLoggedInPage): """The base page for both the all and details page""" @property def in_config(self): """Determine if we're in the config section""" if (self.context['object'].appliance.version >= '5.8' and self.context['object'].type == 'Ansible Tower'): nav_chain = ['Automation', 'Ansible', 'Ansible Tower'] else: nav_chain = ['Configuration', 'Management'] return self.logged_in_as_current_user and self.navigation.currently_selected == nav_chain
[docs]class ConfigManagementAllView(ConfigManagementView): """The main list view""" toolbar = View.nested(ConfigManagementToolbar) sidebar = View.nested(ConfigManagementSideBar) search = View.nested(Search) including_entities = View.include(ConfigManagementEntities, use_parent=True) @property def is_displayed(self): """Is this view being displayed?""" if self.obj.appliance.version >= '5.8' and self.obj.type == 'Ansible Tower': title_text = 'All Ansible Tower Providers' else: title_text = 'All Configuration Manager Providers' return self.in_config and self.entities.title.text == title_text
[docs]class ConfigManagementDetailsView(ConfigManagementView): """The details page""" toolbar = View.nested(ConfigManagementToolbar) sidebar = View.nested(ConfigManagementSideBar) including_entities = View.include(ConfigManagementEntities, use_parent=True) @property def is_displayed(self): """Is this view being displayed?""" titles = [t.format(name=self.obj.name) for t in [ 'Configuration Profiles under Red Hat Satellite Provider "{name} Automation Manager"', 'Inventory Groups under Ansible Tower Provider "{name} Automation Manager"' ]] return self.in_config and self.entities.title.text in titles
[docs]class ConfigManagementProfileView(ConfigManagementView): """The profile page""" toolbar = View.nested(ConfigManagementDetailsToolbar) sidebar = View.nested(ConfigManagementSideBar) entities = View.nested(ConfigManagementProfileEntities) @property def is_displayed(self): """Is this view being displayed?""" title = 'Configured System ({}) "{}"'.format(self.obj.type, self.obj.name) return self.in_config and self.entities.title.text == title
[docs]class ConfigManagementAddView(ConfigManagementView): """The add page""" sidebar = View.nested(ConfigManagementSideBar) entities = View.nested(ConfigManagementAddEntities) @property def is_displayed(self): """Is this view being displayed?""" return False
[docs]class ConfigManagementEditView(ConfigManagementView): """The edit page""" sidebar = View.nested(ConfigManagementSideBar) entities = View.nested(ConfigManagementEditEntities) @property def is_displayed(self): """Is this view being displayed?""" return False
[docs]class ConfigManager(Updateable, Pretty, Navigatable): """ This is base class for Configuration manager objects (Red Hat Satellite, Foreman, Ansible Tower) Args: name: Name of the config. manager url: URL, hostname or IP of the config. manager ssl: Boolean value; `True` if SSL certificate validity should be checked, `False` otherwise credentials: Credentials to access the config. manager key: Key to access the cfme_data yaml data (same as `name` if not specified) Usage: Use Satellite or AnsibleTower classes instead. """ pretty_attr = ['name', 'url'] _param_name = ParamClassName('name') type = None refresh_flash_msg = 'Refresh Provider initiated for 1 provider' def __init__(self, name=None, url=None, ssl=None, credentials=None, key=None, appliance=None): Navigatable.__init__(self, appliance=appliance) self.name = name self.url = url self.ssl = ssl self.credentials = credentials self.key = key or name
[docs] class Credential(BaseCredential, Updateable): pass
@property def ui_name(self): """Return the name used in the UI""" if self.type == 'Ansible Tower': return '{} Automation Manager'.format(self.name) else: return '{} Configuration Manager'.format(self.name)
[docs] def create(self, cancel=False, validate_credentials=True, validate=True, force=False): """Creates the manager through UI Args: cancel (bool): Whether to cancel out of the creation. The cancel is done after all the information present in the manager has been filled in the UI. validate_credentials (bool): Whether to validate credentials - if True and the credentials are invalid, an error will be raised. validate (bool): Whether we want to wait for the manager's data to load and show up in it's detail page. True will also wait, False will only set it up. force (bool): Whether to force the creation even if the manager already exists. True will try anyway; False will check for its existence and leave, if present. """ def config_profiles_loaded(): # Workaround - without this, validation of provider failed config_profiles_names = [prof.name for prof in self.config_profiles] logger.info( "UI: %s\nYAML: %s", set(config_profiles_names), set(self.yaml_data['config_profiles'])) return all( [cp in config_profiles_names for cp in self.yaml_data['config_profiles']]) if not force and self.exists: return form_dict = self.__dict__ form_dict.update(self.credentials.view_value_mapping) if self.appliance.version < '5.8': form_dict['provider_type'] = self.type view = navigate_to(self, 'Add') view.entities.form.fill(form_dict) if validate_credentials: view.entities.form.validate.click() view.flash.assert_success_message('Credential validation was successful') if cancel: view.entities.cancel.click() view.flash.assert_success_message('Add of Provider was cancelled by the user') else: view.entities.add.click() success_message = '{} Provider "{}" was added'.format(self.type, self.name) view.flash.assert_success_message(success_message) view.flash.assert_success_message(self.refresh_flash_msg) if validate: try: self.yaml_data['config_profiles'] except KeyError as e: logger.exception(e) raise wait_for( config_profiles_loaded, fail_func=self.refresh_relationships, handle_exception=True, num_sec=180, delay=30)
[docs] def update(self, updates, cancel=False, validate_credentials=False): """Updates the manager through UI args: updates (dict): Data to change. cancel (bool): Whether to cancel out of the update. The cancel is done after all the new information has been filled in the UI. validate_credentials (bool): Whether to validate credentials - if True and the credentials are invalid, an error will be raised. Note: utils.update use is recommended over use of this method. """ view = navigate_to(self, 'Edit') view.entities.form.fill(updates) if validate_credentials: view.entities.form.validate.click() view.flash.assert_success_message('Credential validation was successful') if cancel: view.entities.cancel.click() view.flash.assert_success_message('Edit of Provider was cancelled by the user') else: view.entities.save.click() view.flash.assert_success_message( '{} Provider "{}" was updated'.format(self.type, updates['name'] or self.name)) self.__dict__.update(**updates)
[docs] def delete(self, cancel=False, wait_deleted=True, force=False): """Deletes the manager through UI Args: cancel (bool): Whether to cancel out of the deletion, when the alert pops up. wait_deleted (bool): Whether we want to wait for the manager to disappear from the UI. True will wait; False will only delete it and move on. force (bool): Whether to try to delete the manager even though it doesn't exist. True will try to delete it anyway; False will check for its existence and leave, if not present. """ if not force and not self.exists: return view = navigate_to(self, 'All') view.toolbar.view_selector.select('List View') row = view.entities.paginator.find_row_on_pages(view.entities.elements, provider_name=self.ui_name) row[0].check() remove_item = VersionPick({ '5.8': 'Remove selected items', '5.9': 'Remove selected items from Inventory' }).pick(self.appliance.version) view.toolbar.configuration.item_select(remove_item, handle_alert=not cancel) if not cancel: view.flash.assert_success_message('Delete initiated for 1 Provider') if wait_deleted: wait_for(func=lambda: self.exists, fail_condition=True, delay=15, num_sec=60)
@property def exists(self): """Returns whether the manager exists in the UI or not""" view = navigate_to(self, 'All') view.toolbar.view_selector.select('List View') try: view.entities.paginator.find_row_on_pages(view.entities.elements, provider_name=self.ui_name) return True except NoSuchElementException: pass return False
[docs] def refresh_relationships(self, cancel=False): """Refreshes relationships and power states of this manager""" view = navigate_to(self, 'All') view.toolbar.view_selector.select('List View') row = view.entities.paginator.find_row_on_pages(view.entities.elements, provider_name=self.ui_name) row[0].check() if view.toolbar.configuration.item_enabled('Refresh Relationships and Power states'): view.toolbar.configuration.item_select('Refresh Relationships and Power states', handle_alert=not cancel) if not cancel: view.flash.assert_success_message(self.refresh_flash_msg)
@property def config_profiles(self): """Returns 'ConfigProfile' configuration profiles (hostgroups) available on this manager""" view = navigate_to(self, 'Details') # TODO - remove it later.Workaround for BZ 1452425 view.toolbar.view_selector.select('List View') view.toolbar.refresh.click() wait_for(lambda: view.entities.elements.is_displayed, fail_func=view.toolbar.refresh.click, handle_exception=True, num_sec=60, delay=5) config_profiles = [] for row in view.entities.elements: if self.type == 'Ansible Tower': name = row.name.text else: name = row.description.text if 'unassigned' in name.lower(): continue config_profiles.append(ConfigProfile(name=name, manager=self)) return config_profiles @property def systems(self): """Returns 'ConfigSystem' configured systems (hosts) available on this manager""" return reduce(lambda x, y: x + y, [prof.systems for prof in self.config_profiles]) @property def yaml_data(self): """Returns yaml data for this manager""" return conf.cfme_data.configuration_managers[self.key] @classmethod
[docs] def load_from_yaml(cls, key): """Returns 'ConfigManager' object loaded from yamls, based on its key""" data = conf.cfme_data.configuration_managers[key] creds = conf.credentials[data['credentials']] return cls( name=data['name'], url=data['url'], ssl=data['ssl'], credentials=cls.Credential( principal=creds['username'], secret=creds['password']), key=key)
@property def quad_name(self): if self.type == 'Ansible Tower': return '{} Automation Manager'.format(self.name) else: return '{} Configuration Manager'.format(self.name)
[docs]def get_config_manager_from_config(cfg_mgr_key): cfg_mgr = conf.cfme_data.get('configuration_managers', {})[cfg_mgr_key] if cfg_mgr['type'] == 'satellite': return Satellite.load_from_yaml(cfg_mgr_key) elif cfg_mgr['type'] == 'ansible': return AnsibleTower.load_from_yaml(cfg_mgr_key) else: raise Exception("Unknown configuration manager key")
[docs]class ConfigProfile(Pretty, Navigatable): """Configuration profile object (foreman-side hostgroup) Args: name: Name of the profile manager: ConfigManager object which this profile is bound to """ pretty_attrs = ['name', 'manager'] def __init__(self, name, manager, appliance=None): Navigatable.__init__(self, appliance=appliance) self.name = name self.manager = manager @property def systems(self): """Returns 'ConfigSystem' objects that are active under this profile""" view = navigate_to(self, 'Details') view.toolbar.view_selector.select('List View') # Unassigned config profile has no tabstrip if 'unassigned' not in self.name.lower(): view.entities.configured_systems.click() if view.entities.configured_systems.elements.is_displayed: return [ConfigSystem(row.hostname.text, self) for row in view.entities.configured_systems.elements] return list()
[docs]class ConfigSystem(Pretty, Navigatable, Taggable): """The tags pages of the config system""" pretty_attrs = ['name', 'manager_key'] def __init__(self, name, profile, appliance=None): Navigatable.__init__(self, appliance=appliance) self.name = name self.profile = profile
[docs] def get_tags(self, tenant="My Company Tags"): """Overridden get_tags method to deal with the fact that configured systems don't have a details view.""" view = navigate_to(self, 'EditTags') return [ self.appliance.collections.categories.instantiate( display_name=r.category.text.replace('*', '').strip()).collections.tags.instantiate( display_name=r.assigned_value.text.strip()) for r in view.form.tags ]
[docs]class Satellite(ConfigManager): """ Configuration manager object (Red Hat Satellite, Foreman) Args: name: Name of the Satellite/Foreman configuration manager url: URL, hostname or IP of the configuration manager ssl: Boolean value; `True` if SSL certificate validity should be checked, `False` otherwise credentials: Credentials to access the config. manager key: Key to access the cfme_data yaml data (same as `name` if not specified) Usage: Create provider: .. code-block:: python satellite_cfg_mgr = Satellite('my_satellite', 'my-satellite.example.com', ssl=False, ConfigManager.Credential(principal='admin', secret='testing'), key='satellite_yaml_key') satellite_cfg_mgr.create() Update provider: .. code-block:: python with update(satellite_cfg_mgr): satellite_cfg_mgr.name = 'new_satellite_name' Delete provider: .. code-block:: python satellite_cfg_mgr.delete() """ type = VersionPicker({ LOWEST: 'Red Hat Satellite', LATEST: 'Foreman' }) def __init__(self, name=None, url=None, ssl=None, credentials=None, key=None): super(Satellite, self).__init__(name=name, url=url, ssl=ssl, credentials=credentials, key=key) self.name = name self.url = url self.ssl = ssl self.credentials = credentials self.key = key or name
[docs]class AnsibleTower(ConfigManager): """ Configuration manager object (Ansible Tower) Args: name: Name of the Ansible Tower configuration manager url: URL, hostname or IP of the configuration manager ssl: Boolean value; `True` if SSL certificate validity should be checked, `False` otherwise credentials: Credentials to access the config. manager key: Key to access the cfme_data yaml data (same as `name` if not specified) Usage: Create provider: .. code-block:: python tower_cfg_mgr = AnsibleTower('my_tower', 'https://my-tower.example.com/api/v1', ssl=False, ConfigManager.Credential(principal='admin', secret='testing'), key='tower_yaml_key') tower_cfg_mgr.create() Update provider: .. code-block:: python with update(tower_cfg_mgr): tower_cfg_mgr.name = 'new_tower_name' Delete provider: .. code-block:: python tower_cfg_mgr.delete() """ type = 'Ansible Tower' def __init__(self, name=None, url=None, ssl=None, credentials=None, key=None): super(AnsibleTower, self).__init__(name=name, url=url, ssl=ssl, credentials=credentials, key=key) self.name = name self.url = url self.ssl = ssl self.credentials = credentials self.key = key or name @property def ui_name(self): """Return the name used in the UI""" return '{} Automation Manager'.format(self.name)
@navigator.register(ConfigManager, 'All')
[docs]class MgrAll(CFMENavigateStep): VIEW = ConfigManagementAllView prerequisite = NavigateToAttribute('appliance.server', 'LoggedIn')
[docs] def step(self): if self.obj.appliance.version < '5.8' or self.obj.type != 'Ansible Tower': self.prerequisite_view.navigation.select('Configuration', 'Management') else: self.prerequisite_view.navigation.select('Automation', 'Ansible Tower', 'Explorer')
[docs] def resetter(self): if self.obj.appliance.version >= '5.8' and self.obj.type == 'Ansible Tower': self.view.sidebar.providers.tree.click_path('All Ansible Tower Providers') else: self.view.sidebar.providers.tree.click_path('All Configuration Manager Providers')
@navigator.register(ConfigManager, 'Add')
[docs]class MgrAdd(CFMENavigateStep): VIEW = ConfigManagementAddView prerequisite = NavigateToSibling('All')
[docs] def step(self): self.prerequisite_view.toolbar.configuration.item_select('Add a new Provider')
@navigator.register(ConfigManager, 'Edit')
[docs]class MgrEdit(CFMENavigateStep): VIEW = ConfigManagementEditView prerequisite = NavigateToSibling('All')
[docs] def step(self): self.prerequisite_view.toolbar.view_selector.select('List View') row = self.prerequisite_view.entities.paginator.find_row_on_pages( self.prerequisite_view.entities.elements, provider_name=self.obj.ui_name) row.click() self.prerequisite_view.toolbar.configuration.item_select('Edit this Provider')
@navigator.register(ConfigManager, 'Details')
[docs]class MgrDetails(CFMENavigateStep): VIEW = ConfigManagementDetailsView prerequisite = NavigateToSibling('All')
[docs] def step(self): self.prerequisite_view.toolbar.view_selector.select('List View') row = self.prerequisite_view.entities.paginator.find_row_on_pages( self.prerequisite_view.entities.elements, provider_name=self.obj.ui_name) row.click()
@navigator.register(ConfigManager, 'EditFromDetails')
[docs]class MgrEditFromDetails(CFMENavigateStep): VIEW = ConfigManagementEditView prerequisite = NavigateToSibling('Details')
[docs] def step(self): self.prerequisite_view.toolbar.configuration.item_select('Edit this Provider')
@navigator.register(ConfigProfile, 'Details')
[docs]class Details(CFMENavigateStep): VIEW = ConfigManagementProfileView prerequisite = NavigateToAttribute('manager', 'Details')
[docs] def step(self): self.prerequisite_view.toolbar.view_selector.select('List View') row = self.prerequisite_view.entities.paginator.find_row_on_pages( self.prerequisite_view.entities.elements, description=self.obj.name) row.click()
@navigator.register(ConfigSystem, 'All')
[docs]class SysAll(CFMENavigateStep): VIEW = ConfigManagementAllView prerequisite = NavigateToAttribute('appliance.server', 'LoggedIn')
[docs] def step(self): self.prerequisite_view.navigation.select('Configuration', 'Management')
[docs] def resetter(self): self.view.sidebar.configured_systems.open() self.view.sidebar.configured_systems.tree.click_path('All Configured Systems')
@navigator.register(ConfigSystem, 'EditTags')
[docs]class SysEditTags(CFMENavigateStep): VIEW = TagPageView prerequisite = NavigateToSibling('All')
[docs] def step(self): self.prerequisite_view.toolbar.view_selector.select('List View') row = self.prerequisite_view.entities.paginator.find_row_on_pages( self.prerequisite_view.entities.elements, hostname=self.obj.name) row[0].check() self.prerequisite_view.toolbar.policy.item_select('Edit Tags')