# -*- coding: utf-8 -*-
import attr
from cached_property import cached_property
from navmazing import NavigateToAttribute, NavigateToSibling
from widgetastic.xpath import quote
from widgetastic.widget import Text, Checkbox
from widgetastic.utils import Fillable
from widgetastic_manageiq import Table, UpDownSelect
from widgetastic_patternfly import CandidateNotFound, Input, Button
from cfme.exceptions import ItemNotFound
from cfme.utils import clear_property_cache
from cfme.modeling.base import BaseCollection, BaseEntity
from cfme.utils.appliance.implementations.ui import navigator, CFMENavigateStep, navigate_to
from . import AutomateExplorerView
[docs]def generate_updown(title):
return './/*[(self::a or self::button) and @title={}]/*[self::img or self::i]'.format(
quote(title))
[docs]class DomainPriorityView(AutomateExplorerView):
title = Text('#explorer_title_text')
domains = UpDownSelect(
'#seq_fields',
generate_updown('Move selected fields up'),
generate_updown('Move selected fields down'))
save_button = Button('Save')
reset_button = Button('Reset')
cancel_button = Button('Cancel')
@property
def is_displayed(self):
return (
self.in_explorer and
self.title.text == 'Datastore' and
self.domains.is_displayed)
[docs]class DomainListView(AutomateExplorerView):
title = Text('#explorer_title_text')
domains = Table('#ns_list_grid')
@property
def is_displayed(self):
return (
self.in_explorer and
self.title.text == 'Datastore' and
self.datastore.is_opened and
self.datastore.tree.currently_selected == ['Datastore'])
[docs]class DomainForm(AutomateExplorerView):
title = Text('#explorer_title_text')
name = Input(name='ns_name')
description = Input(name='ns_description')
enabled = Checkbox(name='ns_enabled')
cancel_button = Button('Cancel')
[docs]class DomainAddView(DomainForm):
add_button = Button('Add')
@property
def is_displayed(self):
return (
self.in_explorer and
self.title.text == 'Adding a new Automate Domain')
[docs]class DomainEditView(DomainForm):
save_button = Button('Save')
@property
def is_displayed(self):
return (
self.in_explorer and
self.title.text == 'Editing Automate Domain "{}"'.format(self.obj.name))
[docs]class Domain(BaseEntity, Fillable):
"""A class representing one Domain in the UI."""
def __init__(
self, collection, name, description=None, enabled=None, locked=None,
git_repository=None, git_checkout_type=None, git_checkout_value=None, db_id=None):
from .namespace import NamespaceCollection
self._collections = {'namespaces': NamespaceCollection}
super(Domain, self).__init__(collection)
self.name = name
self.description = description
if db_id is not None:
self.db_id = db_id
if git_repository is not None:
self.git_repository = git_repository
if git_checkout_type is not None:
self.git_checkout_type = git_checkout_type
if git_checkout_value is not None:
self.git_checkout_value = git_checkout_value
if enabled is not None:
self.enabled = enabled
if locked is not None:
self.locked = locked
__repr__ = object.__repr__
# TODO this needs replacing with something better
def __hash__(self):
return hash((self.name, id(self.parent)))
[docs] def as_fill_value(self):
return self.name
@cached_property
def db_id(self):
table = self.appliance.db.client['miq_ae_namespaces']
try:
return self.appliance.db.client.session.query(table.id).filter(
table.name == self.name,
table.parent_id == None)[0] # noqa
except IndexError:
raise ItemNotFound('Domain named {} not found in the database'.format(self.name))
@cached_property
def git_repository(self):
"""Returns an associated git repository object. None if no git repo associated."""
dbo = self.db_object
if dbo.git_repository_id is None:
return None
from cfme.automate.import_export import AutomateGitRepository
return AutomateGitRepository.from_db(dbo.git_repository_id, appliance=self.appliance)
@cached_property
def git_checkout_type(self):
return self.db_object.ref_type
@cached_property
def git_checkout_value(self):
return self.db_object.ref
@property
def db_object(self):
if self.db_id is None:
return None
table = self.appliance.db.client['miq_ae_namespaces']
return self.appliance.db.client.session.query(table).filter(table.id == self.db_id).first()
@cached_property
def enabled(self):
return self.db_object.enabled
@cached_property
def locked(self):
if self.appliance.version < '5.7':
return self.db_object.system
else:
return self.db_object.source in {'user_locked', 'system', 'remote'}
@property
def domain(self):
return self
@cached_property
def namespaces(self):
return self.collections.namespaces
@property
def tree_display_name(self):
if self.git_repository:
name = '{name} ({ref}) ({name})'.format(name=self.name, ref=self.git_checkout_value)
else:
name = self.name
if self.locked and not self.enabled:
return '{} (Locked & Disabled)'.format(name)
elif self.locked and self.enabled:
return '{} (Locked)'.format(name)
elif not self.locked and not self.enabled:
return '{} (Disabled)'.format(name)
else:
return name
@property
def table_display_name(self):
if self.git_repository:
name = '{name} ({ref})'.format(name=self.name, ref=self.git_checkout_value)
else:
name = self.name
if self.locked and not self.enabled:
return '{} (Locked & Disabled)'.format(name)
elif self.locked and self.enabled:
return '{} (Locked)'.format(name)
elif not self.locked and not self.enabled:
return '{} (Disabled)'.format(name)
else:
return name
@property
def tree_path(self):
return self.parent.tree_path + [self.tree_display_name]
[docs] def delete(self, cancel=False):
# Ensure this has correct data
self.description
# Do it!
details_page = navigate_to(self, 'Details')
details_page.configuration.item_select('Remove this Domain', handle_alert=not cancel)
if cancel:
assert details_page.is_displayed
details_page.flash.assert_no_error()
else:
domains_view = self.create_view(DomainListView)
assert domains_view.is_displayed
domains_view.flash.assert_no_error()
domains_view.flash.assert_message(
'Automate Domain "{}": Delete successful'.format(self.description or self.name))
[docs] def lock(self):
# Ensure this has correct data
self.description
details_page = navigate_to(self, 'Details')
details_page.configuration.item_select('Lock this Domain')
details_page.flash.assert_no_error()
details_page.flash.assert_message('The selected Automate Domain were marked as Locked')
clear_property_cache(self, 'locked')
assert self.locked
[docs] def unlock(self):
# Ensure this has correct data
self.description
details_page = navigate_to(self, 'Details')
details_page.configuration.item_select('Unlock this Domain')
details_page.flash.assert_no_error()
details_page.flash.assert_message('The selected Automate Domain were marked as Unlocked')
clear_property_cache(self, 'locked')
assert not self.locked
[docs] def update(self, updates):
view = navigate_to(self, 'Edit')
changed = view.fill(updates)
if changed:
view.save_button.click()
else:
view.cancel_button.click()
view = self.create_view(DomainDetailsView, override=updates)
assert view.is_displayed
view.flash.assert_no_error()
if changed:
if self.appliance.version >= '5.8.2':
text = (
updates.get('description', self.description) or
updates.get('name', self.name))
else:
text = updates.get('name', self.name)
view.flash.assert_message('Automate Domain "{}" was saved'.format(text))
else:
view.flash.assert_message(
'Edit of Automate Domain "{}" was cancelled by the user'.format(self.name))
@property
def exists(self):
try:
navigate_to(self, 'Details')
return True
except (CandidateNotFound, ItemNotFound):
return False
[docs] def delete_if_exists(self):
if self.exists:
self.delete()
@attr.s
[docs]class DomainCollection(BaseCollection):
"""Collection object for the :py:class:`Domain`."""
tree_path = ['Datastore']
ENTITY = Domain
[docs] def create(self, name=None, description=None, enabled=None, cancel=False):
add_page = navigate_to(self, 'Add')
fill_dict = {
k: v
for k, v in {'name': name, 'description': description, 'enabled': enabled}.items()
if v is not None}
add_page.fill(fill_dict)
if cancel:
add_page.cancel_button.click()
add_page.flash.assert_no_error()
add_page.flash.assert_message('Add of new Automate Domain was cancelled by the user')
return None
else:
add_page.add_button.click()
add_page.flash.assert_no_error()
if self.appliance.version >= '5.8.2':
add_page.flash.assert_message(
'Automate Domain "{}" was added'.format(description or name))
else:
add_page.flash.assert_message(
'Automate Domain "{}" was added'.format(name))
if enabled is None:
# Assume
enabled = False
return self.instantiate(
name=name, description=description, enabled=enabled, locked=False)
[docs] def all(self):
table = self.appliance.db.client['miq_ae_namespaces']
query = self.appliance.db.client.session.query(
table.name, table.description, table.enabled, table.source, table.ref, table.ref_type,
table.git_repository_id)
query = query.filter(table.name != '$', table.parent_id == None) # noqa
result = []
for name, description, enabled, source, ref, ref_type, git_repository_id in query:
if source != 'remote':
result.append(
self.instantiate(
name=name,
description=description or '',
enabled=enabled,
locked=source in {'user_locked', 'system'}))
else:
repo_table = self.appliance.db.client['git_repositories']
repo = self.appliance.db.client.session\
.query(repo_table)\
.filter(repo_table.id == git_repository_id)\
.first()
from cfme.automate.import_export import AutomateGitRepository
agr = AutomateGitRepository(
url=repo.url,
verify_ssl=repo.verify_ssl,
appliance=self.appliance)
result.append(
self.instantiate(
name=name,
description=description,
enabled=enabled,
locked=True,
git_repository=agr,
git_checkout_type=ref_type,
git_checkout_value=ref))
return result
[docs] def delete(self, *domains):
domains = list(domains)
checked_domains = []
all_page = navigate_to(self, 'All')
all_page.domains.uncheck_all()
if not all_page.domains.is_displayed:
raise ValueError('No domain found!')
for row in all_page.domains:
for domain in domains:
if domain.table_display_name == row.name.text:
checked_domains.append(domain)
row[0].check()
break
if set(domains) == set(checked_domains):
break
if set(domains) != set(checked_domains):
raise ValueError('Some of the domains were not found in the UI.')
all_page.configuration.item_select('Remove Domains', handle_alert=True)
all_page.flash.assert_no_error()
for domain in checked_domains:
all_page.flash.assert_message(
'Automate Domain "{}": Delete successful'.format(domain.description or domain.name))
[docs] def set_order(self, items):
if not isinstance(items, (list, tuple)):
items = [items]
processed_items = [Fillable.coerce(item) for item in items]
priority_page = navigate_to(self, 'Priority')
changed = priority_page.domains.fill(processed_items)
if changed:
# Changed
priority_page.save_button.click()
else:
# Not changed
priority_page.cancel_button.click()
domains_view = self.create_view(DomainListView)
assert domains_view.is_displayed
domains_view.flash.assert_no_error()
if changed:
domains_view.flash.assert_message('Priority Order was saved')
else:
domains_view.flash.assert_message('Edit of Priority Order was cancelled by the user')
return changed
@navigator.register(DomainCollection)
[docs]class All(CFMENavigateStep):
VIEW = DomainListView
prerequisite = NavigateToAttribute('appliance.server', 'AutomateExplorer')
[docs] def step(self):
self.prerequisite_view.datastore.tree.click_path(*self.obj.tree_path)
@navigator.register(DomainCollection)
[docs]class Add(CFMENavigateStep):
VIEW = DomainAddView
prerequisite = NavigateToSibling('All')
[docs] def step(self):
self.prerequisite_view.configuration.item_select('Add a New Domain')
@navigator.register(DomainCollection)
[docs]class Priority(CFMENavigateStep):
VIEW = DomainPriorityView
prerequisite = NavigateToSibling('All')
[docs] def step(self):
self.prerequisite_view.configuration.item_select('Edit Priority Order of Domains')
[docs]class DomainDetailsView(AutomateExplorerView):
title = Text('#explorer_title_text')
namespaces = Table('#ns_details_grid')
@property
def is_displayed(self):
return (
self.in_explorer and
self.title.text == 'Automate Domain "{}"'.format(
self.context['object'].table_display_name))
@navigator.register(Domain)
[docs]class Details(CFMENavigateStep):
VIEW = DomainDetailsView
prerequisite = NavigateToAttribute('appliance.server', 'AutomateExplorer')
[docs] def step(self):
try:
self.prerequisite_view.datastore.tree.click_path(*self.obj.tree_path)
except CandidateNotFound:
# Try it with regexp (drop the locked to None)
# That will force reload from database
self.obj.locked = None
self.prerequisite_view.datastore.tree.click_path(*self.obj.tree_path)
@navigator.register(Domain)
[docs]class Edit(CFMENavigateStep):
VIEW = DomainEditView
prerequisite = NavigateToSibling('Details')
[docs] def step(self):
self.prerequisite_view.configuration.item_select('Edit this Domain')