From 03f27f6a40f7df3defb11c129fff2cf3715eeb2c Mon Sep 17 00:00:00 2001 From: Adam Cooper Date: Sat, 23 Nov 2024 12:03:07 -0500 Subject: [PATCH] Remove Twitter integration (#236) Completely removes all references to Twittter integration and tweepy, as Twitter is no longer a functioning website. Resolves #218 --- .gitignore | 1 - README.md | 12 +- docker/template.env | 7 - memebot/client.py | 5 - memebot/config/__init__.py | 47 ------ memebot/integrations/twitter.py | 181 ------------------------ memebot/log/__init__.py | 12 -- memebot/main.py | 5 +- pyproject.toml | 2 - requirements.txt | 1 - tests/conftest.py | 1 - tests/fixtures/general_fixtures.py | 8 +- tests/fixtures/twitter_fixtures.py | 140 ------------------ tests/integrations/test_twitter.py | 220 ----------------------------- 14 files changed, 7 insertions(+), 635 deletions(-) delete mode 100644 memebot/integrations/twitter.py delete mode 100644 tests/fixtures/twitter_fixtures.py delete mode 100644 tests/integrations/test_twitter.py diff --git a/.gitignore b/.gitignore index 3531a9f..b7b5f0f 100644 --- a/.gitignore +++ b/.gitignore @@ -108,7 +108,6 @@ venv.bak/ # project specific client_token data/ -twitter_api_tokens.json .vscode/ .env diff --git a/README.md b/README.md index 60d47ec..5dcae03 100644 --- a/README.md +++ b/README.md @@ -12,29 +12,19 @@ See [Configuration](#configuration) for more context. ``` usage: main.py [-h] [--discord-api-token DISCORD_API_TOKEN] - [--twitter-api-consumer-key TWITTER_API_CONSUMER_KEY] - [--twitter-api-consumer-secret TWITTER_API_CONSUMER_SECRET] - [--twitter-api-bearer-token TWITTER_API_BEARER_TOKEN] [--log-level {DEBUG,INFO,WARNING,ERROR,CRITICAL} | -v] [--log-location {stdout,stderr,syslog,/path/to/file}] - [--no-twitter] [--nodb] [--database-uri DATABASE_URI] + [--nodb] [--database-uri DATABASE_URI] optional arguments: -h, --help show this help message and exit --discord-api-token DISCORD_API_TOKEN The Discord API client token - --twitter-api-consumer-key TWITTER_API_CONSUMER_KEY - (DEPRECATED) The Twitter API consumer key - --twitter-api-consumer-secret TWITTER_API_CONSUMER_SECRET - (DEPRECATED) The Twitter API consumer secret - --twitter-api-bearer-token TWITTER_API_BEARER_TOKEN - The Twitter API OAuth 2.0 bearer token --log-level {DEBUG,INFO,WARNING,ERROR,CRITICAL} Set logging level -v, --verbose Use verbose logging. Equivalent to --log-level DEBUG --log-location {stdout,stderr,syslog,/path/to/file} Set the location for MemeBot's log - --no-twitter Disable Twitter integration --nodb Disable the database connection, and all features which require it. --database-uri DATABASE_URI diff --git a/docker/template.env b/docker/template.env index 3cbf82a..83ce0cc 100644 --- a/docker/template.env +++ b/docker/template.env @@ -1,7 +1,3 @@ -# API Tokens -# MEMEBOT_DISCORD_CLIENT_TOKEN= -# MEMEBOT_TWITTER_BEARER_TOKEN= - # Network configuration # MEMEBOT_NETWORK_SUBNET= # MEMEBOT_NETWORK_DNS_SERVER= @@ -12,9 +8,6 @@ # MEMEBOT_IMAGE_NAME= # MEMEBOT_BUILD_TARGET= -# Integration settings -# MEMEBOT_TWITTER_ENABLED= - # Logging settings # MEMEBOT_LOG_LEVEL= # MEMEBOT_LOG_LOCATION= diff --git a/memebot/client.py b/memebot/client.py index f5c7211..9597e6b 100644 --- a/memebot/client.py +++ b/memebot/client.py @@ -8,7 +8,6 @@ from memebot import config from memebot import db from memebot import log -from memebot.integrations import twitter from memebot.lib import exception, util @@ -22,8 +21,6 @@ async def on_ready() -> None: log.info(f"Logged in as {memebot.user}") synced = await memebot.tree.sync() log.info(f"Synced {len(synced)} command(s)") - if config.twitter_enabled: - twitter.init(memebot.user) if config.database_enabled: db_online = db.test() if db_online: @@ -101,7 +98,5 @@ def get_memebot() -> discord.ext.commands.Bot: new_memebot.add_listener(on_ready) new_memebot.add_listener(on_interaction) new_memebot.tree.error(on_command_error) - if config.twitter_enabled: - new_memebot.add_listener(twitter.process_message_for_interaction, "on_message") return new_memebot diff --git a/memebot/config/__init__.py b/memebot/config/__init__.py index f15a87c..aa5889a 100644 --- a/memebot/config/__init__.py +++ b/memebot/config/__init__.py @@ -7,17 +7,12 @@ # Discord API token discord_api_token: str -# Twitter API tokens -twitter_api_bearer_token: str # The logging level for MemeBot log_level: str # The location for MemeBot's log log_location: logging.Handler -# Flag which tells if Twitter integration is enabled -twitter_enabled: bool - # Flag which tells if a database connection is enabled database_enabled: bool # MongoDB URI @@ -34,30 +29,6 @@ def populate_config_from_command_line() -> None: default=os.getenv("MEMEBOT_DISCORD_CLIENT_TOKEN"), type=str, ) - parser.add_argument( - "--twitter-api-consumer-key", - help="(DEPRECATED) The Twitter API consumer key", - default=os.getenv("MEMEBOT_TWITTER_CONSUMER_KEY"), - type=lambda _: print( - "USING DEPRECATED TWITTER OAUTH 1.0 CREDENTIALS! " - "PLEASE USE BEARER TOKENS INSTEAD!" - ), - ) - parser.add_argument( - "--twitter-api-consumer-secret", - help="(DEPRECATED) The Twitter API consumer secret", - default=os.getenv("MEMEBOT_TWITTER_CONSUMER_SECRET"), - type=lambda _: print( - "USING DEPRECATED TWITTER OAUTH 1.0 CREDENTIALS! " - "PLEASE USE BEARER TOKENS INSTEAD!" - ), - ) - parser.add_argument( - "--twitter-api-bearer-token", - help="The Twitter API OAuth 2.0 bearer token", - default=os.getenv("MEMEBOT_TWITTER_BEARER_TOKEN"), - type=str, - ) # Logging Configuration logging_verbosity_group = parser.add_mutually_exclusive_group() @@ -94,19 +65,6 @@ def populate_config_from_command_line() -> None: type=validators.validate_log_location, ) - # Twitter Integration - parser.add_argument( - "--no-twitter", - help="Disable Twitter integration", - dest="twitter_enabled", - action="store_false", - ) - parser.set_defaults( - twitter_enabled=validators.validate_bool( - os.getenv("MEMEBOT_TWITTER_ENABLED", str(True)) - ) - ) - # Database Configuration parser.add_argument( "--nodb", @@ -132,18 +90,13 @@ def populate_config_from_command_line() -> None: args = parser.parse_args() global discord_api_token - global twitter_api_bearer_token discord_api_token = args.discord_api_token - twitter_api_bearer_token = args.twitter_api_bearer_token global log_level global log_location log_level = args.log_level log_location = args.log_location - global twitter_enabled - twitter_enabled = args.twitter_enabled - global database_enabled global database_uri database_enabled = args.database_enabled diff --git a/memebot/integrations/twitter.py b/memebot/integrations/twitter.py deleted file mode 100644 index 57d3a88..0000000 --- a/memebot/integrations/twitter.py +++ /dev/null @@ -1,181 +0,0 @@ -""" -This is a module for managing our integration with Twitter. -It owns all metadata related to Twitter, -and the only state it maintains is whether the API has been initialized. - -All functions that make any network communication, except for init, -should be asynchronous and stateless. -""" - -import re -from typing import Tuple - -import discord -import emoji -import tweepy - -from memebot import config -from memebot.lib import util -from memebot.lib.exception import MemebotInternalError - -# Regular expression that describes the pattern of a Tweet URL -twitter_url_pattern = re.compile( - r"https://twitter\.com/(\w+|i/web)/status/\d+(\?s=\d+)?" -) - -# Twitter API handle -twitter_api: tweepy.Client - -# Current bot user -bot_user: discord.ClientUser - - -class MemebotTwitterAPIError(MemebotInternalError): - """ - Error type used to reflect errors encountered while interfacing with the Twitter API - """ - - -def init(user: discord.ClientUser) -> None: - """ - Authenticates to the Twitter API. This function is left synchronous, as any further - interaction with Twitter depends on this function executing and returning - successfully, and it should only be run once at startup. - :param user: The user object for the bot user. - """ - global twitter_api - global bot_user - twitter_api = tweepy.Client(bearer_token=config.twitter_api_bearer_token) - bot_user = user - - -def get_twitter_api() -> tweepy.Client: - """ - Utility function for acquiring the module's API object, as to skip the null-checking - :raises ValueError: If init() has not yet been called, making _twitter_api None - :return: _twitter_api, if it has been initialized. - """ - if not twitter_api: - raise ValueError("Twitter integration module has not yet been initialized.") - return twitter_api - - -def get_twitter_url_from_message_content(content: str) -> Tuple[str, bool]: - """ - Loops through each "word" in content, compares the word to a regex - and returns the "word" in the message that matches the twitter URL regex - :param content: message that was sent in discord - """ - match = twitter_url_pattern.search(content) - if match: - return match[0], util.is_spoil(content, match.start()) - else: - return "", False - - -def fetch_tweet_with_expansions( - tweet_id: str, -) -> tuple[tweepy.Tweet, list[tweepy.User], list[tweepy.Media]]: - """ - Fetches a Tweet from the Twitter API, along with relevant user and media detail - """ - # Fetch the Tweet and requested expansions - response = get_twitter_api().get_tweet( - tweet_id, - media_fields=["url", "variants"], - user_fields=["username"], - expansions=["attachments.media_keys", "referenced_tweets.id.author_id"], - ) - - # Verify API response - if not isinstance(response, tweepy.Response): - raise TypeError(f"Unexpected Twitter API response type: ({type(response)})") - tweet_info = response.data - if not isinstance(tweet_info, tweepy.Tweet): - raise TypeError(f"Unexpected response data: {tweet_info}") - tweet_media = response.includes.get("media", []) - tweet_users = response.includes.get("users", []) - - return tweet_info, tweet_users, tweet_media - - -def is_quote_tweet(tweet: tweepy.Tweet) -> bool: - return tweet.referenced_tweets and any( - ref.type == "quoted" for ref in tweet.referenced_tweets - ) - - -def get_quote_tweet_urls(root_tweet: tweepy.Tweet, spoiled: bool) -> str: - """ - Gets URLs of quoteted tweets (nested up to max 3) - :param root_tweet: information of tweet - :param spoiled: whether the quote tweets should be spoiled - :return: URL of quote tweet(s) - """ - output_text = "quoted tweet(s): " - parent_tweet = root_tweet - for _ in range(3): - # Find the ID of the Tweet quoted by the parent - qt_ref = next( - (t for t in parent_tweet.referenced_tweets if t.type == "quoted"), None - ) - if not qt_ref: - raise MemebotTwitterAPIError("Attempted to unroll non-quote-tweet") - qt, qt_users, _ = fetch_tweet_with_expansions(qt_ref.id) - qt_author = next((u for u in qt_users if u.id == qt.author_id), None) - if not qt_author: - raise MemebotTwitterAPIError("Unable to resolve quote-tweet author") - output_text += util.maybe_make_link_spoiler( - f"\nhttps://twitter.com/{qt_author.username}/status/{qt.id}", - spoiled, - ) - if not is_quote_tweet(qt): - break - parent_tweet = qt - return output_text - - -async def process_message_for_interaction(message: discord.Message) -> None: - """ - Processes non-command content of a message to determine if a message contains Tweet - information and requires interaction from MemeBot. Note that this will still affect - command messages, but the content is not processed - as a command by this function. - :param message: The message to process - """ - # Iterate through the message sent, and check if the message contained - # a "word" that was a twitter URL to a status. - twitter_url, spoiled = get_twitter_url_from_message_content(message.content) - if twitter_url: - # Because a twitter URL to a status is oftentimes as follows: - # https://twitter.com/USER/status/xxxxxxxxxxxxxxxxxxx?s=yy - # We need to split up the URL by the "/", and then split it up again - # by the "?" in order to get the ID of the tweet being linked - tweet_id = twitter_url.split("/")[-1].split("?")[0] - - tweet_info, _, tweet_media = fetch_tweet_with_expansions(tweet_id) - - # React with a numeric emoji to Tweets containing multiple images - if (n_images := len(tweet_media)) > 1: - await message.add_reaction(emoji.emojize(f":keycap_{n_images}:")) - # For tweets containing only a single video, we embed the video in Discord. - elif n_images == 1 and tweet_media[0].type in ("video", "animated_gif"): - video_url = max( - # Only select variants with video URLs - # Twitter includes links to x-mpeg playlists as well - filter( - lambda v: v["content_type"].startswith("video"), - tweet_media[0].variants, - ), - # Choose the variant with the highest bitrate - key=lambda v: int(v.get("bitrate", -1)), - )["url"] - await message.channel.send( - f"embedded video:\n" - f"{util.maybe_make_link_spoiler(video_url, spoiled)}" - ) - - # Post quote tweet links. - if is_quote_tweet(tweet_info) and message.author != bot_user: - quote_tweet_urls = get_quote_tweet_urls(tweet_info, spoiled) - await message.channel.send(quote_tweet_urls) diff --git a/memebot/log/__init__.py b/memebot/log/__init__.py index e503dfe..d1c32fc 100644 --- a/memebot/log/__init__.py +++ b/memebot/log/__init__.py @@ -21,18 +21,6 @@ atexit.register(logging.shutdown) -def set_third_party_logging() -> None: - """ - Enable logging on third-party packages that can't be overwritten with MemeBotLogger - """ - # Tweepy does not use a unified logger, so the best we can do is - # enable its debug mode. - import asyncio - - if config.log_level == logging.getLevelName(logging.DEBUG): - asyncio.get_event_loop().set_debug(True) - - # Forward memebot_logger's logging methods as module-level functions debug = memebot_logger.debug info = memebot_logger.info diff --git a/memebot/main.py b/memebot/main.py index ce918fa..519ec5c 100644 --- a/memebot/main.py +++ b/memebot/main.py @@ -8,8 +8,11 @@ def main() -> None: Main function, initializes MemeBot and then loops :return: Exit status of discord.Client.run() """ - log.set_third_party_logging() config.populate_config_from_command_line() + # the ``log`` package should be imported ASAP to ensure our logging shims + # are injected into the runtime before external packages configure + # their logging + log.info("Starting up memebot!") memebot = get_memebot() # !! DO NOT HARDCODE THE TOKEN !! diff --git a/pyproject.toml b/pyproject.toml index edcaf7c..9ce8b57 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,8 +21,6 @@ exclude = [ [[tool.mypy.overrides]] module = [ - # tweepy does not provide type annotations, but it is desired for 4.x - "tweepy", # This module in discord.py does not have full stubs for some reason "discord.ext.commands", ] diff --git a/requirements.txt b/requirements.txt index 3c39e29..8e2df4e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,5 +2,4 @@ discord.py~=2.1.0 emoji~=2.0.0 pymongo~=3.13.0 pymongo-stubs~=0.2.0 -tweepy~=4.12.0 types-emoji~=2.0.0 diff --git a/tests/conftest.py b/tests/conftest.py index e71afce..690f32d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,5 +2,4 @@ "pytest_asyncio", "tests.fixtures.discord_fixtures", "tests.fixtures.general_fixtures", - "tests.fixtures.twitter_fixtures", ] diff --git a/tests/fixtures/general_fixtures.py b/tests/fixtures/general_fixtures.py index e2c2cd2..328fa9d 100644 --- a/tests/fixtures/general_fixtures.py +++ b/tests/fixtures/general_fixtures.py @@ -4,16 +4,14 @@ import pytest from memebot import config -from memebot.integrations import twitter DEFAULT_ENVIRONMENT = os.environ | { "MEMEBOT_DISCORD_CLIENT_TOKEN": "MOCK_TOKEN", - "MEMEBOT_TWITTER_ENABLED": "False", } @pytest.fixture(autouse=True) -def setup_and_teardown(mock_twitter_client: mock.Mock) -> None: +def setup_and_teardown() -> None: """ This is a universal setup and teardown function which is automatically run surrounding each test. The function contains both the setup and teardown behavior. @@ -25,9 +23,7 @@ def setup_and_teardown(mock_twitter_client: mock.Mock) -> None: os.environ = DEFAULT_ENVIRONMENT with mock.patch("argparse.ArgumentParser", mock.MagicMock()): - with mock.patch("tweepy.Client", lambda *_, **__: mock_twitter_client): - config.populate_config_from_command_line() - twitter.init(mock.MagicMock()) + config.populate_config_from_command_line() # Run test yield diff --git a/tests/fixtures/twitter_fixtures.py b/tests/fixtures/twitter_fixtures.py deleted file mode 100644 index 107ad47..0000000 --- a/tests/fixtures/twitter_fixtures.py +++ /dev/null @@ -1,140 +0,0 @@ -from typing import Any -from unittest import mock - -import pytest - - -@pytest.fixture -def mock_twitter_url() -> str: - return "https://twitter.com/dril/status/922321981" - - -@pytest.fixture -@mock.patch("tweepy.ReferencedTweet") -def mock_referenced_tweet(MockReferencedTweet: type[mock.Mock]) -> mock.Mock: - return MockReferencedTweet() - - -@pytest.fixture -@mock.patch("tweepy.ReferencedTweet") -def mock_referenced_tweet_2(MockReferencedTweet: type[mock.Mock]) -> mock.Mock: - return MockReferencedTweet() - - -@pytest.fixture -@mock.patch("tweepy.ReferencedTweet") -def mock_referenced_tweet_3(MockReferencedTweet: type[mock.Mock]) -> mock.Mock: - return MockReferencedTweet() - - -@pytest.fixture -@mock.patch("tweepy.ReferencedTweet") -def mock_referenced_tweet_4(MockReferencedTweet: type[mock.Mock]) -> mock.Mock: - return MockReferencedTweet() - - -@pytest.fixture -@mock.patch("tweepy.Tweet", spec=True) -def mock_tweet(MockTweet: type[mock.Mock], mock_twitter_user: mock.Mock) -> mock.Mock: - tweet = MockTweet() - tweet.referenced_tweets = [] - tweet.author_id = mock_twitter_user.id - return tweet - - -@pytest.fixture -@mock.patch("tweepy.Tweet", spec=True) -def mock_quote_tweet( - MockTweet: type[mock.Mock], - mock_referenced_tweet: mock.Mock, - mock_twitter_user: mock.Mock, -) -> mock.Mock: - tweet = MockTweet() - mock_referenced_tweet.type = "quoted" - tweet.referenced_tweets = [mock_referenced_tweet] - tweet.author_id = mock_twitter_user.id - return tweet - - -@pytest.fixture -@mock.patch("tweepy.Tweet", spec=True) -def mock_quote_tweet_2( - MockTweet: mock.Mock, - mock_referenced_tweet_2: mock.Mock, - mock_twitter_user: mock.Mock, -) -> mock.Mock: - tweet = MockTweet() - mock_referenced_tweet_2.type = "quoted" - tweet.referenced_tweets = [mock_referenced_tweet_2] - tweet.author_id = mock_twitter_user.id - return tweet - - -@pytest.fixture -@mock.patch("tweepy.Tweet", spec=True) -def mock_quote_tweet_3( - MockTweet: mock.Mock, - mock_referenced_tweet_3: mock.Mock, - mock_twitter_user: mock.Mock, -) -> mock.Mock: - tweet = MockTweet() - mock_referenced_tweet_3.type = "quoted" - tweet.referenced_tweets = [mock_referenced_tweet_3] - tweet.author_id = mock_twitter_user.id - return tweet - - -@pytest.fixture -@mock.patch("tweepy.Tweet", spec=True) -def mock_quote_tweet_4( - MockTweet: mock.Mock, - mock_referenced_tweet_4: mock.Mock, - mock_twitter_user: mock.Mock, -) -> mock.Mock: - tweet = MockTweet() - mock_referenced_tweet_4.type = "quoted" - tweet.referenced_tweets = [mock_referenced_tweet_4] - tweet.author_id = mock_twitter_user.id - return tweet - - -@pytest.fixture -@mock.patch("tweepy.User", new_callable=mock.MagicMock) -def mock_twitter_user(MockTwitterUser: type[mock.Mock]) -> mock.Mock: - return MockTwitterUser() - - -@pytest.fixture -@mock.patch("tweepy.Response", autospec=True) -@mock.patch("tweepy.Client", autospec=True) -def mock_twitter_client( - MockTwitterClient: type[mock.Mock], - MockTwitterResponse: type[mock.Mock], - mock_twitter_user: mock.Mock, - mock_tweet: mock.Mock, - mock_quote_tweet: mock.Mock, - mock_quote_tweet_2: mock.Mock, - mock_quote_tweet_3: mock.Mock, - mock_quote_tweet_4: mock.Mock, -) -> mock.Mock: - def _get_tweet(tweet_id: str, *_: Any, **__: Any) -> mock.Mock: - resp = MockTwitterResponse() - resp.data = next( - ( - t - for t in ( - mock_quote_tweet, - mock_quote_tweet_2, - mock_quote_tweet_3, - mock_quote_tweet_4, - ) - if t.id == tweet_id - ), - mock_tweet, - ) - resp.includes = {"users": [mock_twitter_user]} - return resp - - client = MockTwitterClient() - client.get_tweet = _get_tweet - return client diff --git a/tests/integrations/test_twitter.py b/tests/integrations/test_twitter.py deleted file mode 100644 index 9bd03a3..0000000 --- a/tests/integrations/test_twitter.py +++ /dev/null @@ -1,220 +0,0 @@ -from unittest import mock - -import emoji -import pytest - -from memebot.integrations import twitter -from memebot.lib import exception - - -def test_twitter_url_regex(mock_twitter_url: str) -> None: - """ - Tests that the Twitter regex only works on valid tweet URLs - """ - assert twitter.twitter_url_pattern.search(mock_twitter_url) is not None - assert twitter.twitter_url_pattern.search(f"{mock_twitter_url}?s=19") is not None - assert ( - twitter.twitter_url_pattern.search(f"{mock_twitter_url}?s=19&t=12345") - is not None - ) - assert twitter.twitter_url_pattern.search("https://memebot.com") is None - assert twitter.twitter_url_pattern.search("https://twitter.com/foo/bar") is None - assert ( - twitter.twitter_url_pattern.search("https://twitter.com/i/status/12345") - is not None - ) - assert ( - twitter.twitter_url_pattern.search("https://twitter.com/i/web/status/12345") - is not None - ) - assert twitter.twitter_url_pattern.search("https://twitter.com") is None - assert twitter.twitter_url_pattern.search("twitter.com") is None - assert twitter.twitter_url_pattern.search("twitter.com/i/status/12345") is None - - -def test_get_twitter_url_from_message_content(mock_twitter_url: str) -> None: - """ - Test that we're able to successfully pull a Tweet URL from a discord message - """ - assert twitter.get_twitter_url_from_message_content( - f"foo bar baz {mock_twitter_url} 1 2 3" - ) == (mock_twitter_url, False) - assert twitter.get_twitter_url_from_message_content( - f"foo bar baz {mock_twitter_url}" - ) == (mock_twitter_url, False) - assert twitter.get_twitter_url_from_message_content( - f"{mock_twitter_url} 1 2 3" - ) == (mock_twitter_url, False) - - # Test spoiler - assert twitter.get_twitter_url_from_message_content( - f"foo bar baz ||{mock_twitter_url}|| 1 2 3" - ) == (mock_twitter_url, True) - assert twitter.get_twitter_url_from_message_content( - f"foo bar baz ||{mock_twitter_url}||" - ) == (mock_twitter_url, True) - assert twitter.get_twitter_url_from_message_content( - f"||{mock_twitter_url}|| 1 2 3" - ) == (mock_twitter_url, True) - assert twitter.get_twitter_url_from_message_content( - f"||foo bar baz {mock_twitter_url}|| 1 2 3" - ) == (mock_twitter_url, True) - - -def test_is_quote_tweet(mock_quote_tweet: mock.Mock, mock_tweet: mock.Mock) -> None: - """ - Test if we properly evaluate quote tweets - """ - assert twitter.is_quote_tweet(mock_quote_tweet) - assert not twitter.is_quote_tweet(mock_tweet) - - -def test_get_quote_tweet_urls( - mock_quote_tweet: mock.Mock, mock_tweet: mock.Mock, mock_twitter_user: mock.Mock -) -> None: - """ - Test if we properly construct the string which displays nested quote tweet URLs - """ - actual = twitter.get_quote_tweet_urls(mock_quote_tweet, False) - expected = ( - f"quoted tweet(s): \n" - f"https://twitter.com/{mock_twitter_user.username}" - f"/status/{mock_tweet.id}" - ) - assert actual == expected - - -def test_nested_quote_tweet( - mock_quote_tweet: mock.Mock, - mock_quote_tweet_2: mock.Mock, - mock_quote_tweet_3: mock.Mock, - mock_quote_tweet_4: mock.Mock, - mock_twitter_user: mock.Mock, -) -> None: - """ - Test that multiple nested quote tweets come out properly - """ - mock_quote_tweet.referenced_tweets[0].id = mock_quote_tweet_2.id - mock_quote_tweet_2.referenced_tweets[0].id = mock_quote_tweet_3.id - mock_quote_tweet_3.referenced_tweets[0].id = mock_quote_tweet_4.id - actual = twitter.get_quote_tweet_urls(mock_quote_tweet, False) - expected = ( - "quoted tweet(s): \n" - f"https://twitter.com/{mock_twitter_user.username}" - f"/status/{mock_quote_tweet_2.id}\n" - f"https://twitter.com/{mock_twitter_user.username}" - f"/status/{mock_quote_tweet_3.id}\n" - f"https://twitter.com/{mock_twitter_user.username}" - f"/status/{mock_quote_tweet_4.id}" - ) - assert actual == expected - - -def test_quote_tweet_non_quote_failure(mock_tweet: mock.Mock) -> None: - """ - Test that attempting to unroll a non-quote tweet fails - """ - with pytest.raises(exception.MemebotInternalError): - twitter.get_quote_tweet_urls(mock_tweet, False) - - -@pytest.mark.asyncio -@pytest.mark.parametrize("num_media", [2, 3, 4]) -async def test_tweet_media_reaction( - mock_twitter_url: str, - mock_message: mock.Mock, - num_media: int, -) -> None: - """ - Tests that tweets with more than 1 media item get reacted to - """ - old_get_tweet = twitter.twitter_api.get_tweet - - def patched_get_tweet(*args, **kwargs): - response = old_get_tweet(*args, **kwargs) - response.includes |= {"media": [mock.MagicMock()] * num_media} - return response - - with mock.patch.object(twitter.twitter_api, "get_tweet", patched_get_tweet): - mock_message.content = mock_twitter_url - await twitter.process_message_for_interaction(mock_message) - mock_message.add_reaction.assert_awaited_once_with( - emoji.emojize(f":keycap_{num_media}:") - ) - - -@pytest.mark.asyncio -async def test_tweet_media_no_reaction( - mock_twitter_url: str, mock_message: mock.Mock -) -> None: - """ - Tests that tweets with a single piece of media do not get reacted to - """ - old_get_tweet = twitter.twitter_api.get_tweet - - def patched_get_tweet(*args, **kwargs): - response = old_get_tweet(*args, **kwargs) - response.includes |= {"media": [mock.MagicMock()]} - return response - - with mock.patch.object(twitter.twitter_api, "get_tweet", patched_get_tweet): - mock_message.content = mock_twitter_url - await twitter.process_message_for_interaction(mock_message) - mock_message.add_reaction.assert_not_awaited() - - -@pytest.mark.asyncio -@pytest.mark.parametrize("media_type", ["video", "animated_gif"]) -@pytest.mark.parametrize("content_type", ["video/mp4", "video/mpeg"]) -async def test_tweet_embedded_video( - mock_twitter_url: str, mock_message: mock.Mock, media_type: str, content_type: str -) -> None: - """ - Tests that tweets with a video have the video embedded - """ - old_get_tweet = twitter.twitter_api.get_tweet - mock_variant_1 = mock.MagicMock() - mock_variant_2 = mock.MagicMock() - mock_variant_3 = mock.MagicMock() - - with mock.patch("tweepy.media.Media", spec=True) as MockMedia: - - def patched_get_tweet(*args, **kwargs): - response = old_get_tweet(*args, **kwargs) - mock_media = MockMedia() - mock_media.type = media_type - mock_media.variants = [ - {"bitrate": n, "url": variant, "content_type": content_type} - for n, variant in enumerate( - (mock_variant_1, mock_variant_2, mock_variant_3) - ) - ] - response.includes |= {"media": [mock_media]} - return response - - with mock.patch.object(twitter.twitter_api, "get_tweet", patched_get_tweet): - mock_message.content = mock_twitter_url - await twitter.process_message_for_interaction(mock_message) - mock_message.channel.send.assert_awaited_once_with( - f"embedded video:\n{mock_variant_3}" - ) - - -@pytest.mark.asyncio -async def test_single_media_no_embed( - mock_message: mock.Mock, mock_twitter_url: str -) -> None: - """ - Tests that a tweet with a single non-video media does not attempt to embed - """ - old_get_tweet = twitter.twitter_api.get_tweet - - def patched_get_tweet(*args, **kwargs): - response = old_get_tweet(*args, **kwargs) - response.includes |= {"media": [mock.MagicMock()]} - return response - - with mock.patch.object(twitter.twitter_api, "get_tweet", patched_get_tweet): - mock_message.content = mock_twitter_url - await twitter.process_message_for_interaction(mock_message) - mock_message.channel.send.assert_not_awaited()