Mastodon
/
ephemetoot-eliminare-vecchi-messaggi
mirror of https://github.com/hughrun/ephemetoot
1
0
Fork 0
Code Issues Releases Wiki Activity

format with black

This commit is contained in:
Hugh Rundle 2020-09-08 20:05:10 +10:00
parent d52fb4f408
commit 4905e0543d
3 changed files with 294 additions and 215 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
ephemetoot/console.py
View File

@ -126,9 +126,9 @@ else:
def main():
'''
"""
Call ephemetoot.check_toots() on each user in the config file, with options set via flags from command line.
'''
"""
try:
if options.init:

191
ephemetoot/ephemetoot.py
View File

@ -19,6 +19,7 @@ import requests
# local
from ephemetoot import plist
def compulsory_input(tags, name, example):
value = ""
@ -30,6 +31,7 @@ def compulsory_input(tags, name, example):
return value
def digit_input(tags, name, example):
value = ""
@ -41,23 +43,24 @@ def digit_input(tags, name, example):
return value
def yes_no_input(tags, name):
value = ""
while value not in ["y", "n"]:
value = input(
tags[0] + name + tags[1] + "(y or n):" + tags[2]
)
value = input(tags[0] + name + tags[1] + "(y or n):" + tags[2])
return_val = "true" if value == "y" else "false"
return return_val
def optional_input(tags, name, example):
value = input(tags[0] + name + tags[1] + example + tags[2])
return value
def init():
'''
"""
Creates a config.yaml file in the current directory, based on user input.
'''
"""
try:
# text colour markers (beginning, example, end)
@ -65,23 +68,35 @@ def init():
print("\nCreate your config.yaml file.\n")
print(
"For help checkout the docs at ",
tags[0],
"ephemetoot.hugh.run",
tags[2],
"\n",
sep=""
)
"For help check out the docs at ",
tags[0],
"ephemetoot.hugh.run",
tags[2],
"\n",
sep="",
)
conf_token = compulsory_input(tags, "Access token: ", None)
conf_user = compulsory_input(tags, "Username", "(without the '@' - e.g. alice):")
conf_user = compulsory_input(
tags, "Username", "(without the '@' - e.g. alice):"
)
conf_url = compulsory_input(tags, "Base URL", "(e.g. example.social):")
conf_days = digit_input(tags, "Days to keep", "(default 365):")
conf_pinned = yes_no_input(tags, "Keep pinned toots?")
conf_keep_toots = optional_input(tags, "Toots to keep", "(optional list of IDs separated by commas):")
conf_keep_hashtags = optional_input(tags, "Hashtags to keep", "(optional list without '#' e.g. mastodon, gardening, cats):")
conf_keep_visibility = optional_input(tags, "Visibility to keep", "(optional list e.g. 'direct'):")
conf_archive = optional_input(tags, "Archive path", "(optional filepath for archive):")
conf_keep_toots = optional_input(
tags, "Toots to keep", "(optional list of IDs separated by commas):"
)
conf_keep_hashtags = optional_input(
tags,
"Hashtags to keep",
"(optional list without '#' e.g. mastodon, gardening, cats):",
)
conf_keep_visibility = optional_input(
tags, "Visibility to keep", "(optional list e.g. 'direct'):"
)
conf_archive = optional_input(
tags, "Archive path", "(optional filepath for archive):"
)
# write out the config file
with open("config.yaml", "w") as configfile:
@ -119,10 +134,11 @@ def init():
except Exception as e:
print(e)
def version(vnum):
'''
"""
Prints current and latest version numbers to console.
'''
"""
try:
latest = requests.get(
@ -141,11 +157,12 @@ def version(vnum):
except Exception as e:
print("Something went wrong:", e)
def schedule(options):
'''
"""
Creates and loads a plist file for scheduled running with launchd. If --time flag is used, the scheduled time is set accordingly. Note that this is designed for use on MacOS.
'''
"""
try:
if options.schedule == ".":
@ -167,8 +184,8 @@ def schedule(options):
# write out file directly to ~/Library/LaunchAgents
f = open(
os.path.join(
os.path.expanduser("~/Library/LaunchAgents"),
"ephemetoot.scheduler.plist"
os.path.expanduser("~/Library/LaunchAgents"),
"ephemetoot.scheduler.plist",
),
mode="w",
)
@ -197,6 +214,7 @@ def schedule(options):
if options.verbose:
print(e)
def archive_toot(config, toot):
# define archive path
if config["archive"][0] == "~":
@ -215,19 +233,19 @@ def archive_toot(config, toot):
f.write(json.dumps(toot, indent=4, default=jsondefault))
f.close()
def jsondefault(obj):
if isinstance(obj, (date, datetime)):
return obj.isoformat()
def tooted_date(toot):
return toot.created_at.strftime("%d %b %Y")
def datestamp_now():
return str(
datetime.now(timezone.utc).strftime(
"%a %d %b %Y %H:%M:%S %z"
)
)
return str(datetime.now(timezone.utc).strftime("%a %d %b %Y %H:%M:%S %z"))
def console_print(msg, options, skip):
@ -239,6 +257,7 @@ def console_print(msg, options, skip):
print(msg)
def print_rate_limit_message(mastodon):
now = time.time()
@ -249,17 +268,16 @@ def print_rate_limit_message(mastodon):
datestamp_now(),
"- next reset due in",
str(format(diff / 60, ".0f")),
"minutes.\n"
"minutes.\n",
)
def retry_on_error(options, mastodon, toot, attempts):
if attempts < 6:
try:
console_print(
"Attempt " + str(attempts) + " at " + datestamp_now(),
options,
False
"Attempt " + str(attempts) + " at " + datestamp_now(), options, False
)
mastodon.status_delete(toot)
except:
@ -269,6 +287,7 @@ def retry_on_error(options, mastodon, toot, attempts):
else:
raise TimeoutError("Gave up after 5 attempts")
def process_toot(config, options, mastodon, deleted_count, toot):
keep_pinned = "keep_pinned" in config and config["keep_pinned"]
@ -294,39 +313,32 @@ def process_toot(config, options, mastodon, deleted_count, toot):
try:
if keep_pinned and hasattr(toot, "pinned") and toot.pinned:
console_print(
"📌 skipping pinned toot - " + str(toot.id),
options,
True
)
console_print("📌 skipping pinned toot - " + str(toot.id), options, True)
elif toot.id in toots_to_keep:
console_print(
"💾 skipping saved toot - " + str(toot.id),
options,
True
)
console_print("💾 skipping saved toot - " + str(toot.id), options, True)
elif toot.visibility in visibility_to_keep:
console_print(
"👀 skipping " + toot.visibility + " toot - " + str(toot.id),
options,
True
"👀 skipping " + toot.visibility + " toot - " + str(toot.id),
options,
True,
)
elif len(hashtags_to_keep.intersection(toot_tags)) > 0:
console_print(
"#️⃣ skipping toot with hashtag - " + str(toot.id),
options,
True
"#️⃣ skipping toot with hashtag - " + str(toot.id), options, True
)
elif cutoff_date > toot.created_at:
if hasattr(toot, "reblog") and toot.reblog:
console_print(
"👎 unboosting toot " + str(toot.id) + " boosted " + tooted_date(toot),
options,
False
console_print(
"👎 unboosting toot "
+ str(toot.id)
+ " boosted "
+ tooted_date(toot),
options,
False,
)
deleted_count += 1
@ -334,13 +346,13 @@ def process_toot(config, options, mastodon, deleted_count, toot):
if not options.test:
if mastodon.ratelimit_remaining == 0:
console_print(
"Rate limit reached. Waiting for a rate limit reset",
options,
False
"Rate limit reached. Waiting for a rate limit reset",
options,
False,
)
# check for --archive-deleted
if (options.archive_deleted and "id" in toot and "archive" in config):
if options.archive_deleted and "id" in toot and "archive" in config:
# write toot to archive
archive_toot(config, toot)
@ -348,21 +360,23 @@ def process_toot(config, options, mastodon, deleted_count, toot):
else:
console_print(
"❌ deleting toot " + str(toot.id) + " tooted " + tooted_date(toot),
options,
False
"❌ deleting toot " + str(toot.id) + " tooted " + tooted_date(toot),
options,
False,
)
deleted_count += 1
time.sleep(2) # wait 2 secs between deletes to be a bit nicer to the server
time.sleep(
2
) # wait 2 secs between deletes to be a bit nicer to the server
if not options.test:
# deal with rate limits
if (mastodon.ratelimit_remaining == 0 and not options.quiet):
if mastodon.ratelimit_remaining == 0 and not options.quiet:
print_rate_limit_message(mastodon)
# check for --archive-deleted
if (options.archive_deleted and "id" in toot and "archive" in config):
if options.archive_deleted and "id" in toot and "archive" in config:
archive_toot(config, toot)
# finally we actually delete the toot
@ -383,16 +397,24 @@ def process_toot(config, options, mastodon, deleted_count, toot):
if options.verbose:
print("🛑 ERROR deleting toot -", str(toot.id), "\n", e)
else:
print( "🛑 ERROR deleting toot -", str(toot.id), "-", str(e.args[0]), "-", str(e.args[3]) )
print(
"🛑 ERROR deleting toot -",
str(toot.id),
"-",
str(e.args[0]),
"-",
str(e.args[3]),
)
console_print(
"Waiting " + str(options.retry_mins) + " minutes before re-trying",
options,
False
"Waiting " + str(options.retry_mins) + " minutes before re-trying",
options,
False,
)
time.sleep(60 * options.retry_mins)
retry_on_error(options, mastodon, toot, attempts=2)
def check_batch(config, options, mastodon, user_id, timeline, deleted_count=0):
"""
Check a batch of up to 40 toots. This is usually triggered by check_toots, and then recursively calls itself until all toots within the time period specified have been checked.
@ -413,7 +435,7 @@ def check_batch(config, options, mastodon, user_id, timeline, deleted_count=0):
else:
if not options.test:
if options.datestamp:
print( "\n", datestamp_now(), end=" : ")
print("\n", datestamp_now(), end=" : ")
print("Removed " + str(deleted_count) + " toots.\n")
@ -426,9 +448,13 @@ def check_batch(config, options, mastodon, user_id, timeline, deleted_count=0):
if options.quiet:
if options.datestamp:
print( "\n", datestamp_now(), sep="", end=" : ")
print("\n", datestamp_now(), sep="", end=" : ")
print("Test run completed. This would have removed", str(deleted_count), "toots.\n")
print(
"Test run completed. This would have removed",
str(deleted_count),
"toots.\n",
)
else:
print("---------------------------------------")
@ -439,12 +465,19 @@ def check_batch(config, options, mastodon, user_id, timeline, deleted_count=0):
except IndexError:
print("No toots found!\n")
def check_toots(config, options, retry_count=0):
'''
"""
The main function, uses the Mastodon API to check all toots in the user timeline, and delete any that do not meet any of the exclusion criteria from the config file.
'''
"""
try:
print("Fetching account details for @", config["username"], "@", config["base_url"], sep="")
print(
"Fetching account details for @",
config["username"],
"@",
config["base_url"],
sep="",
)
if options.pace:
mastodon = Mastodon(
@ -459,16 +492,16 @@ def check_toots(config, options, retry_count=0):
ratelimit_method="wait",
)
user_id = mastodon.account_verify_credentials().id # verify user and get ID
account = mastodon.account(user_id) # get the account
timeline = mastodon.account_statuses(user_id, limit=40) # initial batch
user_id = mastodon.account_verify_credentials().id # verify user and get ID
account = mastodon.account(user_id) # get the account
timeline = mastodon.account_statuses(user_id, limit=40) # initial batch
if not options.quiet:
print("Checking", str(account.statuses_count), "toots")
# check first batch
# check first batch
# check_batch() then recursively keeps looping until all toots have been checked
check_batch(config, options, mastodon, user_id, timeline)
check_batch(config, options, mastodon, user_id, timeline)
except KeyboardInterrupt:
print("Operation aborted.")
@ -508,4 +541,4 @@ def check_toots(config, options, retry_count=0):
if options.verbose:
print("ERROR:", e)
else:
print("ERROR:", str(e.args[0]), "\n")
print("ERROR:", str(e.args[0]), "\n")

314
tests/test_ephemetoot.py
View File

@ -17,112 +17,108 @@ from ephemetoot import ephemetoot
########################
toot_dict = {
'id': 104136090490756999,
'created_at': datetime.datetime(2020, 5, 9, 2, 17, 18, 598000, tzinfo=tzutc()),
'in_reply_to_id': None,
'in_reply_to_account_id': None,
'sensitive': False,
'spoiler_text': '',
'visibility': 'public',
'language': 'en',
'uri': 'https://example.social/users/testbot/statuses/104136090490756503',
'url': 'https://example.social/@testbot/104136090490756503',
'replies_count': 0,
'reblogs_count': 0,
'favourites_count': 0,
'favourited': False,
'reblogged': False,
'muted': False,
'bookmarked': False,
'pinned': True,
'content': '<p>hello I am testing</p>',
'reblog': None,
'application': None,
'account': {
'id': 16186,
'username': 'testbot',
'acct': 'testbot',
'display_name': 'ephemtoot Testing Bot',
'locked': True,
'bot': True,
'discoverable': False,
'group': False,
'created_at': datetime.datetime(2018, 11, 16, 23, 15, 15, 718000, tzinfo=tzutc()),
'note': '<p>Liable to explode at any time, handle with care.</p>',
'url': 'https://example.social/@testbot',
'avatar': 'https://example.social/system/accounts/avatars/000/016/186/original/66d11c4191332e7a.png?1542410869',
'avatar_static': 'https://example.social/system/accounts/avatars/000/016/186/original/66d11c4191332e7a.png?1542410869',
'header': 'https://example.social/headers/original/header.png',
'header_static': 'https://example.social/headers/original/header.png',
'followers_count': 100,
'following_count': 10,
'statuses_count': 99,
'last_status_at': datetime.datetime(2020, 8, 17, 0, 0),
'emojis': [],
'fields': [
{
'name': 'Fully',
'value': 'Automated',
'verified_at': None
},
{
'name': 'Luxury',
'value': 'Communism',
'verified_at': None
}
]
},
'media_attachments': [],
'mentions': [],
'tags': [],
'emojis': [],
'card': None,
'poll': None
"id": 104136090490756999,
"created_at": datetime.datetime(2020, 5, 9, 2, 17, 18, 598000, tzinfo=tzutc()),
"in_reply_to_id": None,
"in_reply_to_account_id": None,
"sensitive": False,
"spoiler_text": "",
"visibility": "public",
"language": "en",
"uri": "https://example.social/users/testbot/statuses/104136090490756503",
"url": "https://example.social/@testbot/104136090490756503",
"replies_count": 0,
"reblogs_count": 0,
"favourites_count": 0,
"favourited": False,
"reblogged": False,
"muted": False,
"bookmarked": False,
"pinned": True,
"content": "<p>hello I am testing</p>",
"reblog": None,
"application": None,
"account": {
"id": 16186,
"username": "testbot",
"acct": "testbot",
"display_name": "ephemtoot Testing Bot",
"locked": True,
"bot": True,
"discoverable": False,
"group": False,
"created_at": datetime.datetime(
2018, 11, 16, 23, 15, 15, 718000, tzinfo=tzutc()
),
"note": "<p>Liable to explode at any time, handle with care.</p>",
"url": "https://example.social/@testbot",
"avatar": "https://example.social/system/accounts/avatars/000/016/186/original/66d11c4191332e7a.png?1542410869",
"avatar_static": "https://example.social/system/accounts/avatars/000/016/186/original/66d11c4191332e7a.png?1542410869",
"header": "https://example.social/headers/original/header.png",
"header_static": "https://example.social/headers/original/header.png",
"followers_count": 100,
"following_count": 10,
"statuses_count": 99,
"last_status_at": datetime.datetime(2020, 8, 17, 0, 0),
"emojis": [],
"fields": [
{"name": "Fully", "value": "Automated", "verified_at": None},
{"name": "Luxury", "value": "Communism", "verified_at": None},
],
},
"media_attachments": [],
"mentions": [],
"tags": [],
"emojis": [],
"card": None,
"poll": None,
}
# Turn dict into object needed by mastodon.py
# Use this in tests after making any changes
# you need to your dict object
# NOTE: ensure values in the dict object are what you need:
# NOTE: ensure values in the dict object are what you need:
# it can be mutated by any test before your test runs
def dict2obj(d):
# checking whether object d is a
# instance of class list
if isinstance(d, list):
d = [dict2obj(x) for x in d]
# if d is not a instance of dict then
# directly object is returned
if not isinstance(d, dict):
return d
# declaring a class
class C:
def dict2obj(d):
# checking whether object d is a
# instance of class list
if isinstance(d, list):
d = [dict2obj(x) for x in d]
# if d is not a instance of dict then
# directly object is returned
if not isinstance(d, dict):
return d
# declaring a class
class C:
pass
# constructor of the class passed to obj
obj = C()
for k in d:
obj.__dict__[k] = dict2obj(d[k])
return obj
# constructor of the class passed to obj
obj = C()
for k in d:
obj.__dict__[k] = dict2obj(d[k])
return obj
# here is our toot object - use this in tests
toot = dict2obj(toot_dict)
# config file after being parsed by yaml.safe_load
config_file = {
'access_token': 'abcd_1234',
'username': 'alice',
'base_url': 'test.social',
'hashtags_to_keep': ['ephemetoot'],
'days_to_keep': 14,
'keep_pinned': True,
'toots_to_keep': [103996285277439262, 103976473612749097, 103877521458738491],
'visibility_to_keep': [None],
'archive': 'archive'
"access_token": "abcd_1234",
"username": "alice",
"base_url": "test.social",
"hashtags_to_keep": ["ephemetoot"],
"days_to_keep": 14,
"keep_pinned": True,
"toots_to_keep": [103996285277439262, 103976473612749097, 103877521458738491],
"visibility_to_keep": [None],
"archive": "archive",
}
# mock GitHub API call for the version number
@ -131,6 +127,7 @@ class MockGitHub:
def json():
return {"tag_name": "vLATEST_VERSION"}
# mock Mastodon
class Mocktodon:
def __init__(self):
@ -149,15 +146,16 @@ class Mocktodon:
# create 10 statuses
# the first 2 will be checked in the first batch (in production it would be 40)
user_toots = []
def make_toot(i=1):
if i < 11:
keys = ("id", "created_at", "reblog", "tags", "visibility")
vals = (
i,
datetime.datetime(2018, 11, i, 23, 15, 15, 718000, tzinfo=tzutc()),
False,
[],
"public"
i,
datetime.datetime(2018, 11, i, 23, 15, 15, 718000, tzinfo=tzutc()),
False,
[],
"public",
)
user_toot = dict(zip(keys, vals))
user_toots.append(user_toot)
@ -165,17 +163,28 @@ class Mocktodon:
make_toot(total)
user_toots.sort(reverse=True)
make_toot(1) # make the batch of toots
make_toot(1) # make the batch of toots
# ignore user_id
# filter for toots with id smaller than max_id
this_batch = []
# use dict_to_obj to make a toot for each toot in the obj then a list from that
this_batch = [ dict2obj(t) for t in user_toots if t["id"] > max_id][:limit]
this_batch = [dict2obj(t) for t in user_toots if t["id"] > max_id][:limit]
return this_batch
# mock argparse objects (options)
class Namespace:
def __init__(self, archive_deleted=False, datestamp=False, hide_skipped=False, retry_mins=1, schedule=False, test=False, time=False, quiet=False):
def __init__(
self,
archive_deleted=False,
datestamp=False,
hide_skipped=False,
retry_mins=1,
schedule=False,
test=False,
time=False,
quiet=False,
):
self.archive_deleted = archive_deleted
self.datestamp = datestamp
self.schedule = schedule
@ -185,14 +194,15 @@ class Namespace:
self.quiet = quiet
self.retry_mins = retry_mins
@pytest.fixture
def mock_github_response(monkeypatch):
def mock_get(*args, **kwargs):
return MockGitHub()
monkeypatch.setattr(requests, "get", mock_get)
########################
# TESTS #
########################
@ -201,15 +211,17 @@ def mock_github_response(monkeypatch):
# Remember that a previous test may have mutated
# one of the values above: set all values you are using
def test_archive_toot(tmpdir):
p = tmpdir.mkdir("archive")
config_file['archive'] = str(p) # make archive directory a temp test dir
config_file["archive"] = str(p) # make archive directory a temp test dir
ephemetoot.archive_toot(config_file, toot)
file_exists = os.path.exists(p + "/104136090490756999.json")
assert file_exists
def test_check_batch(capfd, monkeypatch):
config = config_file
options = Namespace(archive_deleted=False)
@ -220,8 +232,8 @@ def test_check_batch(capfd, monkeypatch):
# monkeypatch process_toot to add 1 to deleted_count and return
# this simulates what would happen if the toot was being deleted
monkeypatch.setattr(
"ephemetoot.ephemetoot.process_toot",
lambda config, options, mastodon, deleted_count, toot: deleted_count + 1
"ephemetoot.ephemetoot.process_toot",
lambda config, options, mastodon, deleted_count, toot: deleted_count + 1,
)
# run check_batch
ephemetoot.check_batch(config, options, mastodon, user_id, timeline, 0)
@ -229,63 +241,84 @@ def test_check_batch(capfd, monkeypatch):
output = capfd.readouterr().out.split("\n")
assert output[0] == "Removed 10 toots."
def test_console_print(capfd):
ephemetoot.console_print("test123", Namespace(test=False, hide_skipped=False, quiet=False), False)
ephemetoot.console_print(
"test123", Namespace(test=False, hide_skipped=False, quiet=False), False
)
assert capfd.readouterr().out == "test123\n"
def test_console_print_quiet():
result = ephemetoot.console_print("test123", Namespace(test=False, hide_skipped=False, quiet=True), False)
result = ephemetoot.console_print(
"test123", Namespace(test=False, hide_skipped=False, quiet=True), False
)
assert result == None
def test_console_print_skip():
result = ephemetoot.console_print("test123", Namespace(test=False, hide_skipped=True, quiet=False), True)
result = ephemetoot.console_print(
"test123", Namespace(test=False, hide_skipped=True, quiet=False), True
)
assert result == None
def test_datestamp_now():
datestamp = ephemetoot.datestamp_now()
date_object = datetime.datetime.strptime(datestamp, "%a %d %b %Y %H:%M:%S %z")
# use timetuple() to exclude differences in milliseconds
assert datetime.datetime.now(timezone.utc).timetuple() == date_object.timetuple()
assert datetime.datetime.now(timezone.utc).timetuple() == date_object.timetuple()
def test_init(monkeypatch, tmpdir):
# monkeypatch current directory
current_dir = tmpdir.mkdir("current_dir") # temporary directory for testing
current_dir = tmpdir.mkdir("current_dir") # temporary directory for testing
monkeypatch.chdir(current_dir)
# monkeypatch input ...outputs
monkeypatch.setattr("ephemetoot.ephemetoot.compulsory_input", lambda a, b, c: "compulsory")
monkeypatch.setattr(
"ephemetoot.ephemetoot.compulsory_input", lambda a, b, c: "compulsory"
)
monkeypatch.setattr("ephemetoot.ephemetoot.digit_input", lambda a, b, c: "14")
monkeypatch.setattr("ephemetoot.ephemetoot.yes_no_input", lambda a, b: "false")
monkeypatch.setattr("ephemetoot.ephemetoot.optional_input", lambda a, b, c: "optional")
monkeypatch.setattr(
"ephemetoot.ephemetoot.optional_input", lambda a, b, c: "optional"
)
# run init
ephemetoot.init()
assert os.path.exists( os.path.join(current_dir, "config.yaml"))
assert os.path.exists(os.path.join(current_dir, "config.yaml"))
def test_jsondefault():
d = ephemetoot.jsondefault(toot.created_at)
assert d == "2020-05-09T02:17:18.598000+00:00"
def test_process_toot(capfd, tmpdir, monkeypatch):
# config uses config_listed at top of this tests file
p = tmpdir.mkdir("archive") # use temporary test directory
p = tmpdir.mkdir("archive") # use temporary test directory
config_file["archive"] = str(p)
config_file["keep_pinned"] = False
config_file["toots_to_keep"] = []
config_file["toots_to_keep"] = []
config_file["visibility_to_keep"] = []
options = Namespace(archive_deleted=False)
mastodon = Mocktodon()
toot_dict["pinned"] = False
toot_dict["pinned"] = False
toot_dict["visibility"] = "public"
toot_dict["reblog"] = False
toot = dict2obj(toot_dict)
ephemetoot.process_toot(config_file, options, mastodon, 0, toot)
assert capfd.readouterr().out == "❌ deleting toot 104136090490756999 tooted 09 May 2020\n"
assert (
capfd.readouterr().out
== "❌ deleting toot 104136090490756999 tooted 09 May 2020\n"
)
def test_process_toot_pinned(capfd, tmpdir):
# config uses config_listed at top of this tests file
p = tmpdir.mkdir("archive") # use temporary test directory
p = tmpdir.mkdir("archive") # use temporary test directory
config_file["archive"] = str(p)
config_file["keep_pinned"] = True
options = Namespace(archive_deleted=False)
@ -294,9 +327,11 @@ def test_process_toot_pinned(capfd, tmpdir):
toot = dict2obj(toot_dict)
ephemetoot.process_toot(config_file, options, mastodon, 0, toot)
assert capfd.readouterr().out == "📌 skipping pinned toot - 104136090490756999\n"
def test_process_toot_saved(capfd, tmpdir):
# config uses config_listed at top of this tests file
p = tmpdir.mkdir("archive") # use temporary test directory
p = tmpdir.mkdir("archive") # use temporary test directory
config_file["archive"] = str(p)
config_file["keep_pinned"] = False
config_file["toots_to_keep"].append(104136090490756999)
@ -307,45 +342,52 @@ def test_process_toot_saved(capfd, tmpdir):
ephemetoot.process_toot(config_file, options, mastodon, 0, toot)
assert capfd.readouterr().out == "💾 skipping saved toot - 104136090490756999\n"
def test_process_toot_visibility(capfd, tmpdir):
# config uses config_listed at top of this tests file
p = tmpdir.mkdir("archive") # use temporary test directory
p = tmpdir.mkdir("archive") # use temporary test directory
config_file["archive"] = str(p)
config_file["keep_pinned"] = False # is true above so make false
config_file["toots_to_keep"].remove(104136090490756999) # don't keep this toot
config_file["keep_pinned"] = False # is true above so make false
config_file["toots_to_keep"].remove(104136090490756999) # don't keep this toot
config_file["visibility_to_keep"].append("testing")
options = Namespace(archive_deleted=False)
mastodon = Mocktodon()
toot_dict["pinned"] = False # is true above so make false
toot_dict["pinned"] = False # is true above so make false
toot_dict["visibility"] = "testing"
toot = dict2obj(toot_dict)
ephemetoot.process_toot(config_file, options, mastodon, 0, toot)
assert capfd.readouterr().out == "👀 skipping testing toot - 104136090490756999\n"
def test_process_toot_hashtag(capfd, tmpdir, monkeypatch):
# config uses config_listed at top of this tests file
p = tmpdir.mkdir("archive") # use temporary test directory
p = tmpdir.mkdir("archive") # use temporary test directory
config_file["archive"] = str(p)
config_file["keep_pinned"] = False
config_file["toots_to_keep"] = []
config_file["toots_to_keep"] = []
config_file["visibility_to_keep"] = []
options = Namespace(archive_deleted=False)
mastodon = Mocktodon()
toot_dict["pinned"] = False
toot_dict["pinned"] = False
toot_dict["visibility"] = "public"
toot_dict["reblog"] = True
toot = dict2obj(toot_dict)
ephemetoot.process_toot(config_file, options, mastodon, 0, toot)
assert capfd.readouterr().out == "👎 unboosting toot 104136090490756999 boosted 09 May 2020\n"
assert (
capfd.readouterr().out
== "👎 unboosting toot 104136090490756999 boosted 09 May 2020\n"
)
def test_retry_on_error():
# Namespace object constructed from top of tests (representing options)
# toot comes from variable at top of test
mastodon = Mocktodon()
toot = dict2obj(toot_dict)
retry = ephemetoot.retry_on_error(Namespace(retry_mins=True), mastodon, toot, 5 )
assert retry == None # should not return an error
retry = ephemetoot.retry_on_error(Namespace(retry_mins=True), mastodon, toot, 5)
assert retry == None # should not return an error
def test_retry_on_error_max_tries():
# Namespace object constructed from top of tests (representing options)
@ -353,12 +395,13 @@ def test_retry_on_error_max_tries():
with pytest.raises(TimeoutError):
mastodon = Mocktodon()
toot = dict2obj(toot_dict)
retry = ephemetoot.retry_on_error(Namespace(retry_mins=True), mastodon, toot, 7 )
retry = ephemetoot.retry_on_error(Namespace(retry_mins=True), mastodon, toot, 7)
def test_schedule(monkeypatch, tmpdir):
home = tmpdir.mkdir("current_dir") # temporary directory for testing
launch = tmpdir.mkdir("TestAgents") # temporary directory for testing
home = tmpdir.mkdir("current_dir") # temporary directory for testing
launch = tmpdir.mkdir("TestAgents") # temporary directory for testing
# monkeypatch directories and suppress the plist loading process
# NOTE: it may be possible to test the plist loading process
@ -394,10 +437,11 @@ def test_schedule(monkeypatch, tmpdir):
assert plist[15] == " <string>" + str(home) + "/ephemetoot.log</string>\n"
assert plist[17] == " <string>" + str(home) + "/ephemetoot.error.log</string>\n"
def test_schedule_with_time(monkeypatch, tmpdir):
home = tmpdir.mkdir("current_dir") # temporary directory for testing
launch = tmpdir.mkdir("TestAgents") # temporary directory for testing
home = tmpdir.mkdir("current_dir") # temporary directory for testing
launch = tmpdir.mkdir("TestAgents") # temporary directory for testing
# monkeypatch directories and suppress the plist loading process
# NOTE: it may be possible to test the plist loading process
@ -430,6 +474,7 @@ def test_schedule_with_time(monkeypatch, tmpdir):
assert plist[21] == " <integer>10</integer>\n"
assert plist[23] == " <integer>30</integer>\n"
def test_tooted_date():
string = ephemetoot.tooted_date(toot)
created = datetime.datetime(2020, 5, 9, 2, 17, 18, 598000, tzinfo=timezone.utc)
@ -437,6 +482,7 @@ def test_tooted_date():
assert string == test_string
def test_version(mock_github_response, capfd):
ephemetoot.version("TEST_VERSION")
output = capfd.readouterr().out
@ -447,4 +493,4 @@ You are using release: \033[92mvTEST_VERSION\033[0m
The latest release is: \033[92mvLATEST_VERSION\033[0m
To upgrade to the most recent version run \033[92mpip3 install --update ephemetoot\033[0m\n"""
assert output == msg
assert output == msg