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类add()的实例源码
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_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_list_data(self):
responses.add(
responses.GET,
urljoin(
EMARSYS_URI,
'{}/{}'.format(
CONTACT_ENDPOINT, 'query/?return=3&limit=2&1=Squirrel'
)
),
json=EMARSYS_CONTACTS_LIST_DATA_RESPONSE,
status=200,
content_type='application/json',
match_querystring=True
)
connection = SyncConnection(TEST_USERNAME, TEST_SECRET)
contacts = Contact(connection)
response = contacts.query(
3,
query_tuple=(1, 'Squirrel'),
limit=2
)
assert response == EMARSYS_CONTACTS_LIST_DATA_RESPONSE
def test_get_history(self):
responses.add(
responses.POST,
urljoin(
EMARSYS_URI,
'{}/{}/'.format(CONTACT_ENDPOINT, 'getcontacthistory')
),
json=EMARSYS_CONTACTS_GET_CONTACT_HISTORY_RESPONSE,
status=200,
content_type='application/json'
)
connection = SyncConnection(TEST_USERNAME, TEST_SECRET)
contacts = Contact(connection)
response = contacts.get_history(
[723005829]
)
assert response == EMARSYS_CONTACTS_GET_CONTACT_HISTORY_RESPONSE
def test_notify_gihub(faker):
repo = {'owner_name': faker.domain_word(), 'name': faker.domain_word()}
commit = uuid4().hex
state = random.choice(['success', 'failure', 'error', 'pending']),
job_task = faker.domain_word()
job_id = random.randint(1, 10000)
request_json = {
'state': state,
'target_url': 'https://travis-ci/{0}/{1}/jobs/{2}'.format(repo['owner_name'], repo['name'], job_id),
'description': get_description_from_task(job_task),
'context': job_task
}
responses.add(
responses.POST,
'https://api.github.com/repos/{0}/{1}/statuses/{2}'.format(repo['owner_name'], repo['name'], commit),
json=request_json)
notify_github(repo, commit, state, job_task, job_id)
assert len(responses.calls) == 1
def test_do_post(init_kwargs):
"""Exercise the private _do_post() method"""
# create instance of client under test
client = WSMANClient(**init_kwargs)
assert client
# setup responses
expected_url = "https://{host}:{port}/wsman".format(
host=init_kwargs['host'],
port=init_kwargs.get('port', 443))
fake_response_text = b"fake response from wsman"
responses.add(
responses.POST,
expected_url,
body=fake_response_text,
status=200,
content_type='application/xml')
payload = "fake"
reply = client._do_post(payload)
assert reply == fake_response_text
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_get_badges_by_channel():
channel_id = 7236692
response = {
'admin': {
'alpha': 'https://static-cdn.jtvnw.net/chat-badges/admin-alpha.png',
'image': 'https://static-cdn.jtvnw.net/chat-badges/admin.png',
'svg': 'https://static-cdn.jtvnw.net/chat-badges/admin.svg'
}
}
responses.add(responses.GET,
'%schat/%s/badges' % (BASE_URL, channel_id),
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
badges = client.chat.get_badges_by_channel(channel_id)
assert len(responses.calls) == 1
assert isinstance(badges, dict)
assert badges['admin'] == response['admin']
def test_get_emoticons_by_set():
response = {
'emoticon_sets': {
'19151': [example_emote]
}
}
responses.add(responses.GET,
'%schat/emoticon_images' % BASE_URL,
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
emoticon_sets = client.chat.get_emoticons_by_set()
assert len(responses.calls) == 1
assert isinstance(emoticon_sets, dict)
assert emoticon_sets['emoticon_sets'] == response['emoticon_sets']
assert emoticon_sets['emoticon_sets']['19151'][0] == example_emote
def test_get_all_emoticons():
response = {
'emoticons': [example_emote]
}
responses.add(responses.GET,
'%schat/emoticons' % BASE_URL,
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
emoticon_sets = client.chat.get_all_emoticons()
assert len(responses.calls) == 1
assert isinstance(emoticon_sets, dict)
assert emoticon_sets['emoticons'] == response['emoticons']
assert emoticon_sets['emoticons'][0] == example_emote
def test_get_metadata():
collection_id = 'abcd'
responses.add(responses.GET,
'%scollections/%s' % (BASE_URL, collection_id),
body=json.dumps(example_collection),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
collection = client.collections.get_metadata(collection_id)
assert len(responses.calls) == 1
assert isinstance(collection, Collection)
assert collection.id == example_collection['_id']
assert collection.items_count == example_collection['items_count']
def test_get():
collection_id = 'abcd'
response = {
'_id': 'myIbIFkZphQSbQ',
'items': [example_item]
}
responses.add(responses.GET,
'%scollections/%s/items' % (BASE_URL, collection_id),
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
items = client.collections.get(collection_id)
assert len(responses.calls) == 1
assert len(items) == 1
item = items[0]
assert isinstance(item, Item)
assert item.id == example_item['_id']
assert item.title == example_item['title']
def test_create():
channel_id = 'abcd'
responses.add(responses.POST,
'%schannels/%s/collections' % (BASE_URL, channel_id),
body=json.dumps(example_collection),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth client')
collection = client.collections.create(channel_id, 'this is title')
assert len(responses.calls) == 1
assert isinstance(collection, Collection)
assert collection.id == example_collection['_id']
assert collection.items_count == example_collection['items_count']
def test_add_item():
collection_id = 'abcd'
responses.add(responses.PUT,
'%scollections/%s/items' % (BASE_URL, collection_id),
body=json.dumps(example_item),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth client')
item = client.collections.add_item(collection_id, '1234', 'video')
assert len(responses.calls) == 1
assert isinstance(item, Item)
assert item.id == example_item['_id']
assert item.title == example_item['title']
def test_channels():
response = {
'_total': 2147,
'channels': [example_channel]
}
responses.add(responses.GET,
'%ssearch/channels' % BASE_URL,
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
channels = client.search.channels('mah query')
assert len(responses.calls) == 1
assert len(channels) == 1
channel = channels[0]
assert isinstance(channel, Channel)
assert channel.id == example_channel['_id']
assert channel.name == example_channel['name']
def test_games():
response = {
'_total': 2147,
'games': [example_game]
}
responses.add(responses.GET,
'%ssearch/games' % BASE_URL,
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
games = client.search.games('mah query')
assert len(responses.calls) == 1
assert len(games) == 1
game = games[0]
assert isinstance(game, Game)
assert game.id == example_game['_id']
assert game.name == example_game['name']
def test_streams():
response = {
'_total': 2147,
'streams': [example_stream]
}
responses.add(responses.GET,
'%ssearch/streams' % BASE_URL,
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
streams = client.search.streams('mah query')
assert len(responses.calls) == 1
assert len(streams) == 1
stream = streams[0]
assert isinstance(stream, Stream)
assert stream.id == example_stream['_id']
assert stream.game == example_stream['game']
assert isinstance(stream.channel, Channel)
assert stream.channel.id == example_channel['_id']
assert stream.channel.name == example_channel['name']
def test_get_posts():
channel_id = '1234'
response = {
'_cursor': '1479487861147094000',
'_topic': 'feeds.channel.44322889',
'posts': [example_post]
}
responses.add(responses.GET,
'%sfeed/%s/posts' % (BASE_URL, channel_id),
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
posts = client.channel_feed.get_posts(channel_id)
assert len(responses.calls) == 1
assert len(posts) == 1
post = posts[0]
assert isinstance(post, Post)
assert post.id == example_post['id']
assert post.body == example_post['body']
def test_create_post():
channel_id = '1234'
response = {
'post': example_post,
'tweet': None
}
responses.add(responses.POST,
'%sfeed/%s/posts' % (BASE_URL, channel_id),
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
post = client.channel_feed.create_post(channel_id, 'abcd')
assert len(responses.calls) == 1
assert isinstance(post, Post)
assert post.id == example_post['id']
assert post.body == example_post['body']
def test_delete_post():
channel_id = '1234'
post_id = example_post['id']
responses.add(responses.DELETE,
'%sfeed/%s/posts/%s' % (BASE_URL, channel_id, post_id),
body=json.dumps(example_post),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
post = client.channel_feed.delete_post(channel_id, post_id)
assert len(responses.calls) == 1
assert isinstance(post, Post)
assert post.id == example_post['id']
def test_create_reaction_to_post():
channel_id = '1234'
post_id = example_post['id']
response = {
'id': '24989127',
'emote_id': '25',
'user': {},
'created_at': '2016-11-29T15:51:12Z',
}
responses.add(responses.POST,
'%sfeed/%s/posts/%s/reactions' % (BASE_URL, channel_id, post_id),
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
response = client.channel_feed.create_reaction_to_post(
channel_id, post_id, response['emote_id'])
assert len(responses.calls) == 1
assert response['id']
