diff --git a/pyproject.toml b/pyproject.toml index d0ae287..c9b6965 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,9 @@ dev = ["pytest>=8.4.1", "pytest-django>=4.11.1"] DJANGO_SETTINGS_MODULE = "config.settings" python_files = ["test_*.py", "*_test.py"] addopts = ["--reuse-db", "--no-migrations"] +filterwarnings = [ + "ignore:Parsing dates involving a day of month without a year specified is ambiguous:DeprecationWarning", +] [tool.ruff] lint.select = ["ALL"] diff --git a/twitch/schemas.py b/twitch/schemas.py index 9c77f78..b452bd7 100644 --- a/twitch/schemas.py +++ b/twitch/schemas.py @@ -170,7 +170,7 @@ class TimeBasedDropSchema(BaseModel): required_subs: int = Field(alias="requiredSubs") start_at: str | None = Field(alias="startAt") end_at: str | None = Field(alias="endAt") - benefit_edges: list[DropBenefitEdgeSchema] = Field(alias="benefitEdges") + benefit_edges: list[DropBenefitEdgeSchema] = Field(default=[], alias="benefitEdges") type_name: Literal["TimeBasedDrop"] = Field(alias="__typename") # Inventory-specific fields precondition_drops: None = Field(default=None, alias="preconditionDrops") @@ -185,6 +185,19 @@ class TimeBasedDropSchema(BaseModel): "populate_by_name": True, } + @field_validator("benefit_edges", mode="before") + @classmethod + def handle_null_benefit_edges(cls, v: list | None) -> list: + """Convert null benefitEdges to empty list. + + Args: + v: The raw benefit_edges value (list or None). + + Returns: + Empty list if None, otherwise the list itself. + """ + return v or [] + class DropCampaignSchema(BaseModel): """Schema for Twitch DropCampaign objects. @@ -222,7 +235,7 @@ class DropCampaignSchema(BaseModel): class InventorySchema(BaseModel): """Schema for the inventory field in Inventory operation responses.""" - drop_campaigns_in_progress: list[DropCampaignSchema] = Field(alias="dropCampaignsInProgress") + drop_campaigns_in_progress: list[DropCampaignSchema] = Field(default=[], alias="dropCampaignsInProgress") type_name: Literal["Inventory"] = Field(alias="__typename") # gameEventDrops field is present in Inventory but we don't process it yet game_event_drops: list | None = Field(default=None, alias="gameEventDrops") @@ -234,6 +247,19 @@ class InventorySchema(BaseModel): "populate_by_name": True, } + @field_validator("drop_campaigns_in_progress", mode="before") + @classmethod + def handle_null_campaigns(cls, v: list | None) -> list: + """Convert null dropCampaignsInProgress to empty list. + + Args: + v: The raw drop_campaigns_in_progress value (list or None). + + Returns: + Empty list if None, otherwise the list itself. + """ + return v or [] + class CurrentUserSchema(BaseModel): """Schema for Twitch User objects. diff --git a/twitch/tests/test_better_import_drops.py b/twitch/tests/test_better_import_drops.py index ec18428..bb9d5bd 100644 --- a/twitch/tests/test_better_import_drops.py +++ b/twitch/tests/test_better_import_drops.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING from django.test import TestCase from twitch.management.commands.better_import_drops import Command +from twitch.models import DropCampaign from twitch.schemas import DropBenefitSchema if TYPE_CHECKING: @@ -92,3 +93,295 @@ class ExtractCampaignsTests(TestCase): assert len(valid_responses) == 1 assert broken_dir is None assert valid_responses[0].data.user is not None + + def test_imports_inventory_response_and_sets_operation_name(self) -> None: + """Ensure Inventory JSON imports work and operation_name is set correctly.""" + command = Command() + command.pre_fill_cache() + + # Inventory response with dropCampaignsInProgress + payload = { + "data": { + "currentUser": { + "id": "17658559", + "inventory": { + "dropCampaignsInProgress": [ + { + "id": "inventory-campaign-1", + "name": "Test Inventory Campaign", + "description": "Campaign from Inventory operation", + "startAt": "2025-01-01T00:00:00Z", + "endAt": "2025-12-31T23:59:59Z", + "accountLinkURL": "https://example.com/link", + "detailsURL": "https://example.com/details", + "imageURL": "https://example.com/campaign.png", + "status": "ACTIVE", + "self": { + "isAccountConnected": True, + "__typename": "DropCampaignSelfEdge", + }, + "game": { + "id": "inventory-game-1", + "displayName": "Inventory Game", + "boxArtURL": "https://example.com/boxart.png", + "slug": "inventory-game", + "name": "Inventory Game", + "__typename": "Game", + }, + "owner": { + "id": "inventory-org-1", + "name": "Inventory Organization", + "__typename": "Organization", + }, + "timeBasedDrops": [], + "eventBasedDrops": None, + "__typename": "DropCampaign", + }, + ], + "gameEventDrops": None, + "__typename": "Inventory", + }, + "__typename": "User", + }, + }, + "extensions": { + "operationName": "Inventory", + }, + } + + # Validate and process response + success, broken_dir = command.process_responses( + responses=[payload], + file_path=Path("test_inventory.json"), + options={}, + ) + + assert success is True + assert broken_dir is None + + # Check that campaign was created with operation_name + campaign = DropCampaign.objects.get(twitch_id="inventory-campaign-1") + assert campaign.name == "Test Inventory Campaign" + assert campaign.operation_name == "Inventory" + + def test_handles_inventory_with_null_campaigns(self) -> None: + """Ensure Inventory JSON with null dropCampaignsInProgress is handled correctly.""" + command = Command() + command.pre_fill_cache() + + # Inventory response with null dropCampaignsInProgress + payload = { + "data": { + "currentUser": { + "id": "17658559", + "inventory": { + "dropCampaignsInProgress": None, + "gameEventDrops": None, + "__typename": "Inventory", + }, + "__typename": "User", + }, + }, + "extensions": { + "operationName": "Inventory", + }, + } + + # Should validate successfully even with null campaigns + valid_responses, broken_dir = command._validate_responses( + responses=[payload], + file_path=Path("test_inventory_null.json"), + options={}, + ) + + assert len(valid_responses) == 1 + assert broken_dir is None + assert valid_responses[0].data.current_user is not None + assert valid_responses[0].data.current_user.inventory is not None + + +class CampaignStructureDetectionTests(TestCase): + """Tests for campaign structure detection in _detect_campaign_structure method.""" + + def test_detects_inventory_campaigns_structure(self) -> None: + """Ensure inventory campaign structure is correctly detected.""" + command = Command() + + # Inventory format with dropCampaignsInProgress + response = { + "data": { + "currentUser": { + "id": "123", + "inventory": { + "dropCampaignsInProgress": [ + { + "id": "c1", + "name": "Test Campaign", + }, + ], + "__typename": "Inventory", + }, + "__typename": "User", + }, + }, + } + + structure = command._detect_campaign_structure(response) + assert structure == "inventory_campaigns" + + def test_detects_inventory_campaigns_structure_with_null(self) -> None: + """Ensure inventory format is NOT detected when dropCampaignsInProgress is null.""" + command = Command() + + # Inventory format with null dropCampaignsInProgress - should not detect as inventory_campaigns + response = { + "data": { + "currentUser": { + "id": "123", + "inventory": { + "dropCampaignsInProgress": None, + "__typename": "Inventory", + }, + "__typename": "User", + }, + }, + } + + structure = command._detect_campaign_structure(response) + # Should return None since there are no actual campaigns + assert structure is None + + def test_detects_current_user_drop_campaigns_structure(self) -> None: + """Ensure currentUser.dropCampaigns structure is correctly detected.""" + command = Command() + + response = { + "data": { + "currentUser": { + "id": "123", + "dropCampaigns": [ + { + "id": "c1", + "name": "Test Campaign", + }, + ], + "__typename": "User", + }, + }, + } + + structure = command._detect_campaign_structure(response) + assert structure == "current_user_drop_campaigns" + + +class OperationNameFilteringTests(TestCase): + """Tests for filtering campaigns by operation_name field.""" + + def test_can_filter_campaigns_by_operation_name(self) -> None: + """Ensure campaigns can be filtered by operation_name to separate data sources.""" + command = Command() + command.pre_fill_cache() + + # Import a ViewerDropsDashboard campaign + viewer_drops_payload = { + "data": { + "currentUser": { + "id": "123", + "dropCampaigns": [ + { + "id": "viewer-campaign-1", + "name": "Viewer Campaign", + "description": "", + "startAt": "2025-01-01T00:00:00Z", + "endAt": "2025-12-31T23:59:59Z", + "accountLinkURL": "https://example.com", + "detailsURL": "https://example.com", + "imageURL": "", + "status": "ACTIVE", + "self": {"isAccountConnected": False, "__typename": "DropCampaignSelfEdge"}, + "game": { + "id": "game-1", + "displayName": "Game 1", + "boxArtURL": "https://example.com/art.png", + "__typename": "Game", + }, + "owner": { + "id": "org-1", + "name": "Org 1", + "__typename": "Organization", + }, + "timeBasedDrops": [], + "__typename": "DropCampaign", + }, + ], + "__typename": "User", + }, + }, + "extensions": { + "operationName": "ViewerDropsDashboard", + }, + } + + # Import an Inventory campaign + inventory_payload = { + "data": { + "currentUser": { + "id": "123", + "inventory": { + "dropCampaignsInProgress": [ + { + "id": "inventory-campaign-1", + "name": "Inventory Campaign", + "description": "", + "startAt": "2025-01-01T00:00:00Z", + "endAt": "2025-12-31T23:59:59Z", + "accountLinkURL": "https://example.com", + "detailsURL": "https://example.com", + "imageURL": "", + "status": "ACTIVE", + "self": {"isAccountConnected": True, "__typename": "DropCampaignSelfEdge"}, + "game": { + "id": "game-2", + "displayName": "Game 2", + "boxArtURL": "https://example.com/art2.png", + "__typename": "Game", + }, + "owner": { + "id": "org-2", + "name": "Org 2", + "__typename": "Organization", + }, + "timeBasedDrops": [], + "eventBasedDrops": None, + "__typename": "DropCampaign", + }, + ], + "gameEventDrops": None, + "__typename": "Inventory", + }, + "__typename": "User", + }, + }, + "extensions": { + "operationName": "Inventory", + }, + } + + # Process both payloads + command.process_responses([viewer_drops_payload], Path("viewer.json"), {}) + command.process_responses([inventory_payload], Path("inventory.json"), {}) + + # Verify we can filter by operation_name + viewer_campaigns = DropCampaign.objects.filter(operation_name="ViewerDropsDashboard") + inventory_campaigns = DropCampaign.objects.filter(operation_name="Inventory") + + assert viewer_campaigns.count() >= 1 + assert inventory_campaigns.count() >= 1 + + # Verify the correct campaigns are in each queryset + assert viewer_campaigns.filter(twitch_id="viewer-campaign-1").exists() + assert inventory_campaigns.filter(twitch_id="inventory-campaign-1").exists() + + # Cross-check: Inventory campaign should not be in viewer campaigns + assert not viewer_campaigns.filter(twitch_id="inventory-campaign-1").exists() + assert not inventory_campaigns.filter(twitch_id="viewer-campaign-1").exists()