python类sentinel()的实例源码

def test_http_methods(self, mock_create_http_method):
        s = mock.sentinel
        schema = ResourceSchema({}, self.mock_handle_http)
        for method in ('delete', 'get', 'post', 'put'):
            mock_create_http_method.reset_mock()
            fn = getattr(schema, 'http_{}'.format(method))
            result = fn(s.logic, s.request, s.response, s.params, s.required,
                        title=s.title)
            self.assertEqual(result, s.result)
            self.assertEqual(
                mock_create_http_method.call_args_list,
                [mock.call(schema, s.logic, method.upper(), request=s.request,
                           response=s.response, params=s.params,
                           required=s.required, title=s.title, before=None,
                           after=None, allowed_exceptions=None,
                           omit_args=None)])

def test_get_annotation(self):
        """Tests that get_annotation works for badly decorated functions."""
        def decorator(fn):
            def wrapper():
                fn()
            return wrapper
        mock_logic = mock.Mock()
        mock_logic._schema_annotation = mock.sentinel.logic_annotation
        wrapper = decorator(mock_logic)
        self.assertEqual(ResourceSchemaAnnotation.get_annotation(mock_logic),
                         mock.sentinel.logic_annotation)
        self.assertEqual(ResourceSchemaAnnotation.get_annotation(wrapper),
                         mock.sentinel.logic_annotation)
        wrapper._schema_annotation = mock.sentinel.wrapper_annotation
        self.assertEqual(ResourceSchemaAnnotation.get_annotation(wrapper),
                         mock.sentinel.wrapper_annotation)
        delattr(wrapper, '_schema_annotation')
        delattr(mock_logic, '_schema_annotation')
        self.assertIsNone(ResourceSchemaAnnotation.get_annotation(wrapper))

def _get_all_hosts(self):
        host_values = [
            ('host1', 'node1', {'instances': {
                'member1': mock.sentinel,
                'instance13': mock.sentinel
            }}),
            ('host2', 'node2', {'instances': {
                'member2': mock.sentinel,
                'member3': mock.sentinel,
                'member4': mock.sentinel,
                'member5': mock.sentinel,
                'instance14': mock.sentinel
            }}),
            ('host3', 'node3', {'instances': {
                'instance15': mock.sentinel
            }}),
            ('host4', 'node4', {'instances': {
                'member6': mock.sentinel,
                'member7': mock.sentinel,
                'instance16': mock.sentinel
            }})]
        return [fakes.FakeHostState(host, node, values)
                for host, node, values in host_values]

def _test_create_quota_usage_if_missing_created(self, per_project_quotas):
        # Tests that the QuotaUsage is created.
        user_usages = {}
        if per_project_quotas:
            resource = sqlalchemy_api.PER_PROJECT_QUOTAS[0]
        else:
            resource = 'fake-resource'
        project_id = 'fake-project'
        user_id = 'fake_user'
        session = mock.sentinel
        quota_usage = mock.sentinel
        with mock.patch.object(sqlalchemy_api, '_quota_usage_create',
                               return_value=quota_usage) as quc:
            self.assertTrue(sqlalchemy_api._create_quota_usage_if_missing(
                                user_usages, resource, None,
                                project_id, user_id, session))
        self.assertEqual(quota_usage, user_usages[resource])
        # Now test if the QuotaUsage was created with a user_id or not.
        if per_project_quotas:
            quc.assert_called_once_with(
                project_id, None, resource, 0, 0, None, session)
        else:
            quc.assert_called_once_with(
                project_id, user_id, resource, 0, 0, None, session)

def test_compute_node_get_model(self, mock_model_query):

        class FakeFiltered(object):
            def first(self):
                return mock.sentinel.first

        fake_filtered_cn = FakeFiltered()

        class FakeModelQuery(object):
            def filter_by(self, id):
                return fake_filtered_cn

        mock_model_query.return_value = FakeModelQuery()
        result = sqlalchemy_api.compute_node_get_model(self.ctxt,
                                                       self.item["id"])
        self.assertEqual(result, mock.sentinel.first)
        mock_model_query.assert_called_once_with(self.ctxt, models.ComputeNode)

def test_get_domain_info_with_more_return(self, mock_get_domain):
        instance = objects.Instance(**self.test_instance)
        dom_mock = mock.MagicMock()
        dom_mock.info.return_value = [
            1, 2048, 737, 8, 12345, 888888
        ]
        dom_mock.ID.return_value = mock.sentinel.instance_id
        mock_get_domain.return_value = dom_mock
        drvr = libvirt_driver.LibvirtDriver(fake.FakeVirtAPI(), False)
        info = drvr.get_info(instance)
        self.assertEqual(1, info.state)
        self.assertEqual(2048, info.max_mem_kb)
        self.assertEqual(737, info.mem_kb)
        self.assertEqual(8, info.num_cpu)
        self.assertEqual(12345, info.cpu_time_ns)
        self.assertEqual(mock.sentinel.instance_id, info.id)
        dom_mock.info.assert_called_once_with()
        dom_mock.ID.assert_called_once_with()
        mock_get_domain.assert_called_once_with(instance)

