openstack
/
watcher
Code Issues Proposed changes

Add importing modules instead of classes 

This patch removes import of classes and replaces them with import of modules.

Change-Id: Id2502cf96ac7227cf1036cd54a74f3d7acd83479
Closes-Bug: #1594434
This commit is contained in:
Alexandr Stavitskiy 2016-06-21 19:37:01 +03:00
parent e3edc67045
commit cb693e4093
24 changed files with 131 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
watcher/common/ceilometer_helper.py
View File

@ -17,7 +17,7 @@
# limitations under the License.
#
from ceilometerclient.exc import HTTPUnauthorized
from ceilometerclient import exc
from watcher.common import clients
@ -68,7 +68,7 @@ class CeilometerHelper(object):
def query_retry(self, f, *args, **kargs):
try:
return f(*args, **kargs)
except HTTPUnauthorized:
except exc.HTTPUnauthorized:
self.osc.reset_clients()
self.ceilometer = self.osc.ceilometer()
return f(*args, **kargs)

4
watcher/decision_engine/messaging/events.py
View File

@ -17,10 +17,10 @@
# limitations under the License.
#
from enum import Enum
import enum
class Events(Enum):
class Events(enum.Enum):
ALL = '*',
ACTION_PLAN = "action_plan"
TRIGGER_AUDIT = "trigger_audit"

14
watcher/decision_engine/model/hypervisor.py
View File

@ -14,17 +14,17 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from watcher.decision_engine.model.compute_resource import ComputeResource
from watcher.decision_engine.model.hypervisor_state import HypervisorState
from watcher.decision_engine.model.power_state import PowerState
from watcher.decision_engine.model import compute_resource
from watcher.decision_engine.model import hypervisor_state
from watcher.decision_engine.model import power_state
class Hypervisor(ComputeResource):
class Hypervisor(compute_resource.ComputeResource):
def __init__(self):
super(Hypervisor, self).__init__()
self._state = HypervisorState.ONLINE
self._status = HypervisorState.ENABLED
self._power_state = PowerState.g0
self._state = hypervisor_state.HypervisorState.ONLINE
self._status = hypervisor_state.HypervisorState.ENABLED
self._power_state = power_state.PowerState.g0
@property
def state(self):

4
watcher/decision_engine/model/hypervisor_state.py
View File

@ -14,10 +14,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from enum import Enum
import enum
class HypervisorState(Enum):
class HypervisorState(enum.Enum):
ONLINE = 'up'
OFFLINE = 'down'
ENABLED = 'enabled'

4
watcher/decision_engine/model/mapping.py
View File

