Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_submit_image__bad_websocket(self, _, __):
self.reddit.read_only = False
with self.recorder.use_cassette("TestSubreddit.test_submit_image"):
subreddit = self.reddit.subreddit(
pytest.placeholders.test_subreddit
)
for file_name in ("test.png", "test.jpg"):
image = self.image_path(file_name)
with pytest.raises(ClientException):
subreddit.submit_image("Test Title", image)
def test_id_from_url__invalid_urls(self):
urls = [
"",
"1",
"/",
"my.it/2gmzqe",
"http://my.it/_",
"https://redd.it/_/",
"http://reddit.com/comments/_/2gmzqe",
"https://reddit.com/r/wallpapers/",
"https://reddit.com/r/wallpapers",
]
for url in urls:
with pytest.raises(ClientException):
Submission.id_from_url(url)
def run(self):
log.info("Bot {} started.".format(c.BOT_NAME))
try:
started_at = time.time()
subreddit = self.reddit.subreddit(c.SUBREDDITS)
for submission in subreddit.stream.submissions():
# old submissions are discarded
if submission.created_utc < started_at:
continue
self.processSubmission(submission)
except (ClientException, InvalidToken) as e:
# something wrong with the client, this should not happen
# for critical exceptions the bot will stop
# handled by main.py
raise CriticalException("Error processing subreddit stream: {}".format(str(e)))
except (PrawcoreException, APIException) as e:
# for minor exceptions the bot will wait a timeout and retry running again
# handled by main.py
raise MinorException("Error processing subreddit stream: {}".format(str(e)))
def main():
### Login loop
while True:
try:
reddit = praw.Reddit(client_id = c_id, \
client_secret = c_secret, \
user_agent = u_a, \
username = usrnm, \
password = passwd) # Connect to Reddit
break
except praw.exceptions.APIException: # Catch Reddit API error. REVIEW PARAMS
print("Reddit API error. Try again.")
except praw.exceptions.ClientException: # Catch client login error
print("Client-side error. Try again.")
### Scraping loop
while True:
while True:
subs = get_subreddits(reddit)
master = create_dict(subs)
get_settings(subs,master)
confirm = print_settings(master)
if confirm == "y":
break
for sub,settings in master.items():
cat_i = settings[0]
search_for = settings[1]
# reply to lines not marked with prefix '-'
for post in post_lines:
pd = parseRedDumpLine(post)
if pd[0][0] == '-':
continue
# get submission from submission.id stored in scrape dump
submission = reddit.submission(url=pd[1])
# reply with random promotion
try:
submission.reply(getRandRedPromo())
log('Replied: ' + pd[1])
# wait for 45-75 secs to evade spamming flags
wait(45, 75)
except praw.exceptions.ClientException as e:
log("Error: " + e.message)
# wish to step through this section with a debugger: This block
# of code is NOT debugger-friendly. If there is *any*
# significant time between the POST request just above this
# comment and the creation of the websocket connection just
# below, the code will become stuck in an infinite loop at the
# socket.recv() call. I believe this is because only one message is
# sent over the websocket, and if the client doesn't connect
# soon enough, it will miss the message and get stuck forever
# waiting for another.
#
# So if you need to debug this section of code, please let the
# websocket creation happen right after the POST request,
# otherwise you will have trouble.
if not isinstance(response, dict):
raise ClientException(
"Something went wrong with your post: {!r}".format(response)
)
if without_websockets:
return
try:
socket = websocket.create_connection(
response["json"]["data"]["websocket_url"], timeout=timeout
)
ws_update = loads(socket.recv())
socket.close()
except websocket.WebSocketTimeoutException:
raise ClientException(
"Websocket error. Check your media file. "
"Your post may still have been created."
def id_from_url(url):
"""Get the ID of a comment from the full URL."""
parts = RedditBase._url_parts(url)
try:
comment_index = parts.index("comments")
except ValueError:
raise ClientException("Invalid URL: {}".format(url))
if len(parts) - 4 != comment_index:
raise ClientException("Invalid URL: {}".format(url))
return parts[-1]
"""Parse the message item to see what action to take."""
item = inbox_item.item
# always mark the item as read
if constants.MARK_READ:
item.mark_read()
# do nothing if non-moderator calls testing bot
if config.is_testing_environ and item.author not in config.moderators:
logger.info("non-moderator called testing bot")
return
command, comment = inbox_item.get_message_command()
if command == 'delete':
try:
author = item.author
try:
parent = comment.parent()
except ClientException:
logger.debug("Parent comment was deleted.")
return
if author == parent.author or author in config.moderators or author in parent.subreddit.moderator():
logger.info('deleting original comment')
comment.delete()
await log_event('delete', item)
except Exception as e:
logger.exception(e, inbox_item=inbox_item)
elif command == 'edit':
try:
parent = comment.parent()
except ClientException:
logger.debug("Parent comment was deleted.")
return
parent.refresh()
parent.subject = 'username mention'
"""
Handles username mentions and handles the formatting and posting of
those calls as workable jobs to ToR.
:param mention: the Comment object containing the username mention.
:return: None.
"""
try:
i18n = translation()
pm_subject = i18n['responses']['direct_message']['subject']
pm_body = i18n['responses']['direct_message']['body']
# message format is subject, then body
mention.author.message(pm_subject, _(pm_body))
logging.info(f'Message sent to {mention.author.name}!')
except (RedditClientException, AttributeError):
# apparently this crashes with an AttributeError if someone
# calls the bot and immediately deletes their comment. This
# should fix that.
pass
def read_only(self, value):
"""Set or unset the use of the ReadOnlyAuthorizer.
Raise :class:`ClientException` when attempting to unset ``read_only``
and only the ReadOnlyAuthorizer is available.
"""
if value:
self._core = self._read_only_core
elif self._authorized_core is None:
raise ClientException(
"read_only cannot be unset as only the "
"ReadOnlyAuthorizer is available."
)
else:
self._core = self._authorized_core