{}
", desc_text)) + + # Insert start and end date info + insert_date_info(item, parts) + + if drops_data: + parts.append( + format_html( + "{}
", + _construct_drops_summary(drops_data, channel_name=channel_name), + ), + ) + + # Only show channels if drop is not subscription only + if not getattr(item, "is_subscription_only", False) and channels is not None: + game: Game | None = getattr(item, "game", None) + parts.append(_build_channels_html(channels, game=game)) + + details_url: str | None = getattr(item, "details_url", None) + if details_url: + parts.append(format_html('About', details_url)) + + return SafeText("".join(str(p) for p in parts)) + + # MARK: /rss/reward-campaigns/ class RewardCampaignFeed(Feed): """RSS feed for latest reward campaigns (Quest rewards).""" diff --git a/twitch/management/commands/watch_imports.py b/twitch/management/commands/watch_imports.py deleted file mode 100644 index a95bb20..0000000 --- a/twitch/management/commands/watch_imports.py +++ /dev/null @@ -1,116 +0,0 @@ -import logging -from pathlib import Path -from time import sleep -from typing import TYPE_CHECKING - -from django.core.management.base import BaseCommand -from django.core.management.base import CommandError - -from twitch.management.commands.better_import_drops import ( - Command as BetterImportDropsCommand, -) - -if TYPE_CHECKING: - from django.core.management.base import CommandParser - -logger: logging.Logger = logging.getLogger("ttvdrops.watch_imports") - - -class Command(BaseCommand): - """Watch for JSON files in a directory and import them automatically.""" - - help = "Watch a directory for JSON files and import them automatically" - requires_migrations_checks = True - - def add_arguments(self, parser: CommandParser) -> None: - """Populate the command with arguments.""" - parser.add_argument( - "path", - type=str, - help="Path to directory to watch for JSON files", - ) - - def handle(self, *args, **options) -> None: # noqa: ARG002 - """Main entry point for the watch command. - - Args: - *args: Variable length argument list (unused). - **options: Arbitrary keyword arguments. - - Raises: - CommandError: If the provided path does not exist or is not a directory. - """ - watch_path: Path = self.get_watch_path(options) - self.stdout.write( - self.style.SUCCESS(f"Watching {watch_path} for JSON files..."), - ) - self.stdout.write("Press Ctrl+C to stop\n") - - importer_command: BetterImportDropsCommand = BetterImportDropsCommand() - - while True: - try: - sleep(1) - self.import_json_files( - importer_command=importer_command, - watch_path=watch_path, - ) - - except KeyboardInterrupt: - msg = "Received keyboard interrupt. Stopping watch..." - self.stdout.write(self.style.WARNING(msg)) - break - except CommandError as e: - msg = f"Import command error: {e}" - self.stdout.write(self.style.ERROR(msg)) - except Exception as e: - msg: str = f"Error while watching directory: {e}" - raise CommandError(msg) from e - - logger.info("Stopped watching directory: %s", watch_path) - - def import_json_files( - self, - importer_command: BetterImportDropsCommand, - watch_path: Path, - ) -> None: - """Import all JSON files in the watch directory using the provided importer command. - - Args: - importer_command: An instance of the BetterImportDropsCommand to handle the import logic. - watch_path: The directory path to watch for JSON files. - """ - # TODO(TheLovinator): Implement actual file watching using watchdog or similar library. # noqa: TD003 - json_files: list[Path] = [ - f for f in watch_path.iterdir() if f.suffix == ".json" and f.is_file() - ] - if not json_files: - return - - for json_file in json_files: - self.stdout.write(f"Importing {json_file}...") - importer_command.handle(path=json_file) - - def get_watch_path(self, options: dict[str, str]) -> Path: - """Validate and return the watch path from the command options. - - Args: - options: The command options containing the 'path' key. - - Returns: - The validated watch path as a Path object. - - Raises: - CommandError: If the provided path does not exist or is not a directory. - """ - watch_path: Path = Path(options["path"]).resolve() - - if not watch_path.exists(): - msg: str = f"Path does not exist: {watch_path}" - raise CommandError(msg) - - if not watch_path.is_dir(): - msg: str = f"Path is not a directory: {watch_path}, it is a {watch_path.stat().st_mode}" - raise CommandError(msg) - - return watch_path diff --git a/twitch/tests/test_feeds.py b/twitch/tests/test_feeds.py index ed4d7f0..7492691 100644 --- a/twitch/tests/test_feeds.py +++ b/twitch/tests/test_feeds.py @@ -64,14 +64,6 @@ class RSSFeedTestCase(TestCase): response: _MonkeyPatchedWSGIResponse = self.client.get(url) assert response.status_code == 200 assert response["Content-Type"] == "application/rss+xml; charset=utf-8" - content: str = response.content.decode("utf-8") - assert "Test Game by Test Organization" in content - - expected_rss_link: str = reverse( - "twitch:game_campaign_feed", - args=[self.game.twitch_id], - ) - assert expected_rss_link in content def test_campaign_feed(self) -> None: """Test campaign feed returns 200.""" @@ -123,6 +115,19 @@ class RSSFeedTestCase(TestCase): content: str = response.content.decode("utf-8") assert "Test Game" in content + def test_organization_campaign_feed(self) -> None: + """Test organization-specific campaign feed returns 200.""" + url: str = reverse( + "twitch:organization_campaign_feed", + args=[self.org.twitch_id], + ) + response: _MonkeyPatchedWSGIResponse = self.client.get(url) + assert response.status_code == 200 + assert response["Content-Type"] == "application/rss+xml; charset=utf-8" + # Verify the organization name is in the feed + content: str = response.content.decode("utf-8") + assert "Test Organization" in content + def test_game_campaign_feed_filters_correctly(self) -> None: """Test game campaign feed only shows campaigns for that game.""" # Create another game with a campaign @@ -152,6 +157,42 @@ class RSSFeedTestCase(TestCase): # Should NOT contain other campaign assert "Other Campaign" not in content + def test_organization_campaign_feed_filters_correctly(self) -> None: + """Test organization campaign feed only shows campaigns for that organization.""" + # Create another organization with a game and campaign + other_org = Organization.objects.create( + twitch_id="other-org-123", + name="Other Organization", + ) + other_game = Game.objects.create( + twitch_id="other-game-456", + slug="other-game-2", + name="Other Game 2", + display_name="Other Game 2", + ) + other_game.owners.add(other_org) + DropCampaign.objects.create( + twitch_id="other-campaign-456", + name="Other Campaign 2", + game=other_game, + start_at=timezone.now(), + end_at=timezone.now() + timedelta(days=7), + operation_names=["DropCampaignDetails"], + ) + + # Get feed for first organization + url: str = reverse( + "twitch:organization_campaign_feed", + args=[self.org.twitch_id], + ) + response: _MonkeyPatchedWSGIResponse = self.client.get(url) + content: str = response.content.decode("utf-8") + + # Should contain first campaign + assert "Test Campaign" in content + # Should NOT contain other campaign + assert "Other Campaign 2" not in content + QueryAsserter = Callable[..., AbstractContextManager[object]] @@ -400,8 +441,65 @@ def test_game_feed_queries_bounded( game.owners.add(org) url: str = reverse("twitch:game_feed") - # One query for games + one prefetch query for owners. - with django_assert_num_queries(2, exact=True): + with django_assert_num_queries(1, exact=True): + response: _MonkeyPatchedWSGIResponse = client.get(url) + + assert response.status_code == 200 + + +@pytest.mark.django_db +def test_organization_campaign_feed_queries_bounded( + client: Client, + django_assert_num_queries: QueryAsserter, +) -> None: + """Organization campaign feed should not regress in query count.""" + org: Organization = Organization.objects.create( + twitch_id="org-campaign-feed", + name="Org Campaign Feed", + ) + game: Game = Game.objects.create( + twitch_id="org-campaign-game", + slug="org-campaign-game", + name="Org Campaign Game", + display_name="Org Campaign Game", + ) + game.owners.add(org) + + for i in range(3): + _build_campaign(game, i) + + url: str = reverse("twitch:organization_campaign_feed", args=[org.twitch_id]) + # TODO(TheLovinator): 12 queries is still quite high for a feed - we should be able to optimize this further, but this is a good starting point to prevent regressions for now. # noqa: TD003 + with django_assert_num_queries(12, exact=False): + response: _MonkeyPatchedWSGIResponse = client.get(url) + + assert response.status_code == 200 + + +@pytest.mark.django_db +def test_organization_campaign_feed_queries_do_not_scale_with_items( + client: Client, + django_assert_num_queries: QueryAsserter, +) -> None: + """Organization campaign RSS feed query count should remain bounded as item count grows.""" + org: Organization = Organization.objects.create( + twitch_id="test-org-org-scale-queries", + name="Org Scale Query Org", + ) + game: Game = Game.objects.create( + twitch_id="test-game-org-scale-queries", + slug="org-scale-game", + name="Org Scale Game", + display_name="Org Scale Game", + ) + game.owners.add(org) + + for i in range(50): + _build_campaign(game, i) + + url: str = reverse("twitch:organization_campaign_feed", args=[org.twitch_id]) + + with django_assert_num_queries(15, exact=False): response: _MonkeyPatchedWSGIResponse = client.get(url) assert response.status_code == 200 @@ -493,6 +591,7 @@ URL_NAMES: list[tuple[str, dict[str, str]]] = [ ("twitch:game_feed", {}), ("twitch:game_campaign_feed", {"twitch_id": "test-game-123"}), ("twitch:organization_feed", {}), + ("twitch:organization_campaign_feed", {"twitch_id": "test-org-123"}), ("twitch:reward_campaign_feed", {}), ] diff --git a/twitch/tests/test_views.py b/twitch/tests/test_views.py index 1f32860..2cc5c80 100644 --- a/twitch/tests/test_views.py +++ b/twitch/tests/test_views.py @@ -930,36 +930,6 @@ class TestChannelListView: assert response.status_code == 200 assert "game" in response.context - @pytest.mark.django_db - def test_game_detail_view_serializes_owners_field( - self, - client: Client, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - """Game detail JSON payload should use `owners` (M2M), not stale `owner`.""" - org: Organization = Organization.objects.create( - twitch_id="org-game-detail", - name="Org Game Detail", - ) - game: Game = Game.objects.create( - twitch_id="g2-owners", - name="Game2 Owners", - display_name="Game2 Owners", - ) - game.owners.add(org) - - monkeypatch.setattr("twitch.views.format_and_color_json", lambda data: data) - - url: str = reverse("twitch:game_detail", args=[game.twitch_id]) - response: _MonkeyPatchedWSGIResponse = client.get(url) - assert response.status_code == 200 - - game_data: dict[str, Any] = response.context["game_data"] - fields: dict[str, Any] = game_data["fields"] - assert "owners" in fields - assert fields["owners"] == [org.pk] - assert "owner" not in fields - @pytest.mark.django_db def test_org_list_view(self, client: Client) -> None: """Test org list view returns 200 and has orgs in context.""" diff --git a/twitch/tests/test_watch_imports.py b/twitch/tests/test_watch_imports.py deleted file mode 100644 index 37d0005..0000000 --- a/twitch/tests/test_watch_imports.py +++ /dev/null @@ -1,214 +0,0 @@ -from pathlib import Path -from typing import TYPE_CHECKING -from unittest.mock import MagicMock - -import pytest -from _pytest.capture import CaptureResult -from django.core.management.base import CommandError - -from twitch.management.commands.watch_imports import Command - -if TYPE_CHECKING: - from pathlib import Path - - from _pytest.capture import CaptureResult - - -class TestWatchImportsCommand: - """Tests for the watch_imports management command.""" - - def test_get_watch_path_returns_resolved_directory(self, tmp_path: Path) -> None: - """It should return the resolved path when a valid directory is provided.""" - command = Command() - - result: Path = command.get_watch_path({"path": str(tmp_path)}) - - assert result == tmp_path.resolve() - - def test_get_watch_path_raises_for_missing_path(self, tmp_path: Path) -> None: - """It should raise CommandError when the path does not exist.""" - command = Command() - missing_path: Path = tmp_path / "missing" - - with pytest.raises(CommandError, match="Path does not exist"): - command.get_watch_path({"path": str(missing_path)}) - - def test_get_watch_path_raises_for_file_path(self, tmp_path: Path) -> None: - """It should raise CommandError when the path is not a directory.""" - command = Command() - file_path: Path = tmp_path / "data.json" - file_path.write_text("{}", encoding="utf-8") - - with pytest.raises(CommandError, match="Path is not a directory"): - command.get_watch_path({"path": str(file_path)}) - - def test_import_json_files_imports_only_json_files(self, tmp_path: Path) -> None: - """It should call importer for .json files and ignore other entries.""" - command = Command() - importer_command = MagicMock() - - json_file_1: Path = tmp_path / "one.json" - json_file_2: Path = tmp_path / "two.json" - ignored_txt: Path = tmp_path / "notes.txt" - ignored_dir: Path = tmp_path / "nested.json" - - json_file_1.write_text("{}", encoding="utf-8") - json_file_2.write_text("[]", encoding="utf-8") - ignored_txt.write_text("ignored", encoding="utf-8") - ignored_dir.mkdir() - - command.import_json_files( - importer_command=importer_command, - watch_path=tmp_path, - ) - - imported_paths: list[Path] = [ - call.kwargs["path"] for call in importer_command.handle.call_args_list - ] - assert set(imported_paths) == {json_file_1, json_file_2} - - def test_import_json_files_no_json_files_does_nothing(self, tmp_path: Path) -> None: - """It should not call importer when no JSON files are present.""" - command = Command() - importer_command = MagicMock() - - (tmp_path / "notes.txt").write_text("ignored", encoding="utf-8") - - command.import_json_files( - importer_command=importer_command, - watch_path=tmp_path, - ) - - importer_command.handle.assert_not_called() - - def test_handle_stops_cleanly_on_keyboard_interrupt( - self, - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - capsys: pytest.CaptureFixture[str], - ) -> None: - """It should print a warning and stop when interrupted by keyboard.""" - command = Command() - importer_instance = MagicMock() - - monkeypatch.setattr(command, "get_watch_path", lambda options: tmp_path) - monkeypatch.setattr( - "twitch.management.commands.watch_imports.BetterImportDropsCommand", - lambda: importer_instance, - ) - monkeypatch.setattr( - "twitch.management.commands.watch_imports.sleep", - lambda _seconds: (_ for _ in ()).throw(KeyboardInterrupt()), - ) - - command.handle(path=str(tmp_path)) - - captured: CaptureResult[str] = capsys.readouterr() - assert "Watching" in captured.out - assert "Received keyboard interrupt. Stopping watch..." in captured.out - - def test_handle_reports_command_errors_and_continues_until_interrupt( - self, - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - capsys: pytest.CaptureFixture[str], - ) -> None: - """It should report CommandError from import and continue the loop.""" - command = Command() - importer_instance = MagicMock() - sleep_calls: dict[str, int] = {"count": 0} - - def fake_sleep(_seconds: int) -> None: - """Simulate sleep and raise KeyboardInterrupt after 2 calls. - - Args: - _seconds: The number of seconds to sleep (ignored). - - Raises: - KeyboardInterrupt: After being called twice to simulate user interrupt. - """ - sleep_calls["count"] += 1 - if sleep_calls["count"] == 2: - raise KeyboardInterrupt - - def fake_import_json_files( - *, - importer_command: MagicMock, - watch_path: Path, - ) -> None: - """Simulate an import that raises CommandError. - - Args: - importer_command: The mock importer command instance (ignored). - watch_path: The path being watched (ignored). - - Raises: - CommandError: Always raised to simulate an import error. - """ - msg = "bad import" - raise CommandError(msg) - - monkeypatch.setattr( - command, - "get_watch_path", - lambda options: tmp_path, - ) - monkeypatch.setattr( - "twitch.management.commands.watch_imports.BetterImportDropsCommand", - lambda: importer_instance, - ) - monkeypatch.setattr( - "twitch.management.commands.watch_imports.sleep", - fake_sleep, - ) - monkeypatch.setattr( - command, - "import_json_files", - fake_import_json_files, - ) - - command.handle(path=str(tmp_path)) - - captured: CaptureResult[str] = capsys.readouterr() - assert "Import command error: bad import" in captured.out - assert "Received keyboard interrupt. Stopping watch..." in captured.out - - def test_handle_wraps_unexpected_errors_in_command_error( - self, - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - ) -> None: - """It should wrap unexpected exceptions in a CommandError.""" - command = Command() - importer_instance = MagicMock() - - monkeypatch.setattr( - command, - "get_watch_path", - lambda options: tmp_path, - ) - monkeypatch.setattr( - "twitch.management.commands.watch_imports.BetterImportDropsCommand", - lambda: importer_instance, - ) - monkeypatch.setattr( - "twitch.management.commands.watch_imports.sleep", - lambda _seconds: None, - ) - - def fake_import_json_files( - *, - importer_command: MagicMock, - watch_path: Path, - ) -> None: - msg = "boom" - raise RuntimeError(msg) - - monkeypatch.setattr( - command, - "import_json_files", - fake_import_json_files, - ) - - with pytest.raises(CommandError, match="Error while watching directory: boom"): - command.handle(path=str(tmp_path)) diff --git a/twitch/urls.py b/twitch/urls.py index 45543de..f414bb5 100644 --- a/twitch/urls.py +++ b/twitch/urls.py @@ -6,6 +6,7 @@ from twitch import views from twitch.feeds import DropCampaignFeed from twitch.feeds import GameCampaignFeed from twitch.feeds import GameFeed +from twitch.feeds import OrganizationCampaignFeed from twitch.feeds import OrganizationRSSFeed from twitch.feeds import RewardCampaignFeed @@ -82,27 +83,18 @@ urlpatterns: list[URLPattern] = [ views.export_organizations_json, name="export_organizations_json", ), - # RSS feeds - # /rss/campaigns/ - all active campaigns path("rss/campaigns/", DropCampaignFeed(), name="campaign_feed"), - # /rss/games/ - newly added games path("rss/games/", GameFeed(), name="game_feed"), - # /rss/games/