spla/Mastodonplus.py
1
0
Fork 0
Codi Incidències Sol·licituds Pull repo.project Llançaments Paquets Wiki Activitat

Merge pull request #80 from foxmask/master

Explorar la Font 
not pep8 compliant - issue 71
This commit is contained in:
Lorenz Diener 2017-09-08 13:45:00 +02:00 cometido por GitHub
commit d9471c2d7d
S'han modificat 6 arxius amb 229 adicions i 179 eliminacions
Mostrar estadístiques Descarregar el fitxer de pedaç Descarregar el fitxer de diferències Expand all files Collapse all files

4
docs/conf.py
Veure arxiu

@ -30,7 +30,7 @@ import os
import sys import sys
sys.path.insert(0, os.path.abspath('../')) sys.path.insert(0, os.path.abspath('../'))
autodoc_member_order = 'by_source' autodoc_member_order = 'by_source'
#print(sys.path) # print(sys.path)
# Add any Sphinx extension module names here, as strings. They can be # Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
@ -154,7 +154,7 @@ todo_include_todos = False
# html_logo = None # html_logo = None
# The name of an image file (relative to this directory) to use as a favicon of # The name of an image file (relative to this directory) to use as a favicon of
# the docs. This file should be a Windows icon file (.ico) being 16x16 or 32x32 # the docs. This file should be a Windows icon file (.ico) being 16x16 or 32x32
# pixels large. # pixels large.
# #
# html_favicon = None # html_favicon = None

351
mastodon/Mastodon.py
Veure arxiu

