kopia lustrzana https://github.com/halcy/Mastodon.py
Merge branch 'master' into master
Lorenz Diener
GitHub
commit
3b5c4ae6f5
|
|
@ -2,6 +2,12 @@ A note on versioning: This librarys major version will grow with the APIs
|
|||
version number. Breaking changes will be indicated by a change in the minor
|
||||
(or major) version number, and will generally be avoided.
|
||||
|
||||
v1.8.0 (in progress)
|
||||
--------------------
|
||||
* Replace some lambdas with list comprenehsions (thanks eumiro)
|
||||
* Add `resolve` keyword to `account_search` (thanks zevaryx)
|
||||
* Add support for user agent header in `create_app` (thanks jkawamoto)
|
||||
|
||||
v1.8.0
|
||||
------
|
||||
* Overall: Support level is now 3.5.5 (last before 4.0.0)
|
||||
|
|
|
|||
17
Pipfile
17
Pipfile
|
|
@ -1,17 +0,0 @@
|
|||
[[source]]
|
||||
url = "https://pypi.org/simple"
|
||||
verify_ssl = true
|
||||
name = "pypi"
|
||||
|
||||
[packages]
|
||||
|
||||
[dev-packages]
|
||||
mastodon-py = {editable = true, extras = ["tests"], path = "."}
|
||||
pytest = "<4"
|
||||
pytest-runner = "*"
|
||||
pytest-cov = "*"
|
||||
vcrpy = "*"
|
||||
pytest-vcr = "<1"
|
||||
pytest-mock = "*"
|
||||
requests-mock = "*"
|
||||
|
||||
|
|
@ -143,7 +143,7 @@ class Mastodon(Internals):
|
|||
def me(self):
|
||||
"""
|
||||
Get this user's account. Synonym for `account_verify_credentials()`, does exactly
|
||||
the same thing, just exists becase `account_verify_credentials()` has a confusing
|
||||
the same thing, just exists because `account_verify_credentials()` has a confusing
|
||||
name.
|
||||
"""
|
||||
return self.account_verify_credentials()
|
||||
|
|
@ -244,7 +244,7 @@ class Mastodon(Internals):
|
|||
params)
|
||||
|
||||
@api_version("1.0.0", "2.3.0", _DICT_VERSION_ACCOUNT)
|
||||
def account_search(self, q, limit=None, following=False):
|
||||
def account_search(self, q, limit=None, following=False, resolve=False):
|
||||
"""
|
||||
Fetch matching accounts. Will lookup an account remotely if the search term is
|
||||
in the username@domain format and not yet in the database. Set `following` to
|
||||
|
|
|
|||
|
|
@ -43,7 +43,7 @@ class Mastodon(Internals):
|
|||
if role_ids is not None:
|
||||
if not isinstance(role_ids, list):
|
||||
role_ids = [role_ids]
|
||||
role_ids = list(map(self.__unpack_id, role_ids))
|
||||
role_ids = [self.__unpack_id(x) for x in role_ids]
|
||||
|
||||
if invited_by is not None:
|
||||
invited_by = self.__unpack_id(invited_by)
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ import collections
|
|||
|
||||
from .errors import MastodonIllegalArgumentError, MastodonNetworkError, MastodonVersionError, MastodonAPIError
|
||||
from .versions import _DICT_VERSION_APPLICATION
|
||||
from .defaults import _DEFAULT_SCOPES, _SCOPE_SETS, _DEFAULT_TIMEOUT
|
||||
from .defaults import _DEFAULT_SCOPES, _SCOPE_SETS, _DEFAULT_TIMEOUT, _DEFAULT_USER_AGENT
|
||||
from .utility import parse_version_string, api_version
|
||||
|
||||
from .internals import Mastodon as Internals
|
||||
|
|
@ -21,7 +21,7 @@ class Mastodon(Internals):
|
|||
###
|
||||
@staticmethod
|
||||
def create_app(client_name, scopes=_DEFAULT_SCOPES, redirect_uris=None, website=None, to_file=None,
|
||||
api_base_url=None, request_timeout=_DEFAULT_TIMEOUT, session=None):
|
||||
api_base_url=None, request_timeout=_DEFAULT_TIMEOUT, session=None, user_agent=_DEFAULT_USER_AGENT):
|
||||
"""
|
||||
Create a new app with given `client_name` and `scopes` (The basic scopes are "read", "write", "follow" and "push"
|
||||
- more granular scopes are available, please refer to Mastodon documentation for which) on the instance given
|
||||
|
|
@ -37,6 +37,8 @@ class Mastodon(Internals):
|
|||
Specify `session` with a requests.Session for it to be used instead of the default. This can be
|
||||
used to, amongst other things, adjust proxy or SSL certificate settings.
|
||||
|
||||
Specify `user_agent` if you want to use a specific name as `User-Agent` header, otherwise "mastodonpy" will be used.
|
||||
|
||||
Presently, app registration is open by default, but this is not guaranteed to be the case for all
|
||||
Mastodon instances in the future.
|
||||
|
||||
|
|
@ -51,21 +53,24 @@ class Mastodon(Internals):
|
|||
'client_name': client_name,
|
||||
'scopes': " ".join(scopes)
|
||||
}
|
||||
headers = {
|
||||
'User-Agent': user_agent
|
||||
}
|
||||
|
||||
if redirect_uris is not None:
|
||||
if isinstance(redirect_uris, (list, tuple)):
|
||||
redirect_uris = "\n".join(list(redirect_uris))
|
||||
request_data['redirect_uris'] = redirect_uris
|
||||
else:
|
||||
request_data['redirect_uris'] = 'urn:ietf:wg:oauth:2.0:oob'
|
||||
if website is not None:
|
||||
request_data['website'] = website
|
||||
try:
|
||||
if redirect_uris is not None:
|
||||
if isinstance(redirect_uris, (list, tuple)):
|
||||
redirect_uris = "\n".join(list(redirect_uris))
|
||||
request_data['redirect_uris'] = redirect_uris
|
||||
else:
|
||||
request_data['redirect_uris'] = 'urn:ietf:wg:oauth:2.0:oob'
|
||||
if website is not None:
|
||||
request_data['website'] = website
|
||||
if session:
|
||||
ret = session.post(f"{api_base_url}/api/v1/apps", data=request_data, timeout=request_timeout)
|
||||
ret = session.post(f"{api_base_url}/api/v1/apps", data=request_data, headers=headers, timeout=request_timeout)
|
||||
response = ret.json()
|
||||
else:
|
||||
response = requests.post(f"{api_base_url}/api/v1/apps", data=request_data, timeout=request_timeout)
|
||||
response = requests.post(f"{api_base_url}/api/v1/apps", data=request_data, headers=headers, timeout=request_timeout)
|
||||
response = response.json()
|
||||
except Exception as e:
|
||||
raise MastodonNetworkError(f"Could not complete request: {e}")
|
||||
|
|
@ -84,7 +89,7 @@ class Mastodon(Internals):
|
|||
###
|
||||
def __init__(self, client_id=None, client_secret=None, access_token=None, api_base_url=None, debug_requests=False,
|
||||
ratelimit_method="wait", ratelimit_pacefactor=1.1, request_timeout=_DEFAULT_TIMEOUT, mastodon_version=None,
|
||||
version_check_mode="created", session=None, feature_set="mainline", user_agent="mastodonpy", lang=None):
|
||||
version_check_mode="created", session=None, feature_set="mainline", user_agent=_DEFAULT_USER_AGENT, lang=None):
|
||||
"""
|
||||
Create a new API wrapper instance based on the given `client_secret` and `client_id` on the
|
||||
instance given by `api_base_url`. If you give a `client_id` and it is not a file, you must
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
_DEFAULT_TIMEOUT = 300
|
||||
_DEFAULT_STREAM_TIMEOUT = 300
|
||||
_DEFAULT_STREAM_RECONNECT_WAIT_SEC = 5
|
||||
_DEFAULT_USER_AGENT = "mastodonpy"
|
||||
_DEFAULT_SCOPES = ['read', 'write', 'follow', 'push']
|
||||
_SCOPE_SETS = {
|
||||
'read': [
|
||||
|
|
|
|||
|
|
@ -15,6 +15,11 @@ class Mastodon(Internals):
|
|||
"""
|
||||
Fetch the logged-in user's favourited statuses.
|
||||
|
||||
This endpoint uses internal ids for pagination, passing status ids to
|
||||
`max_id`, `min_id`, or `since_id` will not work. Pagination functions
|
||||
:ref:`fetch_next() <fetch_next()>`
|
||||
and :ref:`fetch_previous() <fetch_previous()>` should be used instead.
|
||||
|
||||
Returns a list of :ref:`status dicts <status dicts>`.
|
||||
"""
|
||||
if max_id is not None:
|
||||
|
|
@ -37,6 +42,11 @@ class Mastodon(Internals):
|
|||
"""
|
||||
Get a list of statuses bookmarked by the logged-in user.
|
||||
|
||||
This endpoint uses internal ids for pagination, passing status ids to
|
||||
`max_id`, `min_id`, or `since_id` will not work. Pagination functions
|
||||
:ref:`fetch_next() <fetch_next()>`
|
||||
and :ref:`fetch_previous() <fetch_previous()>` should be used instead.
|
||||
|
||||
Returns a list of :ref:`status dicts <status dicts>`.
|
||||
"""
|
||||
if max_id is not None:
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ class Mastodon(Internals):
|
|||
since_id = self.__unpack_id(since_id, dateconv=True)
|
||||
|
||||
params = self.__generate_params(locals(), ['id'])
|
||||
return self.__api_request('GET', f'/api/v1/lists/{id}/accounts')
|
||||
return self.__api_request('GET', f'/api/v1/lists/{id}/accounts', params)
|
||||
|
||||
###
|
||||
# Writing data: Lists
|
||||
|
|
@ -91,7 +91,7 @@ class Mastodon(Internals):
|
|||
|
||||
if not isinstance(account_ids, list):
|
||||
account_ids = [account_ids]
|
||||
account_ids = list(map(lambda x: self.__unpack_id(x), account_ids))
|
||||
account_ids = [self.__unpack_id(x) for x in account_ids]
|
||||
|
||||
params = self.__generate_params(locals(), ['id'])
|
||||
self.__api_request('POST', f'/api/v1/lists/{id}/accounts', params)
|
||||
|
|
@ -105,7 +105,7 @@ class Mastodon(Internals):
|
|||
|
||||
if not isinstance(account_ids, list):
|
||||
account_ids = [account_ids]
|
||||
account_ids = list(map(lambda x: self.__unpack_id(x), account_ids))
|
||||
account_ids = [self.__unpack_id(x) for x in account_ids]
|
||||
|
||||
params = self.__generate_params(locals(), ['id'])
|
||||
self.__api_request('DELETE', f'/api/v1/lists/{id}/accounts', params)
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ class Mastodon(Internals):
|
|||
if status_ids is not None:
|
||||
if not isinstance(status_ids, list):
|
||||
status_ids = [status_ids]
|
||||
status_ids = list(map(lambda x: self.__unpack_id(x), status_ids))
|
||||
status_ids = [self.__unpack_id(x) for x in status_ids]
|
||||
|
||||
params_initial = locals()
|
||||
if not forward:
|
||||
|
|
|
|||
|
|
@ -321,6 +321,8 @@ class Mastodon(Internals):
|
|||
the users that are being replied to the status text and retains
|
||||
CW and visibility if not explicitly overridden.
|
||||
|
||||
Note that `to_status` should be a :ref:`status dict <status dict>` and not an ID.
|
||||
|
||||
Set `untag` to True if you want the reply to only go to the user you
|
||||
are replying to, removing every other mentioned user from the
|
||||
conversation.
|
||||
|
|
@ -334,7 +336,10 @@ class Mastodon(Internals):
|
|||
|
||||
# Determine users to mention
|
||||
mentioned_accounts = collections.OrderedDict()
|
||||
mentioned_accounts[to_status.account.id] = to_status.account.acct
|
||||
try:
|
||||
mentioned_accounts[to_status.account.id] = to_status.account.acct
|
||||
except AttributeError as e:
|
||||
raise TypeError("to_status must specify a status dict!") from e
|
||||
|
||||
if not untag:
|
||||
for account in to_status.mentions:
|
||||
|
|
@ -342,8 +347,7 @@ class Mastodon(Internals):
|
|||
mentioned_accounts[account.id] = account.acct
|
||||
|
||||
# Join into one piece of text. The space is added inside because of self-replies.
|
||||
status = "".join(map(lambda x: "@" + x + " ",
|
||||
mentioned_accounts.values())) + status
|
||||
status = " ".join(f"@{x}" for x in mentioned_accounts.values()) + " " + status
|
||||
|
||||
# Retain visibility / cw
|
||||
if visibility is None and 'visibility' in to_status:
|
||||
|
|
|
|||
12
setup.py
12
setup.py
|
|
@ -1,3 +1,5 @@
|
|||
from pathlib import Path
|
||||
|
||||
from setuptools import setup
|
||||
|
||||
test_deps = [
|
||||
|
|
@ -26,15 +28,21 @@ extras = {
|
|||
"blurhash": blurhash_deps,
|
||||
}
|
||||
|
||||
this_directory = Path(__file__).parent
|
||||
long_description = (this_directory / "README.rst").read_text()
|
||||
|
||||
setup(name='Mastodon.py',
|
||||
version='1.8.0',
|
||||
description='Python wrapper for the Mastodon API',
|
||||
long_description=long_description,
|
||||
long_description_content_type='text/x-rst',
|
||||
packages=['mastodon'],
|
||||
install_requires=[
|
||||
'requests>=2.4.2',
|
||||
'python-dateutil',
|
||||
'six',
|
||||
'python-magic',
|
||||
'six',
|
||||
'python-magic-bin ; platform_system=="Windows"', # pragma: no cover
|
||||
'python-magic ; platform_system!="Windows"',
|
||||
'decorator>=4.0.0',
|
||||
] + blurhash_deps,
|
||||
tests_require=test_deps,
|
||||
|
|
|
|||
|
|
@ -206,14 +206,14 @@ def test_account_pin_unpin(api, api2):
|
|||
try:
|
||||
assert relationship
|
||||
assert relationship['endorsed']
|
||||
assert user["id"] in map(lambda x: x["id"], endorsed)
|
||||
assert any(x["id"] == user["id"] for x in endorsed)
|
||||
finally:
|
||||
relationship = api.account_unpin(user)
|
||||
endorsed2 = api.endorsements()
|
||||
api.account_unfollow(user)
|
||||
assert relationship
|
||||
assert not relationship['endorsed']
|
||||
assert not user["id"] in map(lambda x: x["id"], endorsed2)
|
||||
assert not any(x["id"] == user["id"] for x in endorsed2)
|
||||
|
||||
@pytest.mark.vcr()
|
||||
def test_preferences(api):
|
||||
|
|
|
|||
|
|
@ -173,7 +173,7 @@ def test_admin_domain_blocks(api2):
|
|||
assert block3.public_comment == "sicko behaviour"
|
||||
assert block3.private_comment == "jk ilu <3"
|
||||
api2.admin_delete_domain_block(block2)
|
||||
assert not block3.id in map(lambda x: x.id, api2.admin_domain_blocks())
|
||||
assert not any(x.id == block3.id for x in api2.admin_domain_blocks())
|
||||
|
||||
@pytest.mark.vcr(match_on=['path'])
|
||||
def test_admin_stats(api2):
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
from mastodon import Mastodon
|
||||
from mastodon import Mastodon, MastodonNetworkError
|
||||
import pytest
|
||||
import requests
|
||||
from requests import HTTPError
|
||||
import time
|
||||
|
||||
try:
|
||||
|
|
@ -8,7 +9,7 @@ try:
|
|||
except ImportError:
|
||||
from unittest.mock import Mock
|
||||
|
||||
def test_create_app(mocker, to_file=None, redirect_uris=None, website=None):
|
||||
def test_create_app(mocker, to_file=None, redirect_uris=None, website=None, user_agent="mastodonpy"):
|
||||
# there is no easy way to delete an anonymously created app so
|
||||
# instead we mock Requests
|
||||
resp = Mock()
|
||||
|
|
@ -22,7 +23,8 @@ def test_create_app(mocker, to_file=None, redirect_uris=None, website=None):
|
|||
api_base_url="example.com",
|
||||
to_file=to_file,
|
||||
redirect_uris=redirect_uris,
|
||||
website=website
|
||||
website=website,
|
||||
user_agent=user_agent
|
||||
)
|
||||
|
||||
assert app == ('foo', 'bar')
|
||||
|
|
@ -38,11 +40,39 @@ def test_create_app_redirect_uris(mocker):
|
|||
kwargs = requests.post.call_args[1]
|
||||
assert kwargs['data']['redirect_uris'] == 'http://example.net'
|
||||
|
||||
def test_create_app_multiple_redirect_uris(mocker):
|
||||
test_create_app(mocker, redirect_uris=['http://example.net', 'https://example.net'])
|
||||
kwargs = requests.post.call_args[1]
|
||||
assert kwargs['data']['redirect_uris'] == 'http://example.net\nhttps://example.net'
|
||||
|
||||
def test_create_app_website(mocker):
|
||||
test_create_app(mocker, website='http://example.net')
|
||||
kwargs = requests.post.call_args[1]
|
||||
assert kwargs['data']['website'] == 'http://example.net'
|
||||
|
||||
def test_create_app_session():
|
||||
resp = Mock(**{'json.return_value': {'client_id': 'foo', 'client_secret': 'bar'}})
|
||||
sess = Mock(**{'post.return_value': resp})
|
||||
|
||||
app = Mastodon.create_app("Mastodon.py test suite", api_base_url="example.com", session=sess)
|
||||
|
||||
assert app == ('foo', 'bar')
|
||||
sess.post.assert_called()
|
||||
|
||||
def test_create_app_error(mocker):
|
||||
def post(_url, **_kwargs):
|
||||
raise HTTPError("Unauthorized")
|
||||
|
||||
mocker.patch('requests.post', side_effect=post)
|
||||
|
||||
with pytest.raises(MastodonNetworkError):
|
||||
Mastodon.create_app("Mastodon.py test suite", api_base_url="example.com")
|
||||
|
||||
def test_create_app_user_agent(mocker):
|
||||
test_create_app(mocker, user_agent="pytest")
|
||||
kwargs = requests.post.call_args[1]
|
||||
assert kwargs['headers']['User-Agent'] == 'pytest'
|
||||
|
||||
@pytest.mark.vcr()
|
||||
def test_app_verify_credentials(api):
|
||||
app = api.app_verify_credentials()
|
||||
|
|
@ -54,7 +84,7 @@ def test_app_account_create():
|
|||
# This leaves behind stuff on the test server, which is unfortunate, but eh.
|
||||
suffix = str(time.time()).replace(".", "")[-5:]
|
||||
|
||||
test_app = test_app = Mastodon.create_app(
|
||||
test_app = Mastodon.create_app(
|
||||
"mastodon.py generated test app",
|
||||
api_base_url="http://localhost:3000/"
|
||||
)
|
||||
|
|
@ -74,7 +104,7 @@ def test_app_account_create():
|
|||
def test_app_account_create_invalid():
|
||||
suffix = str(time.time()).replace(".", "")[-5:]
|
||||
|
||||
test_app = test_app = Mastodon.create_app(
|
||||
test_app = Mastodon.create_app(
|
||||
"mastodon.py generated test app",
|
||||
api_base_url="http://localhost:3000/"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -58,10 +58,11 @@ def test_filter_serverside(api, api2):
|
|||
time.sleep(2)
|
||||
tl = api.timeline_home()
|
||||
try:
|
||||
assert not status_1['id'] in map(lambda st: st['id'], tl)
|
||||
assert not status_2['id'] in map(lambda st: st['id'], tl)
|
||||
assert status_3['id'] in map(lambda st: st['id'], tl)
|
||||
assert status_4['id'] in map(lambda st: st['id'], tl)
|
||||
st_ids = {st["id"] for st in tl}
|
||||
assert status_1["id"] not in st_ids
|
||||
assert status_2["id"] not in st_ids
|
||||
assert status_3["id"] in st_ids
|
||||
assert status_4["id"] in st_ids
|
||||
finally:
|
||||
api.filter_delete(keyword_filter_1)
|
||||
api.filter_delete(keyword_filter_2)
|
||||
|
|
@ -94,16 +95,18 @@ def test_filter_clientside(api, api2):
|
|||
|
||||
tl = api.timeline_home()
|
||||
try:
|
||||
assert status_1['id'] in map(lambda st: st['id'], tl)
|
||||
assert status_2['id'] in map(lambda st: st['id'], tl)
|
||||
assert status_3['id'] in map(lambda st: st['id'], tl)
|
||||
assert status_4['id'] in map(lambda st: st['id'], tl)
|
||||
st_ids = {st["id"] for st in tl}
|
||||
assert status_1['id'] in st_ids
|
||||
assert status_2['id'] in st_ids
|
||||
assert status_3['id'] in st_ids
|
||||
assert status_4['id'] in st_ids
|
||||
|
||||
filtered = api.filters_apply(tl, [keyword_filter_1, keyword_filter_2, keyword_filter_3], 'home')
|
||||
assert not status_1['id'] in map(lambda st: st['id'], filtered)
|
||||
assert not status_2['id'] in map(lambda st: st['id'], filtered)
|
||||
assert status_3['id'] in map(lambda st: st['id'], filtered)
|
||||
assert status_4['id'] in map(lambda st: st['id'], filtered)
|
||||
st_ids = {st["id"] for st in filtered}
|
||||
assert status_1['id'] not in st_ids
|
||||
assert status_2['id'] not in st_ids
|
||||
assert status_3['id'] in st_ids
|
||||
assert status_4['id'] in st_ids
|
||||
finally:
|
||||
api.filter_delete(keyword_filter_1)
|
||||
api.filter_delete(keyword_filter_2)
|
||||
|
|
|
|||
|
|
@ -24,15 +24,15 @@ def test_list_add_remove_account(api, api2, mastodon_list):
|
|||
|
||||
api.account_follow(user)
|
||||
api.list_accounts_add(mastodon_list, user)
|
||||
assert user.id in map(lambda x: x.id, api.list_accounts(mastodon_list))
|
||||
assert any(x.id == user.id for x in api.list_accounts(mastodon_list))
|
||||
|
||||
api.account_unfollow(user)
|
||||
assert len(api.list_accounts(mastodon_list)) == 0
|
||||
|
||||
api.account_follow(user)
|
||||
api.list_accounts_add(mastodon_list, user)
|
||||
assert user.id in map(lambda x: x.id, api.list_accounts(mastodon_list))
|
||||
|
||||
assert any(x.id == user.id for x in api.list_accounts(mastodon_list))
|
||||
|
||||
api.list_accounts_delete(mastodon_list, user)
|
||||
assert len(api.list_accounts(mastodon_list)) == 0
|
||||
|
||||
|
|
@ -56,9 +56,8 @@ def test_list_timeline(api, api2, mastodon_list):
|
|||
|
||||
status = api2.status_post("I have never stolen a ham in my life.", visibility="public")
|
||||
time.sleep(2)
|
||||
list_tl = list(map(lambda x: x.id, api.timeline_list(mastodon_list)))
|
||||
assert status.id in list_tl
|
||||
|
||||
assert any(x.id == status.id for x in api.timeline_list(mastodon_list))
|
||||
|
||||
api2.status_delete(status)
|
||||
api.account_unfollow(user)
|
||||
|
||||
|
|
@ -172,15 +172,15 @@ def test_scheduled_status(api):
|
|||
assert scheduled_toot_2.scheduled_at < scheduled_toot.scheduled_at
|
||||
|
||||
scheduled_toot_list = api.scheduled_statuses()
|
||||
assert scheduled_toot_2.id in map(lambda x: x.id, scheduled_toot_list)
|
||||
assert any(x.id == scheduled_toot_2.id for x in scheduled_toot_list)
|
||||
|
||||
scheduled_toot_3 = api.scheduled_status(scheduled_toot.id)
|
||||
assert scheduled_toot_2.id == scheduled_toot_3.id
|
||||
|
||||
api.scheduled_status_delete(scheduled_toot_2)
|
||||
scheduled_toot_list_2 = api.scheduled_statuses()
|
||||
assert not scheduled_toot_2.id in map(lambda x: x.id, scheduled_toot_list_2)
|
||||
|
||||
assert not any(x.id == scheduled_toot_2.id for x in scheduled_toot_list_2)
|
||||
|
||||
if os.path.exists("tests/cassettes/test_scheduled_status_datetimeobjects.pkl"):
|
||||
the_very_immediate_future = datetime.datetime.fromtimestamp(pickle.load(open("tests/cassettes/test_scheduled_status_datetimeobjects.pkl", 'rb')))
|
||||
else:
|
||||
|
|
@ -190,9 +190,9 @@ def test_scheduled_status(api):
|
|||
time.sleep(15)
|
||||
statuses = api.timeline_home()
|
||||
scheduled_toot_list_3 = api.scheduled_statuses()
|
||||
assert scheduled_toot_4.id in map(lambda x: x.id, statuses)
|
||||
assert not scheduled_toot_4.id in map(lambda x: x.id, scheduled_toot_list_3)
|
||||
|
||||
assert any(x.id == scheduled_toot_4.id for x in statuses)
|
||||
assert not any(x.id == scheduled_toot_4.id for x in scheduled_toot_list_3)
|
||||
|
||||
# The following two tests need to be manually (!) ran 10 minutes apart when recording.
|
||||
# Sorry, I can't think of a better way to test scheduled statuses actually work as intended.
|
||||
@pytest.mark.vcr(match_on=['path'])
|
||||
|
|
@ -205,7 +205,7 @@ def test_scheduled_status_long_part1(api):
|
|||
pickle.dump(the_medium_term_future.timestamp(), open("tests/cassettes_special/test_scheduled_status_long_datetimeobjects.pkl", 'wb'))
|
||||
scheduled_toot = api.status_post(f"please ensure maximum headroom at {the_medium_term_future}", scheduled_at=the_medium_term_future)
|
||||
scheduled_toot_list = api.scheduled_statuses()
|
||||
assert scheduled_toot.id in map(lambda x: x.id, scheduled_toot_list)
|
||||
assert any(x.id == scheduled_toot.id for x in scheduled_toot_list)
|
||||
pickle.dump(scheduled_toot.params.text, open("tests/cassettes_special/test_scheduled_status_long_text.pkl", 'wb'))
|
||||
|
||||
@pytest.mark.vcr(match_on=['path'])
|
||||
|
|
|
|||
|
|
@ -323,17 +323,17 @@ def test_stream_user_direct(api, api2, api3):
|
|||
conversations = []
|
||||
edits = []
|
||||
listener = CallbackStreamListener(
|
||||
update_handler = lambda x: updates.append(x),
|
||||
local_update_handler = lambda x: local_updates.append(x),
|
||||
notification_handler = lambda x: notifications.append(x),
|
||||
delete_handler = lambda x: deletes.append(x),
|
||||
conversation_handler = lambda x: conversations.append(x),
|
||||
status_update_handler = lambda x: edits.append(x),
|
||||
filters_changed_handler = lambda x: 0,
|
||||
announcement_handler = lambda x: 0,
|
||||
announcement_reaction_handler = lambda x: 0,
|
||||
announcement_delete_handler = lambda x: 0,
|
||||
encryted_message_handler = lambda x: 0,
|
||||
update_handler=updates.append,
|
||||
local_update_handler=local_updates.append,
|
||||
notification_handler=notifications.append,
|
||||
delete_handler=deletes.append,
|
||||
conversation_handler=conversations.append,
|
||||
status_update_handler=edits.append,
|
||||
filters_changed_handler=lambda x: 0,
|
||||
announcement_handler=lambda x: 0,
|
||||
announcement_reaction_handler=lambda x: 0,
|
||||
announcement_delete_handler=lambda x: 0,
|
||||
encryted_message_handler=lambda x: 0,
|
||||
)
|
||||
|
||||
posted = []
|
||||
|
|
@ -384,7 +384,7 @@ def test_stream_user_local(api, api2):
|
|||
|
||||
updates = []
|
||||
listener = CallbackStreamListener(
|
||||
local_update_handler = lambda x: updates.append(x),
|
||||
local_update_handler=updates.append,
|
||||
)
|
||||
|
||||
posted = []
|
||||
|
|
@ -412,7 +412,7 @@ def test_stream_direct(api, api2):
|
|||
|
||||
conversations = []
|
||||
listener = CallbackStreamListener(
|
||||
conversation_handler = lambda x: conversations.append(x),
|
||||
conversation_handler=conversations.append,
|
||||
)
|
||||
|
||||
def do_activities():
|
||||
|
|
|
|||
|
|
@ -9,14 +9,14 @@ import os
|
|||
def test_public_tl_anonymous(api_anonymous, status3):
|
||||
time.sleep(3)
|
||||
tl = api_anonymous.timeline_public()
|
||||
assert status3['id'] in list(map(lambda st: st['id'], tl))
|
||||
assert any(st["id"] == status3["id"] for st in tl)
|
||||
|
||||
@pytest.mark.vcr()
|
||||
def test_public_tl(api, status):
|
||||
public = api.timeline_public()
|
||||
local = api.timeline_local()
|
||||
assert status['id'] in map(lambda st: st['id'], public)
|
||||
assert status['id'] in map(lambda st: st['id'], local)
|
||||
assert any(st["id"] == status["id"] for st in public)
|
||||
assert any(st["id"] == status["id"] for st in local)
|
||||
|
||||
@pytest.mark.vcr()
|
||||
def test_unauthed_home_tl_throws(api_anonymous, status):
|
||||
|
|
@ -27,14 +27,14 @@ def test_unauthed_home_tl_throws(api_anonymous, status):
|
|||
def test_home_tl(api, status):
|
||||
time.sleep(3)
|
||||
tl = api.timeline_home()
|
||||
assert status['id'] in map(lambda st: st['id'], tl)
|
||||
assert any(st["id"] == status["id"] for st in tl)
|
||||
|
||||
@pytest.mark.vcr()
|
||||
def test_hashtag_tl(api):
|
||||
status = api.status_post('#hoot (hashtag toot)')
|
||||
tl = api.timeline_hashtag('hoot')
|
||||
try:
|
||||
assert status['id'] in map(lambda st: st['id'], tl)
|
||||
assert any(st["id"] == status["id"] for st in tl)
|
||||
finally:
|
||||
api.status_delete(status['id'])
|
||||
|
||||
|
|
@ -58,8 +58,8 @@ def test_conversations(api, api2):
|
|||
conversations2 = api2.conversations()
|
||||
api.status_delete(status)
|
||||
assert conversations
|
||||
assert status.id in map(lambda x: x.last_status.id, conversations)
|
||||
assert account.id in map(lambda x: x.accounts[0].id, conversations)
|
||||
assert any(x.last_status.id == status.id for x in conversations)
|
||||
assert any(x.accounts[0].id == account.id for x in conversations)
|
||||
assert conversations[0].unread is True
|
||||
assert conversations2[0].unread is False
|
||||
|
||||
|
|
@ -67,16 +67,16 @@ def test_conversations(api, api2):
|
|||
def test_min_max_id(api, status):
|
||||
time.sleep(3)
|
||||
tl = api.timeline_home(min_id = status.id - 1000, max_id = status.id + 1000)
|
||||
assert status['id'] in map(lambda st: st['id'], tl)
|
||||
assert any(st["id"] == status["id"] for st in tl)
|
||||
|
||||
tl = api.timeline_home(min_id = status.id - 2000, max_id = status.id - 1000)
|
||||
assert not status['id'] in map(lambda st: st['id'], tl)
|
||||
assert not any(st["id"] == status["id"] for st in tl)
|
||||
|
||||
tl = api.timeline_home(min_id = status.id + 1000, max_id = status.id + 2000)
|
||||
assert not status['id'] in map(lambda st: st['id'], tl)
|
||||
assert not any(st["id"] == status["id"] for st in tl)
|
||||
|
||||
tl = api.timeline_home(since_id = status.id - 1000)
|
||||
assert status['id'] in map(lambda st: st['id'], tl)
|
||||
assert any(st["id"] == status["id"] for st in tl)
|
||||
|
||||
@pytest.mark.vcr()
|
||||
def test_min_max_id_datetimes(api, status):
|
||||
|
|
@ -99,7 +99,7 @@ def test_min_max_id_datetimes(api, status):
|
|||
|
||||
time.sleep(3)
|
||||
tl = api.timeline_home(min_id = the_past, max_id = the_future)
|
||||
assert status['id'] in map(lambda st: st['id'], tl)
|
||||
assert any(st["id"] == status["id"] for st in tl)
|
||||
|
||||
tl = api.timeline_home(min_id = the_future, max_id = the_far_future)
|
||||
assert not status['id'] in map(lambda st: st['id'], tl)
|
||||
assert not any(st["id"] == status["id"] for st in tl)
|
||||
|
|
|
|||
9
tox.ini
9
tox.ini
|
|
@ -1,11 +1,8 @@
|
|||
[tox]
|
||||
envlist = py36,py37
|
||||
envlist = py36,py37,py38,py39,py310,py311
|
||||
skipsdist = true
|
||||
|
||||
|
||||
[testenv]
|
||||
deps = pipenv==2018.11.14
|
||||
passenv = HOME
|
||||
commands =
|
||||
pipenv sync -d
|
||||
pipenv run pytest
|
||||
deps = .[test]
|
||||
commands = python setup.py test
|
||||
|
|
|
|||
Ładowanie…
Reference in New Issue