TheLovinator/ttvdrops
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor and remove tests, update models and views

Browse source 
- Deleted all test files in accounts and twitch apps to clean up the codebase.
- Updated the DropCampaign, Game, Organization, DropBenefit, TimeBasedDrop, and DropBenefitEdge models to include database indexing for improved query performance.
- Modified the DropCampaignListView and GameDetailView to remove unnecessary status filtering and streamline campaign retrieval logic.
- Enhanced the campaign detail template to properly format campaign descriptions.
- Adjusted the import_drop_campaign management command to increase default worker and batch sizes for improved performance.
- Cleaned up the admin configuration for DropCampaign and TimeBasedDrop models.
This commit is contained in:
Joakim Hellsén 2025-07-24 02:40:59 +02:00
commit 8f4e851fb9
16 changed files with 193 additions and 741 deletions
Download patch file Download diff file Expand all files Collapse all files

8
twitch/admin.py
View file

@ -32,10 +32,10 @@ class TimeBasedDropInline(admin.TabularInline):
class DropCampaignAdmin(admin.ModelAdmin):
"""Admin configuration for DropCampaign model."""
list_display = ("id", "name", "game", "owner", "status", "start_at", "end_at", "is_active")
list_filter = ("status", "game", "owner")
list_display = ("id", "name", "game", "owner", "start_at", "end_at", "is_active")
list_filter = ("game", "owner")
search_fields = ("id", "name", "description")
inlines = [TimeBasedDropInline]
inlines = [TimeBasedDropInline] # noqa: RUF012
readonly_fields = ("created_at", "updated_at")
@ -61,7 +61,7 @@ class TimeBasedDropAdmin(admin.ModelAdmin):
)
list_filter = ("campaign__game", "campaign")
search_fields = ("id", "name")
inlines = [DropBenefitEdgeInline]
inlines = [DropBenefitEdgeInline] # noqa: RUF012
@admin.register(DropBenefit)

6
twitch/management/commands/clean_playback_token_files.py
View file

@ -142,11 +142,7 @@ class Command(BaseCommand):
return
# Set up the deleted directory
if deleted_dir_path:
deleted_dir = Path(str(deleted_dir_path))
else:
# Default to a 'deleted' subdirectory in the source directory
deleted_dir = base_dir / "deleted"
deleted_dir: Path = Path(str(deleted_dir_path)) if deleted_dir_path else base_dir / "deleted"
if not dry_run and not deleted_dir.exists():
deleted_dir.mkdir(parents=True, exist_ok=True)

21
twitch/management/commands/import_drop_campaign.py
View file

