• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python shared.request_with_middleware函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中senlin.tests.unit.api.shared.request_with_middleware函数的典型用法代码示例。如果您正苦于以下问题:Python request_with_middleware函数的具体用法?Python request_with_middleware怎么用?Python request_with_middleware使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了request_with_middleware函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_receiver_create_unapplicable_action

    def test_receiver_create_unapplicable_action(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'create', True)
        action = 'BAD'
        body = {
            'receiver': {
                'name': 'test_receiver',
                'type': 'webhook',
                'cluster_id': 'FAKE_CLUSTER',
                'action': action,
            }
        }
        req = self._post('/receivers', jsonutils.dumps(body))

        msg = 'Action BAD is not applicable clusters.'
        error = senlin_exc.BadRequest(msg=msg)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call',
                                     side_effect=error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.create,
                                              req, body=body)

        expected = body['receiver']
        type_name = expected.pop('type')
        expected['type_name'] = type_name
        expected['actor'] = None
        expected['params'] = None
        mock_call.assert_called_once_with(req.context,
                                          ('receiver_create', expected))
        self.assertEqual(400, resp.json['code'])
        self.assertEqual('BadRequest', resp.json['error']['type'])
开发者ID:jonnary,项目名称:senlin,代码行数:31,代码来源:test_receivers.py


示例2: test_profile_create_with_spec_validation_failed

    def test_profile_create_with_spec_validation_failed(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'create', True)
        body = {
            'profile': {
                'name': 'test_profile',
                'spec': {
                    'type': 'test_profile_type',
                    'version': '1.0',
                    'properties': {'param': 'value'},
                },
                'metadata': {},
            }
        }
        req = self._post('/profiles', jsonutils.dumps(body))

        msg = 'Spec validation error (param): value'
        error = senlin_exc.SpecValidationFailed(message=msg)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call',
                                     side_effect=error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.create,
                                              req, body=body)
        mock_call.assert_called_once_with(req.context,
                                          ('profile_create', body['profile']))
        self.assertEqual(400, resp.json['code'])
        self.assertEqual('SpecValidationFailed', resp.json['error']['type'])
开发者ID:zhaihaifei,项目名称:senlin,代码行数:27,代码来源:test_profiles.py


示例3: test_receiver_create_with_bad_type

    def test_receiver_create_with_bad_type(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'create', True)
        r_type = 'unsupported'
        body = {
            'receiver': {
                'name': 'test_receiver',
                'type': r_type,
                'cluster_id': 'FAKE_CLUSTER',
                'action': 'test_action',
            }
        }
        req = self._post('/receivers', jsonutils.dumps(body))

        msg = _('receiver obj_type (%s) is unsupported.') % r_type
        error = senlin_exc.BadRequest(msg=msg)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call',
                                     side_effect=error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.create,
                                              req, body=body)

        expected_args = body['receiver']
        type_name = expected_args.pop('type')
        expected_args['type_name'] = type_name
        expected_args['actor'] = None
        expected_args['params'] = None
        mock_call.assert_called_once_with(req.context,
                                          ('receiver_create', expected_args))
        self.assertEqual(400, resp.json['code'])
        self.assertEqual('BadRequest', resp.json['error']['type'])
开发者ID:jonnary,项目名称:senlin,代码行数:31,代码来源:test_receivers.py


示例4: test_receiver_create_with_cluster_id_notfound

    def test_receiver_create_with_cluster_id_notfound(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'create', True)
        cluster_id = 'FAKE_ID'
        body = {
            'receiver': {
                'name': 'test_receiver',
                'type': 'webhook',
                'cluster_id': cluster_id,
                'action': 'test_action',
            }
        }
        req = self._post('/receivers', jsonutils.dumps(body))

        error = senlin_exc.ClusterNotFound(cluster=cluster_id)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call',
                                     side_effect=error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.create,
                                              req, body=body)

        expected_args = body['receiver']
        type_name = expected_args.pop('type')
        expected_args['type_name'] = type_name
        expected_args['actor'] = None
        expected_args['params'] = None
        mock_call.assert_called_once_with(req.context,
                                          ('receiver_create', expected_args))
        self.assertEqual(404, resp.json['code'])
        self.assertEqual('ClusterNotFound', resp.json['error']['type'])
开发者ID:jonnary,项目名称:senlin,代码行数:30,代码来源:test_receivers.py


示例5: test_node_create_with_bad_cluster

    def test_node_create_with_bad_cluster(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'create', True)
        body = {
            'node': {
                'name': 'test_node',
                'profile_id': 'xxxx-yyyy-zzzz',
                'cluster_id': 'non-existent-cluster',
                'role': None,
                'metadata': {},
            }
        }
        req = self._post('/nodes', jsonutils.dumps(body))

        error = senlin_exc.ClusterNotFound(cluster='non-existent-cluster')
        mock_call = self.patchobject(rpc_client.EngineClient, 'call',
                                     side_effect=error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.create,
                                              req, body=body)

        mock_call.assert_called_once_with(req.context,
                                          ('node_create', body['node']))
        self.assertEqual(404, resp.json['code'])
        self.assertEqual('ClusterNotFound', resp.json['error']['type'])
