python类sentinel()的实例源码

def test_force_close(self):
        tr = self.create_transport()
        tr._buffer.extend(b'1')
        self.loop.add_reader(7, mock.sentinel)
        self.loop.add_writer(7, mock.sentinel)
        tr._force_close(None)

        self.assertTrue(tr._closing)
        self.assertEqual(tr._buffer, list_to_buffer())
        self.assertFalse(self.loop.readers)
        self.assertFalse(self.loop.writers)

        # second close should not remove reader
        tr._force_close(None)
        self.assertFalse(self.loop.readers)
        self.assertEqual(1, self.loop.remove_reader_count[7])

def test_force_close(self):
        tr = self.create_transport()
        tr._buffer.extend(b'1')
        self.loop.add_reader(7, mock.sentinel)
        self.loop.add_writer(7, mock.sentinel)
        tr._force_close(None)

        self.assertTrue(tr.is_closing())
        self.assertEqual(tr._buffer, list_to_buffer())
        self.assertFalse(self.loop.readers)
        self.assertFalse(self.loop.writers)

        # second close should not remove reader
        tr._force_close(None)
        self.assertFalse(self.loop.readers)
        self.assertEqual(1, self.loop.remove_reader_count[7])

def test_force_close(self):
        tr = _SelectorTransport(self.loop, self.sock, self.protocol, None)
        tr._buffer.extend(b'1')
        self.loop.add_reader(7, mock.sentinel)
        self.loop.add_writer(7, mock.sentinel)
        tr._force_close(None)

        self.assertTrue(tr._closing)
        self.assertEqual(tr._buffer, list_to_buffer())
        self.assertFalse(self.loop.readers)
        self.assertFalse(self.loop.writers)

        # second close should not remove reader
        tr._force_close(None)
        self.assertFalse(self.loop.readers)
        self.assertEqual(1, self.loop.remove_reader_count[7])

def environ(env_name, env_value):
    """Mock environment variable value.

    Example::

        @mock.environ('DJANGO_SETTINGS_MODULE', 'proj.settings')
        def test_other_settings(self):
            ...

    """
    sentinel = object()
    prev_val = os.environ.get(env_name, sentinel)
    os.environ[env_name] = env_value
    try:
        yield
    finally:
        if prev_val is sentinel:
            os.environ.pop(env_name, None)
        else:
            os.environ[env_name] = prev_val

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')