See #170: subscribe to 3d-party RSS feeds in Funkwhale

api/tests/audio/test_tasks.py

@@ -0,0 +1,53 @@
+import datetime
+
+import pytest
+
+from funkwhale_api.audio import tasks
+
+
+def test_fetch_rss_feeds(factories, settings, now, mocker):
+    settings.PODCASTS_RSS_FEED_REFRESH_DELAY = 5
+    prunable_date = now - datetime.timedelta(
+        seconds=settings.PODCASTS_RSS_FEED_REFRESH_DELAY
+    )
+    fetch_rss_feed = mocker.patch.object(tasks.fetch_rss_feed, "delay")
+    channels = [
+        # recent, not fetched
+        factories["audio.Channel"](actor__last_fetch_date=now, external=True),
+        # old but not external, not fetched
+        factories["audio.Channel"](actor__last_fetch_date=prunable_date),
+        # old and external, fetched !
+        factories["audio.Channel"](actor__last_fetch_date=prunable_date, external=True),
+        factories["audio.Channel"](actor__last_fetch_date=prunable_date, external=True),
+    ]
+
+    tasks.fetch_rss_feeds()
+
+    assert fetch_rss_feed.call_count == 2
+    fetch_rss_feed.assert_any_call(rss_url=channels[2].rss_url)
+    fetch_rss_feed.assert_any_call(rss_url=channels[3].rss_url)
+
+
+def test_fetch_rss_feed(factories, mocker):
+    channel = factories["audio.Channel"](external=True)
+
+    get_channel_from_rss_url = mocker.patch.object(
+        tasks.serializers, "get_channel_from_rss_url"
+    )
+    tasks.fetch_rss_feed(channel.rss_url)
+
+    get_channel_from_rss_url.assert_called_once_with(channel.rss_url)
+
+
+def test_fetch_rss_feed_blocked_is_deleted(factories, mocker):
+    channel = factories["audio.Channel"](external=True)
+
+    mocker.patch.object(
+        tasks.serializers,
+        "get_channel_from_rss_url",
+        side_effect=tasks.serializers.BlockedFeedException(),
+    )
+    tasks.fetch_rss_feed(channel.rss_url)
+
+    with pytest.raises(channel.DoesNotExist):
+        channel.refresh_from_db()