@ -38,14 +38,14 @@ class Command(BaseCommand):
parser.add_argument(
"--max-workers",
type=int,
default=4,
help="Maximum number of worker processes to use for parallel importing (default: 4)",
default=100,
help="Maximum number of worker processes to use for parallel importing (default: 100)",
)
parser.add_argument(
"--batch-size",
type=int,
default=100,
help="Number of files to process in each batch (default: 100)",
default=500,
help="Number of files to process in each batch (default: 500)",
)
parser.add_argument(
"--max-retries",
@ -96,7 +96,7 @@ class Command(BaseCommand):
msg = f"Path {path} is neither a file nor a directory"
raise CommandError(msg)
def _process_directory(self, directory: Path, processed_dir: str, max_workers: int = 4, batch_size: int = 100) -> None:
def _process_directory(self, directory: Path, processed_dir: str, max_workers: int = 100, batch_size: int = 1000) -> None:
"""Process all JSON files in a directory using parallel processing.
Args:
@ -189,9 +189,13 @@ class Command(BaseCommand):
try:
with file_path.open(encoding="utf-8") as f:
data = json.load(f)
except json.JSONDecodeError as e:
msg = f"Error decoding JSON: {e}"
raise CommandError(msg) from e
except json.JSONDecodeError:
error_dir_name = "error"
error_dir: Path = file_path.parent / error_dir_name
error_dir.mkdir(exist_ok=True)
self.stdout.write(self.style.WARNING(f"Invalid JSON in '{file_path.name}'. Moving to '{error_dir_name}'."))
shutil.move(str(file_path), str(error_dir / file_path.name))
return 0
# Counter for imported campaigns
campaigns_imported = 0
@ -274,7 +278,6 @@ class Command(BaseCommand):
"image_url": campaign_data.get("imageURL", ""),
"start_at": campaign_data["startAt"],
"end_at": campaign_data["endAt"],
"status": campaign_data["status"],
"is_account_connected": campaign_data["self"]["isAccountConnected"],
"game": game,
"owner": organization,

115
twitch/migrations/0001_initial.py
View file

@ -1,4 +1,4 @@
# Generated by Django 5.2.4 on 2025-07-09 20:44
# Generated by Django 5.2.4 on 2025-07-23 23:51
import django.db.models.deletion
from django.db import migrations, models
@ -16,27 +16,12 @@ class Migration(migrations.Migration):
name='DropBenefit',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('name', models.TextField()),
('name', models.TextField(db_index=True)),
('image_asset_url', models.URLField(blank=True, default='', max_length=500)),
('created_at', models.DateTimeField()),
('created_at', models.DateTimeField(db_index=True)),
('entitlement_limit', models.PositiveIntegerField(default=1)),
('is_ios_available', models.BooleanField(default=False)),
('distribution_type', models.TextField(choices=[('DIRECT_ENTITLEMENT', 'Direct Entitlement'), ('CODE', 'Code')])),
],
),
migrations.CreateModel(
name='Game',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('slug', models.TextField(blank=True, default='')),
('display_name', models.TextField()),
],
),
migrations.CreateModel(
name='Organization',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('name', models.TextField()),
('distribution_type', models.TextField(choices=[('DIRECT_ENTITLEMENT', 'Direct Entitlement'), ('CODE', 'Code')], db_index=True)),
],
),
migrations.CreateModel(
@ -47,23 +32,43 @@ class Migration(migrations.Migration):
('benefit', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='twitch.dropbenefit')),
],
),
migrations.CreateModel(
name='Game',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('slug', models.TextField(blank=True, db_index=True, default='')),
('display_name', models.TextField(db_index=True)),
],
options={
'indexes': [models.Index(fields=['slug'], name='twitch_game_slug_a02d3c_idx'), models.Index(fields=['display_name'], name='twitch_game_display_a35ba3_idx')],
},
),
migrations.AddField(
model_name='dropbenefit',
name='game',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='drop_benefits', to='twitch.game'),
),
migrations.CreateModel(
name='Organization',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('name', models.TextField(db_index=True)),
],
options={
'indexes': [models.Index(fields=['name'], name='twitch_orga_name_febe72_idx')],
},
),
migrations.CreateModel(
name='DropCampaign',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('name', models.TextField()),
('name', models.TextField(db_index=True)),
('description', models.TextField(blank=True)),
('details_url', models.URLField(blank=True, default='', max_length=500)),
('account_link_url', models.URLField(blank=True, default='', max_length=500)),
('image_url', models.URLField(blank=True, default='', max_length=500)),
('start_at', models.DateTimeField()),
('end_at', models.DateTimeField()),
('status', models.TextField(choices=[('ACTIVE', 'Active'), ('UPCOMING', 'Upcoming'), ('EXPIRED', 'Expired')])),
('start_at', models.DateTimeField(db_index=True)),
('end_at', models.DateTimeField(db_index=True)),
('is_account_connected', models.BooleanField(default=False)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
@ -80,11 +85,11 @@ class Migration(migrations.Migration):
name='TimeBasedDrop',
fields=[
('id', models.TextField(primary_key=True, serialize=False)),
('name', models.TextField()),
('required_minutes_watched', models.PositiveIntegerField()),
('name', models.TextField(db_index=True)),
('required_minutes_watched', models.PositiveIntegerField(db_index=True)),
('required_subs', models.PositiveIntegerField(default=0)),
('start_at', models.DateTimeField()),
('end_at', models.DateTimeField()),
('start_at', models.DateTimeField(db_index=True)),
('end_at', models.DateTimeField(db_index=True)),
('benefits', models.ManyToManyField(related_name='drops', through='twitch.DropBenefitEdge', to='twitch.dropbenefit')),
('campaign', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='time_based_drops', to='twitch.dropcampaign')),
],
@ -94,6 +99,62 @@ class Migration(migrations.Migration):
name='drop',
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='twitch.timebaseddrop'),
),
migrations.AddIndex(
model_name='dropcampaign',
index=models.Index(fields=['name'], name='twitch_drop_name_3b70b3_idx'),
),
migrations.AddIndex(
model_name='dropcampaign',
index=models.Index(fields=['start_at', 'end_at'], name='twitch_drop_start_a_6e5fb6_idx'),
),
migrations.AddIndex(
model_name='dropcampaign',
index=models.Index(fields=['game'], name='twitch_drop_game_id_868e70_idx'),
),
migrations.AddIndex(
model_name='dropcampaign',
index=models.Index(fields=['owner'], name='twitch_drop_owner_i_37241d_idx'),
),
migrations.AddIndex(
model_name='dropbenefit',
index=models.Index(fields=['name'], name='twitch_drop_name_7125ff_idx'),
),
migrations.AddIndex(
model_name='dropbenefit',
index=models.Index(fields=['created_at'], name='twitch_drop_created_a3563e_idx'),
),
migrations.AddIndex(
model_name='dropbenefit',
index=models.Index(fields=['distribution_type'], name='twitch_drop_distrib_08b224_idx'),
),
migrations.AddIndex(
model_name='dropbenefit',
index=models.Index(fields=['game'], name='twitch_drop_game_id_a9209e_idx'),
),
migrations.AddIndex(
model_name='dropbenefit',
index=models.Index(fields=['owner_organization'], name='twitch_drop_owner_o_45b4cc_idx'),
),
migrations.AddIndex(
model_name='timebaseddrop',
index=models.Index(fields=['name'], name='twitch_time_name_47c0f4_idx'),
),
migrations.AddIndex(
model_name='timebaseddrop',
index=models.Index(fields=['start_at', 'end_at'], name='twitch_time_start_a_c481f1_idx'),
),
migrations.AddIndex(
model_name='timebaseddrop',
index=models.Index(fields=['campaign'], name='twitch_time_campaig_bbe349_idx'),
),
migrations.AddIndex(
model_name='timebaseddrop',
index=models.Index(fields=['required_minutes_watched'], name='twitch_time_require_82c30c_idx'),
),
migrations.AddIndex(
model_name='dropbenefitedge',
index=models.Index(fields=['drop', 'benefit'], name='twitch_drop_drop_id_5a574c_idx'),
),
migrations.AlterUniqueTogether(
name='dropbenefitedge',
unique_together={('drop', 'benefit')},

88
twitch/models.py
View file

@ -10,8 +10,14 @@ class Game(models.Model):
"""Represents a game on Twitch."""
id = models.TextField(primary_key=True)
slug = models.TextField(blank=True, default="")
display_name = models.TextField()
slug = models.TextField(blank=True, default="", db_index=True)
display_name = models.TextField(db_index=True)
class Meta:
indexes: ClassVar[list] = [
models.Index(fields=["slug"]),
models.Index(fields=["display_name"]),
]
def __str__(self) -> str:
"""Return a string representation of the game."""
@ -22,7 +28,12 @@ class Organization(models.Model):
"""Represents an organization on Twitch that can own drop campaigns."""
id = models.TextField(primary_key=True)
name = models.TextField()
name = models.TextField(db_index=True)
class Meta:
indexes: ClassVar[list] = [
models.Index(fields=["name"]),
]
def __str__(self) -> str:
"""Return a string representation of the organization."""
@ -32,31 +43,32 @@ class Organization(models.Model):
class DropCampaign(models.Model):
"""Represents a Twitch drop campaign."""
STATUS_CHOICES: ClassVar[list[tuple[str, str]]] = [
("ACTIVE", "Active"),
("UPCOMING", "Upcoming"),
("EXPIRED", "Expired"),
]
id = models.TextField(primary_key=True)
name = models.TextField()
name = models.TextField(db_index=True)
description = models.TextField(blank=True)
details_url = models.URLField(max_length=500, blank=True, default="")
account_link_url = models.URLField(max_length=500, blank=True, default="")
image_url = models.URLField(max_length=500, blank=True, default="")
start_at = models.DateTimeField()
end_at = models.DateTimeField()
status = models.TextField(choices=STATUS_CHOICES)
start_at = models.DateTimeField(db_index=True)
end_at = models.DateTimeField(db_index=True)
is_account_connected = models.BooleanField(default=False)
# Foreign keys
game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="drop_campaigns")
owner = models.ForeignKey(Organization, on_delete=models.CASCADE, related_name="drop_campaigns")
game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="drop_campaigns", db_index=True)
owner = models.ForeignKey(Organization, on_delete=models.CASCADE, related_name="drop_campaigns", db_index=True)
# Tracking fields
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
indexes: ClassVar[list] = [
models.Index(fields=["name"]),
models.Index(fields=["start_at", "end_at"]),
models.Index(fields=["game"]),
models.Index(fields=["owner"]),
]
def __str__(self) -> str:
"""Return a string representation of the drop campaign."""
return self.name
@ -65,7 +77,7 @@ class DropCampaign(models.Model):
def is_active(self) -> bool:
"""Check if the campaign is currently active."""
now = timezone.now()
return self.start_at <= now <= self.end_at and self.status == "ACTIVE"
return self.start_at <= now <= self.end_at
@property
def clean_name(self) -> str:
@ -115,16 +127,25 @@ class DropBenefit(models.Model):
]
id = models.TextField(primary_key=True)
name = models.TextField()
name = models.TextField(db_index=True)
image_asset_url = models.URLField(max_length=500, blank=True, default="")
created_at = models.DateTimeField()
created_at = models.DateTimeField(db_index=True)
entitlement_limit = models.PositiveIntegerField(default=1)
is_ios_available = models.BooleanField(default=False)
distribution_type = models.TextField(choices=DISTRIBUTION_TYPES)
distribution_type = models.TextField(choices=DISTRIBUTION_TYPES, db_index=True)
# Foreign keys
game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="drop_benefits")
owner_organization = models.ForeignKey(Organization, on_delete=models.CASCADE, related_name="drop_benefits")
game = models.ForeignKey(Game, on_delete=models.CASCADE, related_name="drop_benefits", db_index=True)
owner_organization = models.ForeignKey(Organization, on_delete=models.CASCADE, related_name="drop_benefits", db_index=True)
class Meta:
indexes: ClassVar[list] = [
models.Index(fields=["name"]),
models.Index(fields=["created_at"]),
models.Index(fields=["distribution_type"]),
models.Index(fields=["game"]),
models.Index(fields=["owner_organization"]),
]
def __str__(self) -> str:
"""Return a string representation of the drop benefit."""
@ -135,16 +156,24 @@ class TimeBasedDrop(models.Model):
"""Represents a time-based drop in a drop campaign."""
id = models.TextField(primary_key=True)
name = models.TextField()
required_minutes_watched = models.PositiveIntegerField()
name = models.TextField(db_index=True)
required_minutes_watched = models.PositiveIntegerField(db_index=True)
required_subs = models.PositiveIntegerField(default=0)
start_at = models.DateTimeField()
end_at = models.DateTimeField()
start_at = models.DateTimeField(db_index=True)
end_at = models.DateTimeField(db_index=True)
# Foreign keys
campaign = models.ForeignKey(DropCampaign, on_delete=models.CASCADE, related_name="time_based_drops")
campaign = models.ForeignKey(DropCampaign, on_delete=models.CASCADE, related_name="time_based_drops", db_index=True)
benefits = models.ManyToManyField(DropBenefit, through="DropBenefitEdge", related_name="drops")
class Meta:
indexes: ClassVar[list] = [
models.Index(fields=["name"]),
models.Index(fields=["start_at", "end_at"]),
models.Index(fields=["campaign"]),
models.Index(fields=["required_minutes_watched"]),
]
def __str__(self) -> str:
"""Return a string representation of the time-based drop."""
return self.name
@ -153,12 +182,15 @@ class TimeBasedDrop(models.Model):
class DropBenefitEdge(models.Model):
"""Represents the relationship between a TimeBasedDrop and a DropBenefit."""
drop = models.ForeignKey(TimeBasedDrop, on_delete=models.CASCADE)
benefit = models.ForeignKey(DropBenefit, on_delete=models.CASCADE)
drop = models.ForeignKey(TimeBasedDrop, on_delete=models.CASCADE, db_index=True)
benefit = models.ForeignKey(DropBenefit, on_delete=models.CASCADE, db_index=True)
entitlement_limit = models.PositiveIntegerField(default=1)
class Meta:
unique_together = ("drop", "benefit")
indexes: ClassVar[list] = [
models.Index(fields=["drop", "benefit"]),
]
def __str__(self) -> str:
"""Return a string representation of the drop benefit edge."""

1
twitch/tests.py
View file

@ -1 +0,0 @@
# Create your tests here.

0
twitch/tests/__init__.py
View file

168
twitch/tests/test_pagination.py
View file

@ -1,168 +0,0 @@
"""Tests for pagination functionality in the campaign list view."""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from django.urls import reverse
from django.utils import timezone
from twitch.models import DropCampaign, Game, Organization
if TYPE_CHECKING:
from django.test import Client
@pytest.mark.django_db
class TestCampaignListPagination:
"""Test cases for campaign list pagination."""
def test_default_pagination(self, client: Client) -> None:
"""Test that pagination works with default settings."""
# Create test data - enough to require pagination
game = Game.objects.create(id="test-game", slug="test-game", display_name="Test Game")
org = Organization.objects.create(id="test-org", name="Test Org")
# Create 30 campaigns to test pagination
now = timezone.now()
campaigns = []
for i in range(30):
campaign = DropCampaign.objects.create(
id=f"campaign-{i}",
name=f"Campaign {i}",
game=game,
owner=org,
start_at=now,
end_at=now + timezone.timedelta(days=1),
status="ACTIVE",
)
campaigns.append(campaign)
# Test first page
response = client.get(reverse("twitch:campaign_list"))
assert response.status_code == 200
assert "page_obj" in response.context
assert response.context["page_obj"].number == 1
assert len(response.context["campaigns"]) == 24 # Default paginate_by
assert response.context["page_obj"].paginator.count == 30
# Test second page
response = client.get(reverse("twitch:campaign_list") + "?page=2")
assert response.status_code == 200
assert response.context["page_obj"].number == 2
assert len(response.context["campaigns"]) == 6 # Remaining campaigns
def test_custom_per_page(self, client: Client) -> None:
"""Test that custom per_page parameter works."""
# Create test data
game = Game.objects.create(id="test-game", slug="test-game", display_name="Test Game")
org = Organization.objects.create(id="test-org", name="Test Org")
now = timezone.now()
for i in range(25):
DropCampaign.objects.create(
id=f"campaign-{i}",
name=f"Campaign {i}",
game=game,
owner=org,
start_at=now,
end_at=now + timezone.timedelta(days=1),
status="ACTIVE",
)
# Test with per_page=12
response = client.get(reverse("twitch:campaign_list") + "?per_page=12")
assert response.status_code == 200
assert len(response.context["campaigns"]) == 12
assert response.context["selected_per_page"] == 12
# Test with per_page=48
response = client.get(reverse("twitch:campaign_list") + "?per_page=48")
assert response.status_code == 200
assert len(response.context["campaigns"]) == 25 # All campaigns fit on one page
def test_invalid_per_page_fallback(self, client: Client) -> None:
"""Test that invalid per_page values fall back to default."""
# Create test data
game = Game.objects.create(id="test-game", slug="test-game", display_name="Test Game")
org = Organization.objects.create(id="test-org", name="Test Org")
now = timezone.now()
for i in range(30):
DropCampaign.objects.create(
id=f"campaign-{i}",
name=f"Campaign {i}",
game=game,
owner=org,
start_at=now,
end_at=now + timezone.timedelta(days=1),
status="ACTIVE",
)
# Test with invalid per_page value
response = client.get(reverse("twitch:campaign_list") + "?per_page=999")
assert response.status_code == 200
assert len(response.context["campaigns"]) == 24 # Falls back to default
assert response.context["selected_per_page"] == 24
# Test with non-numeric per_page value
response = client.get(reverse("twitch:campaign_list") + "?per_page=invalid")
assert response.status_code == 200
assert len(response.context["campaigns"]) == 24 # Falls back to default
def test_pagination_with_filters(self, client: Client) -> None:
"""Test that pagination works correctly with filters."""
# Create test data with different statuses
game = Game.objects.create(id="test-game", slug="test-game", display_name="Test Game")
org = Organization.objects.create(id="test-org", name="Test Org")
now = timezone.now()
# Create 20 active campaigns
for i in range(20):
DropCampaign.objects.create(
id=f"active-{i}",
name=f"Active Campaign {i}",
game=game,
owner=org,
start_at=now,
end_at=now + timezone.timedelta(days=1),
status="ACTIVE",
)
# Create 10 expired campaigns
for i in range(10):
DropCampaign.objects.create(
id=f"expired-{i}",
name=f"Expired Campaign {i}",
game=game,
owner=org,
start_at=now - timezone.timedelta(days=2),
end_at=now - timezone.timedelta(days=1),
status="EXPIRED",
)
# Test filtering by active status with pagination
response = client.get(reverse("twitch:campaign_list") + "?status=ACTIVE&per_page=12")
assert response.status_code == 200
assert len(response.context["campaigns"]) == 12
assert response.context["page_obj"].paginator.count == 20 # Only active campaigns
assert all(c.status == "ACTIVE" for c in response.context["campaigns"])
# Test second page of active campaigns
response = client.get(reverse("twitch:campaign_list") + "?status=ACTIVE&per_page=12&page=2")
assert response.status_code == 200
assert len(response.context["campaigns"]) == 8 # Remaining active campaigns
assert response.context["page_obj"].number == 2
def test_context_variables(self, client: Client) -> None:
"""Test that all necessary context variables are present."""
response = client.get(reverse("twitch:campaign_list"))
assert response.status_code == 200
# Check for pagination-related context
context = response.context
assert "per_page_options" in context
assert "selected_per_page" in context
assert context["per_page_options"] == [12, 24, 48, 96]
assert context["selected_per_page"] == 24 # Default value

83
twitch/tests/test_views.py
View file

@ -1,83 +0,0 @@
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from django.urls import reverse
from django.utils import timezone
from twitch.models import DropCampaign, Game, Organization
if TYPE_CHECKING:
from django.test import Client
@pytest.mark.django_db
class TestGameDetailView:
"""Test cases for GameDetailView."""
def test_expired_campaigns_filtering(self, client: Client) -> None:
"""Test that expired campaigns are correctly filtered."""
# Create test data
game = Game.objects.create(
id="123",
slug="test-game",
display_name="Test Game",
)
organization = Organization.objects.create(
id="456",
name="Test Organization",
)
now = timezone.now()
# Create an active campaign
active_campaign = DropCampaign.objects.create(
id="active-campaign",
name="Active Campaign",
game=game,
owner=organization,
start_at=now - timezone.timedelta(days=1),
end_at=now + timezone.timedelta(days=1),
status="ACTIVE",
)
# Create an expired campaign (end date in the past)
expired_by_date = DropCampaign.objects.create(
id="expired-by-date",
name="Expired By Date",
game=game,
owner=organization,
start_at=now - timezone.timedelta(days=3),
end_at=now - timezone.timedelta(days=1),
status="ACTIVE", # Still marked as active but date is expired
)
# Create an expired campaign (status is EXPIRED)
expired_by_status = DropCampaign.objects.create(
id="expired-by-status",
name="Expired By Status",
game=game,
owner=organization,
start_at=now - timezone.timedelta(days=3),
end_at=now + timezone.timedelta(days=1),
status="EXPIRED", # Explicitly expired
)
# Get the view context
url = reverse("twitch:game_detail", kwargs={"pk": game.id})
response = client.get(url)
# Check that active_campaigns only contains the active campaign
active_campaigns = response.context["active_campaigns"]
assert len(active_campaigns) == 1
assert active_campaigns[0].id == active_campaign.id
# Check that expired_campaigns contains only the expired campaigns
expired_campaigns = response.context["expired_campaigns"]
assert len(expired_campaigns) == 2
expired_campaign_ids = [c.id for c in expired_campaigns]
assert expired_by_date.id in expired_campaign_ids
assert expired_by_status.id in expired_campaign_ids
assert active_campaign.id not in expired_campaign_ids

23
twitch/views.py
View file

@ -26,7 +26,7 @@ class DropCampaignListView(ListView):
context_object_name = "campaigns"
paginate_by = 96 # Default pagination size
def get_paginate_by(self, queryset) -> int:
def get_paginate_by(self, queryset) -> int: # noqa: ANN001, ARG002
"""Get the pagination size, allowing override via URL parameter.
Args:
@ -50,13 +50,8 @@ class DropCampaignListView(ListView):
QuerySet: Filtered drop campaigns.
"""
queryset: QuerySet[DropCampaign] = super().get_queryset()
status_filter: str | None = self.request.GET.get("status")
game_filter: str | None = self.request.GET.get("game")
# Apply filters
if status_filter:
queryset = queryset.filter(status=status_filter)
if game_filter:
queryset = queryset.filter(game__id=game_filter)
@ -78,8 +73,7 @@ class DropCampaignListView(ListView):
context["games"] = Game.objects.all().order_by("display_name")
# Add status options for filtering
context["status_options"] = [status[0] for status in DropCampaign.STATUS_CHOICES]
context["status_options"] = []
# Add selected filters
context["selected_status"] = self.request.GET.get("status", "")
context["selected_game"] = self.request.GET.get("game", "")
@ -177,7 +171,6 @@ class GameListView(ListView):
filter=Q(
drop_campaigns__start_at__lte=now,
drop_campaigns__end_at__gte=now,
drop_campaigns__status="ACTIVE",
),
distinct=True,
),
@ -208,7 +201,7 @@ class GameListView(ListView):
# This query gets all games with their campaign counts and organization info
game_org_relations = DropCampaign.objects.values("game_id", "owner_id", "owner__name").annotate(
campaign_count=Count("id", distinct=True),
active_count=Count("id", filter=Q(start_at__lte=now, end_at__gte=now, status="ACTIVE"), distinct=True),
active_count=Count("id", filter=Q(start_at__lte=now, end_at__gte=now), distinct=True),
)
# Step 3: Get all games in a single query with their display names
@ -279,16 +272,14 @@ class GameDetailView(DetailView):
all_campaigns = DropCampaign.objects.filter(game=game).select_related("owner").order_by("-end_at")
# Filter the campaigns in Python instead of making multiple queries
active_campaigns = [
campaign for campaign in all_campaigns if campaign.start_at <= now and campaign.end_at >= now and campaign.status == "ACTIVE"
]
active_campaigns = [campaign for campaign in all_campaigns if campaign.start_at <= now and campaign.end_at >= now]
active_campaigns.sort(key=lambda c: c.end_at) # Sort by end_at ascending
upcoming_campaigns = [campaign for campaign in all_campaigns if campaign.start_at > now and campaign.status == "UPCOMING"]
upcoming_campaigns = [campaign for campaign in all_campaigns if campaign.start_at > now]
upcoming_campaigns.sort(key=lambda c: c.start_at) # Sort by start_at ascending
# Filter for expired campaigns
expired_campaigns = [campaign for campaign in all_campaigns if campaign.end_at < now or campaign.status == "EXPIRED"]
expired_campaigns = [campaign for campaign in all_campaigns if campaign.end_at < now]
context.update({
"active_campaigns": active_campaigns,
@ -312,7 +303,7 @@ def dashboard(request: HttpRequest) -> HttpResponse:
# Get active campaigns with prefetching to reduce queries
now = timezone.now()
active_campaigns = (
DropCampaign.objects.filter(start_at__lte=now, end_at__gte=now, status="ACTIVE")
DropCampaign.objects.filter(start_at__lte=now, end_at__gte=now)
.select_related("game", "owner")
# Prefetch the time-based drops with their benefits to avoid N+1 queries
.prefetch_related(Prefetch("time_based_drops", queryset=TimeBasedDrop.objects.prefetch_related("benefits")))