python类sentinel()的实例源码

def test_instance_get_all_by_filters_paginated_allows_deleted_marker(
            self, mock_paginate, mock_fill, mock_get):
        ctxt = mock.MagicMock()
        ctxt.elevated.return_value = mock.sentinel.elevated
        sqlalchemy_api.instance_get_all_by_filters_sort(ctxt, {}, marker='foo')
        mock_get.assert_called_once_with(mock.sentinel.elevated, 'foo')
        ctxt.elevated.assert_called_once_with(read_deleted='yes')

def test_create_quota_usage_if_missing_not_created(self):
        # Tests that QuotaUsage isn't created if it's already in user_usages.
        resource = 'fake-resource'
        project_id = 'fake-project'
        user_id = 'fake_user'
        session = mock.sentinel
        quota_usage = mock.sentinel
        user_usages = {resource: quota_usage}
        with mock.patch.object(sqlalchemy_api, '_quota_usage_create') as quc:
            self.assertFalse(sqlalchemy_api._create_quota_usage_if_missing(
                                user_usages, resource, None,
                                project_id, user_id, session))
        self.assertFalse(quc.called)

def test_migrate_disk_and_power_off_resize_copy_disk_info(self,
                                                              mock_disk_info,
                                                              mock_shared,
                                                              mock_path,
                                                              mock_destroy,
                                                              mock_copy,
                                                              mock_execuate):

        instance = self._create_instance()
        disk_info = self._disk_info()
        disk_info_text = jsonutils.loads(disk_info)
        instance_base = os.path.dirname(disk_info_text[0]['path'])
        flavor = {'root_gb': 10, 'ephemeral_gb': 25}
        flavor_obj = objects.Flavor(**flavor)

        mock_disk_info.return_value = disk_info
        mock_path.return_value = instance_base
        mock_shared.return_value = False

        src_disk_info_path = os.path.join(instance_base + '_resize',
                                          'disk.info')

        with mock.patch.object(os.path, 'exists', autospec=True) \
                as mock_exists:
            # disk.info exists on the source
            mock_exists.side_effect = \
                lambda path: path == src_disk_info_path
            self.drvr.migrate_disk_and_power_off(context.get_admin_context(),
                                                 instance, mock.sentinel,
                                                 flavor_obj, None)
            self.assertTrue(mock_exists.called)

        dst_disk_info_path = os.path.join(instance_base, 'disk.info')
        mock_copy.assert_any_call(src_disk_info_path, dst_disk_info_path,
                                  host=mock.sentinel, on_execute=mock.ANY,
                                  on_completion=mock.ANY)

def test_get_by_uuid(self, mock_db_get):
        obj = objects.InstanceGroup.get_by_uuid(mock.sentinel.ctx,
                                                       _DB_UUID)
        mock_db_get.assert_called_once_with(mock.sentinel.ctx, _DB_UUID)
        self.assertEqual(_INST_GROUP_DB['members'], obj.members)
        self.assertEqual(_INST_GROUP_DB['policies'], obj.policies)
        self.assertEqual(_DB_UUID, obj.uuid)
        self.assertEqual(_INST_GROUP_DB['project_id'], obj.project_id)
        self.assertEqual(_INST_GROUP_DB['user_id'], obj.user_id)
        self.assertEqual(_INST_GROUP_DB['name'], obj.name)

def test_get_by_instance_uuid(self, mock_db_get):
        objects.InstanceGroup.get_by_instance_uuid(
                mock.sentinel.ctx, mock.sentinel.instance_uuid)
        mock_db_get.assert_called_once_with(
                mock.sentinel.ctx, mock.sentinel.instance_uuid)

def test_refresh(self, mock_db_get):
        changed_group = copy.deepcopy(_INST_GROUP_DB)
        changed_group['name'] = 'new_name'
        mock_db_get.side_effect = [_INST_GROUP_DB, changed_group]
        obj = objects.InstanceGroup.get_by_uuid(mock.sentinel.ctx,
                                                       _DB_UUID)
        self.assertEqual(_INST_GROUP_DB['name'], obj.name)
        obj.refresh()
        self.assertEqual('new_name', obj.name)
        self.assertEqual(set([]), obj.obj_what_changed())

