Fix up unit tests for new structure

tests/extractors/test_extractor_base.py

@@ -0,0 +1,26 @@
+import pytest
+
+from auto_archiver.core.metadata import Metadata
+from auto_archiver.core.extractor import Extractor
+from auto_archiver.core.module import get_module
+class TestExtractorBase(object):
+
+    extractor_module: str = None
+    config: dict = None
+
+    @pytest.fixture(autouse=True)
+    def setup_archiver(self, setup_module):
+        assert self.extractor_module is not None, "self.extractor_module must be set on the subclass"
+        assert self.config is not None, "self.config must be a dict set on the subclass"
+        self.extractor: Extractor = setup_module(self.extractor_module, self.config)
+    
+    def assertValidResponseMetadata(self, test_response: Metadata, title: str, timestamp: str, status: str = ""):
+        assert test_response is not False
+
+        if not status:
+            assert test_response.is_success()
+        else:
+            assert status == test_response.status
+
+        assert title == test_response.get_title()
+        assert timestamp, test_response.get("timestamp")

tests/extractors/test_generic_extractor.py

@@ -0,0 +1,205 @@
+from pathlib import Path
+import datetime
+import os
+
+from os.path import dirname
+
+import pytest
+
+from auto_archiver.modules.generic_extractor.generic_extractor import GenericExtractor
+from .test_extractor_base import TestExtractorBase
+
+class TestGenericExtractor(TestExtractorBase):
+    """Tests Generic Extractor
+    """
+    extractor_module = 'generic_extractor'
+    extractor: GenericExtractor
+
+    config = {
+        'subtitles': False,
+        'comments': False,
+        'livestreams': False,
+        'live_from_start': False,
+        'end_means_success': True,
+        'allow_playlist': False,
+        'max_downloads': "inf",
+        'proxy': None,
+        'cookies_from_browser': False,
+        'cookie_file': None,
+        }
+    
+    def test_load_dropin(self):
+        # test loading dropins that are in the generic_archiver package
+        package = "auto_archiver.modules.generic_extractor"
+        assert self.extractor.dropin_for_name("bluesky", package=package)
+
+        # test loading dropings via filepath
+        path = os.path.join(dirname(dirname(__file__)), "data/")
+        assert self.extractor.dropin_for_name("dropin", additional_paths=[path])
+
+
+
+    @pytest.mark.parametrize("url, is_suitable", [
+        ("https://www.youtube.com/watch?v=5qap5aO4i9A", True),
+        ("https://www.tiktok.com/@funnycats0ftiktok/video/7345101300750748970?lang=en", True),
+        ("https://www.instagram.com/p/CU1J9JYJ9Zz/", True),
+        ("https://www.facebook.com/nytimes/videos/10160796550110716", True),
+        ("https://www.twitch.tv/videos/1167226570", True),
+        ("https://bellingcat.com/news/2021/10/08/ukrainian-soldiers-are-being-killed-by-landmines-in-the-donbas/", True),
+        ("https://google.com", True)])
+    def test_suitable_urls(self, make_item, url, is_suitable):
+        """
+            Note: expected behaviour is to return True for all URLs, as YoutubeDLArchiver should be able to handle all URLs
+            This behaviour may be changed in the future (e.g. if we want the youtubedl archiver to just handle URLs it has extractors for,
+            and then if and only if all archivers fails, does it fall back to the generic archiver)
+        """
+        assert self.extractor.suitable(url) == is_suitable
+
+    @pytest.mark.download
+    def test_download_tiktok(self, make_item):
+        item = make_item("https://www.tiktok.com/@funnycats0ftiktok/video/7345101300750748970")
+        result = self.extractor.download(item)
+        assert result.get_url() == "https://www.tiktok.com/@funnycats0ftiktok/video/7345101300750748970"
+
+    @pytest.mark.download
+    @pytest.mark.parametrize("url", [
+        "https://bsky.app/profile/colborne.bsky.social/post/3lcxcpgt6j42l",
+        "twitter.com/bellingcat/status/123",
+        "https://www.youtube.com/watch?v=1"
+    ])
+    def test_download_nonexistend_media(self, make_item, url):
+        """
+        Test to make sure that the extractor doesn't break on non-existend posts/media
+
+        It should return 'False'
+        """
+        item = make_item(url)
+        result = self.extractor.download(item)
+        assert not result
+
+
+    @pytest.mark.download
+    def test_youtube_download(self, make_item):
+        # url https://www.youtube.com/watch?v=5qap5aO4i9A
+        item = make_item("https://www.youtube.com/watch?v=J---aiyznGQ")
+        result = self.extractor.download(item)
+        assert result.get_url() == "https://www.youtube.com/watch?v=J---aiyznGQ"
+        assert result.get_title() == "Keyboard Cat! - THE ORIGINAL!"
+        assert result.get('description') == "Buy NEW Keyboard Cat Merch! https://keyboardcat.creator-spring.com\n\nxo Keyboard Cat memes make your day better!\nhttp://www.keyboardcatstore.com/\nhttps://www.facebook.com/thekeyboardcat\nhttp://www.charlieschmidt.com/"
+        assert len(result.media) == 2
+        assert Path(result.media[0].filename).name == "J---aiyznGQ.webm"
+        assert Path(result.media[1].filename).name == "hqdefault.jpg"
+
+    @pytest.mark.download
+    def test_bluesky_download_multiple_images(self, make_item):
+        item = make_item("https://bsky.app/profile/bellingcat.com/post/3lffjoxcu7k2w")
+        result = self.extractor.download(item)
+        assert result is not False
+
+    @pytest.mark.download
+    def test_bluesky_download_single_image(self, make_item):
+        item = make_item("https://bsky.app/profile/bellingcat.com/post/3lfn3hbcxgc2q")
+        result = self.extractor.download(item)
+        assert result is not False
+    
+    @pytest.mark.download
+    def test_bluesky_download_no_media(self, make_item):
+        item = make_item("https://bsky.app/profile/bellingcat.com/post/3lfphwmcs4c2z")
+        result = self.extractor.download(item)
+        assert result is not False
+
+    @pytest.mark.download
+    def test_bluesky_download_video(self, make_item):
+        item = make_item("https://bsky.app/profile/bellingcat.com/post/3le2l4gsxlk2i")
+        result = self.extractor.download(item)
+        assert result is not False
+    
+    @pytest.mark.download
+    def test_truthsocial_download_video(self, make_item):
+        item = make_item("https://truthsocial.com/@DaynaTrueman/posts/110602446619561579")
+        result = self.extractor.download(item)
+        assert len(result.media) == 1
+        assert result is not False
+
+    @pytest.mark.download
+    def test_truthsocial_download_no_media(self, make_item):
+        item = make_item("https://truthsocial.com/@bbcnewa/posts/109598702184774628")
+        result = self.extractor.download(item)
+        assert result is not False
+    
+    @pytest.mark.download
+    def test_truthsocial_download_poll(self, make_item):
+        item = make_item("https://truthsocial.com/@CNN_US/posts/113724326568555098")
+        result = self.extractor.download(item)
+        assert result is not False
+    
+    @pytest.mark.download
+    def test_truthsocial_download_single_image(self, make_item):
+        item = make_item("https://truthsocial.com/@mariabartiromo/posts/113861116433335006")
+        result = self.extractor.download(item)
+        assert len(result.media) == 1
+        assert result is not False
+
+    @pytest.mark.download
+    def test_truthsocial_download_multiple_images(self, make_item):
+        item = make_item("https://truthsocial.com/@trrth/posts/113861302149349135")
+        result = self.extractor.download(item)
+        assert len(result.media) == 3
+
+    @pytest.mark.download
+    def test_twitter_download_nonexistend_tweet(self, make_item):
+        # this tweet does not exist
+        url = "https://x.com/Bellingcat/status/17197025860711058"
+        response = self.extractor.download(make_item(url))
+        assert not response
+    
+    @pytest.mark.download
+    def test_twitter_download_malformed_tweetid(self, make_item):
+        # this tweet does not exist
+        url = "https://x.com/Bellingcat/status/1719702a586071100058"
+        response = self.extractor.download(make_item(url))
+        assert not response
+
+    @pytest.mark.download
+    def test_twitter_download_tweet_no_media(self, make_item):
+        
+        item = make_item("https://twitter.com/MeCookieMonster/status/1617921633456640001?s=20&t=3d0g4ZQis7dCbSDg-mE7-w")
+        post = self.extractor.download(item)
+
+        self.assertValidResponseMetadata(
+            post,
+            "Onion rings are just vegetable donuts.",
+            datetime.datetime(2023, 1, 24, 16, 25, 51, tzinfo=datetime.timezone.utc),
+            "yt-dlp_Twitter: success"
+        )
+    
+    @pytest.mark.download
+    def test_twitter_download_video(self, make_item):
+        url = "https://x.com/bellingcat/status/1871552600346415571"
+        post = self.extractor.download(make_item(url))
+        self.assertValidResponseMetadata(
+            post,
+            "Bellingcat - This month's Bellingchat Premium is with @KolinaKoltai. She reveals how she investigated a platform allowing users to create AI-generated child sexual abuse material and explains why it's crucial to investigate the people behind these services",
+            datetime.datetime(2024, 12, 24, 13, 44, 46, tzinfo=datetime.timezone.utc)
+        )
+
+    @pytest.mark.xfail(reason="Currently failing, sensitive content requires logged in users/cookies - not yet implemented")
+    @pytest.mark.download
+    @pytest.mark.parametrize("url, title, timestamp, image_hash", [
+            ("https://x.com/SozinhoRamalho/status/1876710769913450647", "ignore tweet, testing sensitivity warning nudity", datetime.datetime(2024, 12, 31, 14, 18, 33, tzinfo=datetime.timezone.utc), "image_hash"),
+            ("https://x.com/SozinhoRamalho/status/1876710875475681357", "ignore tweet, testing sensitivity warning violence", datetime.datetime(2024, 12, 31, 14, 18, 33, tzinfo=datetime.timezone.utc), "image_hash"),
+            ("https://x.com/SozinhoRamalho/status/1876711053813227618", "ignore tweet, testing sensitivity warning sensitive", datetime.datetime(2024, 12, 31, 14, 18, 33, tzinfo=datetime.timezone.utc), "image_hash"),
+            ("https://x.com/SozinhoRamalho/status/1876711141314801937", "ignore tweet, testing sensitivity warning nudity, violence, sensitivity", datetime.datetime(2024, 12, 31, 14, 18, 33, tzinfo=datetime.timezone.utc), "image_hash"),
+        ])
+    def test_twitter_download_sensitive_media(self, url, title, timestamp, image_hash, make_item):
+
+        """Download tweets with sensitive media"""
+
+        post = self.extractor.download(make_item(url))
+        self.assertValidResponseMetadata(
+            post,
+            title,
+            timestamp
+        )
+        assert len(post.media) == 1
+        assert post.media[0].hash == image_hash