@ -6,7 +6,6 @@ import mimetypes
import time import time
import random import random
import string import string
import pytz
import datetime import datetime
from contextlib import closing from contextlib import closing
import pytz import pytz
@ -17,6 +16,7 @@ import dateutil.parser
import re import re
import copy import copy
class Mastodon: class Mastodon:
""" """
Super basic but thorough and easy to use Mastodon Super basic but thorough and easy to use Mastodon
@ -28,12 +28,12 @@ class Mastodon:
__DEFAULT_BASE_URL = 'https://mastodon.social' __DEFAULT_BASE_URL = 'https://mastodon.social'
__DEFAULT_TIMEOUT = 300 __DEFAULT_TIMEOUT = 300
### ###
# Registering apps # Registering apps
### ###
@staticmethod @staticmethod
def create_app(client_name, scopes = ['read', 'write', 'follow'], redirect_uris = None, website = None, to_file = None, api_base_url = __DEFAULT_BASE_URL, request_timeout = __DEFAULT_TIMEOUT): def create_app(client_name, scopes=['read', 'write', 'follow'], redirect_uris=None, website=None, to_file=None,
api_base_url=__DEFAULT_BASE_URL, request_timeout=__DEFAULT_TIMEOUT):
""" """
Create a new app with given client_name and scopes (read, write, follow) Create a new app with given client_name and scopes (read, write, follow)
@ -47,7 +47,7 @@ class Mastodon:
Returns client_id and client_secret. Returns client_id and client_secret.
""" """
api_base_url = Mastodon.__protocolize(api_base_url) api_base_url = Mastodon.__protocolize(api_base_url)
request_data = { request_data = {
'client_name': client_name, 'client_name': client_name,
'scopes': " ".join(scopes) 'scopes': " ".join(scopes)
@ -55,18 +55,18 @@ class Mastodon:
try: try:
if redirect_uris is not None: if redirect_uris is not None:
request_data['redirect_uris'] = redirect_uris; request_data['redirect_uris'] = redirect_uris
else: else:
request_data['redirect_uris'] = 'urn:ietf:wg:oauth:2.0:oob'; request_data['redirect_uris'] = 'urn:ietf:wg:oauth:2.0:oob'
if website is not None: if website is not None:
request_data['website'] = website request_data['website'] = website
response = requests.post(api_base_url + '/api/v1/apps', data = request_data, timeout = request_timeout) response = requests.post(api_base_url + '/api/v1/apps', data=request_data, timeout=request_timeout)
response = response.json() response = response.json()
except Exception as e: except Exception as e:
raise MastodonNetworkError("Could not complete request: %s" % e) raise MastodonNetworkError("Could not complete request: %s" % e)
if to_file != None: if to_file is not None:
with open(to_file, 'w') as secret_file: with open(to_file, 'w') as secret_file:
secret_file.write(response['client_id'] + '\n') secret_file.write(response['client_id'] + '\n')
secret_file.write(response['client_secret'] + '\n') secret_file.write(response['client_secret'] + '\n')
@ -76,7 +76,10 @@ class Mastodon:
### ###
# Authentication, including constructor # Authentication, including constructor
### ###
def __init__(self, client_id, client_secret = None, access_token = None, api_base_url = __DEFAULT_BASE_URL, debug_requests = False, ratelimit_method = "wait", ratelimit_pacefactor = 1.1, request_timeout = __DEFAULT_TIMEOUT): def __init__(self, client_id, client_secret=None, access_token=None,
api_base_url=__DEFAULT_BASE_URL, debug_requests=False,
ratelimit_method="wait", ratelimit_pacefactor=1.1,
request_timeout=__DEFAULT_TIMEOUT):
""" """
Create a new API wrapper instance based on the given client_secret and client_id. If you Create a new API wrapper instance based on the given client_secret and client_id. If you
give a client_id and it is not a file, you must also give a secret. give a client_id and it is not a file, you must also give a secret.
@ -115,7 +118,7 @@ class Mastodon:
self.request_timeout = request_timeout self.request_timeout = request_timeout
if not ratelimit_method in ["throw", "wait", "pace"]: if ratelimit_method not in ["throw", "wait", "pace"]:
raise MastodonIllegalArgumentError("Invalid ratelimit method.") raise MastodonIllegalArgumentError("Invalid ratelimit method.")
if os.path.isfile(self.client_id): if os.path.isfile(self.client_id):
@ -123,15 +126,15 @@ class Mastodon:
self.client_id = secret_file.readline().rstrip() self.client_id = secret_file.readline().rstrip()
self.client_secret = secret_file.readline().rstrip() self.client_secret = secret_file.readline().rstrip()
else: else:
if self.client_secret == None: if self.client_secret is None:
raise MastodonIllegalArgumentError('Specified client id directly, but did not supply secret') raise MastodonIllegalArgumentError('Specified client id directly, but did not supply secret')
if self.access_token != None and os.path.isfile(self.access_token): if self.access_token is not None and os.path.isfile(self.access_token):
with open(self.access_token, 'r') as token_file: with open(self.access_token, 'r') as token_file:
self.access_token = token_file.readline().rstrip() self.access_token = token_file.readline().rstrip()
def auth_request_url(self, client_id = None, redirect_uris = "urn:ietf:wg:oauth:2.0:oob", scopes = ['read', 'write', 'follow']): def auth_request_url(self, client_id=None, redirect_uris="urn:ietf:wg:oauth:2.0:oob",
scopes=['read', 'write', 'follow']):
"""Returns the url that a client needs to request the grant from the server. """Returns the url that a client needs to request the grant from the server.
""" """
if client_id is None: if client_id is None:
@ -140,8 +143,8 @@ class Mastodon:
if os.path.isfile(client_id): if os.path.isfile(client_id):
with open(client_id, 'r') as secret_file: with open(client_id, 'r') as secret_file:
client_id = secret_file.readline().rstrip() client_id = secret_file.readline().rstrip()
params = {} params = dict()
params['client_id'] = client_id params['client_id'] = client_id
params['response_type'] = "code" params['response_type'] = "code"
params['redirect_uri'] = redirect_uris params['redirect_uri'] = redirect_uris
@ -149,22 +152,23 @@ class Mastodon:
formatted_params = urlencode(params) formatted_params = urlencode(params)
return "".join([self.api_base_url, "/oauth/authorize?", formatted_params]) return "".join([self.api_base_url, "/oauth/authorize?", formatted_params])
def log_in(self, username = None, password = None,\ def log_in(self, username=None, password=None,
code = None, redirect_uri = "urn:ietf:wg:oauth:2.0:oob", refresh_token = None,\ code=None, redirect_uri="urn:ietf:wg:oauth:2.0:oob", refresh_token=None,
scopes = ['read', 'write', 'follow'], to_file = None): scopes=['read', 'write', 'follow'], to_file=None):
""" """
Your username is the e-mail you use to log in into mastodon. Your username is the e-mail you use to log in into mastodon.
Can persist access token to file, to be used in the constructor. Can persist access token to file, to be used in the constructor.
Supports refresh_token but Mastodon.social doesn't implement it at the moment. Supports refresh_token but Mastodon.social doesn't implement it at the moment.
Handles password, authorization_code, and refresh_token authentication. Handles password, authorization_code, and refresh_token authentication.
Will throw a MastodonIllegalArgumentError if username / password Will throw a MastodonIllegalArgumentError if username / password
are wrong, scopes are not valid or granted scopes differ from requested. are wrong, scopes are not valid or granted scopes differ from requested.
For OAuth2 documentation, compare https://github.com/doorkeeper-gem/doorkeeper/wiki/Interacting-as-an-OAuth-client-with-Doorkeeper For OAuth2 documentation, compare
https://github.com/doorkeeper-gem/doorkeeper/wiki/Interacting-as-an-OAuth-client-with-Doorkeeper
Returns the access token. Returns the access token.
""" """
@ -179,13 +183,13 @@ class Mastodon:
params['grant_type'] = 'refresh_token' params['grant_type'] = 'refresh_token'
else: else:
raise MastodonIllegalArgumentError('Invalid arguments given. username and password or code are required.') raise MastodonIllegalArgumentError('Invalid arguments given. username and password or code are required.')
params['client_id'] = self.client_id params['client_id'] = self.client_id
params['client_secret'] = self.client_secret params['client_secret'] = self.client_secret
params['scope'] = " ".join(scopes) params['scope'] = " ".join(scopes)
try: try:
response = self.__api_request('POST', '/oauth/token', params, do_ratelimiting = False) response = self.__api_request('POST', '/oauth/token', params, do_ratelimiting=False)
self.access_token = response['access_token'] self.access_token = response['access_token']
self.__set_refresh_token(response.get('refresh_token')) self.__set_refresh_token(response.get('refresh_token'))
self.__set_token_expired(int(response.get('expires_in', 0))) self.__set_token_expired(int(response.get('expires_in', 0)))
@ -201,9 +205,10 @@ class Mastodon:
received_scopes = " ".join(sorted(response["scope"].split(" "))) received_scopes = " ".join(sorted(response["scope"].split(" ")))
if requested_scopes != received_scopes: if requested_scopes != received_scopes:
raise MastodonAPIError('Granted scopes "' + received_scopes + '" differ from requested scopes "' + requested_scopes + '".') raise MastodonAPIError(
'Granted scopes "' + received_scopes + '" differ from requested scopes "' + requested_scopes + '".')
if to_file != None: if to_file is not None:
with open(to_file, 'w') as token_file: with open(to_file, 'w') as token_file:
token_file.write(response['access_token'] + '\n') token_file.write(response['access_token'] + '\n')
@ -223,7 +228,7 @@ class Mastodon:
### ###
# Reading data: Timelines # Reading data: Timelines
## ##
def timeline(self, timeline = "home", max_id = None, since_id = None, limit = None): def timeline(self, timeline="home", max_id=None, since_id=None, limit=None):
""" """
Fetch statuses, most recent ones first. Timeline can be home, local, public, Fetch statuses, most recent ones first. Timeline can be home, local, public,
or tag/hashtag. See the following functions documentation for what those do. or tag/hashtag. See the following functions documentation for what those do.
@ -239,39 +244,44 @@ class Mastodon:
params_initial['local'] = True params_initial['local'] = True
params = self.__generate_params(params_initial, ['timeline']) params = self.__generate_params(params_initial, ['timeline'])
return self.__api_request('GET', '/api/v1/timelines/' + timeline, params) url = '/api/v1/timelines/{0}'.format(timeline)
return self.__api_request('GET', url, params)
def timeline_home(self, max_id = None, since_id = None, limit = None): def timeline_home(self, max_id=None, since_id=None, limit=None):
""" """
Fetch the authenticated users home timeline (i.e. followed users and self). Fetch the authenticated users home timeline (i.e. followed users and self).
Returns a list of toot dicts. Returns a list of toot dicts.
""" """
return self.timeline('home', max_id = max_id, since_id = since_id, limit = limit) return self.timeline('home', max_id=max_id, since_id=since_id,
limit=limit)
def timeline_local(self, max_id = None, since_id = None, limit = None): def timeline_local(self, max_id=None, since_id=None, limit=None):
""" """
Fetches the local / instance-wide timeline, not including replies. Fetches the local / instance-wide timeline, not including replies.
Returns a list of toot dicts. Returns a list of toot dicts.
""" """
return self.timeline('local', max_id = max_id, since_id = since_id, limit = limit) return self.timeline('local', max_id=max_id, since_id=since_id,
limit=limit)
def timeline_public(self, max_id = None, since_id = None, limit = None): def timeline_public(self, max_id=None, since_id=None, limit=None):
""" """
Fetches the public / visible-network timeline, not including replies. Fetches the public / visible-network timeline, not including replies.
Returns a list of toot dicts. Returns a list of toot dicts.
""" """
return self.timeline('public', max_id = max_id, since_id = since_id, limit = limit) return self.timeline('public', max_id=max_id, since_id=since_id,
limit=limit)
def timeline_hashtag(self, hashtag, max_id = None, since_id = None, limit = None): def timeline_hashtag(self, hashtag, max_id=None, since_id=None, limit=None):
""" """
Fetch a timeline of toots with a given hashtag. Fetch a timeline of toots with a given hashtag.
Returns a list of toot dicts. Returns a list of toot dicts.
""" """
return self.timeline('tag/' + str(hashtag), max_id = max_id, since_id = since_id, limit = limit) url = 'tag/{0}'.format(str(hashtag))
return self.timeline(url, max_id=max_id, since_id=since_id, limit=limit)
### ###
# Reading data: Statuses # Reading data: Statuses
@ -282,7 +292,8 @@ class Mastodon:
Returns a toot dict. Returns a toot dict.
""" """
return self.__api_request('GET', '/api/v1/statuses/' + str(id)) url = '/api/v1/statuses/{0}'.format(str(id))
return self.__api_request('GET', url)
def status_card(self, id): def status_card(self, id):
""" """
@ -291,7 +302,8 @@ class Mastodon:
Returns a card dict. Returns a card dict.
""" """
return self.__api_request('GET', '/api/v1/statuses/' + str(id) + '/card') url = '/api/v1/statuses/{0}/card'.format(str(id))
return self.__api_request('GET', url)
def status_context(self, id): def status_context(self, id):
""" """
@ -299,7 +311,8 @@ class Mastodon:
Returns a context dict. Returns a context dict.
""" """
return self.__api_request('GET', '/api/v1/statuses/' + str(id) + '/context') url = '/api/v1/statuses/{0}/context'.format(str(id))
return self.__api_request('GET', url)
def status_reblogged_by(self, id): def status_reblogged_by(self, id):
""" """
@ -307,7 +320,8 @@ class Mastodon:
Returns a list of user dicts. Returns a list of user dicts.
""" """
return self.__api_request('GET', '/api/v1/statuses/' + str(id) + '/reblogged_by') url = '/api/v1/statuses/{0}/reblogged_by'.format(str(id))
return self.__api_request('GET', url)
def status_favourited_by(self, id): def status_favourited_by(self, id):
""" """
@ -315,12 +329,13 @@ class Mastodon:
Returns a list of user dicts. Returns a list of user dicts.
""" """
return self.__api_request('GET', '/api/v1/statuses/' + str(id) + '/favourited_by') url = '/api/v1/statuses/{0}/favourited_by'.format(str(id))
return self.__api_request('GET', url)
### ###
# Reading data: Notifications # Reading data: Notifications
### ###
def notifications(self, id = None, max_id = None, since_id = None, limit = None): def notifications(self, id=None, max_id=None, since_id=None, limit=None):
""" """
Fetch notifications (mentions, favourites, reblogs, follows) for the authenticated Fetch notifications (mentions, favourites, reblogs, follows) for the authenticated
user. user.
@ -329,11 +344,12 @@ class Mastodon:
Returns a list of notification dicts. Returns a list of notification dicts.
""" """
if id == None: if id is None:
params = self.__generate_params(locals(), ['id']) params = self.__generate_params(locals(), ['id'])
return self.__api_request('GET', '/api/v1/notifications', params) return self.__api_request('GET', '/api/v1/notifications', params)
else: else:
return self.__api_request('GET', '/api/v1/notifications/' + str(id)) url = '/api/v1/notifications/{0}'.format(str(id))
return self.__api_request('GET', url)
### ###
# Reading data: Accounts # Reading data: Accounts
@ -344,7 +360,8 @@ class Mastodon:
Returns a user dict. Returns a user dict.
""" """
return self.__api_request('GET', '/api/v1/accounts/' + str(id)) url = '/api/v1/accounts/{0}'.format(str(id))
return self.__api_request('GET', url)
def account_verify_credentials(self): def account_verify_credentials(self):
""" """
@ -354,32 +371,35 @@ class Mastodon:
""" """
return self.__api_request('GET', '/api/v1/accounts/verify_credentials') return self.__api_request('GET', '/api/v1/accounts/verify_credentials')
def account_statuses(self, id, max_id = None, since_id = None, limit = None): def account_statuses(self, id, max_id=None, since_id=None, limit=None):
""" """
Fetch statuses by user id. Same options as timeline are permitted. Fetch statuses by user id. Same options as timeline are permitted.
Returns a list of toot dicts. Returns a list of toot dicts.
""" """
params = self.__generate_params(locals(), ['id']) params = self.__generate_params(locals(), ['id'])
return self.__api_request('GET', '/api/v1/accounts/' + str(id) + '/statuses', params) url = '/api/v1/accounts/{0}/statuses'.format(str(id))
return self.__api_request('GET', url, params)
def account_following(self, id, max_id = None, since_id = None, limit = None): def account_following(self, id, max_id=None, since_id=None, limit=None):
""" """
Fetch users the given user is following. Fetch users the given user is following.
Returns a list of user dicts. Returns a list of user dicts.
""" """
params = self.__generate_params(locals(), ['id']) params = self.__generate_params(locals(), ['id'])
return self.__api_request('GET', '/api/v1/accounts/' + str(id) + '/following', params) url = '/api/v1/accounts/{0}/following'.format(str(id))
return self.__api_request('GET', url, params)
def account_followers(self, id, max_id = None, since_id = None, limit = None): def account_followers(self, id, max_id=None, since_id=None, limit=None):
""" """
Fetch users the given user is followed by. Fetch users the given user is followed by.
Returns a list of user dicts. Returns a list of user dicts.
""" """
params = self.__generate_params(locals(), ['id']) params = self.__generate_params(locals(), ['id'])
return self.__api_request('GET', '/api/v1/accounts/' + str(id) + '/followers', params) url = '/api/v1/accounts/{0}/followers'.format(str(id))
return self.__api_request('GET', url, params)
def account_relationships(self, id): def account_relationships(self, id):
""" """
@ -389,9 +409,10 @@ class Mastodon:
Returns a list of relationship dicts. Returns a list of relationship dicts.
""" """
params = self.__generate_params(locals()) params = self.__generate_params(locals())
return self.__api_request('GET', '/api/v1/accounts/relationships', params) return self.__api_request('GET', '/api/v1/accounts/relationships',
params)
def account_search(self, q, limit = None): def account_search(self, q, limit=None):
""" """
Fetch matching accounts. Will lookup an account remotely if the search term is Fetch matching accounts. Will lookup an account remotely if the search term is
in the username@domain format and not yet in the database. in the username@domain format and not yet in the database.
@ -404,7 +425,7 @@ class Mastodon:
### ###
# Reading data: Searching # Reading data: Searching
### ###
def search(self, q, resolve = False): def search(self, q, resolve=False):
""" """
Fetch matching hashtags, accounts and statuses. Will search federated Fetch matching hashtags, accounts and statuses. Will search federated
instances if resolve is True. instances if resolve is True.
@ -417,7 +438,7 @@ class Mastodon:
### ###
# Reading data: Mutes and Blocks # Reading data: Mutes and Blocks
### ###
def mutes(self, max_id = None, since_id = None, limit = None): def mutes(self, max_id=None, since_id=None, limit=None):
""" """
Fetch a list of users muted by the authenticated user. Fetch a list of users muted by the authenticated user.
@ -426,7 +447,7 @@ class Mastodon:
params = self.__generate_params(locals()) params = self.__generate_params(locals())
return self.__api_request('GET', '/api/v1/mutes', params) return self.__api_request('GET', '/api/v1/mutes', params)
def blocks(self, max_id = None, since_id = None, limit = None): def blocks(self, max_id=None, since_id=None, limit=None):
""" """
Fetch a list of users blocked by the authenticated user. Fetch a list of users blocked by the authenticated user.
@ -449,7 +470,7 @@ class Mastodon:
### ###
# Reading data: Favourites # Reading data: Favourites
### ###
def favourites(self, max_id = None, since_id = None, limit = None): def favourites(self, max_id=None, since_id=None, limit=None):
""" """
Fetch the authenticated user's favourited statuses. Fetch the authenticated user's favourited statuses.
@ -461,7 +482,7 @@ class Mastodon:
### ###
# Reading data: Follow requests # Reading data: Follow requests
### ###
def follow_requests(self, max_id = None, since_id = None, limit = None): def follow_requests(self, max_id=None, since_id=None, limit=None):
""" """
Fetch the authenticated user's incoming follow requests. Fetch the authenticated user's incoming follow requests.
@ -473,7 +494,7 @@ class Mastodon:
### ###
# Reading data: Domain blocks # Reading data: Domain blocks
### ###
def domain_blocks(self, max_id = None, since_id = None, limit = None): def domain_blocks(self, max_id=None, since_id=None, limit=None):
""" """
Fetch the authenticated user's blocked domains. Fetch the authenticated user's blocked domains.
@ -481,11 +502,12 @@ class Mastodon:
""" """
params = self.__generate_params(locals()) params = self.__generate_params(locals())
return self.__api_request('GET', '/api/v1/domain_blocks', params) return self.__api_request('GET', '/api/v1/domain_blocks', params)
### ###
# Writing data: Statuses # Writing data: Statuses
### ###
def status_post(self, status, in_reply_to_id = None, media_ids = None, sensitive = False, visibility = '', spoiler_text = None): def status_post(self, status, in_reply_to_id=None, media_ids=None,
sensitive=False, visibility='', spoiler_text=None):
""" """
Post a status. Can optionally be in reply to another status and contain Post a status. Can optionally be in reply to another status and contain
up to four pieces of media (Uploaded via media_post()). media_ids can up to four pieces of media (Uploaded via media_post()). media_ids can
@ -517,12 +539,13 @@ class Mastodon:
# Validate visibility parameter # Validate visibility parameter
valid_visibilities = ['private', 'public', 'unlisted', 'direct', ''] valid_visibilities = ['private', 'public', 'unlisted', 'direct', '']
if params_initial['visibility'].lower() not in valid_visibilities: if params_initial['visibility'].lower() not in valid_visibilities:
raise ValueError('Invalid visibility value! Acceptable values are %s' % valid_visibilities) raise ValueError('Invalid visibility value! Acceptable '
'values are %s' % valid_visibilities)
if params_initial['sensitive'] == False: if params_initial['sensitive'] is False:
del[params_initial['sensitive']] del [params_initial['sensitive']]
if media_ids != None: if media_ids is not None:
try: try:
media_ids_proper = [] media_ids_proper = []
for media_id in media_ids: for media_id in media_ids:
@ -531,7 +554,8 @@ class Mastodon:
else: else:
media_ids_proper.append(media_id) media_ids_proper.append(media_id)
except Exception as e: except Exception as e:
raise MastodonIllegalArgumentError("Invalid media dict: %s" % e) raise MastodonIllegalArgumentError("Invalid media "
"dict: %s" % e)
params_initial["media_ids"] = media_ids_proper params_initial["media_ids"] = media_ids_proper
@ -552,14 +576,16 @@ class Mastodon:
Returns an empty dict for good measure. Returns an empty dict for good measure.
""" """
return self.__api_request('DELETE', '/api/v1/statuses/' + str(id)) url = '/api/v1/statuses/{0}'.format(str(id))
return self.__api_request('DELETE', url)
def status_reblog(self, id): def status_reblog(self, id):
"""Reblog a status. """Reblog a status.
Returns a toot with with a new status that wraps around the reblogged one. Returns a toot with with a new status that wraps around the reblogged one.
""" """
return self.__api_request('POST', '/api/v1/statuses/' + str(id) + "/reblog") url = '/api/v1/statuses/{0}/reblog'.format(str(id))
return self.__api_request('POST', url)
def status_unreblog(self, id): def status_unreblog(self, id):
""" """
@ -567,7 +593,8 @@ class Mastodon:
Returns a toot dict with the status that used to be reblogged. Returns a toot dict with the status that used to be reblogged.
""" """
return self.__api_request('POST', '/api/v1/statuses/' + str(id) + "/unreblog") url = '/api/v1/statuses/{0}/unreblog'.format(str(id))
return self.__api_request('POST', url)
def status_favourite(self, id): def status_favourite(self, id):
""" """
@ -575,7 +602,8 @@ class Mastodon:
Returns a toot dict with the favourited status. Returns a toot dict with the favourited status.
""" """
return self.__api_request('POST', '/api/v1/statuses/' + str(id) + "/favourite") url = '/api/v1/statuses/{0}/favourite'.format(str(id))
return self.__api_request('POST', url)
def status_unfavourite(self, id): def status_unfavourite(self, id):
""" """
@ -583,7 +611,8 @@ class Mastodon:
Returns a toot dict with the un-favourited status. Returns a toot dict with the un-favourited status.
""" """
return self.__api_request('POST', '/api/v1/statuses/' + str(id) + "/unfavourite") url = '/api/v1/statuses/{0}/unfavourite'.format(str(id))
return self.__api_request('POST', url)
### ###
# Writing data: Notifications # Writing data: Notifications
@ -603,7 +632,8 @@ class Mastodon:
Returns a relationship dict containing the updated relationship to the user. Returns a relationship dict containing the updated relationship to the user.
""" """
return self.__api_request('POST', '/api/v1/accounts/' + str(id) + "/follow") url = '/api/v1/accounts/{0}/follow'.format(str(id))
return self.__api_request('POST', url)
def follows(self, uri): def follows(self, uri):
""" """
@ -620,7 +650,8 @@ class Mastodon:
Returns a relationship dict containing the updated relationship to the user. Returns a relationship dict containing the updated relationship to the user.
""" """
return self.__api_request('POST', '/api/v1/accounts/' + str(id) + "/unfollow") url = '/api/v1/accounts/{0}/unfollow'.format(str(id))
return self.__api_request('POST', url)
def account_block(self, id): def account_block(self, id):
""" """
@ -628,7 +659,8 @@ class Mastodon:
Returns a relationship dict containing the updated relationship to the user. Returns a relationship dict containing the updated relationship to the user.
""" """
return self.__api_request('POST', '/api/v1/accounts/' + str(id) + "/block") url = '/api/v1/accounts/{0}/block'.format(str(id))
return self.__api_request('POST', url)
def account_unblock(self, id): def account_unblock(self, id):
""" """
@ -636,7 +668,8 @@ class Mastodon:
Returns a relationship dict containing the updated relationship to the user. Returns a relationship dict containing the updated relationship to the user.
""" """
return self.__api_request('POST', '/api/v1/accounts/' + str(id) + "/unblock") url = '/api/v1/accounts/{0}/unblock'.format(str(id))
return self.__api_request('POST', url)
def account_mute(self, id): def account_mute(self, id):
""" """
@ -644,7 +677,8 @@ class Mastodon:
Returns a relationship dict containing the updated relationship to the user. Returns a relationship dict containing the updated relationship to the user.
""" """
return self.__api_request('POST', '/api/v1/accounts/' + str(id) + "/mute") url = '/api/v1/accounts/{0}/mute'.format(str(id))
return self.__api_request('POST', url)
def account_unmute(self, id): def account_unmute(self, id):
""" """
@ -652,9 +686,11 @@ class Mastodon:
Returns a relationship dict containing the updated relationship to the user. Returns a relationship dict containing the updated relationship to the user.
""" """
return self.__api_request('POST', '/api/v1/accounts/' + str(id) + "/unmute") url = '/api/v1/accounts/{0}/unmute'.format(str(id))
return self.__api_request('POST', url)
def account_update_credentials(self, display_name = None, note = None, avatar = None, header = None): def account_update_credentials(self, display_name=None, note=None,
avatar=None, header=None):
""" """
Update the profile for the currently authenticated user. Update the profile for the currently authenticated user.
@ -689,7 +725,8 @@ class Mastodon:
Returns an empty dict. Returns an empty dict.
""" """
return self.__api_request('POST', '/api/v1/follow_requests/' + str(id) + "/authorize") url = '/api/v1/follow_requests/{0}/authorize'.format(str(id))
return self.__api_request('POST', url)
def follow_request_reject(self, id): def follow_request_reject(self, id):
""" """
@ -697,12 +734,13 @@ class Mastodon:
Returns an empty dict. Returns an empty dict.
""" """
return self.__api_request('POST', '/api/v1/follow_requests/' + str(id) + "/reject") url = '/api/v1/follow_requests/{0}/reject'.format(str(id))
return self.__api_request('POST', url)
### ###
# Writing data: Media # Writing data: Media
### ###
def media_post(self, media_file, mime_type = None): def media_post(self, media_file, mime_type=None):
""" """
Post an image. media_file can either be image data or Post an image. media_file can either be image data or
a file name. If image data is passed directly, the mime a file name. If image data is passed directly, the mime
@ -715,45 +753,49 @@ class Mastodon:
Returns a media dict. This contains the id that can be used in Returns a media dict. This contains the id that can be used in
status_post to attach the media file to a toot. status_post to attach the media file to a toot.
""" """
if mime_type == None and os.path.isfile(media_file): if mime_type is None and os.path.isfile(media_file):
mime_type = mimetypes.guess_type(media_file)[0] mime_type = mimetypes.guess_type(media_file)[0]
media_file = open(media_file, 'rb') media_file = open(media_file, 'rb')
if mime_type == None: if mime_type is None:
raise MastodonIllegalArgumentError('Could not determine mime type or data passed directly without mime type.') raise MastodonIllegalArgumentError('Could not determine mime type'
' or data passed directly '
'without mime type.')
random_suffix = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(10)) random_suffix = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(10))
file_name = "mastodonpyupload_" + str(time.time()) + "_" + str(random_suffix) + mimetypes.guess_extension(mime_type) file_name = "mastodonpyupload_" + str(time.time()) + "_" + str(random_suffix) + mimetypes.guess_extension(
mime_type)
media_file_description = (file_name, media_file, mime_type) media_file_description = (file_name, media_file, mime_type)
return self.__api_request('POST', '/api/v1/media', files = {'file': media_file_description}) return self.__api_request('POST', '/api/v1/media',
files={'file': media_file_description})
### ###
# Writing data: Domain blocks # Writing data: Domain blocks
### ###
def domain_block(self, domain = None): def domain_block(self, domain=None):
""" """
Add a block for all statuses originating from the specified domain for the logged-in user. Add a block for all statuses originating from the specified domain for the logged-in user.
""" """
params = self.__generate_params(locals()) params = self.__generate_params(locals())
return self.__api_request('POST', '/api/v1/domain_blocks', params) return self.__api_request('POST', '/api/v1/domain_blocks', params)
def domain_unblock(self, domain = None): def domain_unblock(self, domain=None):
""" """
Remove a domain block for the logged-in user. Remove a domain block for the logged-in user.
""" """
params = self.__generate_params(locals()) params = self.__generate_params(locals())
return self.__api_request('DELETE', '/api/v1/domain_blocks', params) return self.__api_request('DELETE', '/api/v1/domain_blocks', params)
### ###
# Pagination # Pagination
### ###
def fetch_next(self, previous_page): def fetch_next(self, previous_page):
""" """
Fetches the next page of results of a paginated request. Pass in the Fetches the next page of results of a paginated request. Pass in the
previous page in its entirety, or the pagination information dict previous page in its entirety, or the pagination information dict
returned as a part of that pages last status ('_pagination_next'). returned as a part of that pages last status ('_pagination_next').
Returns the next page or None if no further data is available. Returns the next page or None if no further data is available.
""" """
if isinstance(previous_page, list): if isinstance(previous_page, list):
@ -763,21 +805,21 @@ class Mastodon:
return None return None
else: else:
params = copy.deepcopy(previous_page) params = copy.deepcopy(previous_page)
method = params['_pagination_method'] method = params['_pagination_method']
del params['_pagination_method'] del params['_pagination_method']
endpoint = params['_pagination_endpoint'] endpoint = params['_pagination_endpoint']
del params['_pagination_endpoint'] del params['_pagination_endpoint']
return self.__api_request(method, endpoint, params) return self.__api_request(method, endpoint, params)
def fetch_previous(self, next_page): def fetch_previous(self, next_page):
""" """
Fetches the previous page of results of a paginated request. Pass in the Fetches the previous page of results of a paginated request. Pass in the
previous page in its entirety, or the pagination information dict previous page in its entirety, or the pagination information dict
returned as a part of that pages first status ('_pagination_prev'). returned as a part of that pages first status ('_pagination_prev').
Returns the previous page or None if no further data is available. Returns the previous page or None if no further data is available.
""" """
if isinstance(next_page, list): if isinstance(next_page, list):
@ -787,34 +829,34 @@ class Mastodon:
return None return None
else: else:
params = copy.deepcopy(next_page) params = copy.deepcopy(next_page)
method = params['_pagination_method'] method = params['_pagination_method']
del params['_pagination_method'] del params['_pagination_method']
endpoint = params['_pagination_endpoint'] endpoint = params['_pagination_endpoint']
del params['_pagination_endpoint'] del params['_pagination_endpoint']
return self.__api_request(method, endpoint, params) return self.__api_request(method, endpoint, params)
def fetch_remaining(self, first_page): def fetch_remaining(self, first_page):
""" """
Fetches all the remaining pages of a paginated request starting from a Fetches all the remaining pages of a paginated request starting from a
first page and returns the entire set of results (including the first page first page and returns the entire set of results (including the first page
that was passed in) as a big list. that was passed in) as a big list.
Be careful, as this might generate a lot of requests, depending on what you are Be careful, as this might generate a lot of requests, depending on what you are
fetching, and might cause you to run into rate limits very quickly. fetching, and might cause you to run into rate limits very quickly.
""" """
first_page = copy.deepcopy(first_page) first_page = copy.deepcopy(first_page)
all_pages = [] all_pages = []
current_page = first_page current_page = first_page
while current_page != None and len(current_page) > 0: while current_page is not None and len(current_page) > 0:
all_pages.extend(current_page) all_pages.extend(current_page)
current_page = self.fetch_next(current_page) current_page = self.fetch_next(current_page)
return all_pages return all_pages
### ###
# Streaming # Streaming
### ###
@ -858,7 +900,7 @@ class Mastodon:
incoming events. incoming events.
""" """
return self.__stream("/api/v1/streaming/hashtag?tag={}".format(tag), listener) return self.__stream("/api/v1/streaming/hashtag?tag={}".format(tag), listener)
### ###
# Internal helpers, dragons probably # Internal helpers, dragons probably
### ###
@ -870,22 +912,22 @@ class Mastodon:
Assumes UTC if timezone is not given. Assumes UTC if timezone is not given.
""" """
date_time_utc = None date_time_utc = None
if date_time.tzinfo == None: if date_time.tzinfo is None:
date_time_utc = date_time.replace(tzinfo = pytz.utc) date_time_utc = date_time.replace(tzinfo=pytz.utc)
else: else:
date_time_utc = date_time.astimezone(pytz.utc) date_time_utc = date_time.astimezone(pytz.utc)
epoch_utc = datetime.datetime.utcfromtimestamp(0).replace(tzinfo = pytz.utc) epoch_utc = datetime.datetime.utcfromtimestamp(0).replace(tzinfo=pytz.utc)
return (date_time_utc - epoch_utc).total_seconds() return (date_time_utc - epoch_utc).total_seconds()
def __api_request(self, method, endpoint, params = {}, files = {}, do_ratelimiting = True): def __api_request(self, method, endpoint, params={}, files={}, do_ratelimiting=True):
""" """
Internal API request helper. Internal API request helper.
""" """
response = None response = None
headers = None headers = None
remaining_wait = 0
# "pace" mode ratelimiting: Assume constant rate of requests, sleep a little less long than it # "pace" mode ratelimiting: Assume constant rate of requests, sleep a little less long than it
# would take to not hit the rate limit at that request rate. # would take to not hit the rate limit at that request rate.
if do_ratelimiting and self.ratelimit_method == "pace": if do_ratelimiting and self.ratelimit_method == "pace":
@ -906,10 +948,10 @@ class Mastodon:
time.sleep(to_next) time.sleep(to_next)
# Generate request headers # Generate request headers
if self.access_token != None: if self.access_token is not None:
headers = {'Authorization': 'Bearer ' + self.access_token} headers = {'Authorization': 'Bearer ' + self.access_token}
if self.debug_requests == True: if self.debug_requests:
print('Mastodon: Request to endpoint "' + endpoint + '" using method "' + method + '".') print('Mastodon: Request to endpoint "' + endpoint + '" using method "' + method + '".')
print('Parameters: ' + str(params)) print('Parameters: ' + str(params))
print('Headers: ' + str(headers)) print('Headers: ' + str(headers))
@ -923,24 +965,28 @@ class Mastodon:
response_object = None response_object = None
try: try:
if method == 'GET': if method == 'GET':
response_object = requests.get(self.api_base_url + endpoint, params = params, headers = headers, files = files, timeout = self.request_timeout) response_object = requests.get(self.api_base_url + endpoint, params=params,
headers=headers, files=files,
timeout=self.request_timeout)
if method == 'POST': if method == 'POST':
response_object = requests.post(self.api_base_url + endpoint, data = params, headers = headers, files = files, timeout = self.request_timeout) response_object = requests.post(self.api_base_url + endpoint, data=params, headers=headers,
files=files, timeout=self.request_timeout)
if method == 'PATCH': if method == 'PATCH':
response_object = requests.patch(self.api_base_url + endpoint, data = params, headers = headers, files = files, timeout = self.request_timeout) response_object = requests.patch(self.api_base_url + endpoint, data=params, headers=headers,
files=files, timeout=self.request_timeout)
if method == 'DELETE': if method == 'DELETE':
response_object = requests.delete(self.api_base_url + endpoint, data = params, headers = headers, files = files, timeout = self.request_timeout) response_object = requests.delete(self.api_base_url + endpoint, data=params, headers=headers,
files=files, timeout=self.request_timeout)
except Exception as e: except Exception as e:
raise MastodonNetworkError("Could not complete request: %s" % e) raise MastodonNetworkError("Could not complete request: %s" % e)
if response_object == None: if response_object is None:
raise MastodonIllegalArgumentError("Illegal request.") raise MastodonIllegalArgumentError("Illegal request.")
# Handle response # Handle response
if self.debug_requests == True: if self.debug_requests:
print('Mastodon: Response received with code ' + str(response_object.status_code) + '.') print('Mastodon: Response received with code ' + str(response_object.status_code) + '.')
print('response headers: ' + str(response_object.headers)) print('response headers: ' + str(response_object.headers))
print('Response text content: ' + str(response_object.text)) print('Response text content: ' + str(response_object.text))
@ -954,20 +1000,26 @@ class Mastodon:
try: try:
response = response_object.json() response = response_object.json()
except: except:
raise MastodonAPIError("Could not parse response as JSON, response code was %s, bad json content was '%s'" % (response_object.status_code, response_object.content)) raise MastodonAPIError(
"Could not parse response as JSON, response code was %s, "
"bad json content was '%s'" % (response_object.status_code,
response_object.content))
# Parse link headers # Parse link headers
if isinstance(response, list) and 'Link' in response_object.headers and response_object.headers['Link'] != "": if isinstance(response, list) and \
tmp_urls = requests.utils.parse_header_links(response_object.headers['Link'].rstrip('>').replace('>,<', ',<')) 'Link' in response_object.headers and \
response_object.headers['Link'] != "":
tmp_urls = requests.utils.parse_header_links(
response_object.headers['Link'].rstrip('>').replace('>,<', ',<'))
for url in tmp_urls: for url in tmp_urls:
if not 'rel' in url: if 'rel' not in url:
continue continue
if url['rel'] == 'next': if url['rel'] == 'next':
# Be paranoid and extract max_id specifically # Be paranoid and extract max_id specifically
next_url = url['url'] next_url = url['url']
matchgroups = re.search(r"max_id=([0-9]*)", next_url) matchgroups = re.search(r"max_id=([0-9]*)", next_url)
if matchgroups: if matchgroups:
next_params = copy.deepcopy(params) next_params = copy.deepcopy(params)
next_params['_pagination_method'] = method next_params['_pagination_method'] = method
@ -976,12 +1028,12 @@ class Mastodon:
if "since_id" in next_params: if "since_id" in next_params:
del next_params['since_id'] del next_params['since_id']
response[-1]['_pagination_next'] = next_params response[-1]['_pagination_next'] = next_params
if url['rel'] == 'prev': if url['rel'] == 'prev':
# Be paranoid and extract since_id specifically # Be paranoid and extract since_id specifically
prev_url = url['url'] prev_url = url['url']
matchgroups = re.search(r"since_id=([0-9]*)", prev_url) matchgroups = re.search(r"since_id=([0-9]*)", prev_url)
if matchgroups: if matchgroups:
prev_params = copy.deepcopy(params) prev_params = copy.deepcopy(params)
prev_params['_pagination_method'] = method prev_params['_pagination_method'] = method
@ -990,7 +1042,7 @@ class Mastodon:
if "max_id" in prev_params: if "max_id" in prev_params:
del prev_params['max_id'] del prev_params['max_id']
response[0]['_pagination_prev'] = prev_params response[0]['_pagination_prev'] = prev_params
# Handle rate limiting # Handle rate limiting
if 'X-RateLimit-Remaining' in response_object.headers and do_ratelimiting: if 'X-RateLimit-Remaining' in response_object.headers and do_ratelimiting:
self.ratelimit_remaining = int(response_object.headers['X-RateLimit-Remaining']) self.ratelimit_remaining = int(response_object.headers['X-RateLimit-Remaining'])
@ -1024,21 +1076,20 @@ class Mastodon:
return response return response
def __stream(self, endpoint, listener, params = {}): def __stream(self, endpoint, listener, params={}):
""" """
Internal streaming API helper. Internal streaming API helper.
""" """
headers = {} headers = {}
if self.access_token != None: if self.access_token is not None:
headers = {'Authorization': 'Bearer ' + self.access_token} headers = {'Authorization': 'Bearer ' + self.access_token}
url = self.api_base_url + endpoint url = self.api_base_url + endpoint
with closing(requests.get(url, headers = headers, data = params, stream = True)) as r: with closing(requests.get(url, headers=headers, data=params, stream=True)) as r:
listener.handle_stream(r.iter_lines()) listener.handle_stream(r.iter_lines())
def __generate_params(self, params, exclude=[]):
def __generate_params(self, params, exclude = []):
""" """
Internal named-parameters-to-dict helper. Internal named-parameters-to-dict helper.
@ -1052,7 +1103,7 @@ class Mastodon:
del params['self'] del params['self']
param_keys = list(params.keys()) param_keys = list(params.keys())
for key in param_keys: for key in param_keys:
if params[key] == None or key in exclude: if params[key] is None or key in exclude:
del params[key] del params[key]
param_keys = list(params.keys()) param_keys = list(params.keys())
@ -1063,28 +1114,24 @@ class Mastodon:
return params return params
def __get_token_expired(self): def __get_token_expired(self):
"""Internal helper for oauth code""" """Internal helper for oauth code"""
if self._token_expired < datetime.datetime.now(): return self._token_expired < datetime.datetime.now()
return True
else:
return False
def __set_token_expired(self, value): def __set_token_expired(self, value):
"""Internal helper for oauth code""" """Internal helper for oauth code"""
self._token_expired = datetime.datetime.now() + datetime.timedelta(seconds=value) self._token_expired = datetime.datetime.now() + datetime.timedelta(seconds=value)
return return
def __get_refresh_token(self): def __get_refresh_token(self):
"""Internal helper for oauth code""" """Internal helper for oauth code"""
return self._refresh_token return self._refresh_token
def __set_refresh_token(self, value): def __set_refresh_token(self, value):
"""Internal helper for oauth code""" """Internal helper for oauth code"""
self._refresh_token = value self._refresh_token = value
return return
@staticmethod @staticmethod
def __protocolize(base_url): def __protocolize(base_url):
"""Internal add-protocol-to-url helper""" """Internal add-protocol-to-url helper"""
@ -1095,21 +1142,25 @@ class Mastodon:
base_url = base_url.rstrip("/") base_url = base_url.rstrip("/")
return base_url return base_url
## ##
# Exceptions # Exceptions
## ##
class MastodonIllegalArgumentError(ValueError): class MastodonIllegalArgumentError(ValueError):
pass pass
class MastodonFileNotFoundError(IOError): class MastodonFileNotFoundError(IOError):
pass pass
class MastodonNetworkError(IOError): class MastodonNetworkError(IOError):
pass pass
class MastodonAPIError(Exception): class MastodonAPIError(Exception):
pass pass
class MastodonRatelimitError(Exception): class MastodonRatelimitError(Exception):
pass pass

