python类patch()的实例源码

def update(self, uid, data, token):
        """
        Update a user's data. Requires an auth token.

        :param uid str: User ID to update
        :param data dict: The dictionary of data attributes to change. Just the attributes.
        :param token str: The authorization token for this user
        :return: True or Exception
        :rtype: Bool or ServerError
        """
        final_dict = {"data": {"id": uid, "type": "users", "attributes": data}}
        final_headers = self.header
        final_headers['Authorization'] = "Bearer {}".format(token)
        r = requests.patch(self.apiurl + "/users/{}".format(uid), json=final_dict, headers=final_headers)

        if r.status_code != 200:
            raise ServerError

        return True

def maintain(self, nodename, remove=False):
        """
        maintain node will disable or enable deployment onto the maintained node.
        """
        node = NodeInfo(nodename)
        base_url = "http://deployd.lain:9003/api/constraints"
        operator = "Remove" if remove else "Add"
        if not remove:
            url = base_url + "?type=node&value=%s" % node.name
            info("PATCH %s" % url)
            resp = requests.patch(url)
        else:
            url = base_url + "?type=node&value=%s" % node.name
            info("DELETE %s" % url)
            resp = requests.delete(url)
        if resp.status_code >= 300:
            error("%s constraint on node %s fail: %s" % (operator, node.name, resp.text))
        else:
            info("%s constraint on node %s success." % (operator, node.name))

def partial_update(self, resource_id, raise_exception=True, headers=None,
                       data=None):
        """
        Method for the partial update of resource, i.e. only a part of the
        resource's fields are updated.

        Example: PATCH endpoint/<pk>/
        """
        request_kwargs = {
            'headers': headers,
            'auth': self.auth,
        }
        request_kwargs.update(self.extract_write_data(
            data, raise_exception, partial=True))
        r = requests.patch(self.format_endpoint(resource_id), **request_kwargs)
        return r

def add_or_update(self, name, config=None):

        # does it exist already?
        plugins_response = self.list()
        plugins_list = plugins_response.json().get('data', [])

        data = {
            "name": name,
        }
        if config is not None:
            data.update(config)   

        plugin_id = self._get_plugin_id(name, plugins_list)
        if plugin_id is None:            
            return requests.post(self.base_url, data)
        else:
            url = "{}/{}" . format (self.base_url, plugin_id)
            return requests.patch(url, data)

def test_requests_mocks(self):
        self.assertTrue(isinstance(requests.get, Mock))
        self.assertTrue(isinstance(requests.put, Mock))
        self.assertTrue(isinstance(requests.post, Mock))
        self.assertTrue(isinstance(requests.patch, Mock))
        self.assertTrue(isinstance(requests.delete, Mock))

def _get_url_and_header(self, path):
        url = self._base_url + path
        header = {'Content-Type': 'application/merge-patch+json',
                  'Accept': 'application/json'}
        if self.token:
            header.update({'Authorization': 'Bearer %s' % self.token})

        return url, header

def patch_status(self, path, data):

        LOG.debug("Patch_status %(path)s: %(data)s", {
            'path': path, 'data': data})
        path = path + '/status'
        url, header = self._get_url_and_header(path)

        response = requests.patch(url, json={"status": data},
                                  headers=header, cert=self.cert,
                                  verify=self.verify_server)
        if response.ok:
            return response.json().get('status')
        raise exc.K8sClientException(response.text)

def update(self, job_id, update_doc):
        endpoint = self.endpoint + job_id
        r = requests.patch(endpoint,
                           headers=self.headers,
                           data=json.dumps(update_doc),
                           verify=self.verify)
        if r.status_code != 200:
            raise exceptions.ApiClientException(r)
        return r.json()['version']

def update(self, action_id, update_doc):
        endpoint = self.endpoint + action_id
        r = requests.patch(endpoint,
                           headers=self.headers,
                           data=json.dumps(update_doc), verify=self.verify)
        if r.status_code != 200:
            raise exceptions.ApiClientException(r)
        return r.json()['version']

def update(self, session_id, update_doc):
        endpoint = self.endpoint + session_id
        r = requests.patch(endpoint,
                           headers=self.headers,
                           data=json.dumps(update_doc),
                           verify=self.verify)
        if r.status_code != 200:
            raise exceptions.ApiClientException(r)
        return r.json()['version']

def update(self, job_id, update_doc):
        endpoint = self.endpoint + job_id
        r = requests.patch(endpoint,
                           headers=self.headers,
                           data=json.dumps(update_doc),
                           verify=self.verify)
        if r.status_code != 200:
            raise exceptions.ApiClientException(r)
        return r.json()['version']

def update(self, action_id, update_doc):
        endpoint = self.endpoint + action_id
        r = requests.patch(endpoint,
                           headers=self.headers,
                           data=json.dumps(update_doc), verify=self.verify)
        if r.status_code != 200:
            raise exceptions.ApiClientException(r)
        return r.json()['version']

