# -*- coding: utf-8 -*-
import attr
import itertools
import random
from cached_property import cached_property
from navmazing import NavigateToSibling, NavigateToAttribute
from wrapanapi.containers.image import Image as ApiImage
from cfme.common import (WidgetasticTaggable, PolicyProfileAssignable,
TagPageView)
from cfme.containers.provider import (Labelable, navigate_and_get_rows, ContainerObjectAllBaseView,
ContainerObjectDetailsBaseView, LoadDetailsMixin,
refresh_and_navigate, ContainerObjectDetailsEntities,
ContainersProvider)
from cfme.utils.appliance.implementations.ui import CFMENavigateStep, navigator, navigate_to
from cfme.configure import tasks
from cfme.modeling.base import BaseCollection, BaseEntity
from cfme.utils.wait import wait_for, TimedOutError
from widgetastic_manageiq import SummaryTable, BaseEntitiesView
from widgetastic.widget import View
[docs]class ImageAllView(ContainerObjectAllBaseView):
SUMMARY_TEXT = "Container Images"
# ProviderEntity has its own fields, image view should rather use BaseEntity instead
including_entities = View.include(BaseEntitiesView, use_parent=True)
[docs]class ImageDetailsView(ContainerObjectDetailsBaseView):
@View.nested
class entities(ContainerObjectDetailsEntities): # noqa
configuration = SummaryTable(title='Configuration')
compliance = SummaryTable(title='Compliance')
@attr.s
[docs]class Image(BaseEntity, WidgetasticTaggable, Labelable, LoadDetailsMixin, PolicyProfileAssignable):
PLURAL = 'Container Images'
all_view = ImageAllView
details_view = ImageDetailsView
name = attr.ib()
id = attr.ib()
provider = attr.ib()
@cached_property
def mgmt(self):
return ApiImage(self.provider.mgmt, self.name, self.sha256)
@cached_property
def sha256(self):
return self.id.split('@')[-1]
@classmethod
[docs] def get_random_instances(cls, provider, count=1, appliance=None, docker_only=False):
"""Generating random instances. (docker_only: means for docker images only)"""
# Grab the images from the UI since we have no way to calculate the name by API attributes
rows = navigate_and_get_rows(provider, cls, count=1000)
if docker_only:
docker_image_ids = [img.id for img in provider.mgmt.list_docker_image()]
rows = filter(lambda r: r.id.text.split('@')[-1] in docker_image_ids,
rows)
random.shuffle(rows)
return [cls(row.name.text, row.id.text, provider, appliance=appliance)
for row in itertools.islice(rows, count)]
[docs] def check_compliance(self, wait_for_finish=True, timeout=240):
"""Initiates compliance check and waits for it to finish."""
view = navigate_to(self, 'Details')
original_state = self.compliance_status
view.toolbar.policy.item_select("Check Compliance of Last Known Configuration",
handle_alert=True)
view.flash.assert_no_error()
if wait_for_finish:
wait_for(
lambda: self.compliance_status != original_state, num_sec=timeout, delay=5,
message='compliance state of {} still matches {}'
.format(self.name, original_state)
)
return self.compliant
@property
def compliance_status(self):
view = refresh_and_navigate(self, 'Details')
return view.entities.compliance.read().get('Status').strip()
@property
def compliant(self):
"""Check if the image is compliant
Returns:
:py:class:`NoneType` if the image was never verified, otherwise :py:class:`bool`
"""
text = self.compliance_status.lower()
if text == "never verified":
return None
elif text.startswith("non-compliant"):
return False
elif text.startswith("compliant"):
return True
else:
raise ValueError("{} is not a known state for compliance".format(text))
@attr.s
[docs]class ImageCollection(BaseCollection):
"""Collection object for :py:class:`Image`."""
ENTITY = Image
[docs] def all(self):
image_table = self.appliance.db.client['container_images']
ems_table = self.appliance.db.client['ext_management_systems']
image_query = self.appliance.db.client.session.query(
image_table.name, image_table.image_ref, ems_table.name).join(
ems_table, image_table.ems_id == ems_table.id)
images = []
for name, id, provider_name in image_query.all():
images.append(self.instantiate(name=name, id=id,
provider=ContainersProvider(name=provider_name, appliance=self.appliance)))
return images
@navigator.register(ImageCollection, 'All')
[docs]class All(CFMENavigateStep):
VIEW = ImageAllView
prerequisite = NavigateToAttribute('appliance.server', 'LoggedIn')
[docs] def step(self):
self.prerequisite_view.navigation.select('Compute', 'Containers', 'Container Images')
[docs] def resetter(self):
# Reset view and selection
self.view.toolbar.view_selector.select("List View")
self.view.paginator.set_items_per_page(1000)
@navigator.register(Image, 'Details')
[docs]class Details(CFMENavigateStep):
VIEW = ImageDetailsView
prerequisite = NavigateToAttribute("parent", 'All')
[docs] def step(self):
self.prerequisite_view.entities.get_entity(provider=self.obj.provider.name,
id=self.obj.id).click()
@navigator.register(Image, 'EditTags')