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
)
python类PATCH的实例源码
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/',
# }
# }
]