开发者ID:onesafe,项目名称:senlin,代码行数:25,代码来源:test_nodes.py


示例6: test_node_update_invalid_profile

    def test_node_update_invalid_profile(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'update', True)
        nid = 'aaaa-bbbb-cccc'
        body = {
            'node': {
                'name': 'test_node',
                'profile_id': 'profile-not-exist',
                'role': None,
                'metadata': {},
            }
        }

        req = self._patch('/nodes/%(node_id)s' % {'node_id': nid},
                          jsonutils.dumps(body))

        error = senlin_exc.ProfileNotFound(profile=nid)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call',
                                     side_effect=shared.to_remote_error(error))

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.update,
                                              req, node_id=nid, body=body)
        mock_call.assert_called_with(
            req.context,
            ('node_update', {
                'identity': nid,
                'name': 'test_node',
                'profile_id': 'profile-not-exist',
                'role': None,
                'metadata': {},
            })
        )
        self.assertEqual(404, resp.json['code'])
        self.assertEqual('ProfileNotFound', resp.json['error']['type'])
开发者ID:onesafe,项目名称:senlin,代码行数:34,代码来源:test_nodes.py


示例7: test_profile_create_with_bad_type

    def test_profile_create_with_bad_type(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'create', True)
        type_name = 'unknown_type'
        body = {
            'profile': {
                'name': 'test_profile',
                'spec': {
                    'type': type_name,
                    'version': '1.0',
                    'properties': {'param': 'value'},
                },
                'metadata': {},
            }
        }
        req = self._post('/profiles', jsonutils.dumps(body))

        error = senlin_exc.ProfileTypeNotFound(profile_type=type_name)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call',
                                     side_effect=error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.create,
                                              req, body=body)

        mock_call.assert_called_once_with(req.context,
                                          ('profile_create', body['profile']))
        self.assertEqual(404, resp.json['code'])
        self.assertEqual('ProfileTypeNotFound', resp.json['error']['type'])
开发者ID:zhaihaifei,项目名称:senlin,代码行数:28,代码来源:test_profiles.py


示例8: test_node_index_denied_policy

    def test_node_index_denied_policy(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'index', False)
        req = self._get('/nodes')

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.index, req)
        self.assertEqual(403, resp.status_int)
        self.assertIn('403 Forbidden', six.text_type(resp))
开发者ID:onesafe,项目名称:senlin,代码行数:8,代码来源:test_nodes.py


示例9: test_profile_type_get_err_denied_policy

    def test_profile_type_get_err_denied_policy(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'get', False)
        type_name = 'BogusProfileType'
        req = self._get('/profile_types/%(type)s' % {'type': type_name})

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.get,
                                              req, type_name=type_name)
        self.assertEqual(403, resp.status_int)
        self.assertIn('403 Forbidden', six.text_type(resp))
开发者ID:jonnary,项目名称:senlin,代码行数:10,代码来源:test_profile_types.py


示例10: test_node_delete_err_denied_policy

    def test_node_delete_err_denied_policy(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'delete', False)
        nid = 'aaaa-bbbb-cccc'
        req = self._delete('/nodes/%(node_id)s' % {'node_id': nid})

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.delete,
                                              req, node_id=nid)

        self.assertEqual(403, resp.status_int)
        self.assertIn('403 Forbidden', six.text_type(resp))
开发者ID:onesafe,项目名称:senlin,代码行数:11,代码来源:test_nodes.py


示例11: test_cluster_policy_index_denied_policy

    def test_cluster_policy_index_denied_policy(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'index', False)
        cid = 'FAKE_CLUSTER'
        req = self._get('/cluster_policy/%s' % cid)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.index,
                                              req, cluster_id=cid)

        self.assertEqual(403, resp.status_int)
        self.assertIn('403 Forbidden', six.text_type(resp))
开发者ID:onesafe,项目名称:senlin,代码行数:11,代码来源:test_cluster_policies.py


示例12: test_event_get_denied_policy

    def test_event_get_denied_policy(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'get', False)
        event_id = 'non-existent-event'
        req = self._get('/events/%(event_id)s' % {'event_id': event_id})

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.get,
                                              req, event_id=event_id)

        self.assertEqual(403, resp.status_int)
        self.assertIn('403 Forbidden', six.text_type(resp))
开发者ID:onesafe,项目名称:senlin,代码行数:11,代码来源:test_events.py


示例13: test_node_index_cluster_not_found

    def test_node_index_cluster_not_found(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'index', True)
        cluster_id = 'non-existent'
        req = self._get('/nodes', {'cluster_id': cluster_id})

        error = senlin_exc.ClusterNotFound(cluster=cluster_id)
        self.patchobject(rpc_client.EngineClient, 'call',
                         side_effect=shared.to_remote_error(error))

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.index, req)

        self.assertEqual(404, resp.json['code'])
        self.assertEqual('ClusterNotFound', resp.json['error']['type'])
