我们从Python开源项目中，提取了以下13个代码示例，用于说明如何使用responses.PATCH。
def _setup_mocking(self): def _mock_callback(request): method = request.method.lower() url = request.path_url handler = getattr(self.app, method) r = handler( url, data=request.body, headers=dict(request.headers) ) return (r.status_code, r.headers, r.data) pattern = re.compile("{}/(.*)".format(self.host)) methods = [ responses.GET, responses.POST, responses.PUT, responses.DELETE, responses.PATCH, ] for method in methods: responses.add_callback( method, pattern, callback=_mock_callback )
def test_pr_synchronize_callback(self, config): """Test that a snooze label is removed from PRs when a new commit is pushed.""" responses.add( responses.PATCH, "https://api.github.com/repos/octocat/Hello-World/issues/1347") responses.add( responses.GET, "https://api.github.com/repos/baxterthehacker/public-repo/issues/1", body=github_responses.SNOOZED_ISSUE_GET) r = snooze.github_callback( "pull_request", json.loads(github_responses.PULL_REQUEST), (config["github_username"], config["github_token"]), config["snooze_label"], config["ignore_members_of"]) assert r is True assert len(responses.calls) == 2
def test_does_post_to_create_webapp(self, api_responses, api_token): expected_post_url = API_ENDPOINT.format(username=getpass.getuser()) expected_patch_url = API_ENDPOINT.format(username=getpass.getuser()) + 'mydomain.com/' api_responses.add(responses.POST, expected_post_url, status=201, body=json.dumps({'status': 'OK'})) api_responses.add(responses.PATCH, expected_patch_url, status=200) Webapp('mydomain.com').create('2.7', '/virtualenv/path', '/project/path', nuke=False) post = api_responses.calls[0] assert post.request.url == expected_post_url assert post.request.body == urlencode({ 'domain_name': 'mydomain.com', 'python_version': PYTHON_VERSIONS['2.7'], }) assert post.request.headers['Authorization'] == f'Token {api_token}'
def test_does_patch_to_update_virtualenv_path(self, api_responses, api_token): expected_post_url = API_ENDPOINT.format(username=getpass.getuser()) expected_patch_url = API_ENDPOINT.format(username=getpass.getuser()) + 'mydomain.com/' api_responses.add(responses.POST, expected_post_url, status=201, body=json.dumps({'status': 'OK'})) api_responses.add(responses.PATCH, expected_patch_url, status=200) Webapp('mydomain.com').create('2.7', '/virtualenv/path', '/project/path', nuke=False) patch = api_responses.calls[1] assert patch.request.url == expected_patch_url assert patch.request.body == urlencode({ 'virtualenv_path': '/virtualenv/path' }) assert patch.request.headers['Authorization'] == f'Token {api_token}'
def test_raises_if_patch_does_not_20x(self, api_responses, api_token): expected_post_url = API_ENDPOINT.format(username=getpass.getuser()) expected_patch_url = API_ENDPOINT.format(username=getpass.getuser()) + 'mydomain.com/' api_responses.add(responses.POST, expected_post_url, status=201, body=json.dumps({'status': 'OK'})) api_responses.add(responses.PATCH, expected_patch_url, status=400, json={'message': 'an error'}) with pytest.raises(Exception) as e: Webapp('mydomain.com').create('2.7', '/virtualenv/path', '/project/path', nuke=False) assert 'PATCH to set virtualenv path via API failed' in str(e.value) assert 'an error' in str(e.value)
def test_does_delete_first_for_nuke_call(self, api_responses, api_token): post_url = API_ENDPOINT.format(username=getpass.getuser()) webapp_url = API_ENDPOINT.format(username=getpass.getuser()) + 'mydomain.com/' api_responses.add(responses.DELETE, webapp_url, status=200) api_responses.add(responses.POST, post_url, status=201, body=json.dumps({'status': 'OK'})) api_responses.add(responses.PATCH, webapp_url, status=200) Webapp('mydomain.com').create('2.7', '/virtualenv/path', '/project/path', nuke=True) delete = api_responses.calls[0] assert delete.request.method == 'DELETE' assert delete.request.url == webapp_url assert delete.request.headers['Authorization'] == f'Token {api_token}'
def test_ignores_404_from_delete_call_when_nuking(self, api_responses, api_token): post_url = API_ENDPOINT.format(username=getpass.getuser()) webapp_url = API_ENDPOINT.format(username=getpass.getuser()) + 'mydomain.com/' api_responses.add(responses.DELETE, webapp_url, status=404) api_responses.add(responses.POST, post_url, status=201, body=json.dumps({'status': 'OK'})) api_responses.add(responses.PATCH, webapp_url, status=200) Webapp('mydomain.com').create('2.7', '/virtualenv/path', '/project/path', nuke=True)
def test_upsert_extant(): url_re2 = re.compile(r'https://.*/services/oauth2/token') responses.add( responses.PATCH, 'http://foo/services/data/v35.0/sobjects/Contact/0031700000BHQzBA' 'AX', body='{"errors": [], "id": "a0917000002rZngAAE", "success": true}', status=204, ) responses.add( responses.POST, url_re2, body='{"instance_url": "http://foo", "errors": [], "id": "a0917000' '002rZngAAE", "access_token": "bar", "success": true}', status=200, ) responses.add( responses.POST, 'http://foo/services/data/v35.0/sobjects/Contact', body='{"errors": [], "id": "0031700000F3kcwAAB", "success": true}', status=200, ) responses.add_callback( responses.GET, 'http://foo/services/data/v35.0/query', callback=request_upsert_extant_callback, ) actual = upsert_customer(customer=customer, form=rdo_form) assert actual is True assert len(responses.calls) == 3
def test_issue_comment_callback(self, config): """Test that a snooze label is removed from issues when a new comment is received.""" responses.add( responses.PATCH, "https://api.github.com/repos/baxterthehacker/public-repo/issues/2") r = snooze.github_callback( "issue_comment", json.loads(github_responses.SNOOZED_ISSUE_COMMENT), (config["github_username"], config["github_token"]), config["snooze_label"], config["ignore_members_of"]) assert r is True assert len(responses.calls) == 1 org_url = "https://api.github.com/orgs/fellowship/members/baxterthehacker" responses.add(responses.GET, org_url, status=204) # is a member r = snooze.github_callback( "issue_comment", json.loads(github_responses.SNOOZED_ISSUE_COMMENT), (config["github_username"], config["github_token"]), config["snooze_label"], ignore_members_of="fellowship") assert r is False orc_url = "https://api.github.com/orgs/orcs/members/baxterthehacker" responses.add(responses.GET, orc_url, status=404) # is not a member r = snooze.github_callback( "issue_comment", json.loads(github_responses.SNOOZED_ISSUE_COMMENT), (config["github_username"], config["github_token"]), config["snooze_label"], ignore_members_of="orcs") assert r is True
def test_pr_commit_comment_callback(self, config): """Test that a snooze label is removed from PRs when a new commit is pushed.""" responses.add( responses.PATCH, "https://api.github.com/repos/octocat/Hello-World/issues/1347") responses.add( responses.GET, "https://api.github.com/repos/baxterthehacker/public-repo/issues/1", body=github_responses.SNOOZED_ISSUE_GET) r = snooze.github_callback( "pull_request_review_comment", json.loads(github_responses.PULL_REQUEST_REVIEW_COMMENT), (config["github_username"], config["github_token"]), config["snooze_label"], config["ignore_members_of"]) assert r is True assert len(responses.calls) == 2 org_url = "https://api.github.com/orgs/fellowship/members/baxterthehacker" responses.add(responses.GET, org_url, status=204) # is a member r = snooze.github_callback( "pull_request_review_comment", json.loads(github_responses.PULL_REQUEST_REVIEW_COMMENT), (config["github_username"], config["github_token"]), config["snooze_label"], ignore_members_of="fellowship") assert r is False orc_url = "https://api.github.com/orgs/orcs/members/baxterthehacker" responses.add(responses.GET, orc_url, status=404) # is not a member r = snooze.github_callback( "pull_request_review_comment", json.loads(github_responses.PULL_REQUEST_REVIEW_COMMENT), (config["github_username"], config["github_token"]), config["snooze_label"], ignore_members_of="orcs") assert r is True
def test_methods(self): """ When the HTTP method-specific methods are called, the correct request method is used. """ client = ContainerHttpClient('127.0.0.1', '45678') responses.add(responses.GET, 'http://127.0.0.1:45678/', status=200) responses.add( responses.OPTIONS, 'http://127.0.0.1:45678/foo', status=201) responses.add(responses.HEAD, 'http://127.0.0.1:45678/bar', status=403) responses.add(responses.POST, 'http://127.0.0.1:45678/baz', status=404) responses.add(responses.PUT, 'http://127.0.0.1:45678/test', status=418) responses.add( responses.PATCH, 'http://127.0.0.1:45678/a/b/c', status=501) responses.add( responses.DELETE, 'http://127.0.0.1:45678/d/e/f', status=503) get_response = client.get() options_response = client.options('/foo') head_response = client.head('/bar') post_response = client.post('/baz') put_response = client.put('/test') patch_response = client.patch('/a/b/c') delete_response = client.delete('/d/e/f') self.assertEqual(get_response.status_code, 200) self.assertEqual(options_response.status_code, 201) self.assertEqual(head_response.status_code, 403) self.assertEqual(post_response.status_code, 404) self.assertEqual(put_response.status_code, 418) self.assertEqual(patch_response.status_code, 501) self.assertEqual(delete_response.status_code, 503) self.assertEqual(len(responses.calls), 7)
def setup_recording(self, **kwargs): _logger.info("recording ...") self.responses.reset() all_requests_re = re.compile("http.*") methods = (responses.GET, responses.POST, responses.PUT, responses.PATCH, responses.DELETE, responses.HEAD, responses.OPTIONS) for http_method in methods: self.responses.add_callback( http_method, all_requests_re, match_querystring=False, callback=self.record())
def test_link_issue_with_comment(self): responses.add( responses.PATCH, 'https://fabrikam-fiber-inc.visualstudio.com/DefaultCollection/_apis/wit/workitems/309?api-version=3.0', body=WORK_ITEM_RESPONSE, content_type='application/json', ) self.plugin.set_option( 'instance', 'fabrikam-fiber-inc.visualstudio.com', self.project) group = self.create_group(message='Hello world', culprit='foo.bar') request = self.request.get('/') request.user = AnonymousUser() form_data = { 'item_id': '309', 'comment': 'Fix this.', } with self.assertRaises(PluginError): self.plugin.link_issue(request, group, form_data) request.user = self.user self.login_as(self.user) UserSocialAuth.objects.create( user=self.user, provider=self.plugin.auth_provider, uid='a89e7204-9ca0-4680-ba7a-cfcf6b3c7445', extra_data={ 'access_token': 'foo', 'refresh_token': 'bar', } ) assert self.plugin.link_issue(request, group, form_data) == { 'id': 309, 'title': 'Customer can sign in using their Microsoft Account', 'url': 'https://fabrikam-fiber-inc.visualstudio.com/web/wi.aspx?pcguid=d81542e4-cdfa-4333-b082-1ae2d6c3ad16&id=309', } request = responses.calls[-1].request assert request.headers['Content-Type'] == 'application/json-patch+json' payload = json.loads(request.body) assert payload == [ { 'op': 'add', 'path': '/fields/System.History', 'value': '<p>Fix this.</p>\n', }, # { # "op": "add", # "path": "/relations/-", # "value": { # "rel": "Hyperlink", # "url": 'http://testserver/baz/bar/issues/1/', # } # } ]
