From 0751c6cd0be7e81d66f5eccce36c00d2c85a6c4e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Hells=C3=A9n?= Date: Tue, 6 Jan 2026 23:06:31 +0100 Subject: [PATCH] Enhance DropCampaign handling: default is_enabled to True if missing, add url field to ChannelInfoSchema --- .vscode/settings.json | 1 + .../commands/better_import_drops.py | 6 +- twitch/schemas.py | 3 +- twitch/tests/test_better_import_drops.py | 120 ++++++++++++++++-- twitch/tests/test_schemas.py | 29 +++-- 5 files changed, 131 insertions(+), 28 deletions(-) diff --git a/.vscode/settings.json b/.vscode/settings.json index a8a8a02..6170aae 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -40,6 +40,7 @@ "runserver", "sendgrid", "speculationrules", + "testchannel", "testpass", "tqdm", "ttvdrops", diff --git a/twitch/management/commands/better_import_drops.py b/twitch/management/commands/better_import_drops.py index ce7a06b..7712078 100644 --- a/twitch/management/commands/better_import_drops.py +++ b/twitch/management/commands/better_import_drops.py @@ -793,8 +793,10 @@ class Command(BaseCommand): allow_schema: The DropCampaignACL Pydantic schema. """ # Update the allow_is_enabled flag if changed - if campaign_obj.allow_is_enabled != allow_schema.is_enabled: - campaign_obj.allow_is_enabled = allow_schema.is_enabled + # Default to True if is_enabled is None (API doesn't always provide this field) + is_enabled: bool = allow_schema.is_enabled if allow_schema.is_enabled is not None else True + if campaign_obj.allow_is_enabled != is_enabled: + campaign_obj.allow_is_enabled = is_enabled campaign_obj.save(update_fields=["allow_is_enabled"]) # Get or create all channels and collect them diff --git a/twitch/schemas.py b/twitch/schemas.py index b452bd7..4e3884c 100644 --- a/twitch/schemas.py +++ b/twitch/schemas.py @@ -85,6 +85,7 @@ class ChannelInfoSchema(BaseModel): twitch_id: str = Field(alias="id") display_name: str | None = Field(default=None, alias="displayName") name: str # Channel login name + url: str | None = None type_name: Literal["Channel"] = Field(alias="__typename") model_config = { @@ -102,7 +103,7 @@ class DropCampaignACLSchema(BaseModel): """ channels: list[ChannelInfoSchema] | None = None - is_enabled: bool = Field(alias="isEnabled") + is_enabled: bool | None = Field(default=None, alias="isEnabled") type_name: Literal["DropCampaignACL"] = Field(alias="__typename") model_config = { diff --git a/twitch/tests/test_better_import_drops.py b/twitch/tests/test_better_import_drops.py index bb9d5bd..ea8fdf3 100644 --- a/twitch/tests/test_better_import_drops.py +++ b/twitch/tests/test_better_import_drops.py @@ -6,10 +6,13 @@ from typing import TYPE_CHECKING from django.test import TestCase from twitch.management.commands.better_import_drops import Command +from twitch.models import Channel from twitch.models import DropCampaign from twitch.schemas import DropBenefitSchema if TYPE_CHECKING: + from debug_toolbar.panels.templates.panel import QuerySet + from twitch.models import DropBenefit @@ -46,7 +49,7 @@ class ExtractCampaignsTests(TestCase): command = Command() command.pre_fill_cache() - payload = { + payload: dict[str, object] = { "data": { "user": { "id": "123", @@ -100,7 +103,7 @@ class ExtractCampaignsTests(TestCase): command.pre_fill_cache() # Inventory response with dropCampaignsInProgress - payload = { + payload: dict[str, object] = { "data": { "currentUser": { "id": "17658559", @@ -160,7 +163,7 @@ class ExtractCampaignsTests(TestCase): assert broken_dir is None # Check that campaign was created with operation_name - campaign = DropCampaign.objects.get(twitch_id="inventory-campaign-1") + campaign: DropCampaign = DropCampaign.objects.get(twitch_id="inventory-campaign-1") assert campaign.name == "Test Inventory Campaign" assert campaign.operation_name == "Inventory" @@ -170,7 +173,7 @@ class ExtractCampaignsTests(TestCase): command.pre_fill_cache() # Inventory response with null dropCampaignsInProgress - payload = { + payload: dict[str, object] = { "data": { "currentUser": { "id": "17658559", @@ -199,6 +202,95 @@ class ExtractCampaignsTests(TestCase): assert valid_responses[0].data.current_user is not None assert valid_responses[0].data.current_user.inventory is not None + def test_handles_inventory_with_allow_acl_url_and_missing_is_enabled(self) -> None: + """Ensure ACL with url field and missing isEnabled is handled correctly.""" + command = Command() + command.pre_fill_cache() + + # Inventory response with allow ACL containing url field and no isEnabled + payload: dict[str, object] = { + "data": { + "currentUser": { + "id": "17658559", + "inventory": { + "dropCampaignsInProgress": [ + { + "id": "inventory-campaign-2", + "name": "Test ACL Campaign", + "description": "", + "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-2", + "displayName": "Test Game", + "boxArtURL": "https://example.com/boxart.png", + "slug": "test-game", + "name": "Test Game", + "__typename": "Game", + }, + "owner": { + "id": "inventory-org-2", + "name": "Test Organization", + "__typename": "Organization", + }, + "allow": { + "channels": [ + { + "id": "91070599", + "name": "testchannel", + "url": "https://www.twitch.tv/testchannel", + "__typename": "Channel", + }, + ], + "__typename": "DropCampaignACL", + }, + "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_acl.json"), + options={}, + ) + + assert success is True + assert broken_dir is None + + # Check that campaign was created and allow_is_enabled defaults to True + campaign: DropCampaign = DropCampaign.objects.get(twitch_id="inventory-campaign-2") + assert campaign.name == "Test ACL Campaign" + assert campaign.allow_is_enabled is True # Should default to True + + # Check that the channel was created and linked + assert campaign.allow_channels.count() == 1 + channel: Channel | None = campaign.allow_channels.first() + assert channel is not None + + assert channel.name == "testchannel" + assert channel.twitch_id == "91070599" + class CampaignStructureDetectionTests(TestCase): """Tests for campaign structure detection in _detect_campaign_structure method.""" @@ -208,7 +300,7 @@ class CampaignStructureDetectionTests(TestCase): command = Command() # Inventory format with dropCampaignsInProgress - response = { + response: dict[str, object] = { "data": { "currentUser": { "id": "123", @@ -226,7 +318,7 @@ class CampaignStructureDetectionTests(TestCase): }, } - structure = command._detect_campaign_structure(response) + structure: str | None = command._detect_campaign_structure(response) assert structure == "inventory_campaigns" def test_detects_inventory_campaigns_structure_with_null(self) -> None: @@ -247,7 +339,7 @@ class CampaignStructureDetectionTests(TestCase): }, } - structure = command._detect_campaign_structure(response) + structure: str | None = command._detect_campaign_structure(response) # Should return None since there are no actual campaigns assert structure is None @@ -255,7 +347,7 @@ class CampaignStructureDetectionTests(TestCase): """Ensure currentUser.dropCampaigns structure is correctly detected.""" command = Command() - response = { + response: dict[str, object] = { "data": { "currentUser": { "id": "123", @@ -270,7 +362,7 @@ class CampaignStructureDetectionTests(TestCase): }, } - structure = command._detect_campaign_structure(response) + structure: str | None = command._detect_campaign_structure(response) assert structure == "current_user_drop_campaigns" @@ -323,7 +415,7 @@ class OperationNameFilteringTests(TestCase): } # Import an Inventory campaign - inventory_payload = { + inventory_payload: dict[str, object] = { "data": { "currentUser": { "id": "123", @@ -372,8 +464,12 @@ class OperationNameFilteringTests(TestCase): 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") + viewer_campaigns: QuerySet[DropCampaign, DropCampaign] = DropCampaign.objects.filter( + operation_name="ViewerDropsDashboard", + ) + inventory_campaigns: QuerySet[DropCampaign, DropCampaign] = DropCampaign.objects.filter( + operation_name="Inventory", + ) assert viewer_campaigns.count() >= 1 assert inventory_campaigns.count() >= 1 diff --git a/twitch/tests/test_schemas.py b/twitch/tests/test_schemas.py index 1a704a7..3dc5059 100644 --- a/twitch/tests/test_schemas.py +++ b/twitch/tests/test_schemas.py @@ -2,7 +2,10 @@ from __future__ import annotations +from twitch.schemas import DropBenefitSchema +from twitch.schemas import DropCampaignSchema from twitch.schemas import GraphQLResponse +from twitch.schemas import TimeBasedDropSchema def test_inventory_operation_validation() -> None: @@ -17,7 +20,7 @@ def test_inventory_operation_validation() -> None: - Benefits have 'DropBenefit' as __typename instead of 'Benefit' """ # Minimal valid Inventory operation structure - payload = { + payload: dict[str, object] = { "data": { "currentUser": { "id": "17658559", @@ -90,7 +93,7 @@ def test_inventory_operation_validation() -> None: } # This should not raise ValidationError - response = GraphQLResponse.model_validate(payload) + response: GraphQLResponse = GraphQLResponse.model_validate(payload) # Verify the structure was parsed correctly assert response.data.current_user is not None @@ -100,13 +103,13 @@ def test_inventory_operation_validation() -> None: assert response.data.current_user.drop_campaigns is not None assert len(response.data.current_user.drop_campaigns) == 1 - campaign = response.data.current_user.drop_campaigns[0] + campaign: DropCampaignSchema = response.data.current_user.drop_campaigns[0] assert campaign.name == "Test Campaign" assert campaign.game.display_name == "Test Game" assert len(campaign.time_based_drops) == 1 # Verify time-based drops - first_drop = campaign.time_based_drops[0] + first_drop: TimeBasedDropSchema = campaign.time_based_drops[0] assert first_drop.name == "Test Drop" assert first_drop.required_minutes_watched == 60 @@ -121,7 +124,7 @@ def test_viewer_drops_dashboard_operation_still_works() -> None: This ensures backward compatibility with the existing import data. """ # Minimal valid ViewerDropsDashboard structure - payload = { + payload: dict[str, object] = { "data": { "currentUser": { "id": "12345", @@ -163,7 +166,7 @@ def test_viewer_drops_dashboard_operation_still_works() -> None: } # This should not raise ValidationError - response = GraphQLResponse.model_validate(payload) + response: GraphQLResponse = GraphQLResponse.model_validate(payload) assert response.data.current_user is not None assert response.data.current_user.login == "testuser" @@ -179,7 +182,7 @@ def test_graphql_response_with_errors() -> None: The schema should accept these responses so they can be processed. """ # Real-world example: Inventory operation with service timeout errors - payload = { + payload: dict[str, object] = { "errors": [ { "message": "service timeout", @@ -232,7 +235,7 @@ def test_graphql_response_with_errors() -> None: } # This should not raise ValidationError even with errors field present - response = GraphQLResponse.model_validate(payload) + response: GraphQLResponse = GraphQLResponse.model_validate(payload) # Verify the errors were captured assert response.errors is not None @@ -256,7 +259,7 @@ def test_drop_campaign_details_missing_distribution_type() -> None: This is based on a real-world validation error from file 7720168310842708008.json. """ - payload = { + payload: dict[str, object] = { "data": { "user": { "id": "58162970", @@ -348,7 +351,7 @@ def test_drop_campaign_details_missing_distribution_type() -> None: } # This should not raise ValidationError despite missing distributionType - response = GraphQLResponse.model_validate(payload) + response: GraphQLResponse = GraphQLResponse.model_validate(payload) # Verify the structure was parsed correctly assert response.data.current_user is not None @@ -358,18 +361,18 @@ def test_drop_campaign_details_missing_distribution_type() -> None: assert response.data.current_user.drop_campaigns is not None assert len(response.data.current_user.drop_campaigns) == 1 - campaign = response.data.current_user.drop_campaigns[0] + campaign: DropCampaignSchema = response.data.current_user.drop_campaigns[0] assert campaign.name == "Official Channel Weekly Drop" assert campaign.game.display_name == "World of Warships" assert len(campaign.time_based_drops) == 1 # Verify time-based drops - first_drop = campaign.time_based_drops[0] + first_drop: TimeBasedDropSchema = campaign.time_based_drops[0] assert first_drop.name == "Week 2: 250 Community Tokens" assert first_drop.required_minutes_watched == 60 # Verify benefits - distributionType should be None since it was missing assert len(first_drop.benefit_edges) == 1 - benefit = first_drop.benefit_edges[0].benefit + benefit: DropBenefitSchema = first_drop.benefit_edges[0].benefit assert benefit.name == "13.7 Update: 250 CT" assert benefit.distribution_type is None # This field was missing in the API response