# -*- coding: utf-8 -*-
"""Page model for Control / Explorer"""
from copy import copy
import attr
from navmazing import NavigateToAttribute, NavigateToSibling
from widgetastic.exceptions import NoSuchElementException
from widgetastic.widget import Checkbox, Table, Text, TextInput, View
from widgetastic_patternfly import Button, Input
from .actions import Action
from cfme.modeling.base import BaseCollection, BaseEntity
from cfme.utils import ParamClassName
from cfme.utils.appliance.implementations.ui import navigator, navigate_to, CFMENavigateStep
from cfme.utils.blockers import BZ
from cfme.utils.pretty import Pretty
from cfme.utils.update import Updateable
from cfme.utils.wait import wait_for
from widgetastic_manageiq import BootstrapSwitchSelect, MultiBoxSelect, SummaryFormItem, Dropdown
from widgetastic_manageiq.expression_editor import ExpressionEditor
from . import ControlExplorerView
[docs]class PoliciesAllView(ControlExplorerView):
title = Text("#explorer_title_text")
@property
def is_displayed(self):
return (
self.in_control_explorer and
self.title.text == "All {} {} Policies".format(
self.context["object"].PRETTY,
self.context["object"].TYPE
)
)
[docs]class EditPolicyEventAssignments(ControlExplorerView):
title = Text("#explorer_title_text")
events = BootstrapSwitchSelect("policy_info_div")
cancel_button = Button("Cancel")
save_button = Button("Save")
@property
def supposed_title(self):
if BZ(1531468, forced_streams=["5.9"]).blocks:
return 'Editing {} {} "{}" Event Assignments'
else:
return 'Editing {} {} Policy "{}" Event Assignments'
@property
def is_displayed(self):
return (
self.in_control_explorer and
self.title.text == self.supposed_title.format(
self.context["object"].PRETTY,
self.context["object"].TYPE,
self.context["object"].description
)
)
[docs]class EditPolicyConditionAssignments(ControlExplorerView):
title = Text("#explorer_title_text")
conditions = MultiBoxSelect()
cancel_button = Button("Cancel")
save_button = Button("Save")
@property
def supposed_title(self):
if BZ(1531468, forced_streams=["5.9"]).blocks:
return 'Editing {} {} "{}" Condition Assignments'
else:
return 'Editing {} {} Policy "{}" Condition Assignments'
@property
def is_displayed(self):
return (
self.in_control_explorer and
self.title.text == self.supposed_title.format(
self.context["object"].PRETTY,
self.context["object"].TYPE,
self.context["object"].description
)
)
[docs]class NewPolicyView(PolicyFormCommon):
title = Text("#explorer_title_text")
add_button = Button("Add")
@property
def is_displayed(self):
return (
self.in_control_explorer and
self.title.text == "Adding a new {} {} Policy".format(
self.context["object"].PRETTY, self.context["object"].TYPE) and
self.policies.is_opened and
self.policies.tree.currently_selected == ["All {} {} Policies".format(
self.context["object"].PRETTY, self.context["object"].TYPE)]
)
[docs]class EditPolicyView(PolicyFormCommon):
title = Text("#explorer_title_text")
save_button = Button("Save")
reset_button = Button("Reset")
@property
def is_displayed(self):
return (
self.in_control_explorer and
self.title.text == 'Editing {} {} Policy "{}"'.format(
self.context["object"].PRETTY,
self.context["object"].TYPE,
self.context["object"].description
) and
self.policies.is_opened and
self.policies.tree.currently_selected == [
"All Policies",
"{} Policies".format(self.context["object"].TYPE),
"{} {} Policies".format(self.context["object"].TREE_NODE,
self.context["object"].TYPE),
self.context["object"].description
]
)
[docs]class PolicyDetailsView(ControlExplorerView):
title = Text("#explorer_title_text")
type = SummaryFormItem("Basic Information", "Action Type")
analysis_profile = SummaryFormItem("Analysis Profile", "Assigned Analysis Profile")
@property
def is_displayed(self):
return (
self.in_control_explorer and
self.title.text == '{} {} Policy "{}"'.format(self.context["object"].PRETTY,
self.context["object"].TYPE, self.context["object"].description) and
self.policies.is_opened and
self.policies.tree.currently_selected == [
"All Policies",
"{} Policies".format(self.context["object"].TYPE),
"{} {} Policies".format(self.context["object"].TREE_NODE,
self.context["object"].TYPE),
self.context["object"].description
]
)
[docs]class ConditionDetailsView(ControlExplorerView):
title = Text("#explorer_title_text")
@property
def is_displayed(self):
return (
self.in_control_explorer and
self.title.text == '{} Condition "{}"'.format(self.context["object"].PRETTY,
self.context["object"].testing_condition.description) and
self.policies.is_opened and
self.policies.tree.currently_selected == [
"All Policies",
"{} Policies".format(self.context["object"].TYPE),
"{} {} Policies".format(self.context["object"].TREE_NODE,
self.context["object"].TYPE),
self.context["object"].description,
self.context["object"].testing_condition.description
]
)
[docs]class EventDetailsView(ControlExplorerView):
title = Text("#explorer_title_text")
toolbar = View.nested(EventDetailsToolbar)
true_actions = Table(".//h3[text()='Order of Actions if ALL Conditions are True']/"
"following-sibling::table[1]")
false_actions = Table(".//h3[text()='Order of Actions if ANY Conditions are False']/"
"following-sibling::table[1]")
@property
def is_displayed(self):
return (
self.in_control_explorer and
self.title.text == 'Event "{}"'.format(self.context["object"].context_event)
)
[docs]class EditEventView(ControlExplorerView):
title = Text("#explorer_title_text")
true_actions = MultiBoxSelect(
move_into="choices_chosen_true_div",
move_from="members_chosen_true_div",
available_items="choices_chosen_true",
chosen_items="members_chosen_true"
)
false_actions = MultiBoxSelect(
move_into="choices_chosen_false_div",
move_from="members_chosen_false_div",
available_items="choices_chosen_false",
chosen_items="members_chosen_false"
)
save_button = Button("Save")
reset_button = Button("Reset")
cancel_button = Button("Cancel")
@property
def is_displayed(self):
return (
self.in_control_explorer and
self.title.text == 'Editing Event "{}"'.format(self.context["object"].context_event)
)
@attr.s
[docs]class BasePolicy(BaseEntity, Updateable, Pretty):
"""This class represents a Policy.
Example:
.. code-block:: python
>>> from cfme.control.explorer.policy import VMCompliancePolicy
>>> policy = VMCompliancePolicy("policy_description")
>>> policy.create()
>>> policy.delete()
Args:
description: Policy name.
active: Whether the policy active or not.
scope: Policy scope.
notes: Policy notes.
"""
TYPE = None
TREE_NODE = None
PRETTY = None
_param_name = ParamClassName('description')
description = attr.ib()
active = attr.ib(default=None)
scope = attr.ib(default=None)
notes = attr.ib(default=None)
def __str__(self):
return self.description
@property
def name_for_policy_profile(self):
return "{} {}: {}".format(self.PRETTY, self.TYPE, self.description)
[docs] def update(self, updates):
"""Update this Policy in UI.
Args:
updates: Provided by update() context manager.
cancel: Whether to cancel the update (default False).
"""
view = navigate_to(self, "Edit")
changed = view.fill(updates)
if changed:
view.save_button.click()
else:
view.cancel_button.click()
view = self.create_view(PolicyDetailsView, override=updates)
wait_for(lambda: view.is_displayed, timeout=10, message="waits until the view is displayed")
view.flash.assert_no_error()
if changed:
view.flash.assert_message(
'Policy "{}" was saved'.format(updates.get("description", self.description)))
else:
view.flash.assert_message(
'Edit of Policy "{}" was cancelled by the user'.format(self.description))
[docs] def delete(self, cancel=False):
"""Delete this Policy in UI.
Args:
cancel: Whether to cancel the deletion (default False).
"""
view = navigate_to(self, "Details")
view.configuration.item_select("Delete this {} Policy".format(self.PRETTY),
handle_alert=not cancel)
if cancel:
assert view.is_displayed
view.flash.assert_no_error()
else:
view = self.create_view(PoliciesAllView)
assert view.is_displayed
view.flash.assert_no_error()
[docs] def copy(self, cancel=False):
"""Copy this Policy in UI.
Args:
cancel: Whether to cancel the copying (default False).
"""
view = navigate_to(self, "Details")
view.configuration.item_select("Copy this {} Policy".format(self.PRETTY),
handle_alert=not cancel)
view = self.create_view(PolicyDetailsView)
assert view.is_displayed
view.flash.assert_success_message('Policy "Copy of {}" was added'.format(self.description))
policy_copy = copy(self)
policy_copy.description = "Copy of {}".format(self.description)
return policy_copy
[docs] def assign_events(self, *events, **kwargs):
"""Assign events to this Policy.
Args:
events: Events which need to be assigned.
extend: Do not uncheck existing events.
"""
events = list(events)
extend = kwargs.pop("extend", False)
if extend:
events += self.assigned_events
view = navigate_to(self, "Details")
view.configuration.item_select("Edit this Policy's Event assignments")
view = self.create_view(EditPolicyEventAssignments)
assert view.is_displayed
changed = view.fill({"events": events})
if changed:
view.save_button.click()
else:
view.cancel_button.click()
view.flash.assert_success_message('Policy "{}" was saved'.format(self.description))
[docs] def is_event_assigned(self, event):
return event in self.assigned_events
[docs] def assign_conditions(self, *conditions):
"""Assign conditions to this Policy.
Args:
conditions: Conditions which need to be assigned.
"""
view = navigate_to(self, "Details")
view.configuration.item_select("Edit this Policy's Condition assignments")
view = self.create_view(EditPolicyConditionAssignments)
assert view.is_displayed
changed = view.fill({"conditions": [condition.description for condition in conditions]})
if changed:
view.save_button.click()
flash_message = 'Policy "{}" was saved'.format(self.description)
else:
view.cancel_button.click()
flash_message = 'Edit of Policy "{}" was cancelled by the user'.format(self.description)
view.flash.assert_success_message(flash_message)
[docs] def is_condition_assigned(self, condition):
condition.context_policy = self
view = navigate_to(condition, "Details in policy")
return view.is_displayed
[docs] def assign_actions_to_event(self, event, actions):
"""
This method takes a list or dict of actions, goes into the policy event and assigns them.
Actions can be passed both as the objects, but they can be passed also as a string.
If the specified event is not assigned to the policy, it will be assigned.
Args:
event: Name of the event under which the actions will be assigned.
actions: If :py:class:`list` (or similar), all of these actions will be set under
TRUE section. If :py:class:`dict`, the action is key and value specifies its
placement. If it's True, then it will be put in the TRUE section and so on.
"""
true_actions, false_actions = [], []
if isinstance(actions, Action):
true_actions.append(actions)
elif isinstance(actions, (list, tuple, set)):
true_actions.extend(actions)
elif isinstance(actions, dict):
for action, is_true in actions.items():
if is_true:
true_actions.append(action)
else:
false_actions.append(action)
else:
raise TypeError("assign_actions_to_event expects, list, tuple, set or dict!")
# Check whether we have all necessary events assigned
if not self.is_event_assigned(event):
self.assign_events(event, extend=True)
assert self.is_event_assigned(event), "Could not assign event {}!".format(event)
# And now we can assign actions
self.context_event = event
view = navigate_to(self, "Event Details")
assert view.is_displayed
view.toolbar.configuration.item_select("Edit Actions for this Policy Event")
view = self.create_view(EditEventView)
assert view.is_displayed
changed = view.fill({
"true_actions": [str(action) for action in true_actions],
"false_actions": [str(action) for action in false_actions]
})
if changed:
view.save_button.click()
else:
view.cancel_button.click()
view.flash.assert_success_message('Actions for Policy Event "{}" were saved'.format(
event))
@property
def exists(self):
policies = self.appliance.db.client["miq_policies"]
return self.appliance.db.client.session\
.query(policies.description)\
.filter(policies.description == self.description)\
.count() > 0
@property
def assigned_events(self):
policies = self.appliance.db.client["miq_policies"]
events = self.appliance.db.client["miq_event_definitions"]
policy_contents = self.appliance.db.client["miq_policy_contents"]
session = self.appliance.db.client.session
policy_id = session.query(policies.id).filter(policies.description == self.description)
assigned_events = session.query(policy_contents.miq_event_definition_id).filter(
policy_contents.miq_policy_id == policy_id)
return [event_name[0] for event_name in session.query(events.description).filter(
events.id.in_(assigned_events))]
[docs] def assigned_actions_to_event(self, event):
self.context_event = event
view = navigate_to(self, "Event Details")
try:
true_actions = [row.description.text for row in view.true_actions.rows()]
except NoSuchElementException:
true_actions = []
try:
false_actions = [row.description.text for row in view.false_actions.rows()]
except NoSuchElementException:
false_actions = []
return true_actions + false_actions
[docs]class HostCompliancePolicy(BasePolicy):
TYPE = "Compliance"
TREE_NODE = "Host"
PRETTY = "Host / Node"
[docs]class VMCompliancePolicy(BasePolicy):
TYPE = "Compliance"
TREE_NODE = "Vm"
PRETTY = "VM and Instance"
[docs]class ReplicatorCompliancePolicy(BasePolicy):
TYPE = "Compliance"
TREE_NODE = "Replicator"
PRETTY = "Container Replicator"
[docs]class PodCompliancePolicy(BasePolicy):
TYPE = "Compliance"
TREE_NODE = "Pod"
PRETTY = "Container Pod"
[docs]class ContainerNodeCompliancePolicy(BasePolicy):
TYPE = "Compliance"
TREE_NODE = PRETTY = "Container Node"
[docs]class ContainerImageCompliancePolicy(BasePolicy):
TYPE = "Compliance"
TREE_NODE = PRETTY = "Container Image"
[docs]class ProviderCompliancePolicy(BasePolicy):
TYPE = "Compliance"
TREE_NODE = PRETTY = "Provider"
[docs]class PhysicalInfrastructureCompliancePolicy(BasePolicy):
TYPE = "Compliance"
TREE_NODE = "Physical Infrastructure"
PRETTY = "Physical Server"
[docs]class HostControlPolicy(BasePolicy):
TYPE = "Control"
TREE_NODE = "Host"
PRETTY = "Host / Node"
[docs]class VMControlPolicy(BasePolicy):
TYPE = "Control"
TREE_NODE = "Vm"
PRETTY = "VM and Instance"
[docs]class ReplicatorControlPolicy(BasePolicy):
TYPE = "Control"
TREE_NODE = "Replicator"
PRETTY = "Container Replicator"
[docs]class PodControlPolicy(BasePolicy):
TYPE = "Control"
TREE_NODE = "Pod"
PRETTY = "Container Pod"
[docs]class ContainerNodeControlPolicy(BasePolicy):
TYPE = "Control"
PRETTY = TREE_NODE = "Container Node"
[docs]class ContainerImageControlPolicy(BasePolicy):
TYPE = "Control"
PRETTY = TREE_NODE = "Container Image"
[docs]class ProviderControlPolicy(BasePolicy):
TYPE = "Control"
PRETTY = TREE_NODE = "Provider"
[docs]class PhysicalInfrastructureControlPolicy(BasePolicy):
TYPE = "Control"
TREE_NODE = "Physical Infrastructure"
PRETTY = "Physical Server"
@attr.s
[docs]class PolicyCollection(BaseCollection):
ENTITY = BasePolicy
HOST_COMPLIANCE_POLICY = HostCompliancePolicy
VM_COMPLIANCE_POLICY = VMCompliancePolicy
REPLICATOR_COMPLIANCE_POLICY = ReplicatorCompliancePolicy
POD_COMPLIANCE_POLICY = PodCompliancePolicy
CONTAINERNODE_COMPLIANCE_POLICY = ContainerNodeCompliancePolicy
PROVIDER_COMPLIANCE_POLICY = ProviderCompliancePolicy
PHYSICAL_INFRASTRUCTURE_COMPLIANCE_POLICY = PhysicalInfrastructureCompliancePolicy
HOST_CONTROL_POLICY = HostControlPolicy
VM_CONTROL_POLICY = VMControlPolicy
REPLICATOR_CONTROL_POLICY = ReplicatorControlPolicy
POD_CONTROL_POLICY = PodControlPolicy
CONTAINERNODE_CONTROL_POLICY = ContainerNodeControlPolicy
CONTAINERIMAGE_CONTROL_POLICY = ContainerImageControlPolicy
PROVIDER_CONTROL_POLICY = ProviderControlPolicy
PHYSICAL_INFRASTRUCTURE_CONTROL_POLICY = PhysicalInfrastructureControlPolicy
# A rare collection override of instantiate
[docs] def instantiate(self, policy_class, description, active=True, scope=None, notes=None):
return policy_class.from_collection(
self, description, active=active, scope=scope, notes=notes)
[docs] def create(self, policy_class, description, active=True, scope=None, notes=None):
policy = self.instantiate(policy_class, description, active=active, scope=scope,
notes=notes)
view = navigate_to(policy, "Add")
view.fill({
"description": policy.description,
"active": policy.active,
"scope": policy.scope,
"notes": policy.notes
})
view.add_button.click()
view = policy.create_view(PolicyDetailsView)
view.wait_displayed()
view.flash.assert_success_message('Policy "{}" was added'.format(policy.description))
return policy
[docs] def all(self):
raise NotImplementedError
[docs] def delete(self, *policies):
for policy in policies:
if policy.exists:
policy.delete()
@navigator.register(PolicyCollection, "All")
[docs]class PolicyAll(CFMENavigateStep):
VIEW = PoliciesAllView
prerequisite = NavigateToAttribute("appliance.server", "ControlExplorer")
[docs] def step(self):
self.prerequisite_view.policies.tree.click_path("All Policies")
@navigator.register(BasePolicy, "Add")
[docs]class PolicyNew(CFMENavigateStep):
VIEW = NewPolicyView
prerequisite = NavigateToAttribute("parent", "All")
[docs] def step(self):
self.prerequisite_view.policies.tree.click_path(
"All Policies",
"{} Policies".format(self.obj.TYPE),
"{} {} Policies".format(self.obj.TREE_NODE, self.obj.TYPE)
)
self.prerequisite_view.configuration.item_select("Add a New {} {} Policy".format(
self.obj.PRETTY, self.obj.TYPE))
@navigator.register(BasePolicy, "Edit")
[docs]class PolicyEdit(CFMENavigateStep):
VIEW = EditPolicyView
prerequisite = NavigateToSibling("Details")
[docs] def step(self):
self.prerequisite_view.configuration.item_select("Edit Basic Info, Scope, and Notes")
@navigator.register(BasePolicy, "Details")
[docs]class PolicyDetails(CFMENavigateStep):
VIEW = PolicyDetailsView
prerequisite = NavigateToAttribute("parent", "All")
[docs] def step(self):
self.prerequisite_view.policies.tree.click_path(
"All Policies",
"{} Policies".format(self.obj.TYPE),
"{} {} Policies".format(self.obj.TREE_NODE, self.obj.TYPE),
self.obj.description
)
@navigator.register(BasePolicy, "Event Details")
[docs]class PolicyEventDetails(CFMENavigateStep):
VIEW = EventDetailsView
prerequisite = NavigateToAttribute("appliance.server", "ControlExplorer")
[docs] def step(self):
self.view.policies.tree.click_path(
"All Policies",
"{} Policies".format(self.obj.TYPE),
"{} {} Policies".format(self.obj.TREE_NODE, self.obj.TYPE),
self.obj.description,
self.obj.context_event
)