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类calls()的实例源码
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_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_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_cached_path(self):
url = 'http://fake.datastore.com/glove.txt.gz'
set_up_glove(url, self.glove_bytes)
# non-existent file
with pytest.raises(FileNotFoundError):
filename = cached_path("tests/fixtures/does_not_exist/fake_file.tar.gz")
# unparsable URI
with pytest.raises(ValueError):
filename = cached_path("fakescheme://path/to/fake/file.tar.gz")
# existing file as path
assert cached_path(self.glove_file) == self.glove_file
# caches urls
filename = cached_path(url, cache_dir=self.TEST_DIR)
assert len(responses.calls) == 2
assert filename == os.path.join(self.TEST_DIR, url_to_filename(url, etag="0"))
with open(filename, 'rb') as cached_file:
assert cached_file.read() == self.glove_bytes
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']
def test_delete_reaction_to_post():
channel_id = '1234'
post_id = example_post['id']
body = {
'deleted': True
}
responses.add(responses.DELETE,
'%sfeed/%s/posts/%s/reactions' % (BASE_URL, channel_id, post_id),
body=json.dumps(body),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
response = client.channel_feed.delete_reaction_to_post(channel_id, post_id, '1234')
assert len(responses.calls) == 1
assert response['deleted']
def test_get_post_comments():
channel_id = '1234'
post_id = example_post['id']
response = {
'_cursor': '1480651694954867000',
'_total': 1,
'comments': [example_comment]
}
responses.add(responses.GET,
'%sfeed/%s/posts/%s/comments' % (BASE_URL, channel_id, post_id),
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
comments = client.channel_feed.get_post_comments(channel_id, post_id)
assert len(responses.calls) == 1
assert len(comments) == 1
comment = comments[0]
assert isinstance(comment, Comment)
assert comment.id == example_comment['id']
assert comment.body == example_comment['body']
def test_delete_post_comment():
channel_id = '1234'
post_id = example_post['id']
comment_id = example_comment['id']
responses.add(responses.DELETE,
'%sfeed/%s/posts/%s/comments/%s' % (BASE_URL, channel_id, post_id, comment_id),
body=json.dumps(example_comment),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
comment = client.channel_feed.delete_post_comment(channel_id, post_id, comment_id)
assert len(responses.calls) == 1
assert isinstance(comment, Comment)
assert comment.id == example_comment['id']
def test_create_reaction_to_comment():
channel_id = '1234'
post_id = example_post['id']
comment_id = example_comment['id']
body = {
'created_at': '2016-12-02T04:26:47Z',
'emote_id': '1',
'id': '1341393b-e872-4554-9f6f-acd5f8b669fc',
'user': {}
}
url = '%sfeed/%s/posts/%s/comments/%s/reactions' % (BASE_URL, channel_id, post_id, comment_id)
responses.add(responses.POST,
url,
body=json.dumps(body),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
response = client.channel_feed.create_reaction_to_comment(channel_id, post_id, comment_id, '1')
assert len(responses.calls) == 1
assert response['id']
def test_delete_reaction_to_comment():
channel_id = '1234'
post_id = example_post['id']
comment_id = example_comment['id']
body = {
'deleted': True
}
url = '%sfeed/%s/posts/%s/comments/%s/reactions' % (BASE_URL, channel_id, post_id, comment_id)
responses.add(responses.DELETE,
url,
body=json.dumps(body),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
response = client.channel_feed.delete_reaction_to_comment(channel_id, post_id, comment_id, '1')
assert len(responses.calls) == 1
assert response['deleted']
def test_get_top():
responses.add(responses.GET,
'%sgames/top' % BASE_URL,
body=json.dumps(example_top_games_response),
status=200,
content_type='application/json')
client = TwitchClient('abcd')
games = client.games.get_top()
assert len(responses.calls) == 1
assert len(games) == 1
assert isinstance(games[0], TopGame)
game = games[0].game
assert isinstance(game, Game)
assert game.id == example_top_games_response['top'][0]['game']['_id']
def test_update():
channel_id = example_channel['_id']
responses.add(responses.PUT,
'%schannels/%s' % (BASE_URL, channel_id),
body=json.dumps(example_channel),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
status = 'Spongebob Squarepants'
channel = client.channels.update(channel_id, status=status)
assert len(responses.calls) == 1
expected_body = json.dumps({'channel': {'status': status}}).encode('utf-8')
assert responses.calls[0].request.body == expected_body
assert isinstance(channel, Channel)
assert channel.id == channel_id
assert channel.name == example_channel['name']
def test_get_editors():
channel_id = example_channel['_id']
response = {
'users': [example_user]
}
responses.add(responses.GET,
'%schannels/%s/editors' % (BASE_URL, channel_id),
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id', 'oauth token')
users = client.channels.get_editors(channel_id)
assert len(responses.calls) == 1
assert len(users) == 1
user = users[0]
assert isinstance(user, User)
assert user.id == example_user['_id']
assert user.name == example_user['name']
def test_get_followers():
channel_id = example_channel['_id']
response = {
'_cursor': '1481675542963907000',
'_total': 41,
'follows': [example_follower]
}
responses.add(responses.GET,
'%schannels/%s/follows' % (BASE_URL, channel_id),
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
followers = client.channels.get_followers(channel_id)
assert len(responses.calls) == 1
assert len(followers) == 1
follow = followers[0]
assert isinstance(follow, Follow)
assert follow.notifications == example_follower['notifications']
assert isinstance(follow.user, User)
assert follow.user.id == example_user['_id']
assert follow.user.name == example_user['name']
def test_get_teams():
channel_id = example_channel['_id']
response = {
'teams': [example_team]
}
responses.add(responses.GET,
'%schannels/%s/teams' % (BASE_URL, channel_id),
body=json.dumps(response),
status=200,
content_type='application/json')
client = TwitchClient('client id')
teams = client.channels.get_teams(channel_id)
assert len(responses.calls) == 1
assert len(teams) == 1
team = teams[0]
assert isinstance(team, Team)
assert team.id == example_team['_id']
assert team.name == example_team['name']
