{}
", 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 new file mode 100644 index 0000000..a95bb20 --- /dev/null +++ b/twitch/management/commands/watch_imports.py @@ -0,0 +1,116 @@ +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 7492691..ed4d7f0 100644 --- a/twitch/tests/test_feeds.py +++ b/twitch/tests/test_feeds.py @@ -64,6 +64,14 @@ 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.""" @@ -115,19 +123,6 @@ 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 @@ -157,42 +152,6 @@ 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]] @@ -441,65 +400,8 @@ def test_game_feed_queries_bounded( game.owners.add(org) url: str = reverse("twitch:game_feed") - 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): + # One query for games + one prefetch query for owners. + with django_assert_num_queries(2, exact=True): response: _MonkeyPatchedWSGIResponse = client.get(url) assert response.status_code == 200 @@ -591,7 +493,6 @@ 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 2cc5c80..1f32860 100644 --- a/twitch/tests/test_views.py +++ b/twitch/tests/test_views.py @@ -930,6 +930,36 @@ 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 new file mode 100644 index 0000000..37d0005 --- /dev/null +++ b/twitch/tests/test_watch_imports.py @@ -0,0 +1,214 @@ +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 f414bb5..45543de 100644 --- a/twitch/urls.py +++ b/twitch/urls.py @@ -6,7 +6,6 @@ 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 @@ -83,18 +82,27 @@ 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/