Source code for cfme.configure.tasks

# -*- coding: utf-8 -*-

""" Module dealing with Configure/Tasks section.
import attr

from navmazing import NavigateToAttribute
from widgetastic.exceptions import RowNotFound
from widgetastic.utils import Version, VersionPick
from widgetastic.widget import View
from widgetastic_manageiq import BootstrapSelect, Button, CheckboxSelect, Table
from widgetastic_patternfly import Dropdown, Tab

from cfme.base.login import BaseLoggedInPage
from cfme.modeling.base import BaseCollection, BaseEntity
from cfme.utils.appliance.implementations.ui import navigator, CFMENavigateStep, navigate_to
from cfme.utils.log import logger
from cfme.utils.wait import wait_for, TimedOutError

[docs]class TasksView(BaseLoggedInPage): # Toolbar delete = Dropdown('Delete Tasks') # dropdown just has icon, use element title reload = Button(title=VersionPick({Version.lowest(): 'Reload the current display', '5.9': 'Refresh this page'})) @View.nested class tabs(View): # noqa # Extra Toolbar # Only on 'All' type tabs, but for access it doesn't make sense to access the tab for a # toolbar button cancel = Button(title='Cancel the selected task') # Form Buttons apply = Button('Apply') reset = Button('Reset') default = Button('Default') # Filters zone = BootstrapSelect(id='chosen_zone') period = BootstrapSelect(id='time_period') user = BootstrapSelect(id='user_choice') # This checkbox search_root captures all the filter options # It will break for status if/when there is second checkbox selection field added # It's the lowest level div with an id that captures the status checkboxes status = CheckboxSelect(search_root='tasks_options_div') state = BootstrapSelect(id='state_choice') @View.nested class mytasks(Tab): # noqa TAB_NAME = VersionPick({Version.lowest(): 'My VM and Container Analysis Tasks', '5.9': 'My Tasks'}) table = Table('//div[@id="gtl_div"]//table') @View.nested class myothertasks(Tab): # noqa TAB_NAME = VersionPick({'5.9': 'My Tasks', Version.lowest(): 'My Other UI Tasks'}) table = Table('//div[@id="gtl_div"]//table') @View.nested class alltasks(Tab): # noqa TAB_NAME = VersionPick({'5.9': 'All Tasks', Version.lowest(): "All VM and Container Analysis Tasks"}) table = Table('//div[@id="gtl_div"]//table') @View.nested class allothertasks(Tab): # noqa TAB_NAME = "All Other Tasks" table = Table('//div[@id="gtl_div"]//table') @property def is_displayed(self): return ( self.tabs.mytasks.is_displayed and self.tabs.myothertasks.is_displayed and self.tabs.alltasks.is_displayed and self.tabs.allothertasks.is_displayed)
[docs]class Task(BaseEntity): """Model of an tasks in cfme. Args: name: Name of the task. user: User who initiated the task. server: Cfme server. tab: Tab where current task located. """ OK = 'ok' ERROR = 'error' IN_PROGRESS = 'in_progress' name = attr.ib() user = attr.ib(default='admin') tab = attr.ib(default='AllTasks') def _is_row_present(self, row_name): view = navigate_to(self.parent, tab_view = getattr(view.tabs, for row in tab_view.table.rows(): if row_name in row.task_name.text: return True return False @property def _row(self): view = navigate_to(self.parent, tab_view = getattr(view.tabs, wait_for( lambda: self._is_row_present(, delay=5, timeout='15m', row = tab_view.table.row( return row @property def status(self): # status column doen't have text id col = self._row[1] if col.browser.is_displayed('i[@class="pficon pficon-ok"]', parent=col): return self.OK elif col.browser.is_displayed('i[@class="pficon pficon-error-circle-o"]', parent=col): return self.ERROR else: return self.IN_PROGRESS @property def state(self): return self._row.state.text @property def updated(self): return self._row.updated.text @property def started(self): return self._row.started.text @property def queued(self): if self.appliance.version >= '5.9': return self._row.queued.text return None @property def message(self): return self._row.message.text @property def exists(self): try: self._row return True except RowNotFound: return False @property def is_successfully_finished(self): message = self.message.lower() if self.status == self.ERROR: if 'timed out' in message: raise TimedOutError("Task {} timed out: {}".format(, message)) else: raise Exception("Task {} error: {}".format(, message)) if self.state.lower() == 'finished' and self.status == self.OK: return True return False
[docs] def wait_for_finished(self, delay=5, timeout='15m'): view = navigate_to(self.parent, wait_for( lambda: self.is_successfully_finished, delay=delay, timeout=timeout,
[docs]class TasksCollection(BaseCollection): """Collection object for :py:class:`cfme.configure.tasks.Task`.""" ENTITY = Task DEFAULT_TAB = 'AllTasks' @property def tab(self): return self.filters.get('tab') or self.DEFAULT_TAB
[docs] def set_task_filter(self, values, cancel=False): view = navigate_to(self, view.fill(values) if cancel:
[docs] def set_default_task_filter(self): view = navigate_to(self,
[docs] def find(self, name): tasks = self.all() for task in tasks: return task if == name else None
[docs] def all(self): view = navigate_to(self, tab_view = getattr(view.tabs, tasks = [self.instantiate(, for row in tab_view.table.rows()] return tasks
[docs] def delete(self, *tasks): view = navigate_to(self, tab_view = getattr(view.tabs, for row in tab_view.table.rows(): if in tasks: row.check() view.delete.item_select('Delete', handle_alert=True)
[docs] def delete_all(self): """Deletes all currently visible on page""" view = navigate_to(self, view.delete.item_select('Delete All', handle_alert=True)
[docs] def is_finished(self, *tasks): view = navigate_to(self, tab_view = getattr(view.tabs, for row in tab_view.table.rows: if in tasks and row.state.lower() != "finished": return False return True
[docs] def is_successfully_finished(self, silent_failure=False, *tasks): view = navigate_to(self, tab_view = getattr(view.tabs, rows = [] for task in tasks: rows.append(tab_view.table.rows(task_name=task, state='finished')) for row in rows: message = row.message.text.lower() if row[1].browser.is_displayed('i[@class="pficon pficon-error-circle-o"]', parent=row[1]): if silent_failure: logger.warning("Task {} error: {}".format(row.task_name.text, message)) return False elif 'timed out' in message: raise TimedOutError("Task {} timed out: {}".format(row.task_name.text, message)) else: Exception("Task {} error: {}".format(row.task_name.text, message)) return True
[docs] def wait_for_finished(self, delay=5, timeout='5m', *tasks): view = navigate_to(self.parent, wait_for( lambda: self.is_finished(tasks), delay=delay, timeout=timeout,
@navigator.register(TasksCollection, 'MyTasks')
[docs]class MyTasks(CFMENavigateStep): VIEW = TasksView prerequisite = NavigateToAttribute('appliance.server', 'Tasks')
[docs] def step(self, *args, **kwargs):
[docs] def am_i_here(self): tasks = self.view.tabs.mytasks return tasks.is_displayed and tasks.is_active()
@navigator.register(TasksCollection, 'MyOtherTasks')
[docs]class MyOtherTasks(CFMENavigateStep): VIEW = TasksView prerequisite = NavigateToAttribute('appliance.server', 'Tasks')
[docs] def step(self, *args, **kwargs):
[docs] def am_i_here(self): tasks = self.view.tabs.myothertasks return tasks.is_displayed and tasks.is_active()
@navigator.register(TasksCollection, 'AllTasks')
[docs]class AllTasks(CFMENavigateStep): VIEW = TasksView prerequisite = NavigateToAttribute('appliance.server', 'Tasks')
[docs] def step(self, *args, **kwargs):
[docs] def am_i_here(self): tasks = self.view.tabs.alltasks return tasks.is_displayed and tasks.is_active()
@navigator.register(TasksCollection, 'AllOtherTasks')
[docs]class AllOtherTasks(CFMENavigateStep): VIEW = TasksView prerequisite = NavigateToAttribute('appliance.server', 'Tasks')
[docs] def step(self, *args, **kwargs):
[docs] def am_i_here(self): tasks = self.view.tabs.allothertasks return tasks.is_displayed and tasks.is_active()