def test_redirect(self):
        url = "url"
        decorated = requires(B, get=AAA, post=AA)(undecorated_view)
        self.assertEqual(decorated._gate.login_url, settings.LOGIN_URL)
        decorated = requires(B, login_url=url, get=AAA, post=AA)(
            undecorated_view)
        self.assertEqual(decorated._gate.login_url, url)

        with patch('django.contrib.auth.views.redirect_to_login',
                   return_value=sentinel) as do_redirect:
            request = self.mock_get_request()
            redirect = decorated(request)
            do_redirect.assert_called_once_with(request.get_full_path(), url)
        self.assertIs(redirect, sentinel)

def test_redirect(self):
        url = "url"
        decorated = requires(B, get=AAA, post=AA)(undecorated_view)
        self.assertEqual(decorated._gate.login_url, settings.LOGIN_URL)
        decorated = requires(B, login_url=url, get=AAA, post=AA)(
            undecorated_view)
        self.assertEqual(decorated._gate.login_url, url)

        with patch('django.contrib.auth.views.redirect_to_login',
                   return_value=sentinel) as do_redirect:
            request = self.mock_get_request()
            redirect = decorated(request)
            do_redirect.assert_called_once_with(request.get_full_path(), url)
        self.assertIs(redirect, sentinel)

def setUp(self):
        self.mock_handle_http = mock.Mock(autospec=True, spec=handle_http,
                                          return_value=mock.sentinel.result)

def test_create_request_schema(self):
        schema = ResourceSchema({'definitions': mock.sentinel.definitions},
                                self.mock_handle_http)
        request_schema = schema._create_request_schema(params=('a', 'b', 'c'),
                                                       required=('b', 'c'))
        self.assertEqual(request_schema, {
            'additionalProperties': True,
            'definitions': mock.sentinel.definitions,
            'properties': {'a': {'$ref': '#/definitions/a'},
                           'b': {'$ref': '#/definitions/b'},
                           'c': {'$ref': '#/definitions/c'}},
            'required': ('b', 'c'),
            'type': 'object',
        })

def test_init(self):
        s = mock.sentinel
        annotation = ResourceSchemaAnnotation(
            s.logic, 'POST', s.schema, s.request_schema,
            s.response_schema)
        self.assertEqual(annotation.logic, s.logic)
        self.assertEqual(annotation.http_method, 'POST')
        self.assertEqual(annotation.schema, s.schema)
        self.assertEqual(annotation.request_schema, s.request_schema)
        self.assertEqual(annotation.response_schema, s.response_schema)
        self.assertEqual(annotation.title, 'Create')

def test_init(self):
        s = mock.sentinel
        router = Router(s.schema_dir, s.resource_schema_class,
                        s.default_base_handler)
        self.assertEqual(s.schema_dir, router.schema_dir)
        self.assertEqual(s.resource_schema_class, router.resource_schema_class)
        self.assertEqual(s.default_base_handler, router.default_base_handler)

        # ensure default_base_handler is `object` if not passed
        router = Router(s.schema_dir, s.resource_schema_class)
        self.assertEqual(object, router.default_base_handler)

def testSplitVariablesGCD(self, group_variables_mock):
        group = {'a': [([-1], [0])],
                 'b': [([-2], [1])],
                 'c': [([0, 6], [2, 5, 6, 7, 8])],
                 'd': [([1], [3, 4, 5, 6, 7])],
                 'ret': [([3, 8], [9])]}
        group_variables_mock.return_value = group
        dataflow.split_variables(
            mock.sentinel, [0, 1, 2, 3, 4], mock.sentinel, mock.sentinel)

def test_msg_str_is_xml_msg(self):
        self.get_error_msg.return_value = sentinel.err
        res = exceptions.get_api_exception_message('a message', 401, True)
        assert_equal(res, sentinel.err)

def setup(self):
        self.p_Session = patch.object(generic, 'Session')
        Session = self.p_Session.start()
        self.p_requests = patch.object(generic, 'requests')
        rquests = self.p_requests.start()
        Session.return_value = rquests

        self.response = MagicMock()
        self.response.status_code = 200
        self.response.text = sentinel.response_text
        self.request = rquests.request
        self.request.return_value = self.response

        self.client = python_kemptech_api.generic.HttpClient()
        self.client.endpoint = 'ep/'

def test_file_parameter_set(self):
        open_ = mock_open(read_data='myData')
        with patch.object(generic, "open", open_, create=True): # as my_open:
           self.client._do_request('GET','MyCommand',
                                                parameters=sentinel.params,
                                                file='my_filename')
           args = self.request.call_args
           # check positional arguments
           assert_equal(args[0], ('GET', 'ep/MyCommand?'))
           # check kwargs
           kw = args[1]
           assert_equal(kw['params'], sentinel.params)
           assert_in('data', kw)

