Update ipervisors selection by attribute

Change-Id: I7087a61e39a2e3e3831fcbbfe1e8f02a89bd54ed

tobiko/__init__.py

@@ -19,6 +19,7 @@ from tobiko.common import _find
 from tobiko.common import _fixture
 from tobiko.common.managers import testcase as testcase_manager
 from tobiko.common.managers import loader as loader_manager
+from tobiko.common import _select
 from tobiko.common import _skip
 
 
@@ -53,6 +54,11 @@ load_module = loader_manager.load_module
 
 discover_testcases = testcase_manager.discover_testcases
 
+Selection = _select.Selection
+select = _select.select
+ObjectNotFound = _select.ObjectNotFound
+MultipleObjectsFound = _select.MultipleObjectsFound
+
 SkipException = _skip.SkipException
 skip = _skip.skip
 skip_if = _skip.skip_if

tobiko/common/_select.py

@@ -0,0 +1,87 @@
+# Copyright 2019 Red Hat
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+from __future__ import absolute_import
+
+from tobiko import _exception
+
+
+class Selection(list):
+
+    def with_attributes(self, **attributes):
+        return self.create(
+            filter_by_attributes(self, exclude=False, **attributes))
+
+    def without_attributes(self, **attributes):
+        return self.create(
+            filter_by_attributes(self, exclude=True, **attributes))
+
+    def with_items(self, **items):
+        return self.create(filter_by_items(self, exclude=False, **items))
+
+    def without_items(self, **items):
+        return self.create(filter_by_items(self, exclude=True, **items))
+
+    @classmethod
+    def create(cls, objects):
+        return cls(objects)
+
+    @property
+    def first(self):
+        if self:
+            return self[0]
+        else:
+            raise ObjectNotFound()
+
+    @property
+    def unique(self):
+        if len(self) > 1:
+            raise MultipleObjectsFound(list(self))
+        else:
+            return self.first
+
+    def __repr__(self):
+        return '{!s}({!r})'.format(type(self).__name__, list(self))
+
+
+select = Selection.create
+
+
+def filter_by_attributes(objects, exclude=False, **attributes):
+    exclude = bool(exclude)
+    for obj in objects:
+        for key, value in attributes.items():
+            matching = value == getattr(obj, key)
+            if matching is exclude:
+                break
+        else:
+            yield obj
+
+
+def filter_by_items(dictionaries, exclude=False, **items):
+    exclude = bool(exclude)
+    for dictionary in dictionaries:
+        for key, value in items.items():
+            matching = value == dictionary[key]
+            if matching is exclude:
+                break
+        else:
+            yield dictionary
+
+
+class ObjectNotFound(_exception.TobikoException):
+    "Object not found."
+
+
+class MultipleObjectsFound(_exception.TobikoException):
+    "Multiple objects found: {objects!r}"

tobiko/openstack/nova/__init__.py

@@ -22,5 +22,6 @@ get_nova_client = _client.get_nova_client
 list_hypervisors = _client.list_hypervisors
 nova_client = _client.nova_client
 NovaClientFixture = _client.NovaClientFixture
+find_hypervisor = _client.find_hypervisor
 
 skip_if_missing_hypervisors = _hypervisor.skip_if_missing_hypervisors

tobiko/openstack/nova/_client.py

@@ -65,4 +65,12 @@ def get_nova_client(session=None, shared=True, init_client=None,
 def list_hypervisors(client=None, detailed=True, **params):
     client = nova_client(client)
     hypervisors = client.hypervisors.list(detailed=detailed)
-    return tobiko.find_by_attributes(hypervisors, **params)
+    return tobiko.select(hypervisors).with_attributes(**params)
+
+
+def find_hypervisor(client=None, unique=False, **params):
+    hypervisors = list_hypervisors(client=client, **params)
+    if unique:
+        return hypervisors.unique
+    else:
+        return hypervisors.first

tobiko/openstack/nova/_hypervisor.py

@@ -35,7 +35,7 @@ class HypervisorListFixture(tobiko.SharedFixture):
 
 def get_hypervisors(**params):
     hypervisors = tobiko.setup_fixture(HypervisorListFixture).hypervisors
-    return tobiko.find_by_attributes(hypervisors, **params)
+    return tobiko.select(hypervisors).with_attributes(**params)
 
 
 def missing_hypervisors(count=1, **params):

tobiko/tests/functional/openstack/test_nova.py

@@ -36,24 +36,42 @@ class KeyPairTest(testtools.TestCase):
 
 class ClientTest(testtools.TestCase):
 
+    @nova.skip_if_missing_hypervisors(count=1)
     def test_list_hypervisors(self):
         hypervisors = nova.list_hypervisors()
         self.assertTrue(hypervisors)
         self.assertEqual(1, hypervisors[0].id)
         self.assertTrue(hasattr(hypervisors[0], 'cpu_info'))
 
+    @nova.skip_if_missing_hypervisors(count=1)
     def test_list_hypervisors_without_details(self):
         hypervisors = nova.list_hypervisors(detailed=False)
         self.assertTrue(hypervisors)
         self.assertEqual(1, hypervisors[0].id)
         self.assertFalse(hasattr(hypervisors[0], 'cpu_info'))
 
+    @nova.skip_if_missing_hypervisors(count=1)
     def test_list_hypervisors_with_hypervisor_hostname(self):
         hypervisor = nova.list_hypervisors()[0]
         hypervisors = nova.list_hypervisors(
             hypervisor_hostname=hypervisor.hypervisor_hostname)
         self.assertEqual([hypervisor], hypervisors)
 
+    @nova.skip_if_missing_hypervisors(count=1)
+    def test_find_hypervisor(self):
+        hypervisor = nova.find_hypervisor()
+        self.assertIsNotNone(hypervisor)
+
+    @nova.skip_if_missing_hypervisors(count=2)
+    def test_find_hypervisor_with_unique(self):
+        self.assertRaises(tobiko.MultipleObjectsFound, nova.find_hypervisor,
+                          unique=True)
+
+    @nova.skip_if_missing_hypervisors(count=2)
+    def test_find_hypervisor_without_unique(self):
+        hypervisor = nova.find_hypervisor()
+        self.assertIsNotNone(hypervisor)
+
 
 class HypervisorTest(testtools.TestCase):