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])
python类NO_CONTENT的实例源码
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])
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)
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='[]')
def webhook():
ds_webhook.webhook_listener()
return ("", httplib.NO_CONTENT) # no content
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)
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_localization.py 文件源码
项目：zendesk-help-center-localization
作者: mykola-mokhnach
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
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_localization.py 文件源码
项目：zendesk-help-center-localization
作者: mykola-mokhnach
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
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)
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, {}, '')
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)
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)
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)
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)