def test_save_without_hosts(self, mock_db_get, mock_db_update,
                                mock_notify):
        mock_db_get.side_effect = [_INST_GROUP_DB, _INST_GROUP_DB]
        obj = objects.InstanceGroup.get_by_uuid(mock.sentinel.ctx, _DB_UUID)
        obj.hosts = ['fake-host1']
        self.assertRaises(exception.InstanceGroupSaveException,
                          obj.save)
        # make sure that we can save by removing hosts from what is updated
        obj.obj_reset_changes(['hosts'])
        obj.save()
        # since hosts was the only update, there is no actual call
        self.assertFalse(mock_db_update.called)
        self.assertFalse(mock_notify.called)

def test_create(self, mock_db_create, mock_notify):
        obj = objects.InstanceGroup(context=mock.sentinel.ctx)
        obj.uuid = _DB_UUID
        obj.name = _INST_GROUP_DB['name']
        obj.user_id = _INST_GROUP_DB['user_id']
        obj.project_id = _INST_GROUP_DB['project_id']
        obj.members = _INST_GROUP_DB['members']
        obj.policies = _INST_GROUP_DB['policies']
        obj.updated_at = _TS_NOW
        obj.created_at = _TS_NOW
        obj.deleted_at = None
        obj.deleted = False
        obj.create()
        mock_db_create.assert_called_once_with(
            mock.sentinel.ctx,
            {'uuid': _DB_UUID,
             'name': _INST_GROUP_DB['name'],
             'user_id': _INST_GROUP_DB['user_id'],
             'project_id': _INST_GROUP_DB['project_id'],
             'created_at': _TS_NOW,
             'updated_at': _TS_NOW,
             'deleted_at': None,
             'deleted': False,
             },
            members=_INST_GROUP_DB['members'],
            policies=_INST_GROUP_DB['policies'])
        mock_notify.assert_called_once_with(
            mock.sentinel.ctx, "create",
            {'uuid': _DB_UUID,
             'name': _INST_GROUP_DB['name'],
             'user_id': _INST_GROUP_DB['user_id'],
             'project_id': _INST_GROUP_DB['project_id'],
             'created_at': _TS_NOW,
             'updated_at': _TS_NOW,
             'deleted_at': None,
             'deleted': False,
             'members': _INST_GROUP_DB['members'],
             'policies': _INST_GROUP_DB['policies'],
             'server_group_id': _DB_UUID})

        self.assertRaises(exception.ObjectActionError, obj.create)

def test_destroy(self, mock_db_delete, mock_notify):
        obj = objects.InstanceGroup(context=mock.sentinel.ctx)
        obj.uuid = _DB_UUID
        obj.destroy()
        mock_db_delete.assert_called_once_with(mock.sentinel.ctx, _DB_UUID)
        mock_notify.assert_called_once_with(mock.sentinel.ctx, "delete",
                                            {'server_group_id': _DB_UUID})

def test_add_members(self, mock_members_add_db, mock_notify):
        mock_members_add_db.return_value = [mock.sentinel.members]
        members = objects.InstanceGroup.add_members(mock.sentinel.ctx,
                                                    _DB_UUID,
                                                    mock.sentinel.members)
        self.assertEqual([mock.sentinel.members], members)
        mock_members_add_db.assert_called_once_with(
                mock.sentinel.ctx,
                _DB_UUID,
                mock.sentinel.members)
        mock_notify.assert_called_once_with(
                mock.sentinel.ctx, "addmember",
                {'instance_uuids': mock.sentinel.members,
                 'server_group_id': _DB_UUID})

def test_count_members_by_user(self, mock_get_db, mock_il_get):
        mock_il_get.return_value = [mock.ANY]
        obj = objects.InstanceGroup.get_by_uuid(mock.sentinel.ctx, _DB_UUID)
        expected_filters = {
            'uuid': ['instance_id1', 'instance_id2'],
            'user_id': 'fake_user',
            'deleted': False
        }
        self.assertEqual(1, obj.count_members_by_user('fake_user'))
        mock_il_get.assert_called_once_with(mock.sentinel.ctx,
                                            filters=expected_filters)