Python httplib 模块，ACCEPTED 实例源码

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

def _put_manifest(
      self,
      image,
      use_digest=False
  ):
    """Upload the manifest for this image."""
    if use_digest:
      tag_or_digest = image.digest()
    else:
      tag_or_digest = _tag_or_digest(self._name)

    self._transport.Request(
        '{base_url}/manifests/{tag_or_digest}'.format(
            base_url=self._base_url(),
            tag_or_digest=tag_or_digest),
        method='PUT',
        body=image.manifest(),
        content_type=image.media_type(),
        accepted_codes=[httplib.OK, httplib.CREATED, httplib.ACCEPTED])

def Delete(
    name,
    creds,
    transport
):
  """Delete a tag or digest.

  Args:
    name: a tag or digest to be deleted.
    creds: the creds to use for deletion.
    transport: the transport to use to contact the registry.
  """
  docker_transport = docker_http.Transport(
      name, creds, transport, docker_http.DELETE)

  resp, unused_content = docker_transport.Request(
      '{scheme}://{registry}/v2/{repository}/manifests/{entity}'.format(
          scheme=docker_http.Scheme(name.registry),
          registry=name.registry,
          repository=name.repository,
          entity=_tag_or_digest(name)),
      method='DELETE',
      accepted_codes=[httplib.OK, httplib.ACCEPTED])

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 _start_upload(
      self,
      digest,
      mount=None
  ):
    """POST to begin the upload process with optional cross-repo mount param."""
    if not mount:
      # Do a normal POST to initiate an upload if mount is missing.
      url = '{base_url}/blobs/uploads/'.format(base_url=self._base_url())
      accepted_codes = [httplib.ACCEPTED]
    else:
      # If we have a mount parameter, try to mount the blob from another repo.
      mount_from = '&'.join(
          ['from=' + urllib.quote(repo.repository, '') for repo in self._mount])
      url = '{base_url}/blobs/uploads/?mount={digest}&{mount_from}'.format(
          base_url=self._base_url(),
          digest=digest,
          mount_from=mount_from)
      accepted_codes = [httplib.CREATED, httplib.ACCEPTED]

    resp, unused_content = self._transport.Request(
        url, method='POST', body=None,
        accepted_codes=accepted_codes)
    return resp.status == httplib.CREATED, resp.get('location')  # type: ignore

def Delete(
    name,
    creds,
    transport
):
  """Delete a tag or digest.

  Args:
    name: a tag or digest to be deleted.
    creds: the credentials to use for deletion.
    transport: the transport to use to contact the registry.
  """
  docker_transport = docker_http.Transport(
      name, creds, transport, docker_http.DELETE)

  resp, unused_content = docker_transport.Request(
      '{scheme}://{registry}/v2/{repository}/manifests/{entity}'.format(
          scheme=docker_http.Scheme(name.registry),
          registry=name.registry,
          repository=name.repository,
          entity=_tag_or_digest(name)),
      method='DELETE',
      accepted_codes=[httplib.OK, httplib.ACCEPTED])

def test__raise_for_status(self, mock_resp):
        """
        Verify that an exceptions gets raised for unexpected responses.

        :param mock_resp: mocked httplib Response
        """
        #
        # ok_statuses should not raise
        #
        mock_resp.status = httplib.CREATED
        self.up.resp = mock_resp
        self.up.content = ''
        try:
            self.up._raise_for_status([httplib.OK, httplib.CREATED])
        except Exception as exc:
            self.fail('_raise_for_status unexpectedly threw {}'.format(exc))

        #
        # Anything else should raise.
        #
        mock_resp.status = httplib.ACCEPTED
        self.assertRaises(
            upapi.exceptions.UnexpectedAPIResponse,
            self.up._raise_for_status,
            [httplib.OK, httplib.CREATED])

def _start_upload(
      self,
      digest,
      mount=None
  ):
    """POST to begin the upload process with optional cross-repo mount param."""
    if not mount:
      # Do a normal POST to initiate an upload if mount is missing.
      url = '{base_url}/blobs/uploads/'.format(base_url=self._base_url())
      accepted_codes = [httplib.ACCEPTED]
    else:
      # If we have a mount parameter, try to mount the blob from another repo.
      mount_from = '&'.join(
          ['from=' + urllib.quote(repo.repository, '') for repo in self._mount])
      url = '{base_url}/blobs/uploads/?mount={digest}&{mount_from}'.format(
          base_url=self._base_url(),
          digest=digest,
          mount_from=mount_from)
      accepted_codes = [httplib.CREATED, httplib.ACCEPTED]

    resp, unused_content = self._transport.Request(
        url, method='POST', body=None,
        accepted_codes=accepted_codes)
    # pytype: disable=attribute-error
    return resp.status == httplib.CREATED, resp.get('location')
    # pytype: enable=attribute-error