438 lines
17 KiB
Python
438 lines
17 KiB
Python
from datetime import timedelta
|
|
|
|
from django.db import connection
|
|
from django.test import Client
|
|
from django.test import TestCase
|
|
from django.test.utils import CaptureQueriesContext
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from twitch import api as twitch_api
|
|
from twitch.models import Channel
|
|
from twitch.models import ChatBadge
|
|
from twitch.models import ChatBadgeSet
|
|
from twitch.models import DropBenefit
|
|
from twitch.models import DropCampaign
|
|
from twitch.models import Game
|
|
from twitch.models import Organization
|
|
from twitch.models import RewardCampaign
|
|
from twitch.models import TimeBasedDrop
|
|
|
|
|
|
class TwitchApiV1TestCase(TestCase):
|
|
"""Tests for the versioned Twitch API."""
|
|
|
|
def setUp(self) -> None:
|
|
"""Create representative Twitch API fixture data."""
|
|
self.client = Client()
|
|
now = timezone.now()
|
|
|
|
self.org = Organization.objects.create(
|
|
twitch_id="org123",
|
|
name="Test Organization",
|
|
)
|
|
self.game = Game.objects.create(
|
|
twitch_id="game123",
|
|
slug="test-game",
|
|
name="Test Game",
|
|
display_name="Test Game",
|
|
box_art="https://example.com/game.png",
|
|
)
|
|
self.game.owners.add(self.org)
|
|
|
|
self.channel = Channel.objects.create(
|
|
twitch_id="channel123",
|
|
name="testchannel",
|
|
display_name="TestChannel",
|
|
allowed_campaign_count=1,
|
|
)
|
|
|
|
self.campaign = DropCampaign.objects.create(
|
|
twitch_id="campaign123",
|
|
name="Test Campaign",
|
|
description="A test campaign",
|
|
details_url="https://example.com/details",
|
|
account_link_url="https://example.com/link",
|
|
image_url="https://example.com/campaign.png",
|
|
game=self.game,
|
|
start_at=now - timedelta(days=1),
|
|
end_at=now + timedelta(days=1),
|
|
operation_names=["DropCampaignDetails"],
|
|
is_fully_imported=True,
|
|
)
|
|
self.campaign.allow_channels.add(self.channel)
|
|
|
|
self.drop = TimeBasedDrop.objects.create(
|
|
twitch_id="drop123",
|
|
name="Test Drop",
|
|
campaign=self.campaign,
|
|
required_minutes_watched=30,
|
|
start_at=now - timedelta(days=1),
|
|
end_at=now + timedelta(days=1),
|
|
)
|
|
self.benefit = DropBenefit.objects.create(
|
|
twitch_id="benefit123",
|
|
name="Test Benefit",
|
|
image_asset_url="https://example.com/benefit.png",
|
|
distribution_type="ITEM",
|
|
)
|
|
self.drop.benefits.add(self.benefit)
|
|
|
|
self.reward_campaign = RewardCampaign.objects.create(
|
|
twitch_id="reward123",
|
|
name="Test Reward",
|
|
brand="Test Brand",
|
|
starts_at=now - timedelta(days=1),
|
|
ends_at=now + timedelta(days=1),
|
|
status="ACTIVE",
|
|
summary="Reward summary",
|
|
external_url="https://example.com/reward",
|
|
game=self.game,
|
|
)
|
|
|
|
self.badge_set = ChatBadgeSet.objects.create(set_id="test-badge-set")
|
|
ChatBadge.objects.create(
|
|
badge_set=self.badge_set,
|
|
badge_id="1",
|
|
image_url_1x="https://example.com/badge-1x.png",
|
|
image_url_2x="https://example.com/badge-2x.png",
|
|
image_url_4x="https://example.com/badge-4x.png",
|
|
title="Test Badge",
|
|
description="Test badge description",
|
|
)
|
|
|
|
def _create_secondary_api_fixture(self) -> None:
|
|
now = timezone.now()
|
|
org = Organization.objects.create(
|
|
twitch_id="org456",
|
|
name="Second Organization",
|
|
)
|
|
game = Game.objects.create(
|
|
twitch_id="game456",
|
|
slug="second-game",
|
|
name="Second Game",
|
|
display_name="Second Game",
|
|
box_art="https://example.com/second-game.png",
|
|
)
|
|
game.owners.add(org)
|
|
|
|
channel = Channel.objects.create(
|
|
twitch_id="channel456",
|
|
name="secondchannel",
|
|
display_name="SecondChannel",
|
|
allowed_campaign_count=1,
|
|
)
|
|
|
|
campaign = DropCampaign.objects.create(
|
|
twitch_id="campaign456",
|
|
name="Second Campaign",
|
|
description="Another test campaign",
|
|
details_url="https://example.com/second-details",
|
|
account_link_url="https://example.com/second-link",
|
|
image_url="https://example.com/second-campaign.png",
|
|
game=game,
|
|
start_at=now - timedelta(days=2),
|
|
end_at=now + timedelta(days=2),
|
|
operation_names=["DropCampaignDetails"],
|
|
is_fully_imported=True,
|
|
)
|
|
campaign.allow_channels.add(channel)
|
|
|
|
drop = TimeBasedDrop.objects.create(
|
|
twitch_id="drop456",
|
|
name="Second Drop",
|
|
campaign=campaign,
|
|
required_minutes_watched=60,
|
|
start_at=now - timedelta(days=2),
|
|
end_at=now + timedelta(days=2),
|
|
)
|
|
benefit = DropBenefit.objects.create(
|
|
twitch_id="benefit456",
|
|
name="Second Benefit",
|
|
image_asset_url="https://example.com/second-benefit.png",
|
|
distribution_type="ITEM",
|
|
)
|
|
drop.benefits.add(benefit)
|
|
|
|
RewardCampaign.objects.create(
|
|
twitch_id="reward456",
|
|
name="Second Reward",
|
|
brand="Second Brand",
|
|
starts_at=now - timedelta(days=2),
|
|
ends_at=now + timedelta(days=2),
|
|
status="ACTIVE",
|
|
summary="Second reward summary",
|
|
external_url="https://example.com/second-reward",
|
|
game=game,
|
|
)
|
|
|
|
badge_set = ChatBadgeSet.objects.create(set_id="second-badge-set")
|
|
ChatBadge.objects.create(
|
|
badge_set=badge_set,
|
|
badge_id="1",
|
|
image_url_1x="https://example.com/second-badge-1x.png",
|
|
image_url_2x="https://example.com/second-badge-2x.png",
|
|
image_url_4x="https://example.com/second-badge-4x.png",
|
|
title="Second Badge",
|
|
description="Second badge description",
|
|
)
|
|
|
|
def test_v1_campaign_list(self) -> None:
|
|
"""Return active campaigns from the v1 list endpoint."""
|
|
response = self.client.get("/twitch/api/v1/campaigns/?status=active")
|
|
|
|
assert response.status_code == 200
|
|
assert "Content-Disposition" not in response
|
|
data = response.json()
|
|
assert data["total"] == 1
|
|
assert data["page"] == 1
|
|
assert data["items"][0]["twitch_id"] == "campaign123"
|
|
assert data["items"][0]["status"] == "active"
|
|
assert data["items"][0]["game"]["twitch_id"] == "game123"
|
|
|
|
def test_v1_campaign_detail(self) -> None:
|
|
"""Return nested campaign detail data from the v1 endpoint."""
|
|
response = self.client.get("/twitch/api/v1/campaigns/campaign123/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["operation_names"] == ["DropCampaignDetails"]
|
|
assert data["game"]["box_art_url"] == "https://example.com/game.png"
|
|
assert data["allowed_channels"][0]["twitch_id"] == "channel123"
|
|
assert data["drops"][0]["benefits"][0]["twitch_id"] == "benefit123"
|
|
|
|
def test_v1_campaign_detail_game_box_art_does_not_load_deferred_file(self) -> None:
|
|
"""Serialize campaign game box art without lazy-loading ImageField data."""
|
|
campaign = DropCampaign.for_detail_view("campaign123")
|
|
|
|
image_fields = {"box_art_file", "box_art_width", "box_art_height"}
|
|
assert campaign.game.get_deferred_fields().isdisjoint(image_fields)
|
|
with CaptureQueriesContext(connection) as capture:
|
|
box_art_url = twitch_api._game_box_art_url(campaign.game)
|
|
|
|
assert box_art_url == "https://example.com/game.png"
|
|
assert len(capture) == 0
|
|
|
|
def test_v1_campaign_detail_uses_local_game_box_art(self) -> None:
|
|
"""Return locally cached game box art from campaign detail responses."""
|
|
self.game.box_art_file = "games/box_art/local.png"
|
|
self.game.box_art_width = 285
|
|
self.game.box_art_height = 380
|
|
self.game.save(
|
|
update_fields=["box_art_file", "box_art_width", "box_art_height"],
|
|
)
|
|
|
|
response = self.client.get("/twitch/api/v1/campaigns/campaign123/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["game"]["box_art_url"] == self.game.box_art_file.url
|
|
|
|
def test_v1_all_endpoints_handle_multiple_rows(self) -> None:
|
|
"""Exercise all v1 routes with enough rows to catch deferred loads."""
|
|
self._create_secondary_api_fixture()
|
|
list_urls: list[tuple[str, int]] = [
|
|
("/twitch/api/v1/campaigns/?page_size=50", 2),
|
|
("/twitch/api/v1/games/?page_size=50", 2),
|
|
("/twitch/api/v1/organizations/?page_size=50", 2),
|
|
("/twitch/api/v1/channels/?page_size=50", 2),
|
|
("/twitch/api/v1/reward-campaigns/?page_size=50", 2),
|
|
("/twitch/api/v1/badges/?page_size=50", 2),
|
|
]
|
|
detail_urls = [
|
|
"/twitch/api/v1/campaigns/campaign123/",
|
|
"/twitch/api/v1/campaigns/campaign456/",
|
|
"/twitch/api/v1/games/game123/",
|
|
"/twitch/api/v1/games/game456/",
|
|
"/twitch/api/v1/organizations/org123/",
|
|
"/twitch/api/v1/organizations/org456/",
|
|
"/twitch/api/v1/channels/channel123/",
|
|
"/twitch/api/v1/channels/channel456/",
|
|
"/twitch/api/v1/reward-campaigns/reward123/",
|
|
"/twitch/api/v1/reward-campaigns/reward456/",
|
|
"/twitch/api/v1/badges/test-badge-set/",
|
|
"/twitch/api/v1/badges/second-badge-set/",
|
|
]
|
|
|
|
for url, expected_total in list_urls:
|
|
response = self.client.get(url)
|
|
assert response.status_code == 200, url
|
|
data = response.json()
|
|
assert data["total"] == expected_total
|
|
assert len(data["items"]) == expected_total
|
|
|
|
for url in detail_urls:
|
|
response = self.client.get(url)
|
|
assert response.status_code == 200, url
|
|
assert response.json()
|
|
|
|
schema_response = self.client.get(reverse("twitch:twitch-api-v1:openapi-json"))
|
|
assert schema_response.status_code == 200
|
|
assert schema_response.json()
|
|
|
|
docs_response = self.client.get(reverse("twitch:twitch-api-v1:openapi-view"))
|
|
assert docs_response.status_code == 200
|
|
|
|
def test_v1_collection_endpoints(self) -> None:
|
|
"""Return v1 list responses for all Twitch API collections."""
|
|
checks = [
|
|
("/twitch/api/v1/games/", "game123"),
|
|
("/twitch/api/v1/organizations/", "org123"),
|
|
("/twitch/api/v1/channels/", "channel123"),
|
|
("/twitch/api/v1/reward-campaigns/", "reward123"),
|
|
("/twitch/api/v1/badges/", "test-badge-set"),
|
|
]
|
|
|
|
for url, expected_id in checks:
|
|
response = self.client.get(url)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
actual_id = data["items"][0].get(
|
|
"twitch_id",
|
|
data["items"][0].get("set_id"),
|
|
)
|
|
assert actual_id == expected_id
|
|
|
|
games_response = self.client.get("/twitch/api/v1/games/")
|
|
games_data = games_response.json()
|
|
assert games_data["items"][0]["campaign_count"] == 1
|
|
assert games_data["items"][0]["active_campaign_count"] == 1
|
|
|
|
def test_v1_organization_detail_includes_games_and_campaigns(self) -> None:
|
|
"""Return concrete game counts and detailed organization campaigns."""
|
|
response = self.client.get("/twitch/api/v1/organizations/org123/")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["games"][0]["twitch_id"] == "game123"
|
|
assert data["games"][0]["campaign_count"] == 1
|
|
assert data["games"][0]["active_campaign_count"] == 1
|
|
assert data["campaigns"][0]["twitch_id"] == "campaign123"
|
|
assert data["campaigns"][0]["operation_names"] == ["DropCampaignDetails"]
|
|
assert data["campaigns"][0]["allowed_channels"][0]["twitch_id"] == "channel123"
|
|
assert data["campaigns"][0]["drops"][0]["twitch_id"] == "drop123"
|
|
assert (
|
|
data["campaigns"][0]["drops"][0]["benefits"][0]["twitch_id"] == "benefit123"
|
|
)
|
|
|
|
def test_v1_game_and_channel_detail_include_campaign_data(self) -> None:
|
|
"""Return campaign API fields on game and channel detail responses."""
|
|
checks = [
|
|
"/twitch/api/v1/games/game123/",
|
|
"/twitch/api/v1/channels/channel123/",
|
|
]
|
|
|
|
for url in checks:
|
|
response = self.client.get(url)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
campaign = data["campaigns"][0]
|
|
assert campaign["description"] == "A test campaign"
|
|
assert campaign["details_url"] == "https://example.com/details"
|
|
assert campaign["account_link_url"] == "https://example.com/link"
|
|
assert campaign["image_url"] == "https://example.com/campaign.png"
|
|
|
|
def test_v1_detail_not_found(self) -> None:
|
|
"""Return 404 for missing v1 campaign detail records."""
|
|
response = self.client.get("/twitch/api/v1/campaigns/missing/")
|
|
|
|
assert response.status_code == 404
|
|
|
|
def test_v1_docs_endpoint(self) -> None:
|
|
"""Render the versioned Twitch API documentation page."""
|
|
response = self.client.get("/twitch/api/v1/docs")
|
|
|
|
assert response.status_code == 200
|
|
assert reverse("twitch:twitch-api-v1:openapi-json") in response.content.decode()
|
|
|
|
def test_v1_docs_links_render_on_twitch_pages(self) -> None:
|
|
"""Expose API docs in nav and resource API links in feed link groups."""
|
|
checks = [
|
|
(
|
|
reverse("core:docs_rss"),
|
|
"API Docs",
|
|
"/twitch/api/v1/docs",
|
|
reverse("twitch:twitch-api-v1:openapi-view"),
|
|
),
|
|
(
|
|
reverse("twitch:dashboard"),
|
|
"API Docs",
|
|
"[api]",
|
|
reverse("twitch:twitch-api-v1:list_campaigns"),
|
|
),
|
|
(
|
|
reverse("twitch:campaign_list"),
|
|
"API Docs",
|
|
"[api]",
|
|
reverse("twitch:twitch-api-v1:list_campaigns"),
|
|
),
|
|
(
|
|
reverse("twitch:campaign_detail", args=[self.campaign.twitch_id]),
|
|
"API Docs",
|
|
"[api]",
|
|
reverse(
|
|
"twitch:twitch-api-v1:get_campaign",
|
|
args=[self.campaign.twitch_id],
|
|
),
|
|
),
|
|
(
|
|
reverse("twitch:game_detail", args=[self.game.twitch_id]),
|
|
"API Docs",
|
|
"[api]",
|
|
reverse("twitch:twitch-api-v1:get_game", args=[self.game.twitch_id]),
|
|
),
|
|
(
|
|
reverse("twitch:games_grid"),
|
|
"API Docs",
|
|
"[api]",
|
|
reverse("twitch:twitch-api-v1:list_games"),
|
|
),
|
|
(
|
|
reverse("twitch:org_list"),
|
|
"API Docs",
|
|
"[api]",
|
|
reverse("twitch:twitch-api-v1:list_organizations"),
|
|
),
|
|
(
|
|
reverse("twitch:reward_campaign_list"),
|
|
"API Docs",
|
|
"[api]",
|
|
reverse("twitch:twitch-api-v1:list_reward_campaigns"),
|
|
),
|
|
(
|
|
reverse(
|
|
"twitch:reward_campaign_detail",
|
|
args=[self.reward_campaign.twitch_id],
|
|
),
|
|
"API Docs",
|
|
"[api]",
|
|
reverse(
|
|
"twitch:twitch-api-v1:get_reward_campaign",
|
|
args=[self.reward_campaign.twitch_id],
|
|
),
|
|
),
|
|
]
|
|
|
|
for url, nav_text, feed_text, api_href in checks:
|
|
response = self.client.get(url)
|
|
assert response.status_code == 200
|
|
content = response.content.decode()
|
|
assert reverse("twitch:twitch-api-v1:openapi-view") in content
|
|
assert api_href in content
|
|
assert nav_text in content
|
|
assert feed_text in content
|
|
|
|
def test_campaign_detail_api_link_targets_campaign_endpoint(self) -> None:
|
|
"""Link campaign detail [api] directly to that campaign JSON endpoint."""
|
|
response = self.client.get(
|
|
reverse("twitch:campaign_detail", args=[self.campaign.twitch_id]),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
content = response.content.decode()
|
|
campaign_api_url = reverse(
|
|
"twitch:twitch-api-v1:get_campaign",
|
|
args=[self.campaign.twitch_id],
|
|
)
|
|
assert f'href="{campaign_api_url}"' in content
|
|
assert 'title="Twitch campaign API">[api]</a>' in content
|