开发者ID:onesafe,项目名称:senlin,代码行数:14,代码来源:test_nodes.py


示例14: test_action_get_denied_policy

    def test_action_get_denied_policy(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'get', False)
        cid = 'FAKE_CLUSTER'
        pid = 'FAKE_POLICY'
        req = self._get('/cluster_policies/%(cid)s/%(pid)s'
                        '' % {'cid': cid, 'pid': pid})

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.get,
                                              req, cluster_id=cid,
                                              policy_id=pid)

        self.assertEqual(403, resp.status_int)
        self.assertIn('403 Forbidden', six.text_type(resp))
开发者ID:onesafe,项目名称:senlin,代码行数:14,代码来源:test_cluster_policies.py


示例15: test_profile_get_not_found

    def test_profile_get_not_found(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'get', True)
        pid = 'non-existent-profile'
        req = self._get('/profiles/%(profile_id)s' % {'profile_id': pid})

        error = senlin_exc.ProfileNotFound(profile=pid)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call')
        mock_call.side_effect = shared.to_remote_error(error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.get,
                                              req, profile_id=pid)

        self.assertEqual(404, resp.json['code'])
        self.assertEqual('ProfileNotFound', resp.json['error']['type'])
开发者ID:zhaihaifei,项目名称:senlin,代码行数:15,代码来源:test_profiles.py


示例16: test_policy_update_denied_policy

    def test_policy_update_denied_policy(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'update', False)
        pid = 'aaaa-bbbb-cccc'
        body = {
            'policy': {'name': 'test_policy', 'spec': {'param5': 'value5'}},
        }
        req = self._put('/policies/%(policy_id)s' % {'policy_id': pid},
                        jsonutils.dumps(body))

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.update,
                                              req, policy_id=pid, body=body)

        self.assertEqual(403, resp.status_int)
        self.assertIn('403 Forbidden', six.text_type(resp))
开发者ID:jonnary,项目名称:senlin,代码行数:15,代码来源:test_policies.py


示例17: test_node_delete_not_found

    def test_node_delete_not_found(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'delete', True)
        nid = 'aaaa-bbbb-cccc'
        req = self._delete('/nodes/%(node_id)s' % {'node_id': nid})

        error = senlin_exc.NodeNotFound(node=nid)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call')
        mock_call.side_effect = shared.to_remote_error(error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.delete,
                                              req, node_id=nid)

        self.assertEqual(404, resp.json['code'])
        self.assertEqual('NodeNotFound', resp.json['error']['type'])
开发者ID:onesafe,项目名称:senlin,代码行数:15,代码来源:test_nodes.py


示例18: test_event_get_not_found

    def test_event_get_not_found(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'get', True)
        event_id = 'non-existent-event'
        req = self._get('/events/%(event_id)s' % {'event_id': event_id})

        error = senlin_exc.EventNotFound(event=event_id)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call')
        mock_call.side_effect = shared.to_remote_error(error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.get,
                                              req, event_id=event_id)

        self.assertEqual(404, resp.json['code'])
        self.assertEqual('EventNotFound', resp.json['error']['type'])
开发者ID:onesafe,项目名称:senlin,代码行数:15,代码来源:test_events.py


示例19: test_profile_delete_resource_in_use

    def test_profile_delete_resource_in_use(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'delete', True)
        pid = 'aaaa-bbbb-cccc'
        req = self._delete('/profiles/%(profile_id)s' % {'profile_id': pid})

        error = senlin_exc.ResourceInUse(resource_type='profile',
                                         resource_id=pid)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call')
        mock_call.side_effect = shared.to_remote_error(error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.delete,
                                              req, profile_id=pid)

        self.assertEqual(409, resp.json['code'])
        self.assertEqual('ResourceInUse', resp.json['error']['type'])
开发者ID:zhaihaifei,项目名称:senlin,代码行数:16,代码来源:test_profiles.py


示例20: test_node_action_check_node_not_found

    def test_node_action_check_node_not_found(self, mock_enforce):
        self._mock_enforce_setup(mock_enforce, 'action', True)
        node_id = 'unknown-node'
        body = {'check': {}}
        req = self._post('/nodes/%(node_id)s/actions' % {'node_id': node_id},
                         jsonutils.dumps(body))

        error = senlin_exc.NodeNotFound(node=node_id)
        mock_call = self.patchobject(rpc_client.EngineClient, 'call')
        mock_call.side_effect = shared.to_remote_error(error)

        resp = shared.request_with_middleware(fault.FaultWrapper,
                                              self.controller.action,
                                              req, node_id=node_id, body=body)

        self.assertEqual(404, resp.json['code'])
        self.assertEqual('NodeNotFound', resp.json['error']['type'])
开发者ID:onesafe,项目名称:senlin,代码行数:17,代码来源:test_nodes.py



注:本文中的senlin.tests.unit.api.shared.request_with_middleware函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python shared.request_with_middleware函数代码示例发布时间:2022-05-27
下一篇:
Python utils.create_a_profile函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap