Python httplib 模块,NO_CONTENT 实例源码

我们从Python开源项目中,提取了以下14个代码示例,用于说明如何使用httplib.NO_CONTENT

项目:containerregistry    作者:google    | 项目源码 | 文件源码
def _patch_upload(
      self,
      image,
      digest
  ):
    mounted, location = self._start_upload(digest, self._mount)

    if mounted:
      logging.info('Layer %s mounted.', digest)
      return

    location = self._get_absolute_url(location)

    resp, unused_content = self._transport.Request(
        location, method='PATCH', body=self._get_blob(image, digest),
        content_type='application/octet-stream',
        accepted_codes=[httplib.NO_CONTENT, httplib.ACCEPTED, httplib.CREATED])

    location = self._add_digest(resp['location'], digest)
    location = self._get_absolute_url(location)
    self._transport.Request(
        location, method='PUT', body=None,
        accepted_codes=[httplib.CREATED])
项目:containerregistry    作者:google    | 项目源码 | 文件源码
def _patch_upload(
      self,
      image,
      digest
  ):
    mounted, location = self._start_upload(digest, self._mount)

    if mounted:
      logging.info('Layer %s mounted.', digest)
      return

    location = self._get_absolute_url(location)

    resp, unused_content = self._transport.Request(
        location, method='PATCH', body=image.blob(digest),
        content_type='application/octet-stream',
        accepted_codes=[httplib.NO_CONTENT, httplib.ACCEPTED, httplib.CREATED])

    location = self._add_digest(resp['location'], digest)
    location = self._get_absolute_url(location)
    self._transport.Request(
        location, method='PUT', body=None,
        accepted_codes=[httplib.CREATED])
项目:pykit    作者:baishancloud    | 项目源码 | 文件源码
def _handle_server_response(self, response):

        if response.status in (httplib.OK, httplib.CREATED,
                               httplib.NO_CONTENT):
            return response

        logger.debug('invalid response status:{st} body:{body}'.format(
                     st=response.status, body=response.data))

        EtcdError.handle(response)
项目:containerregistry    作者:google    | 项目源码 | 文件源码
def _put_images(self):
    """Close the session by putting to the .../images endpoint."""
    docker_http.Request(
        self._transport,
        '{scheme}://{registry}/v1/repositories/{repository}/images'
        .format(scheme=docker_http.Scheme(self._name.registry),
                registry=self._name.registry,
                repository=self._name.repository),
        self._basic_creds,
        accepted_codes=[httplib.NO_CONTENT],
        body='[]')
项目:esignature-recipes-python    作者:docusign    | 项目源码 | 文件源码
def webhook():
    ds_webhook.webhook_listener()
    return ("", httplib.NO_CONTENT) # no content
项目:orangecloud-client    作者:antechrestos    | 项目源码 | 文件源码
def test_delete(self):
        self.client.delete.return_value = mock_api_response('/files/file-id',
                                                            httplib.NO_CONTENT,
                                                            None)
        self.files.delete('file-id')
        self.client.delete.assert_called_with(self.client.delete.return_value.url)
项目:orangecloud-client    作者:antechrestos    | 项目源码 | 文件源码
def test_delete(self):
        self.client.delete.return_value = mock_api_response('/folders/folder-id',
                                                            httplib.NO_CONTENT,
                                                            None)
        self.folders.delete('folder-id')
        self.client.delete.assert_called_with(self.client.delete.return_value.url)
项目:zendesk-help-center-localization    作者:mykola-mokhnach    | 项目源码 | 文件源码
def _delete(self, endpoint):
        response = requests.request('DELETE',
                                    url='{}/{}'.format(self._api_root, endpoint),
                                    auth=(self._login + '/token', self._token))
        if response.status_code != httplib.NO_CONTENT:
            raise APIError(response.text, response.status_code)
项目:zendesk-help-center-localization    作者:mykola-mokhnach    | 项目源码 | 文件源码
def _upload_file(self, endpoint, path, form_data=None):
        multipart_data = {}
        if form_data is not None:
            for key, value in form_data.iteritems():
                multipart_data[key] = (None, value)
        with open(path, 'rb') as src_file:
            multipart_data['file'] = src_file
            response = requests.request('POST',
                                        url='{}/{}'.format(self._api_root, endpoint),
                                        headers={'Accept': 'application/json'},
                                        auth=(self._login + '/token', self._token),
                                        files=multipart_data)
        if response.status_code in (httplib.NO_CONTENT, httplib.CREATED):
            return response.json()
        raise APIError(response.text, response.status_code)
项目:Deploy_XXNET_Server    作者:jzp820927    | 项目源码 | 文件源码
def _handle_delete(gcs_stub, filename):
  """Handle DELETE object."""
  if gcs_stub.delete_object(filename):
    return _FakeUrlFetchResult(httplib.NO_CONTENT, {}, '')
  else:
    return _FakeUrlFetchResult(httplib.NOT_FOUND, {}, '')
项目:os-xenapi    作者:openstack    | 项目源码 | 文件源码
def test_upload_tarball_by_url_http_v2(self):
        fake_conn = mock.Mock()
        mock_HTTPConn = self.mock_patch_object(
            self.glance, '_create_connection', fake_conn)
        mock_validate_image = self.mock_patch_object(
            self.glance, 'validate_image_status_before_upload_v2')
        mock_create_tarball = self.mock_patch_object(
            self.glance.utils, 'create_tarball')
        mock_check_resp_status = self.mock_patch_object(
            self.glance, 'check_resp_status_and_retry')
        mock_update_image_meta = self.mock_patch_object(
            self.glance, '_update_image_meta_v2')
        self.glance._create_connection().getresponse = mock.Mock()
        self.glance._create_connection().getresponse().status = \
            httplib.NO_CONTENT
        fake_extra_headers = {}
        fake_properties = {}
        fake_endpoint = 'http://fake_netloc/fake_path'
        fake_image_id = 'fake_image_id'
        expected_url = "%(glance_endpoint)s/v2/images/%(image_id)s/file" % {
            'glance_endpoint': fake_endpoint,
            'image_id': fake_image_id}
        expected_wsgi_path = '/fake_path/v2/images/%s' % fake_image_id
        expect_url_parts = urlparse(expected_url)

        expected_mgt_url = "%(glance_endpoint)s/v2/images/%(image_id)s" % {
            'glance_endpoint': fake_endpoint,
            'image_id': fake_image_id}
        fake_mgt_parts = urlparse(expected_mgt_url)
        fake_mgt_path = fake_mgt_parts[2]

        self.glance._upload_tarball_by_url_v2(
            'fake_staging_path', fake_image_id, fake_endpoint,
            fake_extra_headers, fake_properties)

        mock_HTTPConn.assert_called_with(expect_url_parts[0],
                                         expect_url_parts[1])
        mock_validate_image.assert_called_with(fake_conn,
                                               expected_url,
                                               fake_extra_headers,
                                               expected_wsgi_path)
        mock_update_image_meta.assert_called_with(fake_conn,
                                                  fake_extra_headers,
                                                  fake_properties,
                                                  fake_mgt_path)

        self.assertTrue(mock_create_tarball.called)
        self.assertTrue(
            mock_HTTPConn.return_value.getresponse.called)
        self.assertFalse(mock_check_resp_status.called)
项目:os-xenapi    作者:openstack    | 项目源码 | 文件源码
def test_upload_tarball_by_url_https_v2(self):
        fake_conn = mock.Mock()
        mock_HTTPSConn = self.mock_patch_object(
            self.glance, '_create_connection', fake_conn)
        mock_validate_image = self.mock_patch_object(
            self.glance, 'validate_image_status_before_upload_v2')
        mock_create_tarball = self.mock_patch_object(
            self.glance.utils, 'create_tarball')
        mock_check_resp_status = self.mock_patch_object(
            self.glance, 'check_resp_status_and_retry')
        mock_update_image_meta = self.mock_patch_object(
            self.glance, '_update_image_meta_v2')
        self.glance._create_connection().getresponse = mock.Mock()
        self.glance._create_connection().getresponse().status = \
            httplib.NO_CONTENT
        fake_extra_headers = {}
        fake_properties = {}
        fake_endpoint = 'https://fake_netloc/fake_path'
        fake_image_id = 'fake_image_id'
        expected_url = "%(glance_endpoint)s/v2/images/%(image_id)s/file" % {
            'glance_endpoint': fake_endpoint,
            'image_id': fake_image_id}
        expect_url_parts = urlparse(expected_url)
        expected_wsgi_path = '/fake_path/v2/images/%s' % fake_image_id

        expected_mgt_url = "%(glance_endpoint)s/v2/images/%(image_id)s" % {
            'glance_endpoint': fake_endpoint,
            'image_id': fake_image_id}
        fake_mgt_parts = urlparse(expected_mgt_url)
        fake_mgt_path = fake_mgt_parts[2]

        self.glance._upload_tarball_by_url_v2(
            'fake_staging_path', fake_image_id, fake_endpoint,
            fake_extra_headers, fake_properties)

        mock_update_image_meta.assert_called_with(fake_conn,
                                                  fake_extra_headers,
                                                  fake_properties,
                                                  fake_mgt_path)
        mock_HTTPSConn.assert_called_with(expect_url_parts[0],
                                          expect_url_parts[1])
        mock_validate_image.assert_called_with(fake_conn,
                                               expected_url,
                                               fake_extra_headers,
                                               expected_wsgi_path)
        self.assertTrue(mock_create_tarball.called)
        self.assertTrue(
            mock_HTTPSConn.return_value.getresponse.called)
        self.assertFalse(mock_check_resp_status.called)
项目:os-xenapi    作者:openstack    | 项目源码 | 文件源码
def test_upload_tarball_by_url_v2_with_api_endpoint(self):
        fake_conn = mock.Mock()
        mock_Conn = self.mock_patch_object(
            self.glance, '_create_connection', fake_conn)
        mock_validate_image = self.mock_patch_object(
            self.glance, 'validate_image_status_before_upload_v2')
        mock_create_tarball = self.mock_patch_object(
            self.glance.utils, 'create_tarball')
        mock_check_resp_status = self.mock_patch_object(
            self.glance, 'check_resp_status_and_retry')
        mock_update_image_meta = self.mock_patch_object(
            self.glance, '_update_image_meta_v2')
        self.glance._create_connection().getresponse = mock.Mock()
        self.glance._create_connection().getresponse().status = \
            httplib.NO_CONTENT
        fake_extra_headers = {}
        fake_properties = {}
        fake_endpoint = 'https://fake_netloc:fake_port'
        fake_image_id = 'fake_image_id'
        expected_url = "%(glance_endpoint)s/v2/images/%(image_id)s/file" % {
            'glance_endpoint': fake_endpoint,
            'image_id': fake_image_id}
        expect_url_parts = urlparse(expected_url)
        expected_api_path = '/v2/images/%s' % fake_image_id

        expected_mgt_url = "%(glance_endpoint)s/v2/images/%(image_id)s" % {
            'glance_endpoint': fake_endpoint,
            'image_id': fake_image_id}
        fake_mgt_parts = urlparse(expected_mgt_url)
        fake_mgt_path = fake_mgt_parts[2]

        self.glance._upload_tarball_by_url_v2(
            'fake_staging_path', fake_image_id, fake_endpoint,
            fake_extra_headers, fake_properties)

        mock_update_image_meta.assert_called_with(fake_conn,
                                                  fake_extra_headers,
                                                  fake_properties,
                                                  fake_mgt_path)
        mock_Conn.assert_called_with(expect_url_parts[0], expect_url_parts[1])
        mock_validate_image.assert_called_with(fake_conn,
                                               expected_url,
                                               fake_extra_headers,
                                               expected_api_path)
        self.assertTrue(mock_create_tarball.called)
        self.assertTrue(
            mock_Conn.return_value.getresponse.called)
        self.assertFalse(mock_check_resp_status.called)
项目:os-xenapi    作者:openstack    | 项目源码 | 文件源码
def test_upload_tarball_by_url_v2_with_wsgi_endpoint(self):
        fake_conn = mock.Mock()
        mock_Conn = self.mock_patch_object(
            self.glance, '_create_connection', fake_conn)
        mock_validate_image = self.mock_patch_object(
            self.glance, 'validate_image_status_before_upload_v2')
        mock_create_tarball = self.mock_patch_object(
            self.glance.utils, 'create_tarball')
        mock_check_resp_status = self.mock_patch_object(
            self.glance, 'check_resp_status_and_retry')
        mock_update_image_meta = self.mock_patch_object(
            self.glance, '_update_image_meta_v2')
        self.glance._create_connection().getresponse = mock.Mock()
        self.glance._create_connection().getresponse().status = \
            httplib.NO_CONTENT
        fake_extra_headers = {}
        fake_properties = {}
        fake_endpoint = 'https://fake_netloc/fake_path'
        fake_image_id = 'fake_image_id'
        expected_url = "%(glance_endpoint)s/v2/images/%(image_id)s/file" % {
            'glance_endpoint': fake_endpoint,
            'image_id': fake_image_id}
        expect_url_parts = urlparse(expected_url)
        expected_wsgi_path = '/fake_path/v2/images/%s' % fake_image_id

        expected_mgt_url = "%(glance_endpoint)s/v2/images/%(image_id)s" % {
            'glance_endpoint': fake_endpoint,
            'image_id': fake_image_id}
        fake_mgt_parts = urlparse(expected_mgt_url)
        fake_mgt_path = fake_mgt_parts[2]

        self.glance._upload_tarball_by_url_v2(
            'fake_staging_path', fake_image_id, fake_endpoint,
            fake_extra_headers, fake_properties)

        mock_update_image_meta.assert_called_with(fake_conn,
                                                  fake_extra_headers,
                                                  fake_properties,
                                                  fake_mgt_path)
        mock_Conn.assert_called_with(expect_url_parts[0], expect_url_parts[1])
        mock_validate_image.assert_called_with(fake_conn,
                                               expected_url,
                                               fake_extra_headers,
                                               expected_wsgi_path)
        self.assertTrue(mock_create_tarball.called)
        self.assertTrue(
            mock_Conn.return_value.getresponse.called)
        self.assertFalse(mock_check_resp_status.called)