def test_get_velocity_success(service):
responses.add(
responses.GET,
'https://www.pivotaltracker.com/services/v5/projects/123/iterations'
'/456?fields=%3Adefault%2Cvelocity%2Cstories',
json={
'velocity': 10,
'stories': [
{'current_state': 'foo', 'estimate': 5},
{'current_state': 'foo'},
],
}
)
result = service.details(456)
assert result == {'velocity': 10, 'stories': {'foo': 5}}
assert responses.calls[0].request.headers['X-TrackerToken'] == 'foobar'
python类GET的实例源码
def test_update_success(debug, service):
service.current_iteration = 1
service.project_version = 2
service._cached = {'foo': 'bar'}
responses.add(
responses.GET,
'https://www.pivotaltracker.com/services/v5/projects/123',
json={'foo': 'bar', 'current_iteration_number': 0},
adding_headers={'X-Tracker-Project-Version': '1'},
)
result = service.update()
debug.assert_called_once_with('fetching Tracker project data')
assert result == {'foo': 'bar'}
assert responses.calls[0].request.headers['X-TrackerToken'] == 'foobar'
def test_update_details(debug, service):
service.current_iteration = 1
service.project_version = 1
service._cached = {'foo': 'bar'}
name = 'foo'
responses.add(
responses.GET,
'https://www.pivotaltracker.com/services/v5/projects/123',
json={'current_iteration_number': 1, 'name': name},
adding_headers={'X-Tracker-Project-Version': '2'},
)
responses.add(
responses.GET,
'https://www.pivotaltracker.com/services/v5/projects/123/iterations'
'/1?fields=%3Adefault%2Cvelocity%2Cstories',
json={'velocity': 10, 'stories': []}
)
result = service.update()
debug.assert_has_calls([
mock.call('fetching Tracker project data'),
mock.call('project updated, fetching iteration details'),
])
assert result == dict(velocity=10, stories={}, name=name)
def test_update_enterprise_success(debug):
responses.add(
responses.GET,
'http://dummy.url/repos/foo/bar/issues?state=all&access_token=foobar',
headers={'User-Agent': 'bar'},
json=[],
)
service = GitHubEnterpriseIssues(
api_token='foobar',
account='foo',
repo='bar',
root='http://dummy.url',
)
result = service.update()
debug.assert_called_once_with('fetching GitHub issue data')
assert result == {'issues': {}, 'name': 'foo/bar', 'health': 'neutral', 'halflife': None}
def test_format_build_missing_data(service):
responses.add(
responses.GET,
'https://coveralls.io/foo/bar/baz.json?page=1',
json=dict(builds=[{}]),
)
result = service.update()
assert result['builds'][0] == dict(
author='<no author>',
committed='time not available',
coverage=None,
message_text=None,
raw_coverage=None,
)
def test_update_success(debug, service):
responses.add(
responses.GET,
'https://api.github.com/repos/foo/bar/commits?access_token=foobar',
json=[{'commit': {
'author': {'name': 'alice'},
'committer': {'name': 'bob', 'date': TWO_DAYS_AGO},
'message': 'commit message',
}}],
)
result = service.update()
debug.assert_called_once_with('fetching GitHub project data')
assert result == {'commits': [{
'message': 'commit message',
'author': 'alice [bob]',
'committed': 'two days ago'
}], 'name': 'foo/bar'}
assert responses.calls[0].request.headers['User-Agent'] == 'bar'
def test_update_enterprise_success(debug):
responses.add(
responses.GET,
'http://dummy.url/repos/foo/bar/commits?access_token=foobar',
json=[{'commit': {
'author': {'name': 'alice'},
'committer': {'name': 'bob', 'date': TWO_DAYS_AGO},
'message': 'commit message',
}}],
)
service = GitHubEnterprise(api_token='foobar', account='foo', repo='bar',
root='http://dummy.url')
result = service.update()
debug.assert_called_once_with('fetching GitHub project data')
assert result == {'commits': [{
'message': 'commit message',
'author': 'alice [bob]',
'committed': 'two days ago'
}], 'name': 'foo/bar'}
assert responses.calls[0].request.headers['User-Agent'] == 'bar'
def test_estimated_formatting(service, url):
response = dict(name='foo', builds=[
dict(duration=0, description=None, timestamp=1481387964313, result=None),
dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
dict(duration=10000, description=None, timestamp=1481387964313, result='SUCCESS'),
])
responses.add(responses.GET, url, json=response)
result = service.update()
assert len(result['builds']) == 4
assert result['builds'][0] == dict(
author='<no author>',
duration=5,
elapsed='five seconds left',
message='<no message>',
outcome='working',
started_at=1481387964,
)
def test_create_issue(self):
responses.add(
responses.GET,
'https://getsentry.atlassian.net/rest/api/2/issue/createmeta',
json=create_meta_response)
responses.add(responses.POST, 'https://getsentry.atlassian.net/rest/api/2/issue', json={
'key': 'SEN-1'
})
self.plugin.set_option('instance_url', 'https://getsentry.atlassian.net', self.project)
group = self.create_group(message='Hello world', culprit='foo.bar')
request = self.request.get('/')
request.user = AnonymousUser()
form_data = {
'title': 'Hello',
'description': 'Fix this.',
'issuetype': 'bug',
'project': 'SEN'
}
assert self.plugin.create_issue(request, group, form_data) == 'SEN-1'
def test_make_call(self):
connection = AsyncConnection(
TEST_USERNAME,
TEST_SECRET,
EMARSYS_URI
)
with aioresponses() as m:
m.get(
urljoin(EMARSYS_URI, 'api/v2/settings'),
status=200,
payload=EMARSYS_SETTINGS_RESPONSE
)
coroutine = connection.make_call('GET', 'api/v2/settings')
loop = asyncio.get_event_loop()
response = loop.run_until_complete(coroutine)
assert response == EMARSYS_SETTINGS_RESPONSE
def test_list_choice(self):
responses.add(
responses.GET,
urljoin(
EMARSYS_URI, '{}/{}/{}'.format(
CONTACT_FIELDS_ENDPOINT,
5,
'choice'
)
),
json=EMARSYS_CONTACT_FIELDS_LIST_CHOICE_RESPONSE,
status=200,
content_type='application/json'
)
connection = SyncConnection(TEST_USERNAME, TEST_SECRET)
contact_fields = ContactField(connection)
response = contact_fields.list_choice(5)
assert response == EMARSYS_CONTACT_FIELDS_LIST_CHOICE_RESPONSE
def test_downloader_conn_error(self):
exception = ConnectionError()
with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
max_retry = 3
for _ in range(max_retry + 1):
rsps.add(responses.GET, self.TEST_MPD_URL, body=exception)
dl = live.Downloader(
mpd=self.TEST_MPD_URL,
output_dir='output_connerror',
duplicate_etag_retry=2,
singlethreaded=True,
max_connection_error_retry=max_retry)
dl.run()
dl.stream_id = '17875351285037717'
output_file = 'output_connerror.mp4'
dl.stitch(output_file, cleartempfiles=True)
self.assertFalse(os.path.isfile(output_file), '{0!s} not generated'.format(output_file))
def test_existing_resource():
url_expr = re.compile(r'https?://api\.test\.tst/v1/videos/show\?.*'
'video_key=VideoKey.*')
responses.add(
responses.GET, url_expr,
status=200,
content_type='application/json',
body='{"status": "ok", '
'"rate_limit": {"reset": 1478929300, "limit": 50, "remaining": 47},'
'"video": {"status": "ready", "expires_date": null, "description": null, '
'"title": "Title", "views": 179, "tags": "", "sourceformat": null, '
'"mediatype": "video", "upload_session_id": null, "custom": {}, '
'"duration": "817.56", "sourceurl": null, "link": null, "author": null, '
'"key": "VideoKey", "error": null, "date": 1464754765, '
'"md5": "653bc15b6cba7319c2df9b5cf869b5b8", "sourcetype": "file", '
'"size": "904237686"}}')
jwp_client = jwplatform.Client('api_key', 'api_secret', host='api.test.tst')
resp = jwp_client.videos.show(video_key='VideoKey')
assert resp['status'] == 'ok'
assert 'status' in resp['video']
assert resp['video']['key'] == 'VideoKey'
def test_nonexisting_resource():
url_expr = re.compile(r'https?://api\.test\.tst/v1/videos/abcd/show\?.*'
'abcd_key=AbcdKey.*')
responses.add(
responses.GET, url_expr,
status=404,
content_type='application/json',
body='{"status": "error", '
'"message": "API method `/videos/abcd/show` not found", '
'"code": "NotFound", "title": "Not Found"}')
jwp_client = jwplatform.Client('api_key', 'api_secret', host='api.test.tst')
with pytest.raises(jwplatform.errors.JWPlatformNotFoundError) as err:
jwp_client.videos.abcd.show(abcd_key='AbcdKey')
assert err.value.message == 'API method `/videos/abcd/show` not found'
def test_wrong_auth(bot):
with controlled_responses() as rsps:
rsps.rsps.add(
responses.GET,
'http://host/rest/api/2/issue/JIRA-1',
status=401)
message = get_message('JIRA-1')
bot.display_issues(message)
assert len(message.send_webapi.call_args_list) == 1
args, kwargs = message.send_webapi.call_args_list[0]
assert args[0] is ''
assert json.loads(args[1]) == [{
'color': 'danger',
'fallback': 'Jira authentication error',
'text': ':exclamation: Jira authentication error'}]
def __add_responses(self):
for request in self.requests_get:
body = None
content_type = 'application/json'
if 'content_type' in request:
content_type = request['content_type']
if 'text' in request:
if content_type.startswith('image/'):
body = base64.b64decode(request['text'])
else:
body = json.dumps(request['text'])
self.rsps.add(
responses.GET,
request['url'],
status=request['code'],
body=body,
content_type=content_type,
match_querystring=True)
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_get_by_slug():
slug = 'OpenUglySnoodVoteNay'
responses.add(
responses.GET,
'%sclips/%s' % (BASE_URL, slug),
body=json.dumps(example_clip),
status=200,
content_type='application/json'
)
client = TwitchClient('client id')
clip = client.clips.get_by_slug(slug)
assert isinstance(clip, Clip)
assert clip.broadcast_id == example_clip['broadcast_id']
def test_get_top():
params = {'limit': 1, 'period': 'month'}
responses.add(
responses.GET,
'%sclips/top' % BASE_URL,
body=json.dumps(example_clips),
status=200,
content_type='application/json'
)
client = TwitchClient('client id')
clips = client.clips.get_top(**params)
assert len(clips) == len(example_clips)
assert isinstance(clips[0], Clip)
assert clips[0].broadcast_id == example_clips['clips'][0]['broadcast_id']
def setup_versions(self, stable, devel, nightly, esr, esr_next=None):
"""
Add a mock response for official versions
"""
self.cleanup()
body = {
"FIREFOX_NIGHTLY": nightly,
"FIREFOX_ESR": esr,
"FIREFOX_ESR_NEXT": esr_next,
"LATEST_FIREFOX_DEVEL_VERSION": devel,
"LATEST_FIREFOX_OLDER_VERSION": "3.6.28",
"LATEST_FIREFOX_RELEASED_DEVEL_VERSION": devel,
"LATEST_FIREFOX_VERSION": stable,
}
local_mock = responses.RequestsMock()
local_mock.add(responses.GET, versions.URL_VERSIONS, json=body)
local_mock.start()
yield local_mock
local_mock.stop()
def test_get_target_group_name(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/groups/GetTargetGroupName',
callback=get_target_group_name_callback,
content_type='application/json'
)
client = Client('username', 'password')
data = client.groups.get_target_group_name(42)
self.assertEqual(data, 'Ireland VIPs Played Last 30 Days')
def test_get_target_group_name_with_empty_target_group_id(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/groups/GetTargetGroupName',
callback=get_target_group_name_callback,
content_type='application/json'
)
client = Client('username', 'password')
self.assertRaises(Exception, client.groups.get_target_group_name, None)
def test_get_target_group_name_with_wrong_name(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/groups/GetTargetGroupName',
callback=get_target_group_name_callback,
content_type='application/json'
)
client = Client('username', 'password')
data = client.groups.get_target_group_name(24)
self.assertFalse(data)
def test_get_target_group_id(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/groups/GetTargetGroupID',
callback=get_target_group_id_callback,
content_type='application/json'
)
client = Client('username', 'password')
data = client.groups.get_target_group_id('UK 20VIPs')
self.assertEqual(data, 26)
def test_get_target_group_id_with_empty_name(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/groups/GetTargetGroupID',
callback=get_target_group_id_callback,
content_type='application/json'
)
client = Client('username', 'password')
self.assertRaises(Exception, client.groups.get_target_group_id, None)
def test_get_target_groups_by_date(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/groups/GetTargetGroupsByDate',
callback=get_target_groups_by_date_callback,
content_type='application/json'
)
client = Client('username', 'password')
data = client.groups.get_target_groups_by_date('2015-05-31')
self.assertEqual(data, [9, 24, 31, 36])
def test_get_target_groups_by_date_with_empty_date(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/groups/GetTargetGroupsByDate',
callback=get_target_groups_by_date_callback,
content_type='application/json'
)
client = Client('username', 'password')
self.assertRaises(Exception, client.groups.get_target_groups_by_date, None)
def test_get_target_groups_by_date_with_wrong_date(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/groups/GetTargetGroupsByDate',
callback=get_target_groups_by_date_callback,
content_type='application/json'
)
client = Client('username', 'password')
data = client.groups.get_target_groups_by_date('3015-05-31')
self.assertFalse(data)
def test_get_value_segment_name(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/segments/GetValueSegmentName',
callback=get_value_segment_name_callback,
content_type='application/json'
)
client = Client('username', 'password')
data = client.segments.get_value_segment_name(1)
self.assertEqual(data, 'Diamond')
def test_get_value_segment_name_with_wrong_segment_id(self):
responses.add_callback(
responses.POST,
'https://api.optimove.net/v3.0/general/login',
callback=login_callback,
content_type='application/json'
)
responses.add_callback(
responses.GET,
'https://api.optimove.net/v3.0/segments/GetValueSegmentName',
callback=get_value_segment_name_callback,
content_type='application/json'
)
client = Client('username', 'password')
data = client.segments.get_value_segment_name(2)
self.assertFalse(data)