29
mastodon/streaming.py
Veure arxiu

@ -1,7 +1,7 @@
''' """
Handlers for the Streaming API: Handlers for the Streaming API:
https://github.com/tootsuite/mastodon/blob/master/docs/Using-the-API/Streaming-API.md https://github.com/tootsuite/mastodon/blob/master/docs/Using-the-API/Streaming-API.md
''' """
import json import json
import logging import logging
@ -12,43 +12,43 @@ log = logging.getLogger(__name__)
class MalformedEventError(Exception): class MalformedEventError(Exception):
'''Raised when the server-sent event stream is malformed.''' """Raised when the server-sent event stream is malformed."""
pass pass
class StreamListener(object): class StreamListener(object):
'''Callbacks for the streaming API. Create a subclass, override the on_xxx """Callbacks for the streaming API. Create a subclass, override the on_xxx
methods for the kinds of events you're interested in, then pass an instance methods for the kinds of events you're interested in, then pass an instance
of your subclass to Mastodon.user_stream(), Mastodon.public_stream(), or of your subclass to Mastodon.user_stream(), Mastodon.public_stream(), or
Mastodon.hashtag_stream().''' Mastodon.hashtag_stream()."""
def on_update(self, status): def on_update(self, status):
'''A new status has appeared! 'status' is the parsed JSON dictionary """A new status has appeared! 'status' is the parsed JSON dictionary
describing the status.''' describing the status."""
pass pass
def on_notification(self, notification): def on_notification(self, notification):
'''A new notification. 'notification' is the parsed JSON dictionary """A new notification. 'notification' is the parsed JSON dictionary
describing the notification.''' describing the notification."""
pass pass
def on_delete(self, status_id): def on_delete(self, status_id):
'''A status has been deleted. status_id is the status' integer ID.''' """A status has been deleted. status_id is the status' integer ID."""
pass pass
def handle_heartbeat(self): def handle_heartbeat(self):
'''The server has sent us a keep-alive message. This callback may be """The server has sent us a keep-alive message. This callback may be
useful to carry out periodic housekeeping tasks, or just to confirm useful to carry out periodic housekeeping tasks, or just to confirm
that the connection is still open.''' that the connection is still open."""
def handle_stream(self, lines): def handle_stream(self, lines):
''' """
Handles a stream of events from the Mastodon server. When each event Handles a stream of events from the Mastodon server. When each event
is received, the corresponding .on_[name]() method is called. is received, the corresponding .on_[name]() method is called.
lines: an iterable of lines of bytes sent by the Mastodon server, as lines: an iterable of lines of bytes sent by the Mastodon server, as
returned by requests.Response.iter_lines(). returned by requests.Response.iter_lines().
''' """
event = {} event = {}
for raw_line in lines: for raw_line in lines:
try: try:
@ -104,4 +104,3 @@ class StreamListener(object):
else: else:
# TODO: allow handlers to return/raise to stop streaming cleanly # TODO: allow handlers to return/raise to stop streaming cleanly
handler(payload) handler(payload)

7
setup.py
Veure arxiu

@ -1,4 +1,4 @@
from setuptools import setup, find_packages from setuptools import setup
setup(name='Mastodon.py', setup(name='Mastodon.py',
version='1.0.8', version='1.0.8',
@ -6,7 +6,7 @@ setup(name='Mastodon.py',
packages=['mastodon'], packages=['mastodon'],
setup_requires=['pytest-runner'], setup_requires=['pytest-runner'],
tests_require=['pytest'], tests_require=['pytest'],
install_requires=['requests', 'python-dateutil', 'six'], install_requires=['requests', 'python-dateutil', 'six', 'pytz'],
url='https://github.com/halcy/Mastodon.py', url='https://github.com/halcy/Mastodon.py',
author='Lorenz Diener', author='Lorenz Diener',
author_email='lorenzd+mastodonpypypi@gmail.com', author_email='lorenzd+mastodonpypypi@gmail.com',
@ -19,5 +19,4 @@ setup(name='Mastodon.py',
'License :: OSI Approved :: MIT License', 'License :: OSI Approved :: MIT License',
'Programming Language :: Python :: 2', 'Programming Language :: Python :: 2',
'Programming Language :: Python :: 3', 'Programming Language :: Python :: 3',
] ])
)

0
tests/__init__.py Normal file
Veure arxiu

17
tests/test_streaming.py
Veure arxiu

@ -24,9 +24,10 @@ class Listener(StreamListener):
self.heartbeats += 1 self.heartbeats += 1
def handle_stream_(self, lines): def handle_stream_(self, lines):
'''Test helper to avoid littering all tests with six.b().''' """Test helper to avoid littering all tests with six.b()."""
return self.handle_stream(map(six.b, lines)) return self.handle_stream(map(six.b, lines))
def test_heartbeat(): def test_heartbeat():
listener = Listener() listener = Listener()
listener.handle_stream_([':one', ':two']) listener.handle_stream_([':one', ':two'])
@ -85,7 +86,7 @@ def test_many(events):
def test_unknown_event(): def test_unknown_event():
'''Be tolerant of new event types''' """Be tolerant of new event types"""
listener = Listener() listener = Listener()
listener.handle_stream_([ listener.handle_stream_([
'event: blahblah', 'event: blahblah',
@ -137,11 +138,11 @@ def test_sse_order_doesnt_matter():
def test_extra_keys_ignored(): def test_extra_keys_ignored():
''' """
https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Event_stream_format https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Event_stream_format
defines 'id' and 'retry' keys which the Mastodon streaming API doesn't use, defines 'id' and 'retry' keys which the Mastodon streaming API doesn't use,
and alleges that "All other field names are ignored". and alleges that "All other field names are ignored".
''' """
listener = Listener() listener = Listener()
listener.handle_stream_([ listener.handle_stream_([
'event: update', 'event: update',
@ -155,7 +156,7 @@ def test_extra_keys_ignored():
def test_valid_utf8(): def test_valid_utf8():
'''Snowman Cat Face With Tears Of Joy''' """Snowman Cat Face With Tears Of Joy"""
listener = Listener() listener = Listener()
listener.handle_stream_([ listener.handle_stream_([
'event: update', 'event: update',
@ -166,7 +167,7 @@ def test_valid_utf8():
def test_invalid_utf8(): def test_invalid_utf8():
'''Cat Face With Tears O''' """Cat Face With Tears O"""
listener = Listener() listener = Listener()
with pytest.raises(MalformedEventError): with pytest.raises(MalformedEventError):
listener.handle_stream_([ listener.handle_stream_([
@ -177,13 +178,13 @@ def test_invalid_utf8():
def test_multiline_payload(): def test_multiline_payload():
''' """
https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Data-only_messages https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Data-only_messages
says that newlines in the 'data' field can be encoded by sending the field says that newlines in the 'data' field can be encoded by sending the field
twice! This would be really pathological for Mastodon because the payload twice! This would be really pathological for Mastodon because the payload
is JSON, but technically literal newlines are permissible (outside strings) is JSON, but technically literal newlines are permissible (outside strings)
so let's handle this case. so let's handle this case.
''' """
listener = Listener() listener = Listener()
listener.handle_stream_([ listener.handle_stream_([
'event: update', 'event: update',