Only show current drops

twitch/feeds.py

@@ -58,6 +58,28 @@ def _with_campaign_related(queryset: QuerySet[DropCampaign]) -> QuerySet[DropCam
     )
 
 
+def _active_drop_campaigns(queryset: QuerySet[DropCampaign]) -> QuerySet[DropCampaign]:
+    """Filter a campaign queryset down to campaigns active right now.
+
+    Returns:
+        QuerySet[DropCampaign]: Queryset with only active drop campaigns.
+    """
+    now: datetime.datetime = timezone.now()
+    return queryset.filter(start_at__lte=now, end_at__gte=now)
+
+
+def _active_reward_campaigns(
+    queryset: QuerySet[RewardCampaign],
+) -> QuerySet[RewardCampaign]:
+    """Filter a reward campaign queryset down to campaigns active right now.
+
+    Returns:
+        QuerySet[RewardCampaign]: Queryset with only active reward campaigns.
+    """
+    now: datetime.datetime = timezone.now()
+    return queryset.filter(starts_at__lte=now, ends_at__gte=now)
+
+
 def genereate_details_link_html(item: DropCampaign) -> list[SafeText]:
     """Helper method to append a details link to the description if available.
 
@@ -733,7 +755,9 @@ class DropCampaignFeed(Feed):
     def items(self) -> list[DropCampaign]:
         """Return the latest drop campaigns ordered by most recent start date (default 200, or limited by ?limit query param)."""
         limit: int = self._limit if self._limit is not None else 200
-        queryset: QuerySet[DropCampaign] = DropCampaign.objects.order_by("-start_at")
+        queryset: QuerySet[DropCampaign] = _active_drop_campaigns(
+            DropCampaign.objects.order_by("-start_at"),
+        )
         return list(_with_campaign_related(queryset)[:limit])
 
     def item_title(self, item: DropCampaign) -> SafeText:
@@ -859,9 +883,11 @@ class GameCampaignFeed(Feed):
     def items(self, obj: Game) -> list[DropCampaign]:
         """Return the latest drop campaigns for this game, ordered by most recent start date (default 200, or limited by ?limit query param)."""
         limit: int = self._limit if self._limit is not None else 200
-        queryset: QuerySet[DropCampaign] = DropCampaign.objects.filter(
-            game=obj,
-        ).order_by("-start_at")
+        queryset: QuerySet[DropCampaign] = _active_drop_campaigns(
+            DropCampaign.objects.filter(
+                game=obj,
+            ).order_by("-start_at"),
+        )
         return list(_with_campaign_related(queryset)[:limit])
 
     def item_title(self, item: DropCampaign) -> SafeText:
@@ -963,8 +989,11 @@ class RewardCampaignFeed(Feed):
     def items(self) -> list[RewardCampaign]:
         """Return the latest reward campaigns (default 200, or limited by ?limit query param)."""
         limit: int = self._limit if self._limit is not None else 200
+        queryset: QuerySet[RewardCampaign] = _active_reward_campaigns(
+            RewardCampaign.objects.select_related("game").order_by("-added_at"),
+        )
         return list(
-            RewardCampaign.objects.select_related("game").order_by("-added_at")[:limit],
+            queryset[:limit],
         )
 
     def item_title(self, item: RewardCampaign) -> SafeText:

twitch/tests/test_feeds.py

@@ -29,6 +29,8 @@ from twitch.models import TimeBasedDrop
 logger: logging.Logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
+    import datetime
+
     from django.test.client import _MonkeyPatchedWSGIResponse
 
     from twitch.tests.test_badge_views import Client
@@ -146,6 +148,35 @@ class RSSFeedTestCase(TestCase):
         assert 'length="314"' in content
         assert 'type="image/gif"' in content
 
+    def test_campaign_feed_only_includes_active_campaigns(self) -> None:
+        """Campaign feed should exclude past and upcoming campaigns."""
+        now: datetime.datetime = timezone.now()
+        DropCampaign.objects.create(
+            twitch_id="past-campaign-123",
+            name="Past Campaign",
+            game=self.game,
+            start_at=now - timedelta(days=10),
+            end_at=now - timedelta(days=1),
+            operation_names=["DropCampaignDetails"],
+        )
+        DropCampaign.objects.create(
+            twitch_id="upcoming-campaign-123",
+            name="Upcoming Campaign",
+            game=self.game,
+            start_at=now + timedelta(days=1),
+            end_at=now + timedelta(days=10),
+            operation_names=["DropCampaignDetails"],
+        )
+
+        url: str = reverse("twitch:campaign_feed")
+        response: _MonkeyPatchedWSGIResponse = self.client.get(url)
+        assert response.status_code == 200
+        content: str = response.content.decode("utf-8")
+
+        assert "Test Campaign" in content
+        assert "Past Campaign" not in content
+        assert "Upcoming Campaign" not in content
+
     def test_campaign_feed_enclosure_helpers(self) -> None:
         """Helper methods for campaigns should respect new fields."""
         feed = DropCampaignFeed()
@@ -228,6 +259,90 @@ class RSSFeedTestCase(TestCase):
         assert 'length="314"' in content
         assert 'type="image/gif"' in content
 
+    def test_game_campaign_feed_only_includes_active_campaigns(self) -> None:
+        """Game campaign feed should exclude old and upcoming campaigns."""
+        now: datetime.datetime = timezone.now()
+        DropCampaign.objects.create(
+            twitch_id="game-past-campaign-123",
+            name="Game Past Campaign",
+            game=self.game,
+            start_at=now - timedelta(days=10),
+            end_at=now - timedelta(days=1),
+            operation_names=["DropCampaignDetails"],
+        )
+        DropCampaign.objects.create(
+            twitch_id="game-upcoming-campaign-123",
+            name="Game Upcoming Campaign",
+            game=self.game,
+            start_at=now + timedelta(days=1),
+            end_at=now + timedelta(days=10),
+            operation_names=["DropCampaignDetails"],
+        )
+
+        url: str = reverse("twitch:game_campaign_feed", args=[self.game.twitch_id])
+        response: _MonkeyPatchedWSGIResponse = self.client.get(url)
+        assert response.status_code == 200
+        content: str = response.content.decode("utf-8")
+
+        assert "Test Campaign" in content
+        assert "Game Past Campaign" not in content
+        assert "Game Upcoming Campaign" not in content
+
+    def test_reward_campaign_feed_only_includes_active_campaigns(self) -> None:
+        """Reward campaign feed should exclude old and upcoming campaigns."""
+        now: datetime.datetime = timezone.now()
+        RewardCampaign.objects.create(
+            twitch_id="active-reward-123",
+            name="Active Reward Campaign",
+            brand="Test Brand",
+            starts_at=now - timedelta(days=1),
+            ends_at=now + timedelta(days=1),
+            status="ACTIVE",
+            summary="Active reward",
+            instructions="Do things",
+            external_url="https://example.com/active-reward",
+            about_url="https://example.com/about-active-reward",
+            is_sitewide=False,
+            game=self.game,
+        )
+        RewardCampaign.objects.create(
+            twitch_id="past-reward-123",
+            name="Past Reward Campaign",
+            brand="Test Brand",
+            starts_at=now - timedelta(days=10),
+            ends_at=now - timedelta(days=1),
+            status="EXPIRED",
+            summary="Past reward",
+            instructions="Was active",
+            external_url="https://example.com/past-reward",
+            about_url="https://example.com/about-past-reward",
+            is_sitewide=False,
+            game=self.game,
+        )
+        RewardCampaign.objects.create(
+            twitch_id="upcoming-reward-123",
+            name="Upcoming Reward Campaign",
+            brand="Test Brand",
+            starts_at=now + timedelta(days=1),
+            ends_at=now + timedelta(days=10),
+            status="UPCOMING",
+            summary="Upcoming reward",
+            instructions="Wait",
+            external_url="https://example.com/upcoming-reward",
+            about_url="https://example.com/about-upcoming-reward",
+            is_sitewide=False,
+            game=self.game,
+        )
+
+        url: str = reverse("twitch:reward_campaign_feed")
+        response: _MonkeyPatchedWSGIResponse = self.client.get(url)
+        assert response.status_code == 200
+        content: str = response.content.decode("utf-8")
+
+        assert "Active Reward Campaign" in content
+        assert "Past Reward Campaign" not in content
+        assert "Upcoming Reward Campaign" not in content
+
     def test_game_campaign_feed_enclosure_helpers(self) -> None:
         """GameCampaignFeed helper methods should pull from the model fields."""
         feed = GameCampaignFeed()