def test_400_status_code(self):
        self.response.status_code = 400
        res = self.client._do_request('GET','MyCommand')
        assert_equal(res, sentinel.response_text)

def test_with_index(self):
        with patch.object(LoadMaster, 'build_virtual_service') as build_virtual_service:
            with patch.object(models, 'get_data'):
                with patch.object(LoadMaster, '_get'):
                    build_virtual_service.return_value = sentinel.vs
                    res = self.lm.get_virtual_service(index=12)
        assert_equal(res, sentinel.vs)

def test_get_data_ok():
    with patch.object(api_xml, '_get_xml_field') as _get_xml_field:
        _get_xml_field.return_value = {'Data': sentinel.data}
        res = api_xml.get_data('anyxml')
        assert_equal(sentinel.data, res)

# Broken by performance imporovements, need to fix later
#def test_get_data_no_Data_key():
#    with patch.object(api_xml, '_get_xml_field') as _get_xml_field:
#        _get_xml_field.return_value = {'junk': 'anything'}
#        res = api_xml.get_data('anyxml')
#        assert_equal(res, {})

def test_get_success_msg():
    with patch.object(api_xml, '_get_xml_field') as _get_xml_field:
        _get_xml_field.return_value = {'Success': sentinel.data}
        res = api_xml.get_success_msg('anyxml')
        assert_equal("{'Success': sentinel.data}", res)

def test_get_error_msg():
    with patch.object(api_xml, '_get_xml_field') as _get_xml_field:
        _get_xml_field.return_value = {'Error': sentinel.data}
        res = api_xml.get_error_msg('anyxml')
        assert_equal("{'Error': sentinel.data}", res)

def test_get_data_field():
    with patch.object(api_xml, '_get_xml_field') as _get_xml_field:
        _get_xml_field.return_value = {'Data': sentinel.data}
        res = api_xml.get_data_field('any_xml', 'any_field')
        assert_equal({'Data': sentinel.data}, res)

def test_with_index_ok(self):
        with patch.object(VirtualService, 'build_real_server') as build_real_server:
            with patch.object(objects, 'get_data'):
                with patch.object(VirtualService, '_get'):
                    self.vs.index = self
                    build_real_server.return_value = sentinel.rs
                    res =  self.vs.get_real_server('1.1.1.1', 80)
        assert_equal(res, sentinel.rs)

def test_without_index_ok(self):
        with patch.object(VirtualService, 'build_real_server') as build_real_server:
            with patch.object(objects, 'get_data'):
                with patch.object(VirtualService, '_get'):
                    self.vs.index = None
                    build_real_server.return_value = sentinel.rs
                    res =  self.vs.get_real_server('1.1.1.1', 80)
        assert_equal(res, sentinel.rs)

def test_without_index_invalid_port(self):
        with patch.object(VirtualService, 'build_real_server') as build_real_server:
            with patch.object(objects, 'get_data'):
                with patch.object(VirtualService, '_get'):
                    self.vs.index = None
                    build_real_server.return_value = sentinel.rs
                    with assert_raises(ValidationError):
                        self.vs.get_real_server('1.1.1.1.', 'junk')

def testSplitVariablesGCD(self, group_variables_mock):
        group = {'a': [([-1], [0])],
                 'b': [([-2], [1])],
                 'c': [([0, 6], [2, 5, 6, 7, 8])],
                 'd': [([1], [3, 4, 5, 6, 7])],
                 'ret': [([3, 8], [9])]}
        group_variables_mock.return_value = group
        dataflow.split_variables(
            mock.sentinel, [0, 1, 2, 3, 4], mock.sentinel, mock.sentinel)

def test_route_token_required(self):
    request = requestMock(b"/", headers={b'Authorization': [str(mock.sentinel)]})
    returned = self.app.execute_endpoint("token_required_endpoint", request)
    self.assertEqual(returned, self.userinfo)

    self.auth.decode_header_value.assert_called_once_with(str(mock.sentinel))

def test_process(self):
        m = CPUs().process(sentinel, self.results, Mock()).maps[0]
        for i in ('cacheSizeL1', 'cacheSizeL2', 'cacheSizeL3', 'cacheSpeedL2', 'cacheSpeedL3',
                  'clockspeed', 'cores', 'extspeed', 'socket', 'threads'):
            self.assertEquals(getattr(m, i), 1)
        self.assertEquals(m.description, 'Description')
        self.assertEquals(m.id, 'DeviceID')
        self.assertEquals(m.title, 'Name')

def test_commit_raises_exception_on_timeout(self):
        transaction = impl_idl.OvsVsctlTransaction(mock.sentinel,
                                                   mock.Mock(), 1)
        with testtools.ExpectedException(exceptions.TimeoutException):
            transaction.commit()

def test_post_commit_does_not_raise_exception(self):
        with mock.patch.object(impl_idl.OvsVsctlTransaction,
                               "do_post_commit", side_effect=Exception):
            transaction = impl_idl.OvsVsctlTransaction(mock.sentinel,
                                                       mock.Mock(), 0)
            transaction.post_commit(mock.Mock())