def update(api_key_id, active):
    """
    Enable or disable an api key with given ID
    """
    action, url = _url((api_key_id,))
    payload = {
        "apikey":
            {
                "active": active
            }
    }

    headers = api_utils.generate_headers('owner', method='PATCH', body=json.dumps(payload),
                                         action=action)
    try:
        response = requests.patch(url, json=payload, headers=headers)
        if response_utils.response_error(response):
            sys.stderr.write('Failed to %s api key with id: %s \n' %
                             ('enable' if active else 'disable', api_key_id))
            sys.exit(1)
        elif response.status_code == 200:
            sys.stdout.write('%s api key with id: %s\n' %
                             ('Enabled' if active else 'Disabled', api_key_id))
            handle_api_key_response(response)
    except requests.exceptions.RequestException as error:
        sys.stderr.write(error)
        sys.exit(1)

def rename_team(team_id, team_name):
    """
    Rename team with the provided team_id.
    """
    params = {
        'team': {
            'name': team_name,
            # as this is a patch request, it won't modify users in the team.
            # what we want is to update the name of the team only.
            'users': [
                {'id': ''}
            ]
        }
    }
    headers = api_utils.generate_headers('rw')

    try:
        response = requests.patch(_url((team_id,))[1], json=params, headers=headers)
        if response_utils.response_error(response):  # Check response has no errors
            click.echo('Renaming team with id: %s failed.' % team_id, err=True)
            sys.exit(1)
        elif response.status_code == 200:
            click.echo("Team: '%s' renamed to: '%s'" % (team_id, team_name))
    except requests.exceptions.RequestException as error:
        click.echo(error, err=True)
        sys.exit(1)

def add_user_to_team(team_id, user_key):
    """
    Add user with the provided user_key to team with provided team_id.
    """
    headers = api_utils.generate_headers('rw')
    params = {'teamid': team_id}
    try:
        response = requests.get(_url((team_id,))[1], params=params, headers=headers)
        if response.status_code == 200:
            params = {
                'team': {
                    'name': response.json()['team']['name'],
                    'users': [
                        # we are doing a patch request here so it's safe to include the user_key
                        # we want to add here
                        {'id': user_key}
                    ]
                }
            }
            headers = api_utils.generate_headers('rw')
            try:
                response = requests.patch(_url((team_id,))[1], json=params, headers=headers)
                if response_utils.response_error(response):  # Check response has no errors
                    click.echo('Adding user to team with key: %s failed.' % team_id, err=True)
                    sys.exit(1)
                elif response.status_code == 200:
                    click.echo('Added user with key: %s to team.' % user_key)
            except requests.exceptions.RequestException as error:
                click.echo(error, err=True)
                sys.exit(1)
        elif response_utils.response_error(response):
            click.echo('Cannot find team. Adding user to team %s failed.' % team_id, err=True)
            sys.exit(1)
    except requests.exceptions.RequestException as error:
        click.echo(error, err=True)
        sys.exit(1)

def test_handler_calls_many_times(server: FakeServer):
    server. \
        on_("patch", "/channels"). \
        response(status=204)

    response_0 = requests.patch(server.base_uri + "/channels")
    response_1 = requests.patch(server.base_uri + "/channels")
    assert response_0.status_code == 204
    assert response_1.status_code == 204

def test_handler_called_nth_times(server: FakeServer):
    server. \
        on_("patch", "/songs"). \
        response(status=204)._3_times()

    response_0 = requests.patch(server.base_uri + "/songs")
    response_1 = requests.patch(server.base_uri + "/songs")
    response_2 = requests.patch(server.base_uri + "/songs")
    response_3 = requests.patch(server.base_uri + "/songs")
    assert response_0.status_code == 204
    assert response_1.status_code == 204
    assert response_2.status_code == 204
    assert response_3.status_code == 500
    assert response_3.text == "Server has not responses for [PATCH] http://localhost:8081/songs"
    assert response_3.headers["content-type"] == "text/plain"

def test_explicit_content_type_remove_application_json(server: FakeServer):
    (server.
        on_("patch", "/games").
        response(status=200, json={"change": "content-type"}, content_type="text/plain"))

    response = requests.patch(server.base_uri + "/games")
    assert response.status_code == 200
    assert response.headers["Content-Type"] == "text/plain"

def test_with_body_not_raise_error(server: FakeServer):
    requests.patch(server.base_uri + "/sleep", data="Good night!")

    expect_that(server.was_requested("patch", "/sleep").
                for_the_first_time().
                with_body("Good night!"))

def test_chain_of_different_with_errors(server: FakeServer):
    requests.patch(server.base_uri + "/users", json={"name": "new_name"})

    with pytest.raises(AssertionError) as error:
        expect_that(server.was_requested("patch", "/users").
                    for_the_first_time().
                    with_content_type("text/plain").
                    with_body("old name"))

    assert str(error.value) == ("Expect that server was requested with [PATCH] http://localhost:8081/users.\n"
                                "For the 1 time: with content type 'text/plain'.\n"
                                "But for the 1 time: content type was 'application/json'.\n"
                                "For the 1 time: with body 'old name'.\n"
                                "But for the 1 time: body was '{\"name\": \"new_name\"}'.")