375 lines
16 KiB
Python
375 lines
16 KiB
Python
"""Tests for Pydantic schemas used in the import process."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from twitch.schemas import GraphQLResponse
|
|
|
|
|
|
def test_inventory_operation_validation() -> None:
|
|
"""Test that the Inventory operation format validates correctly.
|
|
|
|
The Inventory operation has a different structure than ViewerDropsDashboard:
|
|
- No 'login' field in currentUser
|
|
- No 'dropCampaigns' field in currentUser
|
|
- Has 'inventory.dropCampaignsInProgress' instead
|
|
- Campaign data has extra fields that are now explicitly defined
|
|
- Game uses 'name' instead of 'displayName'
|
|
- Benefits have 'DropBenefit' as __typename instead of 'Benefit'
|
|
"""
|
|
# Minimal valid Inventory operation structure
|
|
payload = {
|
|
"data": {
|
|
"currentUser": {
|
|
"id": "17658559",
|
|
"inventory": {
|
|
"__typename": "Inventory",
|
|
"dropCampaignsInProgress": [
|
|
{
|
|
"id": "campaign-1",
|
|
"name": "Test Campaign",
|
|
"game": {
|
|
"id": "game-1",
|
|
"name": "Test Game",
|
|
"boxArtURL": "https://example.com/boxart.jpg",
|
|
"__typename": "Game",
|
|
},
|
|
"status": "ACTIVE",
|
|
"startAt": "2025-01-01T00:00:00Z",
|
|
"endAt": "2025-12-31T23:59:59Z",
|
|
"detailsURL": "https://example.com/details",
|
|
"accountLinkURL": "https://example.com/link",
|
|
"self": {
|
|
"isAccountConnected": True,
|
|
"__typename": "DropCampaignSelfEdge",
|
|
},
|
|
"imageURL": "https://example.com/image.png",
|
|
"allow": None,
|
|
"eventBasedDrops": [],
|
|
"timeBasedDrops": [
|
|
{
|
|
"id": "drop-1",
|
|
"name": "Test Drop",
|
|
"requiredMinutesWatched": 60,
|
|
"requiredSubs": 0,
|
|
"startAt": "2025-01-01T00:00:00Z",
|
|
"endAt": "2025-12-31T23:59:59Z",
|
|
"benefitEdges": [
|
|
{
|
|
"benefit": {
|
|
"id": "benefit-1",
|
|
"name": "Test Benefit",
|
|
"imageAssetURL": "https://example.com/benefit.png",
|
|
"entitlementLimit": 1,
|
|
"isIosAvailable": False,
|
|
"distributionType": "DIRECT_ENTITLEMENT",
|
|
"__typename": "DropBenefit",
|
|
},
|
|
"entitlementLimit": 1,
|
|
"claimCount": 0,
|
|
"__typename": "DropBenefitEdge",
|
|
},
|
|
],
|
|
"preconditionDrops": None,
|
|
"self": {},
|
|
"campaign": {},
|
|
"localizedContent": {},
|
|
"__typename": "TimeBasedDrop",
|
|
},
|
|
],
|
|
"__typename": "DropCampaign",
|
|
},
|
|
],
|
|
"gameEventDrops": [],
|
|
},
|
|
"__typename": "User",
|
|
},
|
|
},
|
|
"extensions": {
|
|
"operationName": "Inventory",
|
|
},
|
|
}
|
|
|
|
# This should not raise ValidationError
|
|
response = GraphQLResponse.model_validate(payload)
|
|
|
|
# Verify the structure was parsed correctly
|
|
assert response.data.current_user is not None
|
|
assert response.data.current_user.twitch_id == "17658559"
|
|
|
|
# Verify drop_campaigns was normalized from inventory
|
|
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]
|
|
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]
|
|
assert first_drop.name == "Test Drop"
|
|
assert first_drop.required_minutes_watched == 60
|
|
|
|
# Verify benefits
|
|
assert len(first_drop.benefit_edges) == 1
|
|
assert first_drop.benefit_edges[0].benefit.name == "Test Benefit"
|
|
|
|
|
|
def test_viewer_drops_dashboard_operation_still_works() -> None:
|
|
"""Test that the original ViewerDropsDashboard format still validates.
|
|
|
|
This ensures backward compatibility with the existing import data.
|
|
"""
|
|
# Minimal valid ViewerDropsDashboard structure
|
|
payload = {
|
|
"data": {
|
|
"currentUser": {
|
|
"id": "12345",
|
|
"login": "testuser",
|
|
"dropCampaigns": [
|
|
{
|
|
"id": "campaign-1",
|
|
"name": "Test Campaign",
|
|
"owner": {
|
|
"id": "org-1",
|
|
"name": "Test Org",
|
|
"__typename": "Organization",
|
|
},
|
|
"game": {
|
|
"id": "game-1",
|
|
"displayName": "Test Game",
|
|
"boxArtURL": "https://example.com/boxart.jpg",
|
|
"__typename": "Game",
|
|
},
|
|
"status": "ACTIVE",
|
|
"startAt": "2025-01-01T00:00:00Z",
|
|
"endAt": "2025-12-31T23:59:59Z",
|
|
"detailsURL": "https://example.com/details",
|
|
"accountLinkURL": "https://example.com/link",
|
|
"self": {
|
|
"isAccountConnected": True,
|
|
"__typename": "DropCampaignSelfEdge",
|
|
},
|
|
"timeBasedDrops": [],
|
|
"__typename": "DropCampaign",
|
|
},
|
|
],
|
|
"__typename": "User",
|
|
},
|
|
},
|
|
"extensions": {
|
|
"operationName": "ViewerDropsDashboard",
|
|
},
|
|
}
|
|
|
|
# This should not raise ValidationError
|
|
response = GraphQLResponse.model_validate(payload)
|
|
|
|
assert response.data.current_user is not None
|
|
assert response.data.current_user.login == "testuser"
|
|
|
|
if response.data.current_user.drop_campaigns is not None:
|
|
assert len(response.data.current_user.drop_campaigns) == 1
|
|
|
|
|
|
def test_graphql_response_with_errors() -> None:
|
|
"""Test that GraphQL responses with errors field validate correctly.
|
|
|
|
Some API responses include partial data with errors (e.g., service timeouts).
|
|
The schema should accept these responses so they can be processed.
|
|
"""
|
|
# Real-world example: Inventory operation with service timeout errors
|
|
payload = {
|
|
"errors": [
|
|
{
|
|
"message": "service timeout",
|
|
"path": ["currentUser", "inventory", "dropCampaignsInProgress", 7, "allow", "channels"],
|
|
},
|
|
{
|
|
"message": "service timeout",
|
|
"path": ["currentUser", "inventory", "dropCampaignsInProgress", 10, "allow", "channels"],
|
|
},
|
|
],
|
|
"data": {
|
|
"currentUser": {
|
|
"id": "17658559",
|
|
"inventory": {
|
|
"__typename": "Inventory",
|
|
"dropCampaignsInProgress": [
|
|
{
|
|
"id": "campaign-1",
|
|
"name": "Test Campaign",
|
|
"game": {
|
|
"id": "game-1",
|
|
"name": "Test Game",
|
|
"boxArtURL": "https://example.com/boxart.jpg",
|
|
"__typename": "Game",
|
|
},
|
|
"status": "ACTIVE",
|
|
"startAt": "2025-01-01T00:00:00Z",
|
|
"endAt": "2025-12-31T23:59:59Z",
|
|
"detailsURL": "https://example.com/details",
|
|
"accountLinkURL": "https://example.com/link",
|
|
"self": {
|
|
"isAccountConnected": True,
|
|
"__typename": "DropCampaignSelfEdge",
|
|
},
|
|
"imageURL": "https://example.com/image.png",
|
|
"allow": None,
|
|
"eventBasedDrops": [],
|
|
"timeBasedDrops": [],
|
|
"__typename": "DropCampaign",
|
|
},
|
|
],
|
|
"gameEventDrops": [],
|
|
},
|
|
"__typename": "User",
|
|
},
|
|
},
|
|
"extensions": {
|
|
"operationName": "Inventory",
|
|
},
|
|
}
|
|
|
|
# This should not raise ValidationError even with errors field present
|
|
response = GraphQLResponse.model_validate(payload)
|
|
|
|
# Verify the errors were captured
|
|
assert response.errors is not None
|
|
assert len(response.errors) == 2
|
|
assert response.errors[0].message == "service timeout"
|
|
assert response.errors[0].path == ["currentUser", "inventory", "dropCampaignsInProgress", 7, "allow", "channels"]
|
|
|
|
# Verify the data is still accessible and valid
|
|
assert response.data.current_user is not None
|
|
assert response.data.current_user.twitch_id == "17658559"
|
|
assert response.data.current_user.drop_campaigns is not None
|
|
assert len(response.data.current_user.drop_campaigns) == 1
|
|
|
|
|
|
def test_drop_campaign_details_missing_distribution_type() -> None:
|
|
"""Test that DropCampaignDetails operation validates when distributionType is missing.
|
|
|
|
Some API responses (e.g., from DropCampaignDetails operation) don't include
|
|
the distributionType field in the benefit object. The schema should handle this
|
|
gracefully by making the field optional.
|
|
|
|
This is based on a real-world validation error from file 7720168310842708008.json.
|
|
"""
|
|
payload = {
|
|
"data": {
|
|
"user": {
|
|
"id": "58162970",
|
|
"dropCampaign": {
|
|
"id": "1cf86f25-540f-11ef-90d0-0a58a9feac02",
|
|
"self": {
|
|
"isAccountConnected": False,
|
|
"__typename": "DropCampaignSelfEdge",
|
|
},
|
|
"allow": {
|
|
"channels": [
|
|
{
|
|
"id": "182565961",
|
|
"displayName": "WorldofWarships",
|
|
"name": "worldofwarships",
|
|
"__typename": "Channel",
|
|
},
|
|
],
|
|
"isEnabled": True,
|
|
"__typename": "DropCampaignACL",
|
|
},
|
|
"accountLinkURL": "https://wargaming.net/personal/",
|
|
"description": "New Drops!",
|
|
"detailsURL": "https://worldofwarships.eu/news/general-news/stream-rewards-137/",
|
|
"endAt": "2024-08-21T21:59:59.996Z",
|
|
"eventBasedDrops": [],
|
|
"game": {
|
|
"id": "32502",
|
|
"slug": "world-of-warships",
|
|
"displayName": "World of Warships",
|
|
"__typename": "Game",
|
|
},
|
|
"imageURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/CAMPAIGN/51ec1931-6792-4b9f-850a-e6b9c454eefc.png",
|
|
"name": "Official Channel Weekly Drop",
|
|
"owner": {
|
|
"id": "6948a129-2c6d-4d88-9444-6b96918a19f8",
|
|
"name": "Wargaming",
|
|
"__typename": "Organization",
|
|
},
|
|
"startAt": "2024-08-14T22:00:00Z",
|
|
"status": "ACTIVE",
|
|
"timeBasedDrops": [
|
|
{
|
|
"id": "79a0a4db-5a4d-11ef-91b7-0a58a9feac02",
|
|
"requiredSubs": 0,
|
|
"benefitEdges": [
|
|
{
|
|
"benefit": {
|
|
"id": "6948a129-2c6d-4d88-9444-6b96918a19f8_CUSTOM_ID_WOWS_TwitchDrops_1307_250ct", # noqa: E501
|
|
"createdAt": "2024-08-06T16:03:15.89Z",
|
|
"entitlementLimit": 1,
|
|
"game": {
|
|
"id": "32502",
|
|
"name": "World of Warships",
|
|
"__typename": "Game",
|
|
},
|
|
"imageAssetURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/REWARD/a049fb1a-ab63-40e5-9a43-bd6b94e52f36.png",
|
|
"isIosAvailable": False,
|
|
"name": "13.7 Update: 250 CT",
|
|
"ownerOrganization": {
|
|
"id": "6948a129-2c6d-4d88-9444-6b96918a19f8",
|
|
"name": "Wargaming",
|
|
"__typename": "Organization",
|
|
},
|
|
"__typename": "DropBenefit",
|
|
},
|
|
"entitlementLimit": 1,
|
|
"__typename": "DropBenefitEdge",
|
|
},
|
|
],
|
|
"endAt": "2024-08-21T21:59:59.996Z",
|
|
"name": "Week 2: 250 Community Tokens",
|
|
"preconditionDrops": None,
|
|
"requiredMinutesWatched": 60,
|
|
"startAt": "2024-08-14T22:00:00Z",
|
|
"__typename": "TimeBasedDrop",
|
|
},
|
|
],
|
|
"__typename": "DropCampaign",
|
|
},
|
|
"__typename": "User",
|
|
},
|
|
},
|
|
"extensions": {
|
|
"durationMilliseconds": 75,
|
|
"operationName": "DropCampaignDetails",
|
|
"requestID": "01J5RDFXF7VMMDWVBZKNZ52ED6",
|
|
},
|
|
}
|
|
|
|
# This should not raise ValidationError despite missing distributionType
|
|
response = GraphQLResponse.model_validate(payload)
|
|
|
|
# Verify the structure was parsed correctly
|
|
assert response.data.current_user is not None
|
|
assert response.data.current_user.twitch_id == "58162970"
|
|
|
|
# Verify drop_campaigns was normalized from dropCampaign (singular)
|
|
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]
|
|
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]
|
|
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
|
|
assert benefit.name == "13.7 Update: 250 CT"
|
|
assert benefit.distribution_type is None # This field was missing in the API response
|