Add API for Twitch data

This commit is contained in:
Joakim Hellsén 2026-05-05 05:01:48 +02:00
commit e960b09084
Signed by: Joakim Hellsén
SSH key fingerprint: SHA256:/9h/CsExpFp+PRhsfA0xznFx2CGfTT5R/kpuFfUgEQk
18 changed files with 1526 additions and 1 deletions

View file

@ -12,7 +12,9 @@ dependencies = [
"django-celery-beat",
"django-celery-results",
"django-debug-toolbar",
"django-ninja",
"django-silk",
"django-zeal",
"django",
"flower",
"gunicorn",
@ -31,7 +33,6 @@ dependencies = [
"setproctitle",
"sitemap-parser",
"tqdm",
"django-zeal>=2.1.0",
]

View file

@ -220,6 +220,7 @@
<a href="{% url 'twitch:channel_list' %}">Channels</a> |
<a href="{% url 'twitch:badge_list' %}">Badges</a> |
<a href="{% url 'twitch:emote_gallery' %}">Emotes</a> |
<a href="{% url 'twitch:twitch-api-v1:openapi-view' %}">API Docs</a> |
<a href="https://www.twitch.tv/drops/inventory">Inventory</a>
<br />
<strong>Kick</strong>

View file

@ -48,6 +48,8 @@
title="Atom feed for Twitch campaigns">[atom]</a>
<a href="{% url 'core:campaign_feed_discord' %}"
title="Discord feed for Twitch campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_campaigns' %}"
title="Twitch campaigns API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div>
</header>

View file

@ -16,6 +16,10 @@
that include Discord relative timestamps (e.g., <code>&lt;t:1773450272:R&gt;</code>) for dates,
making them ideal for Discord bots and integrations. Future enhancements may include Discord-specific formatting or content.
</p>
<p>
Twitch JSON API documentation is available at
<a href="{% url 'twitch:twitch-api-v1:openapi-view' %}">/twitch/api/v1/docs</a>.
</p>
<section>
<h2>Global RSS Feeds</h2>
<table>

View file

@ -103,6 +103,8 @@
title="Atom feed for {{ campaign.game.display_name }} campaigns">[atom]</a>
<a href="{% url 'core:game_campaign_feed_discord' campaign.game.twitch_id %}"
title="Discord feed for {{ campaign.game.display_name }} campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:get_campaign' campaign.twitch_id %}"
title="Twitch campaign API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
{% endif %}
</div>

View file

@ -33,6 +33,8 @@
title="Atom feed for all campaigns">[atom]</a>
<a href="{% url 'core:campaign_feed_discord' %}"
title="Discord feed for all campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_campaigns' %}"
title="Twitch campaigns API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:export_campaigns_csv' %}{% if request.GET %}?{{ request.GET.urlencode }}{% endif %}"
title="Export campaigns as CSV">[csv]</a>

View file

@ -36,6 +36,8 @@
title="Atom feed for campaigns">[atom]</a>
<a href="{% url 'core:campaign_feed_discord' %}"
title="Discord feed for campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_campaigns' %}"
title="Twitch campaigns API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div>
<hr />

View file

@ -65,6 +65,8 @@
title="Atom feed for {{ game.display_name }} campaigns">[atom]</a>
<a href="{% url 'core:game_campaign_feed_discord' game.twitch_id %}"
title="Discord feed for {{ game.display_name }} campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:get_game' game.twitch_id %}"
title="Twitch game API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div>
</div>

View file

@ -32,6 +32,8 @@
title="Atom feed for all games">[atom]</a>
<a href="{% url 'core:game_feed_discord' %}"
title="Discord feed for all games">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_games' %}"
title="Twitch games API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:export_games_csv' %}"
title="Export all games as CSV">[csv]</a>

View file

@ -30,6 +30,8 @@
title="Atom feed for all games">[atom]</a>
<a href="{% url 'core:game_feed_discord' %}"
title="Discord feed for all games">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_games' %}"
title="Twitch games API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:export_games_csv' %}"
title="Export all games as CSV">[csv]</a>

View file

@ -15,6 +15,8 @@
title="Atom feed for all organizations">[atom]</a>
<a href="{% url 'core:organization_feed_discord' %}"
title="Discord feed for all organizations">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_organizations' %}"
title="Twitch organizations API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
<a href="{% url 'twitch:export_organizations_csv' %}"
title="Export all organizations as CSV">[csv]</a>

View file

@ -110,6 +110,8 @@
title="Atom feed for all reward campaigns">[atom]</a>
<a href="{% url 'core:reward_campaign_feed_discord' %}"
title="Discord feed for all reward campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:get_reward_campaign' reward_campaign.twitch_id %}"
title="Twitch reward campaign API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
{% if reward_campaign.external_url %}
<a href="{{ reward_campaign.external_url }}"

View file

@ -36,6 +36,8 @@
title="Atom feed for all reward campaigns">[atom]</a>
<a href="{% url 'core:reward_campaign_feed_discord' %}"
title="Discord feed for all reward campaigns">[discord]</a>
<a href="{% url 'twitch:twitch-api-v1:list_reward_campaigns' %}"
title="Twitch reward campaigns API">[api]</a>
<a href="{% url 'core:docs_rss' %}" title="RSS feed documentation">[explain]</a>
</div>
{% if reward_campaigns %}

822
twitch/api.py Normal file
View file

@ -0,0 +1,822 @@
from __future__ import annotations
import datetime # noqa: TC003
from dataclasses import dataclass
from typing import TYPE_CHECKING
from typing import Literal
from django.db import models
from django.http import Http404
from django.shortcuts import get_object_or_404
from django.utils import timezone
from ninja import NinjaAPI
from ninja import Schema
from twitch.models import Channel
from twitch.models import ChatBadgeSet
from twitch.models import DropCampaign
from twitch.models import Game
from twitch.models import Organization
from twitch.models import RewardCampaign
from twitch.utils import normalize_twitch_box_art_url
if TYPE_CHECKING:
from django.db.models import QuerySet
from django.http import HttpRequest
from twitch.models import ChatBadge
from twitch.models import DropBenefit
from twitch.models import TimeBasedDrop
V1StatusFilter = Literal["active", "upcoming", "expired"]
DEFAULT_PAGE_SIZE = 100
MAX_PAGE_SIZE = 500
@dataclass(frozen=True)
class V1Page[ModelT: models.Model]:
"""Typed slice metadata for v1 list responses."""
items: list[ModelT]
total: int
page: int
page_size: int
class V1PaginationSchema(Schema):
"""Common pagination fields for v1 list responses."""
total: int
page: int
page_size: int
class V1OrganizationSummarySchema(Schema):
"""Compact Twitch organization representation."""
twitch_id: str
name: str
class V1GameSummarySchema(Schema):
"""Compact Twitch game representation."""
twitch_id: str
slug: str
name: str
display_name: str
box_art_url: str
organizations: list[V1OrganizationSummarySchema]
campaign_count: int
active_campaign_count: int
class V1OrganizationSchema(V1OrganizationSummarySchema):
"""Twitch organization response."""
added_at: datetime.datetime
updated_at: datetime.datetime
class V1GameSchema(V1GameSummarySchema):
"""Twitch game response."""
added_at: datetime.datetime
updated_at: datetime.datetime
class V1ChannelSummarySchema(Schema):
"""Compact Twitch channel representation."""
twitch_id: str
name: str
display_name: str
class V1ChannelSchema(V1ChannelSummarySchema):
"""Twitch channel response."""
allowed_campaign_count: int
added_at: datetime.datetime
updated_at: datetime.datetime
class V1DropBenefitSchema(Schema):
"""Twitch drop benefit response."""
twitch_id: str
name: str
image_url: str
distribution_type: str
created_at: datetime.datetime | None
entitlement_limit: int
is_ios_available: bool
class V1TimeBasedDropSchema(Schema):
"""Twitch time-based drop response."""
twitch_id: str
name: str
required_minutes_watched: int | None
required_subs: int
start_at: datetime.datetime | None
end_at: datetime.datetime | None
benefits: list[V1DropBenefitSchema]
class V1DropCampaignSummarySchema(Schema):
"""Compact Twitch drop campaign representation."""
twitch_id: str
name: str
description: str
status: str
image_url: str
details_url: str
account_link_url: str
start_at: datetime.datetime | None
end_at: datetime.datetime | None
game: V1GameSummarySchema
allow_is_enabled: bool
is_fully_imported: bool
added_at: datetime.datetime
updated_at: datetime.datetime
class V1DropCampaignDetailSchema(V1DropCampaignSummarySchema):
"""Twitch drop campaign detail response."""
operation_names: list[str]
allowed_channels: list[V1ChannelSummarySchema]
drops: list[V1TimeBasedDropSchema]
class V1OrganizationDetailSchema(V1OrganizationSchema):
"""Twitch organization detail response."""
games: list[V1GameSummarySchema]
campaigns: list[V1DropCampaignDetailSchema]
class V1GameDetailSchema(V1GameSchema):
"""Twitch game detail response."""
campaigns: list[V1DropCampaignSummarySchema]
class V1ChannelDetailSchema(V1ChannelSchema):
"""Twitch channel detail response."""
campaigns: list[V1DropCampaignSummarySchema]
class V1RewardCampaignSchema(Schema):
"""Twitch reward campaign response."""
twitch_id: str
name: str
brand: str
status: str
computed_status: str
summary: str
instructions: str
external_url: str
reward_value_url_param: str
about_url: str
image_url: str
is_sitewide: bool
starts_at: datetime.datetime | None
ends_at: datetime.datetime | None
game: V1GameSummarySchema | None
added_at: datetime.datetime
updated_at: datetime.datetime
class V1ChatBadgeSchema(Schema):
"""Twitch chat badge response."""
badge_id: str
image_url_1x: str
image_url_2x: str
image_url_4x: str
title: str
description: str
click_action: str | None
click_url: str | None
class V1ChatBadgeSetSchema(Schema):
"""Twitch chat badge set response."""
set_id: str
badges: list[V1ChatBadgeSchema]
added_at: datetime.datetime
updated_at: datetime.datetime
class V1DropCampaignListSchema(V1PaginationSchema):
"""Paginated Twitch drop campaign list."""
items: list[V1DropCampaignSummarySchema]
class V1GameListSchema(V1PaginationSchema):
"""Paginated Twitch game list."""
items: list[V1GameSchema]
class V1OrganizationListSchema(V1PaginationSchema):
"""Paginated Twitch organization list."""
items: list[V1OrganizationSchema]
class V1ChannelListSchema(V1PaginationSchema):
"""Paginated Twitch channel list."""
items: list[V1ChannelSchema]
class V1RewardCampaignListSchema(V1PaginationSchema):
"""Paginated Twitch reward campaign list."""
items: list[V1RewardCampaignSchema]
class V1ChatBadgeSetListSchema(V1PaginationSchema):
"""Paginated Twitch chat badge set list."""
items: list[V1ChatBadgeSetSchema]
api = NinjaAPI(
title="TTVDrops Twitch API",
version="1.0.0",
urls_namespace="twitch:twitch-api-v1",
)
def _paginate[ModelT: models.Model](
queryset: QuerySet[ModelT, ModelT],
*,
page: int,
page_size: int,
) -> V1Page[ModelT]:
page = max(page, 1)
page_size = min(max(page_size, 1), MAX_PAGE_SIZE)
offset = (page - 1) * page_size
return V1Page(
items=list(queryset[offset : offset + page_size]),
page=page,
page_size=page_size,
total=queryset.count(),
)
def _campaign_status(
start_at: datetime.datetime | None,
end_at: datetime.datetime | None,
now: datetime.datetime,
) -> str:
if start_at and end_at:
if start_at <= now <= end_at:
return "active"
if start_at > now:
return "upcoming"
return "expired"
return "unknown"
def _apply_status_filter[ModelT: models.Model](
queryset: QuerySet[ModelT, ModelT],
status: V1StatusFilter | None,
now: datetime.datetime,
*,
start_field: str,
end_field: str,
) -> QuerySet[ModelT, ModelT]:
if status == "active":
return queryset.filter(**{f"{start_field}__lte": now, f"{end_field}__gte": now})
if status == "upcoming":
return queryset.filter(**{f"{start_field}__gt": now})
if status == "expired":
return queryset.filter(**{f"{end_field}__lt": now})
return queryset
def _serialize_organization_summary(org: Organization) -> V1OrganizationSummarySchema:
return V1OrganizationSummarySchema(twitch_id=org.twitch_id, name=org.name)
def _serialize_organization(org: Organization) -> V1OrganizationSchema:
return V1OrganizationSchema(
twitch_id=org.twitch_id,
name=org.name,
added_at=org.added_at,
updated_at=org.updated_at,
)
def _game_campaign_count(game: Game) -> int:
return DropCampaign.objects.filter(game=game).count()
def _game_active_campaign_count(game: Game, now: datetime.datetime) -> int:
return DropCampaign.objects.filter(
game=game,
start_at__lte=now,
end_at__gte=now,
).count()
def _game_box_art_url(game: Game) -> str:
deferred_fields: set[str] = game.get_deferred_fields()
local_image_fields = {"box_art_file", "box_art_width", "box_art_height"}
if deferred_fields.isdisjoint(local_image_fields):
return game.box_art_best_url
if "box_art" not in deferred_fields:
return normalize_twitch_box_art_url(game.box_art or "")
return ""
def _campaign_summary_queryset(
queryset: QuerySet[DropCampaign, DropCampaign],
) -> QuerySet[DropCampaign, DropCampaign]:
return (
queryset
.select_related("game")
.only(
"twitch_id",
"name",
"description",
"details_url",
"account_link_url",
"image_url",
"image_file",
"image_width",
"image_height",
"start_at",
"end_at",
"allow_is_enabled",
"is_fully_imported",
"added_at",
"updated_at",
"game",
"game__twitch_id",
"game__name",
"game__display_name",
"game__slug",
"game__box_art",
"game__box_art_file",
"game__box_art_width",
"game__box_art_height",
)
.prefetch_related("game__owners")
)
def _channel_list_queryset(search_query: str | None) -> QuerySet[Channel, Channel]:
queryset = Channel.objects.only(
"twitch_id",
"name",
"display_name",
"allowed_campaign_count",
"added_at",
"updated_at",
)
normalized_query = (search_query or "").strip()
if normalized_query:
queryset = queryset.filter(
models.Q(name__icontains=normalized_query)
| models.Q(display_name__icontains=normalized_query),
)
return queryset.annotate(
campaign_count=models.F("allowed_campaign_count"),
).order_by("-campaign_count", "name")
def _serialize_game_summary(
game: Game,
now: datetime.datetime,
) -> V1GameSummarySchema:
return V1GameSummarySchema(
twitch_id=game.twitch_id,
slug=game.slug,
name=game.name,
display_name=game.display_name,
box_art_url=_game_box_art_url(game),
organizations=[
_serialize_organization_summary(org) for org in game.owners.all()
],
campaign_count=_game_campaign_count(game),
active_campaign_count=_game_active_campaign_count(game, now),
)
def _serialize_game(game: Game, now: datetime.datetime) -> V1GameSchema:
return V1GameSchema(
twitch_id=game.twitch_id,
slug=game.slug,
name=game.name,
display_name=game.display_name,
box_art_url=_game_box_art_url(game),
organizations=[
_serialize_organization_summary(org) for org in game.owners.all()
],
campaign_count=_game_campaign_count(game),
active_campaign_count=_game_active_campaign_count(game, now),
added_at=game.added_at,
updated_at=game.updated_at,
)
def _serialize_channel_summary(channel: Channel) -> V1ChannelSummarySchema:
return V1ChannelSummarySchema(
twitch_id=channel.twitch_id,
name=channel.name,
display_name=channel.display_name,
)
def _serialize_channel(channel: Channel) -> V1ChannelSchema:
return V1ChannelSchema(
twitch_id=channel.twitch_id,
name=channel.name,
display_name=channel.display_name,
allowed_campaign_count=channel.allowed_campaign_count,
added_at=channel.added_at,
updated_at=channel.updated_at,
)
def _serialize_benefit(benefit: DropBenefit) -> V1DropBenefitSchema:
return V1DropBenefitSchema(
twitch_id=benefit.twitch_id,
name=benefit.name,
image_url=benefit.image_best_url,
distribution_type=benefit.distribution_type,
created_at=benefit.created_at,
entitlement_limit=benefit.entitlement_limit,
is_ios_available=benefit.is_ios_available,
)
def _serialize_drop(drop: TimeBasedDrop) -> V1TimeBasedDropSchema:
return V1TimeBasedDropSchema(
twitch_id=drop.twitch_id,
name=drop.name,
required_minutes_watched=drop.required_minutes_watched,
required_subs=drop.required_subs,
start_at=drop.start_at,
end_at=drop.end_at,
benefits=[_serialize_benefit(benefit) for benefit in drop.benefits.all()],
)
def _serialize_campaign_summary(
campaign: DropCampaign,
now: datetime.datetime,
) -> V1DropCampaignSummarySchema:
return V1DropCampaignSummarySchema(
twitch_id=campaign.twitch_id,
name=campaign.name,
description=campaign.description,
status=_campaign_status(campaign.start_at, campaign.end_at, now),
image_url=campaign.listing_image_url,
details_url=campaign.details_url,
account_link_url=campaign.account_link_url,
start_at=campaign.start_at,
end_at=campaign.end_at,
game=_serialize_game_summary(campaign.game, now),
allow_is_enabled=campaign.allow_is_enabled,
is_fully_imported=campaign.is_fully_imported,
added_at=campaign.added_at,
updated_at=campaign.updated_at,
)
def _serialize_campaign_detail(
campaign: DropCampaign,
now: datetime.datetime,
) -> V1DropCampaignDetailSchema:
return V1DropCampaignDetailSchema(
twitch_id=campaign.twitch_id,
name=campaign.name,
description=campaign.description,
status=_campaign_status(campaign.start_at, campaign.end_at, now),
image_url=campaign.listing_image_url,
details_url=campaign.details_url,
account_link_url=campaign.account_link_url,
start_at=campaign.start_at,
end_at=campaign.end_at,
game=_serialize_game_summary(campaign.game, now),
allow_is_enabled=campaign.allow_is_enabled,
is_fully_imported=campaign.is_fully_imported,
added_at=campaign.added_at,
updated_at=campaign.updated_at,
operation_names=campaign.operation_names,
allowed_channels=[
_serialize_channel_summary(channel)
for channel in campaign.allow_channels.all()
],
drops=[_serialize_drop(drop) for drop in campaign.time_based_drops.all()], # pyright: ignore[reportAttributeAccessIssue]
)
def _serialize_reward_campaign(
campaign: RewardCampaign,
now: datetime.datetime,
) -> V1RewardCampaignSchema:
return V1RewardCampaignSchema(
twitch_id=campaign.twitch_id,
name=campaign.name,
brand=campaign.brand,
status=campaign.status,
computed_status=_campaign_status(campaign.starts_at, campaign.ends_at, now),
summary=campaign.summary,
instructions=campaign.instructions,
external_url=campaign.external_url,
reward_value_url_param=campaign.reward_value_url_param,
about_url=campaign.about_url,
image_url=campaign.image_best_url,
is_sitewide=campaign.is_sitewide,
starts_at=campaign.starts_at,
ends_at=campaign.ends_at,
game=_serialize_game_summary(campaign.game, now) if campaign.game else None,
added_at=campaign.added_at,
updated_at=campaign.updated_at,
)
def _serialize_badge(badge: ChatBadge) -> V1ChatBadgeSchema:
return V1ChatBadgeSchema(
badge_id=badge.badge_id,
image_url_1x=badge.image_url_1x,
image_url_2x=badge.image_url_2x,
image_url_4x=badge.image_url_4x,
title=badge.title,
description=badge.description,
click_action=badge.click_action,
click_url=badge.click_url,
)
def _serialize_badge_set(badge_set: ChatBadgeSet) -> V1ChatBadgeSetSchema:
return V1ChatBadgeSetSchema(
set_id=badge_set.set_id,
badges=[_serialize_badge(badge) for badge in badge_set.badges.all()], # pyright: ignore[reportAttributeAccessIssue]
added_at=badge_set.added_at,
updated_at=badge_set.updated_at,
)
@api.get("/campaigns/", response=V1DropCampaignListSchema)
def list_campaigns(
request: HttpRequest,
page: int = 1,
page_size: int = DEFAULT_PAGE_SIZE,
game: str | None = None,
status: V1StatusFilter | None = None,
) -> V1DropCampaignListSchema:
"""Return paginated Twitch drop campaigns."""
now = timezone.now()
queryset = DropCampaign.for_campaign_list(now, game_twitch_id=game, status=status)
page_data = _paginate(queryset, page=page, page_size=page_size)
return V1DropCampaignListSchema(
total=page_data.total,
page=page_data.page,
page_size=page_data.page_size,
items=[
_serialize_campaign_summary(campaign, now) for campaign in page_data.items
],
)
@api.get("/campaigns/{twitch_id}/", response=V1DropCampaignDetailSchema)
def get_campaign(request: HttpRequest, twitch_id: str) -> V1DropCampaignDetailSchema:
"""Return one Twitch drop campaign.
Raises:
Http404: if not found
"""
try:
campaign = DropCampaign.for_detail_view(twitch_id)
except DropCampaign.DoesNotExist as exc:
msg = "Campaign not found"
raise Http404(msg) from exc
return _serialize_campaign_detail(campaign, timezone.now())
@api.get("/games/", response=V1GameListSchema)
def list_games(
request: HttpRequest,
page: int = 1,
page_size: int = DEFAULT_PAGE_SIZE,
) -> V1GameListSchema:
"""Return paginated Twitch games."""
now = timezone.now()
queryset = Game.with_campaign_counts(now).order_by("display_name")
page_data = _paginate(queryset, page=page, page_size=page_size)
return V1GameListSchema(
total=page_data.total,
page=page_data.page,
page_size=page_data.page_size,
items=[_serialize_game(game, now) for game in page_data.items],
)
@api.get("/games/{twitch_id}/", response=V1GameDetailSchema)
def get_game(request: HttpRequest, twitch_id: str) -> V1GameDetailSchema:
"""Return one Twitch game."""
game = get_object_or_404(Game.for_detail_view(), twitch_id=twitch_id)
campaigns = _campaign_summary_queryset(
DropCampaign.objects.filter(game=game),
).order_by("-end_at")
now = timezone.now()
return V1GameDetailSchema(
twitch_id=game.twitch_id,
slug=game.slug,
name=game.name,
display_name=game.display_name,
box_art_url=_game_box_art_url(game),
organizations=[
_serialize_organization_summary(org) for org in game.owners.all()
],
campaign_count=_game_campaign_count(game),
active_campaign_count=_game_active_campaign_count(game, now),
added_at=game.added_at,
updated_at=game.updated_at,
campaigns=[
_serialize_campaign_summary(campaign, now) for campaign in campaigns
],
)
@api.get("/organizations/", response=V1OrganizationListSchema)
def list_organizations(
request: HttpRequest,
page: int = 1,
page_size: int = DEFAULT_PAGE_SIZE,
) -> V1OrganizationListSchema:
"""Return paginated Twitch organizations."""
page_data = _paginate(
Organization.objects.all().order_by("name"),
page=page,
page_size=page_size,
)
return V1OrganizationListSchema(
total=page_data.total,
page=page_data.page,
page_size=page_data.page_size,
items=[_serialize_organization(org) for org in page_data.items],
)
@api.get("/organizations/{twitch_id}/", response=V1OrganizationDetailSchema)
def get_organization(
request: HttpRequest,
twitch_id: str,
) -> V1OrganizationDetailSchema:
"""Return one Twitch organization."""
org = get_object_or_404(Organization.for_detail_view(), twitch_id=twitch_id)
now = timezone.now()
campaigns = (
DropCampaign.objects
.filter(game__owners=org)
.select_related("game")
.prefetch_related(
"game__owners",
"allow_channels",
"time_based_drops__benefits",
)
.order_by("-start_at")
.distinct()
)
return V1OrganizationDetailSchema(
twitch_id=org.twitch_id,
name=org.name,
added_at=org.added_at,
updated_at=org.updated_at,
games=[_serialize_game_summary(game, now) for game in org.games.all()], # pyright: ignore[reportAttributeAccessIssue]
campaigns=[_serialize_campaign_detail(campaign, now) for campaign in campaigns],
)
@api.get("/channels/", response=V1ChannelListSchema)
def list_channels(
request: HttpRequest,
page: int = 1,
page_size: int = DEFAULT_PAGE_SIZE,
search: str | None = None,
) -> V1ChannelListSchema:
"""Return paginated Twitch channels."""
page_data = _paginate(
_channel_list_queryset(search),
page=page,
page_size=page_size,
)
return V1ChannelListSchema(
total=page_data.total,
page=page_data.page,
page_size=page_data.page_size,
items=[_serialize_channel(channel) for channel in page_data.items],
)
@api.get("/channels/{twitch_id}/", response=V1ChannelDetailSchema)
def get_channel(request: HttpRequest, twitch_id: str) -> V1ChannelDetailSchema:
"""Return one Twitch channel."""
channel = get_object_or_404(Channel.for_detail_view(), twitch_id=twitch_id)
campaigns = _campaign_summary_queryset(
DropCampaign.objects.filter(allow_channels=channel),
).order_by("-start_at")
now = timezone.now()
return V1ChannelDetailSchema(
twitch_id=channel.twitch_id,
name=channel.name,
display_name=channel.display_name,
allowed_campaign_count=channel.allowed_campaign_count,
added_at=channel.added_at,
updated_at=channel.updated_at,
campaigns=[
_serialize_campaign_summary(campaign, now) for campaign in campaigns
],
)
@api.get("/reward-campaigns/", response=V1RewardCampaignListSchema)
def list_reward_campaigns(
request: HttpRequest,
page: int = 1,
page_size: int = DEFAULT_PAGE_SIZE,
game: str | None = None,
status: V1StatusFilter | None = None,
) -> V1RewardCampaignListSchema:
"""Return paginated Twitch reward campaigns."""
now = timezone.now()
queryset = RewardCampaign.objects.select_related("game").prefetch_related(
"game__owners",
)
if game:
queryset = queryset.filter(game__twitch_id=game)
queryset = _apply_status_filter(
queryset.order_by("-starts_at"),
status,
now,
start_field="starts_at",
end_field="ends_at",
)
page_data = _paginate(queryset, page=page, page_size=page_size)
return V1RewardCampaignListSchema(
total=page_data.total,
page=page_data.page,
page_size=page_data.page_size,
items=[
_serialize_reward_campaign(campaign, now) for campaign in page_data.items
],
)
@api.get("/reward-campaigns/{twitch_id}/", response=V1RewardCampaignSchema)
def get_reward_campaign(
request: HttpRequest,
twitch_id: str,
) -> V1RewardCampaignSchema:
"""Return one Twitch reward campaign."""
campaign = get_object_or_404(
RewardCampaign.objects.select_related("game").prefetch_related(
"game__owners",
),
twitch_id=twitch_id,
)
return _serialize_reward_campaign(campaign, timezone.now())
@api.get("/badges/", response=V1ChatBadgeSetListSchema)
def list_badges(
request: HttpRequest,
page: int = 1,
page_size: int = DEFAULT_PAGE_SIZE,
) -> V1ChatBadgeSetListSchema:
"""Return paginated Twitch chat badge sets."""
page_data = _paginate(
ChatBadgeSet.for_list_view(),
page=page,
page_size=page_size,
)
return V1ChatBadgeSetListSchema(
total=page_data.total,
page=page_data.page,
page_size=page_data.page_size,
items=[_serialize_badge_set(badge_set) for badge_set in page_data.items],
)
@api.get("/badges/{set_id}/", response=V1ChatBadgeSetSchema)
def get_badge_set(request: HttpRequest, set_id: str) -> V1ChatBadgeSetSchema:
"""Return one Twitch chat badge set."""
badge_set = get_object_or_404(ChatBadgeSet.for_list_view(), set_id=set_id)
return _serialize_badge_set(badge_set)

View file

@ -91,6 +91,10 @@ class Organization(auto_prefetch.Model):
"name",
"display_name",
"slug",
"box_art",
"box_art_file",
"box_art_width",
"box_art_height",
).order_by("display_name"),
to_attr="games_for_detail",
),
@ -340,6 +344,8 @@ class Game(auto_prefetch.Model):
"box_art_file",
"box_art_width",
"box_art_height",
"added_at",
"updated_at",
)
.prefetch_related(
Prefetch(
@ -803,6 +809,8 @@ class DropCampaign(auto_prefetch.Model):
"start_at",
"end_at",
"game",
"game__twitch_id",
"game__name",
"game__display_name",
)
.prefetch_related(
@ -854,12 +862,19 @@ class DropCampaign(auto_prefetch.Model):
"image_height",
"start_at",
"end_at",
"allow_is_enabled",
"operation_names",
"added_at",
"updated_at",
"is_fully_imported",
"game__twitch_id",
"game__name",
"game__display_name",
"game__slug",
"game__box_art",
"game__box_art_file",
"game__box_art_width",
"game__box_art_height",
)
.prefetch_related(
Prefetch(

438
twitch/tests/test_api.py Normal file
View file

@ -0,0 +1,438 @@
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

View file

@ -2504,6 +2504,227 @@ class TestChannelListView:
assert "g-rss" in response.content.decode("utf-8")
@pytest.mark.django_db
class TestRewardCampaignViews:
"""Tests for Twitch reward campaign list and detail views."""
def _create_game(self, twitch_id: str, display_name: str) -> Game:
game: Game = Game.objects.create(
twitch_id=twitch_id,
slug=twitch_id,
name=display_name,
display_name=display_name,
box_art=f"https://example.com/{twitch_id}.png",
)
org: Organization = Organization.objects.create(
twitch_id=f"{twitch_id}-org",
name=f"{display_name} Org",
)
game.owners.add(org)
return game
def _create_reward_campaign( # noqa: PLR0913
self,
twitch_id: str,
*,
brand: str,
name: str,
game: Game | None,
starts_delta: timedelta,
ends_delta: timedelta,
) -> RewardCampaign:
now: datetime.datetime = timezone.now()
return RewardCampaign.objects.create(
twitch_id=twitch_id,
brand=brand,
name=name,
summary=f"{name} summary",
instructions=f"{name} instructions",
external_url=f"https://example.com/{twitch_id}/external",
about_url=f"https://example.com/{twitch_id}/about",
image_url=f"https://example.com/{twitch_id}.png",
starts_at=now + starts_delta,
ends_at=now + ends_delta,
status="ACTIVE",
is_sitewide=game is None,
game=game,
)
def test_reward_campaign_list_renders_expired_campaigns(
self,
client: Client,
) -> None:
"""Render expired reward campaigns with feed and API links."""
game: Game = self._create_game("reward-list-game", "Reward List Game")
expired = self._create_reward_campaign(
"reward-list-expired",
brand="Expired Brand",
name="Expired Reward",
game=game,
starts_delta=-timedelta(days=4),
ends_delta=-timedelta(days=1),
)
self._create_reward_campaign(
"reward-list-active",
brand="Active Brand",
name="Active Reward",
game=game,
starts_delta=-timedelta(days=1),
ends_delta=timedelta(days=1),
)
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:reward_campaign_list"),
)
assert response.status_code == 200
content: str = response.content.decode()
assert "Expired Brand: Expired Reward" in content
assert (
reverse("twitch:reward_campaign_detail", args=[expired.twitch_id])
in content
)
assert reverse("twitch:game_detail", args=[game.twitch_id]) in content
assert reverse("core:reward_campaign_feed") in content
assert reverse("twitch:twitch-api-v1:list_reward_campaigns") in content
assert response.context["reward_campaigns"].paginator.count == 2
def test_reward_campaign_list_filters_status_and_game(
self,
client: Client,
) -> None:
"""Filter reward campaign context by status and game."""
selected_game: Game = self._create_game("reward-filter-game", "Reward Filter")
other_game: Game = self._create_game("reward-filter-other", "Reward Other")
active = self._create_reward_campaign(
"reward-filter-active",
brand="Filter Brand",
name="Active Filter Reward",
game=selected_game,
starts_delta=-timedelta(days=1),
ends_delta=timedelta(days=1),
)
self._create_reward_campaign(
"reward-filter-other-active",
brand="Other Brand",
name="Other Active Reward",
game=other_game,
starts_delta=-timedelta(days=1),
ends_delta=timedelta(days=1),
)
self._create_reward_campaign(
"reward-filter-expired",
brand="Expired Brand",
name="Expired Filter Reward",
game=selected_game,
starts_delta=-timedelta(days=4),
ends_delta=-timedelta(days=1),
)
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:reward_campaign_list")
+ f"?status=active&game={selected_game.twitch_id}",
)
assert response.status_code == 200
campaigns = list(response.context["reward_campaigns"])
assert campaigns == [active]
assert response.context["selected_status"] == "active"
assert response.context["selected_game"] == selected_game.twitch_id
def test_reward_campaign_detail_renders_campaign_data(
self,
client: Client,
) -> None:
"""Render reward campaign detail fields and resource links."""
game: Game = self._create_game("reward-detail-game", "Reward Detail Game")
reward = self._create_reward_campaign(
"reward-detail",
brand="Detail Brand",
name="Detail Reward",
game=game,
starts_delta=-timedelta(days=1),
ends_delta=timedelta(days=1),
)
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:reward_campaign_detail", args=[reward.twitch_id]),
)
assert response.status_code == 200
content: str = response.content.decode()
assert "Detail Brand: Detail Reward" in content
assert "Detail Reward summary" in content
assert "Detail Reward instructions" in content
assert reverse("twitch:game_detail", args=[game.twitch_id]) in content
assert (
reverse(
"twitch:twitch-api-v1:get_reward_campaign",
args=[reward.twitch_id],
)
in content
)
assert reward.external_url in content
assert reward.about_url in content
assert response.context["is_active"] is True
def test_reward_campaign_detail_404_for_missing_campaign(
self,
client: Client,
) -> None:
"""Return 404 for missing reward campaign detail pages."""
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:reward_campaign_detail", args=["missing-reward"]),
)
assert response.status_code == 404
def test_reward_campaign_list_query_count_stays_flat(
self,
client: Client,
) -> None:
"""Reward campaign list should not issue N+1 queries as rows grow."""
game: Game = self._create_game("reward-flat-game", "Reward Flat Game")
def _select_count() -> int:
with CaptureQueriesContext(connection) as ctx:
response: _MonkeyPatchedWSGIResponse = client.get(
reverse("twitch:reward_campaign_list"),
)
assert response.status_code == 200
return sum(
1
for query in ctx.captured_queries
if query["sql"].lstrip().upper().startswith("SELECT")
)
self._create_reward_campaign(
"reward-flat-base",
brand="Flat Brand",
name="Flat Base Reward",
game=game,
starts_delta=-timedelta(days=4),
ends_delta=-timedelta(days=1),
)
baseline: int = _select_count()
for index in range(10):
self._create_reward_campaign(
f"reward-flat-extra-{index}",
brand="Flat Brand",
name=f"Flat Extra Reward {index}",
game=game,
starts_delta=-timedelta(days=4),
ends_delta=-timedelta(days=1),
)
scaled: int = _select_count()
assert scaled <= baseline + 2, (
"Reward campaign list SELECT count grew; possible N+1. "
f"baseline={baseline}, scaled={scaled}"
)
@pytest.mark.django_db
class TestSEOHelperFunctions:
"""Tests for SEO helper functions."""

View file

@ -3,6 +3,7 @@ from typing import TYPE_CHECKING
from django.urls import path
from twitch import views
from twitch.api import api as twitch_api_v1
if TYPE_CHECKING:
from django.urls.resolvers import URLPattern
@ -12,6 +13,8 @@ app_name = "twitch"
urlpatterns: list[URLPattern | URLResolver] = [
# /twitch/api/v1/
path("api/v1/", twitch_api_v1.urls),
# /twitch/
path("", views.dashboard, name="dashboard"),
# /twitch/badges/