@ -14,7 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from oslo_log import log
from threading import Lock
import threading
LOG = log.getLogger(__name__)
@ -24,7 +24,7 @@ class Mapping(object):
self.model = model
self._mapping_hypervisors = {}
self.mapping_vm = {}
self.lock = Lock()
self.lock = threading.Lock()
def map(self, hypervisor, vm):
"""Select the hypervisor where the instance are launched

4
watcher/decision_engine/model/power_state.py
View File

@ -10,10 +10,10 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from enum import Enum
import enum
class PowerState(Enum):
class PowerState(enum.Enum):
# away mode
g0 = "g0"
# power on suspend (processor caches are flushed)

4
watcher/decision_engine/model/resource.py
View File

@ -14,10 +14,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from enum import Enum
import enum
class ResourceType(Enum):
class ResourceType(enum.Enum):
cpu_cores = 'num_cores'
memory = 'memory'
disk = 'disk'

8
watcher/decision_engine/model/vm.py
View File

@ -13,14 +13,14 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from watcher.decision_engine.model.compute_resource import ComputeResource
from watcher.decision_engine.model.vm_state import VMState
from watcher.decision_engine.model import compute_resource
from watcher.decision_engine.model import vm_state
class VM(ComputeResource):
class VM(compute_resource.ComputeResource):
def __init__(self):
super(VM, self).__init__()
self._state = VMState.ACTIVE.value
self._state = vm_state.VMState.ACTIVE.value
@property
def state(self):

4
watcher/decision_engine/model/vm_state.py
View File

@ -14,10 +14,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from enum import Enum
import enum
class VMState(Enum):
class VMState(enum.Enum):
ACTIVE = 'active' # VM is running
BUILDING = 'building' # VM only exists in DB
PAUSED = 'paused'

6
watcher/tests/applier/test_rpcapi.py
View File

@ -19,7 +19,7 @@
import mock
import oslo_messaging as om
from watcher.applier.rpcapi import ApplierAPI
from watcher.applier import rpcapi
from watcher.common import exception
from watcher.common import utils
@ -30,7 +30,7 @@ class TestApplierAPI(base.TestCase):
def setUp(self):
super(TestApplierAPI, self).setUp()
api = ApplierAPI()
api = rpcapi.ApplierAPI()
def test_get_version(self):
expected_version = self.api.API_VERSION
@ -43,7 +43,7 @@ class TestApplierAPI(base.TestCase):
mock_call.assert_called_once_with(
expected_context.to_dict(),
'check_api_version',
api_version=ApplierAPI().API_VERSION)
api_version=rpcapi.ApplierAPI().API_VERSION)
def test_execute_audit_without_error(self):
with mock.patch.object(om.RPCClient, 'call') as mock_call:

4
watcher/tests/cmd/test_applier.py
View File

@ -24,10 +24,10 @@ from oslo_config import cfg
from oslo_service import service
from watcher.cmd import applier
from watcher.tests.base import BaseTestCase
from watcher.tests import base
class TestApplier(BaseTestCase):
class TestApplier(base.BaseTestCase):
def setUp(self):
super(TestApplier, self).setUp()

6
watcher/tests/common/loader/test_loader.py
View File

@ -20,7 +20,7 @@ import mock
from oslo_config import cfg
from stevedore import driver as drivermanager
from stevedore.extension import Extension
from stevedore import extension as stevedore_extension
from watcher.common import exception
from watcher.common.loader import default
@ -56,7 +56,7 @@ class TestLoader(base.TestCase):
def test_load_loadable_no_opt(self):
fake_driver = drivermanager.DriverManager.make_test_instance(
extension=Extension(
extension=stevedore_extension.Extension(
name="fake",
entry_point="%s:%s" % (FakeLoadable.__module__,
FakeLoadable.__name__),
@ -82,7 +82,7 @@ class TestLoader(base.TestCase):
def test_load_loadable_with_opts(self):
fake_driver = drivermanager.DriverManager.make_test_instance(
extension=Extension(
extension=stevedore_extension.Extension(
name="fake",
entry_point="%s:%s" % (FakeLoadableWithOpts.__module__,
FakeLoadableWithOpts.__name__),

46
watcher/tests/common/messaging/event/test_event_dispatcher.py
View File

@ -13,11 +13,13 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import mock
from mock import call
from mock import MagicMock
from watcher.common.messaging.events.event import Event
from watcher.common.messaging.events.event_dispatcher import EventDispatcher
from watcher.decision_engine.messaging.events import Events
from watcher.common.messaging.events import event as messaging_event
from watcher.common.messaging.events import event_dispatcher
from watcher.decision_engine.messaging import events as messaging_events
from watcher.tests import base
@ -25,54 +27,56 @@ class TestEventDispatcher(base.TestCase):
def setUp(self):
super(TestEventDispatcher, self).setUp()
self.event_dispatcher = EventDispatcher()
self.event_dispatcher = event_dispatcher.EventDispatcher()
def fake_listener(self):
return MagicMock()
return mock.MagicMock()
def fake_event(self, event_type):
event = Event()
event = messaging_event.Event()
event.type = event_type
return event
def test_add_listener(self):
listener = self.fake_listener()
self.event_dispatcher.add_event_listener(Events.ALL,
self.event_dispatcher.add_event_listener(messaging_events.Events.ALL,
listener)
self.assertTrue(self.event_dispatcher.has_listener(
Events.ALL, listener))
messaging_events.Events.ALL, listener))
def test_remove_listener(self):
listener = self.fake_listener()
self.event_dispatcher.add_event_listener(Events.ALL,
self.event_dispatcher.add_event_listener(messaging_events.Events.ALL,
listener)
self.event_dispatcher.remove_event_listener(Events.ALL, listener)
self.event_dispatcher.remove_event_listener(
messaging_events.Events.ALL, listener)
self.assertFalse(self.event_dispatcher.has_listener(
Events.TRIGGER_AUDIT, listener))
messaging_events.Events.TRIGGER_AUDIT, listener))
def test_dispatch_event(self):
listener = self.fake_listener()
event = self.fake_event(Events.TRIGGER_AUDIT)
self.event_dispatcher.add_event_listener(Events.TRIGGER_AUDIT,
listener)
event = self.fake_event(messaging_events.Events.TRIGGER_AUDIT)
self.event_dispatcher.add_event_listener(
messaging_events.Events.TRIGGER_AUDIT, listener)
self.event_dispatcher.dispatch_event(event)
listener.assert_has_calls(calls=[call(event)])
def test_dispatch_event_to_all_listener(self):
event = self.fake_event(Events.ACTION_PLAN)
event = self.fake_event(messaging_events.Events.ACTION_PLAN)
listener_all = self.fake_listener()
listener_action_plan = self.fake_listener()
listener_trigger_audit = self.fake_listener()
self.event_dispatcher.add_event_listener(Events.ALL, listener_all)
self.event_dispatcher.add_event_listener(Events.ACTION_PLAN,
listener_action_plan)
self.event_dispatcher.add_event_listener(
messaging_events.Events.ALL, listener_all)
self.event_dispatcher.add_event_listener(
messaging_events.Events.ACTION_PLAN, listener_action_plan)
self.event_dispatcher.add_event_listener(Events.TRIGGER_AUDIT,
listener_trigger_audit)
self.event_dispatcher.add_event_listener(
messaging_events.Events.TRIGGER_AUDIT, listener_trigger_audit)
self.event_dispatcher.dispatch_event(event)
listener_all.assert_has_calls(calls=[call(event)])

9
watcher/tests/common/messaging/test_notification_handler.py
View File

@ -17,8 +17,8 @@
import mock
import oslo_messaging as messaging
from watcher.common.messaging.notification_handler import NotificationHandler
from watcher.common.messaging.utils.observable import Observable
from watcher.common.messaging import notification_handler
from watcher.common.messaging.utils import observable
from watcher.tests import base
PUBLISHER_ID = 'TEST_API'
@ -28,7 +28,8 @@ class TestNotificationHandler(base.TestCase):
def setUp(self):
super(TestNotificationHandler, self).setUp()
self.notification_handler = NotificationHandler(PUBLISHER_ID)
self.notification_handler = notification_handler.NotificationHandler(
PUBLISHER_ID)
def _test_notify(self, level_to_call):
ctx = {}
@ -37,7 +38,7 @@ class TestNotificationHandler(base.TestCase):
payload = {}
metadata = {}
with mock.patch.object(Observable, 'notify') as mock_call:
with mock.patch.object(observable.Observable, 'notify') as mock_call:
notification_result = level_to_call(ctx, publisher_id, event_type,
payload, metadata)
self.assertEqual(messaging.NotificationResult.HANDLED,

36
watcher/tests/decision_engine/messaging/test_audit_endpoint.py
View File

@ -17,17 +17,16 @@
import mock
from watcher.common import utils
from watcher.decision_engine.audit.default import DefaultAuditHandler
from watcher.decision_engine.messaging.audit_endpoint import AuditEndpoint
from watcher.metrics_engine.cluster_model_collector.manager import \
CollectorManager
from watcher.tests.db.base import DbTestCase
from watcher.tests.decision_engine.strategy.strategies.faker_cluster_state \
import FakerModelCollector
from watcher.decision_engine.audit import default
from watcher.decision_engine.messaging import audit_endpoint
from watcher.metrics_engine.cluster_model_collector import manager
from watcher.tests.db import base
from watcher.tests.decision_engine.strategy.strategies \
import faker_cluster_state
from watcher.tests.objects import utils as obj_utils
class TestAuditEndpoint(DbTestCase):
class TestAuditEndpoint(base.DbTestCase):
def setUp(self):
super(TestAuditEndpoint, self).setUp()
self.audit_template = obj_utils.create_test_audit_template(
@ -36,28 +35,29 @@ class TestAuditEndpoint(DbTestCase):
self.context,
audit_template_id=self.audit_template.id)
@mock.patch.object(CollectorManager, "get_cluster_model_collector")
@mock.patch.object(manager.CollectorManager, "get_cluster_model_collector")
def test_do_trigger_audit(self, mock_collector):
mock_collector.return_value = FakerModelCollector()
mock_collector.return_value = faker_cluster_state.FakerModelCollector()
audit_uuid = utils.generate_uuid()
audit_handler = DefaultAuditHandler(mock.MagicMock())
endpoint = AuditEndpoint(audit_handler)
audit_handler = default.DefaultAuditHandler(mock.MagicMock())
endpoint = audit_endpoint.AuditEndpoint(audit_handler)
with mock.patch.object(DefaultAuditHandler, 'execute') as mock_call:
with mock.patch.object(default.DefaultAuditHandler,
'execute') as mock_call:
mock_call.return_value = 0
endpoint.do_trigger_audit(audit_handler, audit_uuid)
mock_call.assert_called_once_with(audit_uuid, audit_handler)
@mock.patch.object(CollectorManager, "get_cluster_model_collector")
@mock.patch.object(manager.CollectorManager, "get_cluster_model_collector")
def test_trigger_audit(self, mock_collector):
mock_collector.return_value = FakerModelCollector()
mock_collector.return_value = faker_cluster_state.FakerModelCollector()
audit_uuid = utils.generate_uuid()
audit_handler = DefaultAuditHandler(mock.MagicMock())
endpoint = AuditEndpoint(audit_handler)
audit_handler = default.DefaultAuditHandler(mock.MagicMock())
endpoint = audit_endpoint.AuditEndpoint(audit_handler)
with mock.patch.object(DefaultAuditHandler, 'execute') \
with mock.patch.object(default.DefaultAuditHandler, 'execute') \
as mock_call:
mock_call.return_value = 0
endpoint.trigger_audit(audit_handler, audit_uuid)

4
watcher/tests/decision_engine/model/test_disk_info.py
View File

@ -17,13 +17,13 @@
# limitations under the License.
#
from watcher.decision_engine.model.disk_info import DiskInfo
from watcher.decision_engine.model import disk_info
from watcher.tests import base
class TestDiskInfo(base.BaseTestCase):
def test_all(self):
disk_information = DiskInfo()
disk_information = disk_info.DiskInfo()
disk_information.set_size(1024)
self.assertEqual(1024, disk_information.get_size())

58
watcher/tests/decision_engine/model/test_model.py
View File

@ -19,18 +19,17 @@
import uuid
from watcher.common import exception
from watcher.common.exception import IllegalArgumentException
from watcher.decision_engine.model.hypervisor import Hypervisor
from watcher.decision_engine.model.hypervisor_state import HypervisorState
from watcher.decision_engine.model.model_root import ModelRoot
from watcher.decision_engine.model import hypervisor as hypervisor_model
from watcher.decision_engine.model import hypervisor_state
from watcher.decision_engine.model import model_root
from watcher.tests import base
from watcher.tests.decision_engine.strategy.strategies.faker_cluster_state import \
FakerModelCollector
from watcher.tests.decision_engine.strategy.strategies \
import faker_cluster_state
class TestModel(base.BaseTestCase):
def test_model(self):
fake_cluster = FakerModelCollector()
fake_cluster = faker_cluster_state.FakerModelCollector()
model = fake_cluster.generate_scenario_1()
self.assertEqual(5, len(model._hypervisors))
@ -38,17 +37,17 @@ class TestModel(base.BaseTestCase):
self.assertEqual(5, len(model.get_mapping().get_mapping()))
def test_add_hypervisor(self):
model = ModelRoot()
model = model_root.ModelRoot()
id = "{0}".format(uuid.uuid4())
hypervisor = Hypervisor()
hypervisor = hypervisor_model.Hypervisor()
hypervisor.uuid = id
model.add_hypervisor(hypervisor)
self.assertEqual(hypervisor, model.get_hypervisor_from_id(id))
def test_delete_hypervisor(self):
model = ModelRoot()
model = model_root.ModelRoot()
id = "{0}".format(uuid.uuid4())
hypervisor = Hypervisor()
hypervisor = hypervisor_model.Hypervisor()
hypervisor.uuid = id
model.add_hypervisor(hypervisor)
self.assertEqual(hypervisor, model.get_hypervisor_from_id(id))
@ -57,10 +56,10 @@ class TestModel(base.BaseTestCase):
model.get_hypervisor_from_id, id)
def test_get_all_hypervisors(self):
model = ModelRoot()
model = model_root.ModelRoot()
for i in range(10):
id = "{0}".format(uuid.uuid4())
hypervisor = Hypervisor()
hypervisor = hypervisor_model.Hypervisor()
hypervisor.uuid = id
model.add_hypervisor(hypervisor)
all_hypervisors = model.get_all_hypervisors()
@ -69,17 +68,18 @@ class TestModel(base.BaseTestCase):
model.assert_hypervisor(hyp)
def test_set_get_state_hypervisors(self):
model = ModelRoot()
model = model_root.ModelRoot()
id = "{0}".format(uuid.uuid4())
hypervisor = Hypervisor()
hypervisor = hypervisor_model.Hypervisor()
hypervisor.uuid = id
model.add_hypervisor(hypervisor)
self.assertIsInstance(hypervisor.state, HypervisorState)
self.assertIsInstance(hypervisor.state,
hypervisor_state.HypervisorState)
hyp = model.get_hypervisor_from_id(id)
hyp.state = HypervisorState.OFFLINE
self.assertIsInstance(hyp.state, HypervisorState)
hyp.state = hypervisor_state.HypervisorState.OFFLINE
self.assertIsInstance(hyp.state, hypervisor_state.HypervisorState)
# /watcher/decision_engine/framework/model/hypervisor.py
# set_state accept any char chain.
@ -93,9 +93,9 @@ class TestModel(base.BaseTestCase):
# vms = model.get_all_vms()
# self.assert(len(model._vms))
def test_hypervisor_from_id_raise(self):
model = ModelRoot()
model = model_root.ModelRoot()
id = "{0}".format(uuid.uuid4())
hypervisor = Hypervisor()
hypervisor = hypervisor_model.Hypervisor()
hypervisor.uuid = id
model.add_hypervisor(hypervisor)
@ -104,35 +104,35 @@ class TestModel(base.BaseTestCase):
model.get_hypervisor_from_id, id2)
def test_remove_hypervisor_raise(self):
model = ModelRoot()
model = model_root.ModelRoot()
id = "{0}".format(uuid.uuid4())
hypervisor = Hypervisor()
hypervisor = hypervisor_model.Hypervisor()
hypervisor.uuid = id
model.add_hypervisor(hypervisor)
id2 = "{0}".format(uuid.uuid4())
hypervisor2 = Hypervisor()
hypervisor2 = hypervisor_model.Hypervisor()
hypervisor2.uuid = id2
self.assertRaises(exception.HypervisorNotFound,
model.remove_hypervisor, hypervisor2)
def test_assert_hypervisor_raise(self):
model = ModelRoot()
model = model_root.ModelRoot()
id = "{0}".format(uuid.uuid4())
hypervisor = Hypervisor()
hypervisor = hypervisor_model.Hypervisor()
hypervisor.uuid = id
model.add_hypervisor(hypervisor)
self.assertRaises(IllegalArgumentException,
self.assertRaises(exception.IllegalArgumentException,
model.assert_hypervisor, "objet_qcq")
def test_vm_from_id_raise(self):
fake_cluster = FakerModelCollector()
fake_cluster = faker_cluster_state.FakerModelCollector()
model = fake_cluster.generate_scenario_1()
self.assertRaises(exception.InstanceNotFound,
model.get_vm_from_id, "valeur_qcq")
def test_assert_vm_raise(self):
model = ModelRoot()
self.assertRaises(IllegalArgumentException,
model = model_root.ModelRoot()
self.assertRaises(exception.IllegalArgumentException,
model.assert_vm, "valeur_qcq")

6
watcher/tests/decision_engine/model/test_named_element.py
View File

@ -16,17 +16,17 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
from watcher.decision_engine.model.compute_resource import ComputeResource
from watcher.decision_engine.model import compute_resource
from watcher.tests import base
class TestNamedElement(base.BaseTestCase):
def test_namedelement(self):
id = ComputeResource()
id = compute_resource.ComputeResource()
id.uuid = "BLABLABLA"
self.assertEqual("BLABLABLA", id.uuid)
def test_set_get_human_id(self):
id = ComputeResource()
id = compute_resource.ComputeResource()
id.human_id = "BLABLABLA"
self.assertEqual("BLABLABLA", id.human_id)

10
watcher/tests/decision_engine/model/test_vm.py
View File

@ -16,15 +16,15 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
from watcher.decision_engine.model.vm import VM
from watcher.decision_engine.model.vm_state import VMState
from watcher.decision_engine.model import vm as vm_model
from watcher.decision_engine.model import vm_state
from watcher.tests import base
class TestVm(base.BaseTestCase):
def test_namedelement(self):
vm = VM()
vm.state = VMState.ACTIVE
self.assertEqual(VMState.ACTIVE, vm.state)
vm = vm_model.VM()
vm.state = vm_state.VMState.ACTIVE
self.assertEqual(vm_state.VMState.ACTIVE, vm.state)
vm.human_id = "human_05"
self.assertEqual("human_05", vm.human_id)

6
watcher/tests/decision_engine/test_rpcapi.py
View File

@ -18,7 +18,7 @@ import mock
import oslo_messaging as om
from watcher.common import exception
from watcher.common import utils
from watcher.decision_engine.rpcapi import DecisionEngineAPI
from watcher.decision_engine import rpcapi
from watcher.tests import base
@ -27,7 +27,7 @@ class TestDecisionEngineAPI(base.TestCase):
def setUp(self):
super(TestDecisionEngineAPI, self).setUp()
api = DecisionEngineAPI()
api = rpcapi.DecisionEngineAPI()
def test_get_version(self):
expected_version = self.api.API_VERSION
@ -40,7 +40,7 @@ class TestDecisionEngineAPI(base.TestCase):
mock_call.assert_called_once_with(
expected_context.to_dict(),
'check_api_version',
api_version=DecisionEngineAPI().api_version)
api_version=rpcapi.DecisionEngineAPI().api_version)
def test_execute_audit_throw_exception(self):
audit_uuid = "uuid"

4
watcher/tests/objects/test_action.py
View File

@ -14,7 +14,7 @@
# under the License.
import mock
from testtools.matchers import HasLength
from testtools import matchers
from watcher.common import exception
# from watcher.common import utils as w_utils
@ -58,7 +58,7 @@ class TestActionObject(base.DbTestCase):
mock_get_list.return_value = [self.fake_action]
actions = objects.Action.list(self.context)
self.assertEqual(1, mock_get_list.call_count)
self.assertThat(actions, HasLength(1))
self.assertThat(actions, matchers.HasLength(1))
self.assertIsInstance(actions[0], objects.Action)
self.assertEqual(self.context, actions[0]._context)

4
watcher/tests/objects/test_action_plan.py
View File

@ -14,7 +14,7 @@
# under the License.
import mock
from testtools.matchers import HasLength
from testtools import matchers
from watcher.common import exception
# from watcher.common import utils as w_utils
@ -58,7 +58,7 @@ class TestActionPlanObject(base.DbTestCase):
mock_get_list.return_value = [self.fake_action_plan]
action_plans = objects.ActionPlan.list(self.context)
self.assertEqual(1, mock_get_list.call_count)
self.assertThat(action_plans, HasLength(1))
self.assertThat(action_plans, matchers.HasLength(1))
self.assertIsInstance(action_plans[0], objects.ActionPlan)
self.assertEqual(self.context, action_plans[0]._context)

4
watcher/tests/objects/test_audit.py
View File

@ -14,7 +14,7 @@
# under the License.
import mock
from testtools.matchers import HasLength
from testtools import matchers
from watcher.common import exception
# from watcher.common import utils as w_utils
@ -58,7 +58,7 @@ class TestAuditObject(base.DbTestCase):
mock_get_list.return_value = [self.fake_audit]
audits = objects.Audit.list(self.context)
self.assertEqual(1, mock_get_list.call_count, 1)
self.assertThat(audits, HasLength(1))
self.assertThat(audits, matchers.HasLength(1))
self.assertIsInstance(audits[0], objects.Audit)
self.assertEqual(self.context, audits[0]._context)

4
watcher/tests/objects/test_efficacy_indicator.py
View File

@ -14,7 +14,7 @@
# under the License.
import mock
from testtools.matchers import HasLength
from testtools import matchers
from watcher.common import exception
# from watcher.common import utils as w_utils
@ -64,7 +64,7 @@ class TestEfficacyIndicatorObject(base.DbTestCase):
mock_get_list.return_value = [self.fake_efficacy_indicator]
efficacy_indicators = objects.EfficacyIndicator.list(self.context)
self.assertEqual(1, mock_get_list.call_count)
self.assertThat(efficacy_indicators, HasLength(1))
self.assertThat(efficacy_indicators, matchers.HasLength(1))
self.assertIsInstance(
efficacy_indicators[0], objects.EfficacyIndicator)
self.assertEqual(self.context, efficacy_indicators[0]._context)