tests/extractors/test_twitter_api_extractor.py

@@ -0,0 +1,118 @@
+import os
+import datetime
+import hashlib
+import pytest
+
+from pytwitter.models.media import MediaVariant
+from .test_extractor_base import TestExtractorBase
+from auto_archiver.modules.twitter_api_extractor import TwitterApiExtractor
+
+
+@pytest.mark.incremental
+class TestTwitterApiExtractor(TestExtractorBase):
+
+    extractor_module = 'twitter_api_extractor'
+
+    config = {
+        "bearer_tokens": [],
+        "bearer_token": os.environ.get("TWITTER_BEARER_TOKEN", "TEST_KEY"),
+        "consumer_key": os.environ.get("TWITTER_CONSUMER_KEY"),
+        "consumer_secret": os.environ.get("TWITTER_CONSUMER_SECRET"),
+        "access_token": os.environ.get("TWITTER_ACCESS_TOKEN"),
+        "access_secret": os.environ.get("TWITTER_ACCESS_SECRET"),
+    }
+
+    @pytest.mark.parametrize("url, expected", [
+        ("https://t.co/yl3oOJatFp", "https://www.bellingcat.com/category/resources/"),  # t.co URL
+        ("https://x.com/bellingcat/status/1874097816571961839", "https://x.com/bellingcat/status/1874097816571961839"), # x.com urls unchanged
+        ("https://twitter.com/bellingcat/status/1874097816571961839", "https://twitter.com/bellingcat/status/1874097816571961839"), # twitter urls unchanged
+        ("https://twitter.com/bellingcat/status/1874097816571961839?s=20&t=3d0g4ZQis7dCbSDg-mE7-w", "https://twitter.com/bellingcat/status/1874097816571961839?s=20&t=3d0g4ZQis7dCbSDg-mE7-w"), # don't strip params from twitter urls (changed Jan 2025)
+        ("https://www.bellingcat.com/category/resources/", "https://www.bellingcat.com/category/resources/"), # non-twitter/x urls unchanged
+        ("https://www.bellingcat.com/category/resources/?s=20&t=3d0g4ZQis7dCbSDg-mE7-w", "https://www.bellingcat.com/category/resources/?s=20&t=3d0g4ZQis7dCbSDg-mE7-w"), # shouldn't strip params from non-twitter/x URLs
+    ])
+    def test_sanitize_url(self, url, expected):
+        assert expected == self.extractor.sanitize_url(url)
+    
+    @pytest.mark.parametrize("url, exptected_username, exptected_tweetid", [
+        ("https://twitter.com/bellingcat/status/1874097816571961839", "bellingcat", "1874097816571961839"),
+        ("https://x.com/bellingcat/status/1874097816571961839", "bellingcat", "1874097816571961839"),
+        ("https://www.bellingcat.com/category/resources/", False, False)
+        ])
+    def test_get_username_tweet_id_from_url(self, url, exptected_username, exptected_tweetid):
+    
+        username, tweet_id = self.extractor.get_username_tweet_id(url)
+        assert exptected_username == username
+        assert exptected_tweetid == tweet_id
+
+    def test_choose_variants(self):
+        # taken from the response for url https://x.com/bellingcat/status/1871552600346415571
+        variant_list = [MediaVariant(content_type='application/x-mpegURL', url='https://video.twimg.com/ext_tw_video/1871551993677852672/pu/pl/ovWo7ux-bKROwYIC.m3u8?tag=12&v=e1b'),
+                        MediaVariant(bit_rate=256000, content_type='video/mp4', url='https://video.twimg.com/ext_tw_video/1871551993677852672/pu/vid/avc1/480x270/OqZIrKV0LFswMvxS.mp4?tag=12'),
+                        MediaVariant(bit_rate=832000, content_type='video/mp4', url='https://video.twimg.com/ext_tw_video/1871551993677852672/pu/vid/avc1/640x360/uiDZDSmZ8MZn9hsi.mp4?tag=12'),
+                        MediaVariant(bit_rate=2176000, content_type='video/mp4', url='https://video.twimg.com/ext_tw_video/1871551993677852672/pu/vid/avc1/1280x720/6Y340Esh568WZnRZ.mp4?tag=12')
+                        ]
+        chosen_variant = self.extractor.choose_variant(variant_list)
+        assert chosen_variant == variant_list[3]
+    
+    @pytest.mark.skipif(not os.environ.get("TWITTER_BEARER_TOKEN"), reason="No Twitter bearer token provided")
+    @pytest.mark.download
+    def test_download_nonexistent_tweet(self, make_item):
+        # this tweet does not exist
+        url = "https://x.com/Bellingcat/status/17197025860711058"
+        response = self.extractor.download(make_item(url))
+        assert not response
+
+    @pytest.mark.skipif(not os.environ.get("TWITTER_BEARER_TOKEN"), reason="No Twitter bearer token provided")
+    @pytest.mark.download
+    def test_download_malformed_tweetid(self, make_item):
+        # this tweet does not exist
+        url = "https://x.com/Bellingcat/status/1719702586071100058"
+        response = self.extractor.download(make_item(url))
+        assert not response
+
+    @pytest.mark.skipif(not os.environ.get("TWITTER_BEARER_TOKEN"), reason="No Twitter bearer token provided")
+    @pytest.mark.download
+    def test_download_tweet_no_media(self, make_item):
+        
+        item = make_item("https://twitter.com/MeCookieMonster/status/1617921633456640001?s=20&t=3d0g4ZQis7dCbSDg-mE7-w")
+        post = self.extractor.download(item)
+
+        self.assertValidResponseMetadata(
+            post,
+            "Onion rings are just vegetable donuts.",
+            datetime.datetime(2023, 1, 24, 16, 25, 51, tzinfo=datetime.timezone.utc),
+            "twitter-api: success"
+        )
+
+    @pytest.mark.skipif(not os.environ.get("TWITTER_BEARER_TOKEN"), reason="No Twitter bearer token provided")
+    @pytest.mark.download
+    def test_download_video(self, make_item):
+        url = "https://x.com/bellingcat/status/1871552600346415571"
+        post = self.extractor.download(make_item(url))
+        self.assertValidResponseMetadata(
+            post,
+            "This month's Bellingchat Premium is with @KolinaKoltai. She reveals how she investigated a platform allowing users to create AI-generated child sexual abuse material and explains why it's crucial to investigate the people behind these services https://t.co/SfBUq0hSD0 https://t.co/rIHx0WlKp8",
+            datetime.datetime(2024, 12, 24, 13, 44, 46, tzinfo=datetime.timezone.utc)
+        )
+
+    @pytest.mark.skipif(not os.environ.get("TWITTER_BEARER_TOKEN"), reason="No Twitter bearer token provided")
+    @pytest.mark.parametrize("url, title, timestamp", [
+            ("https://x.com/SozinhoRamalho/status/1876710769913450647", "ignore tweet, testing sensitivity warning nudity https://t.co/t3u0hQsSB1", datetime.datetime(2024, 12, 31, 14, 18, 33, tzinfo=datetime.timezone.utc)),
+            ("https://x.com/SozinhoRamalho/status/1876710875475681357", "ignore tweet, testing sensitivity warning violence https://t.co/syYDSkpjZD", datetime.datetime(2024, 12, 31, 14, 18, 33, tzinfo=datetime.timezone.utc)),
+            ("https://x.com/SozinhoRamalho/status/1876711053813227618", "ignore tweet, testing sensitivity warning sensitive https://t.co/XE7cRdjzYq", datetime.datetime(2024, 12, 31, 14, 18, 33, tzinfo=datetime.timezone.utc)),
+            ("https://x.com/SozinhoRamalho/status/1876711141314801937", "ignore tweet, testing sensitivity warning nudity, violence, sensitivity https://t.co/YxCFbbhYE3", datetime.datetime(2024, 12, 31, 14, 18, 33, tzinfo=datetime.timezone.utc)),
+        ])
+    @pytest.mark.download
+    def test_download_sensitive_media(self, url, title, timestamp, check_hash, make_item):
+
+        """Download tweets with sensitive media"""
+
+        post = self.extractor.download(make_item(url))
+        self.assertValidResponseMetadata(
+            post,
+            title,
+            timestamp
+        )
+        assert len(post.media) == 1
+        # check the SHA1 hash (quick) of the media, to make sure it's valid
+        check_hash(post.media[0].filename, "3eea9c03b2dcedd1eb9a169d8bfd1cf877996fab4961de019a96eb9d32d2d733")