Add VSCode tasks for Django management commands

Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>

.github/workflows/ruff.yml

@@ -12,7 +12,7 @@ env:
   DEBUG: "True"
 
 jobs:
-  build:
+  ruff:
     runs-on: ubuntu-latest
     steps:
       # Download the source code
@@ -20,7 +20,7 @@ jobs:
 
       # https://docs.astral.sh/uv/
       - name: Install uv
-        uses: astral-sh/setup-uv@v4
+        uses: astral-sh/setup-uv@v5
 
       # Set up our GitHub Actions workflow with the Python version specified in the .python-version file
       - name: "Set up Python"

.pre-commit-config.yaml

@@ -50,14 +50,14 @@ repos:
 
   # A tool (and pre-commit hook) to automatically upgrade syntax for newer versions of the language.
   - repo: https://github.com/asottile/pyupgrade
-    rev: v3.19.0
+    rev: v3.19.1
     hooks:
       - id: pyupgrade
         args: ["--py311-plus"]
 
   # An extremely fast Python linter and formatter.
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.3
+    rev: v0.8.4
     hooks:
       - id: ruff-format
       - id: ruff

.vscode/launch.json

@@ -1,6 +1,13 @@
 {
   "version": "0.2.0",
   "configurations": [
+    {
+      "name": "Python Debugger: Current File",
+      "type": "debugpy",
+      "request": "launch",
+      "program": "${file}",
+      "console": "integratedTerminal"
+    },
     {
       "name": "Django: Runserver",
       "type": "debugpy",

.vscode/tasks.json

@@ -0,0 +1,29 @@
+{
+    "version": "2.0.0",
+    "tasks": [
+        {
+            "label": "Django: Make migrations.",
+            "type": "shell",
+            "command": "uv run python manage.py makemigrations",
+            "problemMatcher": [
+                "$python"
+            ]
+        },
+        {
+            "label": "Django: Migrate.",
+            "type": "shell",
+            "command": "uv run python manage.py migrate",
+            "problemMatcher": [
+                "$python"
+            ]
+        },
+        {
+            "label": "Django: Run pytest.",
+            "type": "shell",
+            "command": "uv run pytest",
+            "problemMatcher": [
+                "$python"
+            ]
+        }
+    ]
+}

Dockerfile

@@ -1,5 +1,5 @@
 # Stage 1: Build the requirements.txt using Poetry
-FROM python:3.12-slim AS builder
+FROM python:3.13-slim AS builder
 
 # Set environment variables for Python
 ENV PYTHONDONTWRITEBYTECODE 1
@@ -24,7 +24,7 @@ COPY pyproject.toml poetry.lock /app/
 RUN poetry self add poetry-plugin-export && poetry export --format=requirements.txt --output=requirements.txt --only=main --without-hashes
 
 # Stage 2: Install dependencies and run the Django application
-FROM python:3.12-slim AS runner
+FROM python:3.13-slim AS runner
 
 # Set environment variables for Python
 ENV PYTHONDONTWRITEBYTECODE 1

core/import_json.py

@@ -1,11 +1,187 @@
 from __future__ import annotations
 
-from typing import Any
+import logging
+from typing import Any, Literal
+
+from core.models import Benefit, DropCampaign, Game, Owner, TimeBasedDrop
+
+logger: logging.Logger = logging.getLogger(__name__)
 
 
-def import_data_from_view(data: dict[str, Any]) -> None:
-    """Import data that are sent from Twitch Drop Miner.
+type_names = Literal["Organization", "Game", "DropCampaign", "TimeBasedDrop", "DropBenefit", "RewardCampaign", "Reward"]
+
+
+def import_data(data: dict[str, Any]) -> None:
+    """Import the data from the JSON object.
+
+    This looks for '__typename' with the value 'DropCampaign' in the JSON object and imports the data.
 
     Args:
         data (dict[str, Any]): The data to import.
     """
+    drop_campaigns: list[dict[str, Any]] = find_typename_in_json(json_obj=data, typename_to_find="DropCampaign")
+    for drop_campaign_json in drop_campaigns:
+        import_drop_campaigns(drop_campaigns=drop_campaign_json)
+
+
+def import_drop_campaigns(drop_campaigns: dict[str, Any]) -> DropCampaign | None:
+    """Import the drop campaigns from the data.
+
+    Args:
+        drop_campaigns (dict[str, Any]): The drop campaign data.
+
+    Returns:
+        DropCampaign | None: The drop campaign instance if created, otherwise None
+    """
+    twitch_id: str = drop_campaigns.get("id", "")
+    logger.info("\tProcessing drop campaign: %s", twitch_id)
+
+    if not twitch_id:
+        logger.error("\tDrop campaign has no ID: %s", drop_campaigns)
+        return None
+
+    drop_campaign, created = DropCampaign.objects.get_or_create(twitch_id=twitch_id)
+    if created:
+        logger.info("\tCreated drop campaign: %s", drop_campaign)
+
+    owner: Owner = import_owner_data(drop_campaign=drop_campaigns)
+    game: Game = import_game_data(drop_campaign=drop_campaigns, owner=owner)
+    drop_campaign.import_json(data=drop_campaigns, game=game)
+
+    import_time_based_drops(drop_campaigns, drop_campaign)
+
+    return drop_campaign
+
+
+def import_time_based_drops(drop_campaign_json: dict[str, Any], drop_campaign: DropCampaign) -> list[TimeBasedDrop]:
+    """Import the time-based drops from a drop campaign.
+
+    Args:
+        drop_campaign_json (dict[str, Any]): The drop campaign data.
+        drop_campaign (DropCampaign): The drop campaign instance.
+
+    Returns:
+        list[TimeBasedDrop]: The imported time-based drops.
+    """
+    imported_drops: list[TimeBasedDrop] = []
+    time_based_drops: list[dict[str, Any]] = find_typename_in_json(drop_campaign_json, "TimeBasedDrop")
+    for time_based_drop_json in time_based_drops:
+        time_based_drop_id: str = time_based_drop_json.get("id", "")
+        if not time_based_drop_id:
+            logger.error("\tTime-based drop has no ID: %s", time_based_drop_json)
+            continue
+
+        time_based_drop, created = TimeBasedDrop.objects.get_or_create(twitch_id=time_based_drop_id)
+        if created:
+            logger.info("\tCreated time-based drop: %s", time_based_drop)
+
+        time_based_drop.import_json(time_based_drop_json, drop_campaign)
+
+        import_drop_benefits(time_based_drop_json, time_based_drop)
+        imported_drops.append(time_based_drop)
+
+    return imported_drops
+
+
+def import_drop_benefits(time_based_drop_json: dict[str, Any], time_based_drop: TimeBasedDrop) -> list[Benefit]:
+    """Import the drop benefits from a time-based drop.
+
+    Args:
+        time_based_drop_json (dict[str, Any]): The time-based drop data.
+        time_based_drop (TimeBasedDrop): The time-based drop instance.
+
+    Returns:
+        list[Benefit]: The imported drop benefits.
+    """
+    drop_benefits: list[Benefit] = []
+    benefits: list[dict[str, Any]] = find_typename_in_json(time_based_drop_json, "DropBenefit")
+    for benefit_json in benefits:
+        benefit_id: str = benefit_json.get("id", "")
+        if not benefit_id:
+            logger.error("\tBenefit has no ID: %s", benefit_json)
+            continue
+
+        benefit, created = Benefit.objects.get_or_create(twitch_id=benefit_id)
+        if created:
+            logger.info("\tCreated benefit: %s", benefit)
+
+        benefit.import_json(benefit_json, time_based_drop)
+        drop_benefits.append(benefit)
+
+    return drop_benefits
+
+
+def import_owner_data(drop_campaign: dict[str, Any]) -> Owner:
+    """Import the owner data from a drop campaign.
+
+    Args:
+        drop_campaign (dict[str, Any]): The drop campaign data.
+
+    Returns:
+        Owner: The owner instance.
+    """
+    owner_data_list: list[dict[str, Any]] = find_typename_in_json(drop_campaign, "Organization")
+    for owner_data in owner_data_list:
+        owner_id: str = owner_data.get("id", "")
+        if not owner_id:
+            logger.error("\tOwner has no ID: %s", owner_data)
+            continue
+
+        owner, created = Owner.objects.get_or_create(twitch_id=owner_id)
+        if created:
+            logger.info("\tCreated owner: %s", owner)
+
+        owner.import_json(owner_data)
+    return owner
+
+
+def import_game_data(drop_campaign: dict[str, Any], owner: Owner) -> Game:
+    """Import the game data from a drop campaign.
+
+    Args:
+        drop_campaign (dict[str, Any]): The drop campaign data.
+        owner (Owner): The owner of the game.
+
+    Returns:
+        Game: The game instance.
+    """
+    game_data_list: list[dict[str, Any]] = find_typename_in_json(drop_campaign, "Game")
+    for game_data in game_data_list:
+        game_id: str = game_data.get("id", "")
+        if not game_id:
+            logger.error("\tGame has no ID: %s", game_data)
+            continue
+
+        game, created = Game.objects.get_or_create(twitch_id=game_id)
+        if created:
+            logger.info("\tCreated game: %s", game)
+
+        game.import_json(game_data, owner)
+    return game
+
+
+def find_typename_in_json(json_obj: list | dict, typename_to_find: type_names) -> list[dict[str, Any]]:
+    """Recursively search for '__typename' in a JSON object and return dictionaries where '__typename' equals the specified value.
+
+    Args:
+        json_obj (list | dict): The JSON object to search.
+        typename_to_find (str): The '__typename' value to search for.
+
+    Returns:
+        A list of dictionaries where '__typename' equals the specified value.
+    """  # noqa: E501
+    matching_dicts: list[dict[str, Any]] = []
+
+    if isinstance(json_obj, dict):
+        # Check if '__typename' exists and matches the value
+        if json_obj.get("__typename") == typename_to_find:
+            matching_dicts.append(json_obj)
+        # Recurse into the dictionary
+        for value in json_obj.values():
+            matching_dicts.extend(find_typename_in_json(value, typename_to_find))
+    elif isinstance(json_obj, list):
+        # Recurse into each item in the list
+        for item in json_obj:
+            matching_dicts.extend(find_typename_in_json(item, typename_to_find))
+
+    return matching_dicts

core/migrations/0002_alter_user_options.py

@@ -0,0 +1,24 @@
+# Generated by Django 5.1.4 on 2024-12-16 18:28
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from django.db import migrations
+
+if TYPE_CHECKING:
+    from django.db.migrations.operations.base import Operation
+
+
+class Migration(migrations.Migration):
+    """This migration alters the options of the User model to order by username."""
+
+    dependencies: list[tuple[str, str]] = [
+        ("core", "0001_initial"),
+    ]
+
+    operations: list[Operation] = [
+        migrations.AlterModelOptions(
+            name="user",
+            options={"ordering": ["username"]},
+        ),
+    ]

core/migrations/0003_alter_benefit_created_at_and_more.py

@@ -0,0 +1,48 @@
+# Generated by Django 5.1.4 on 2024-12-16 18:31
+from __future__ import annotations
+
+import datetime
+from typing import TYPE_CHECKING
+
+from django.db import migrations, models
+
+if TYPE_CHECKING:
+    from django.db.migrations.operations.base import Operation
+
+
+class Migration(migrations.Migration):
+    """This migration alters the created_at field on the Benefit, DropCampaign, Owner, and TimeBasedDrop models."""
+
+    dependencies: list[tuple[str, str]] = [
+        ("core", "0002_alter_user_options"),
+    ]
+
+    operations: list[Operation] = [
+        migrations.AlterField(
+            model_name="benefit",
+            name="created_at",
+            field=models.DateTimeField(
+                auto_now_add=True,
+                default=datetime.datetime(2024, 12, 16, 18, 31, 14, 851533, tzinfo=datetime.UTC),
+            ),
+            preserve_default=False,
+        ),
+        migrations.AlterField(
+            model_name="dropcampaign",
+            name="created_at",
+            field=models.DateTimeField(
+                auto_now_add=True,
+                help_text="When the drop campaign was first added to the database.",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="owner",
+            name="created_at",
+            field=models.DateTimeField(auto_now_add=True),
+        ),
+        migrations.AlterField(
+            model_name="timebaseddrop",
+            name="created_at",
+            field=models.DateTimeField(auto_now_add=True, help_text="When the drop was first added to the database."),
+        ),
+    ]

core/migrations/0004_alter_game_created_at.py

@@ -0,0 +1,25 @@
+# Generated by Django 5.1.4 on 2024-12-18 22:35
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from django.db import migrations, models
+
+if TYPE_CHECKING:
+    from django.db.migrations.operations.base import Operation
+
+
+class Migration(migrations.Migration):
+    """Fix created_at using auto_created instead of auto_now_add."""
+
+    dependencies: list[tuple[str, str]] = [
+        ("core", "0003_alter_benefit_created_at_and_more"),
+    ]
+
+    operations: list[Operation] = [
+        migrations.AlterField(
+            model_name="game",
+            name="created_at",
+            field=models.DateTimeField(auto_now_add=True, help_text="When the game was first added to the database."),
+        ),
+    ]

core/models.py

@@ -69,7 +69,7 @@ class Owner(auto_prefetch.Model):
     # Django fields
     # Example: "36c4e21d-bdf3-410c-97c3-5a5a4bf1399b"
     twitch_id = models.TextField(primary_key=True, help_text="The Twitch ID of the owner.")
-    created_at = models.DateTimeField(auto_created=True)
+    created_at = models.DateTimeField(auto_now_add=True)
     modified_at = models.DateTimeField(auto_now=True)
 
     # Twitch fields
@@ -104,74 +104,36 @@ class Game(auto_prefetch.Model):
     """The game the drop campaign is for. Note that some reward campaigns are not tied to a game.
 
     JSON:
-        {
-            "data": {
-                "user": {
-                    "dropCampaign": {
-                        "game": {
-                            "id": "155409827",
-                            "slug": "pokemon-trading-card-game-live",
-                            "displayName": "Pok\u00e9mon Trading Card Game Live",
-                            "__typename": "Game"
-                        }
-                    }
-                }
-            }
+        "game": {
+            "id": "155409827",
+            "slug": "pokemon-trading-card-game-live",
+            "displayName": "Pok\u00e9mon Trading Card Game Live",
+            "__typename": "Game"
         }
 
+
     Secondary JSON:
-        {
-            "data": {
-                "currentUser": {
-                    "dropCampaigns": [
-                        {
-                            "game": {
-                                "id": "155409827",
-                                "displayName": "Pok\u00e9mon Trading Card Game Live",
-                                "boxArtURL": "https://static-cdn.jtvnw.net/ttv-boxart/155409827_IGDB-120x160.jpg",
-                                "__typename": "Game"
-                            }
-                        }
-                    ]
-                }
-            }
+        "game": {
+            "id": "155409827",
+            "displayName": "Pok\u00e9mon Trading Card Game Live",
+            "boxArtURL": "https://static-cdn.jtvnw.net/ttv-boxart/155409827_IGDB-120x160.jpg",
+            "__typename": "Game"
         }
 
+
     Tertiary JSON:
-        [
-            {
-                "data": {
-                    "user": {
-                        "dropCampaign": {
-                            "timeBasedDrops": [
-                                {
-                                    "benefitEdges": [
-                                        {
-                                            "benefit": {
-                                                "id": "ea74f727-a52f-11ef-811f-0a58a9feac02",
-                                                "createdAt": "2024-11-17T22:04:28.735Z",
-                                                "entitlementLimit": 1,
-                                                "game": {
-                                                    "id": "155409827",
-                                                    "name": "Pok\u00e9mon Trading Card Game Live",
-                                                    "__typename": "Game"
-                                                }
-                                            }
-                                        }
-                                    ]
-                                }
-                            ]
-                        }
-                    }
-                }
-            }
-        ]
+        "game": {
+            "id": "155409827",
+            "name": "Pok\u00e9mon Trading Card Game Live",
+            "__typename": "Game"
+        }
+
     """
 
     # Django fields
     # "155409827"
     twitch_id = models.TextField(primary_key=True, help_text="The Twitch ID of the game.")
-    created_at = models.DateTimeField(auto_created=True, help_text="When the game was first added to the database.")
+    created_at = models.DateTimeField(auto_now_add=True, help_text="When the game was first added to the database.")
     modified_at = models.DateTimeField(auto_now=True, help_text="When the game was last modified.")
 
     # Twitch fields
@@ -212,6 +174,10 @@ class Game(auto_prefetch.Model):
         if wrong_typename(data, "Game"):
             return self
 
+        if not owner:
+            logger.error("Owner is required for %s: %s", self, data)
+            return self
+
         # Map the fields from the JSON data to the Django model fields.
         field_mapping: dict[str, str] = {
             "displayName": "display_name",
@@ -224,15 +190,14 @@ class Game(auto_prefetch.Model):
         if updated > 0:
             logger.info("Updated %s fields for %s", updated, self)
 
-        if not owner:
-            logger.error("Owner is required for %s", self)
-            return self
-
         # Update the owner if the owner is different or not set.
         if owner != self.org:
             self.org = owner
             logger.info("Updated owner %s for %s", owner, self)
-            self.save()
+
+        self.game_url = f"https://www.twitch.tv/directory/category/{self.slug}"
+
+        self.save()
 
         return self
 
@@ -244,7 +209,7 @@ class DropCampaign(auto_prefetch.Model):
     # "f257ce6e-502a-11ef-816e-0a58a9feac02"
     twitch_id = models.TextField(primary_key=True, help_text="The Twitch ID of the drop campaign.")
     created_at = models.DateTimeField(
-        auto_created=True,
+        auto_now_add=True,
         help_text="When the drop campaign was first added to the database.",
     )
     modified_at = models.DateTimeField(auto_now=True, help_text="When the drop campaign was last modified.")
@@ -346,59 +311,47 @@ class TimeBasedDrop(auto_prefetch.Model):
 
     JSON:
         {
-            "data": {
-                "user": {
-                    "dropCampaign": {
-                        "timeBasedDrops": [
-                            {
-                                "id": "bd663e10-b297-11ef-a6a3-0a58a9feac02",
-                                "requiredSubs": 0,
-                                "benefitEdges": [
-                                    {
-                                        "benefit": {
-                                            "id": "f751ba67-7c8b-4c41-b6df-bcea0914f3ad_CUSTOM_ID_EnergisingBoltFlaskEffect",
-                                            "createdAt": "2024-12-04T23:25:50.995Z",
-                                            "entitlementLimit": 1,
-                                            "game": {
-                                                "id": "1702520304",
-                                                "name": "Path of Exile 2",
-                                                "__typename": "Game"
-                                            },
-                                            "imageAssetURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/REWARD/d70e4e75-7237-4730-9a10-b6016aaaa795.png",
-                                            "isIosAvailable": false,
-                                            "name": "Energising Bolt Flask",
-                                            "ownerOrganization": {
-                                                "id": "f751ba67-7c8b-4c41-b6df-bcea0914f3ad",
-                                                "name": "Grinding Gear Games",
-                                                "__typename": "Organization"
-                                            },
-                                            "distributionType": "DIRECT_ENTITLEMENT",
-                                            "__typename": "DropBenefit"
-                                        },
-                                        "entitlementLimit": 1,
-                                        "__typename": "DropBenefitEdge"
-                                    }
-                                ],
-                                "endAt": "2024-12-14T07:59:59.996Z",
-                                "name": "Early Access Bundle",
-                                "preconditionDrops": null,
-                                "requiredMinutesWatched": 180,
-                                "startAt": "2024-12-06T19:00:00Z",
-                                "__typename": "TimeBasedDrop"
-                            }
-                        ],
-                        "__typename": "DropCampaign"
+            "id": "bd663e10-b297-11ef-a6a3-0a58a9feac02",
+            "requiredSubs": 0,
+            "benefitEdges": [
+                {
+                    "benefit": {
+                        "id": "f751ba67-7c8b-4c41-b6df-bcea0914f3ad_CUSTOM_ID_EnergisingBoltFlaskEffect",
+                        "createdAt": "2024-12-04T23:25:50.995Z",
+                        "entitlementLimit": 1,
+                        "game": {
+                            "id": "1702520304",
+                            "name": "Path of Exile 2",
+                            "__typename": "Game"
+                        },
+                        "imageAssetURL": "https://static-cdn.jtvnw.net/twitch-quests-assets/REWARD/d70e4e75-7237-4730-9a10-b6016aaaa795.png",
+                        "isIosAvailable": false,
+                        "name": "Energising Bolt Flask",
+                        "ownerOrganization": {
+                            "id": "f751ba67-7c8b-4c41-b6df-bcea0914f3ad",
+                            "name": "Grinding Gear Games",
+                            "__typename": "Organization"
+                        },
+                        "distributionType": "DIRECT_ENTITLEMENT",
+                        "__typename": "DropBenefit"
                     },
-                    "__typename": "User"
+                    "entitlementLimit": 1,
+                    "__typename": "DropBenefitEdge"
                 }
-            }
+            ],
+            "endAt": "2024-12-14T07:59:59.996Z",
+            "name": "Early Access Bundle",
+            "preconditionDrops": null,
+            "requiredMinutesWatched": 180,
+            "startAt": "2024-12-06T19:00:00Z",
+            "__typename": "TimeBasedDrop"
         }
-    """  # noqa: E501
+    """
 
     # Django fields
     # "d5cdf372-502b-11ef-bafd-0a58a9feac02"
     twitch_id = models.TextField(primary_key=True, help_text="The Twitch ID of the drop.")
-    created_at = models.DateTimeField(auto_created=True, help_text="When the drop was first added to the database.")
+    created_at = models.DateTimeField(auto_now_add=True, help_text="When the drop was first added to the database.")
     modified_at = models.DateTimeField(auto_now=True, help_text="When the drop was last modified.")
 
     # Twitch fields
@@ -476,7 +429,7 @@ class Benefit(auto_prefetch.Model):
     # Django fields
     # "d5cdf372-502b-11ef-bafd-0a58a9feac02"
     twitch_id = models.TextField(primary_key=True)
-    created_at = models.DateTimeField(null=True, auto_created=True)
+    created_at = models.DateTimeField(auto_now_add=True)
     modified_at = models.DateTimeField(auto_now=True)
 
     # Twitch fields

core/models_utils.py

@@ -78,12 +78,13 @@ def get_value(data: dict, key: str) -> datetime | str | None:
     """
     data_key: Any | None = data.get(key)
     if not data_key:
-        logger.error("Key %s not found in %s", key, data)
+        logger.warning("Key %s not found in %s", key, data)
         return None
 
     # Dates are in the format "2024-08-12T05:59:59.999Z"
     dates: list[str] = ["endAt", "endsAt,", "startAt", "startsAt", "createdAt", "earnableUntil"]
     if key in dates:
+        logger.debug("Converting %s to datetime", data_key)
         return datetime.fromisoformat(data_key.replace("Z", "+00:00"))
 
     return data_key

core/views.py

@@ -10,7 +10,7 @@ from django.template.response import TemplateResponse
 from django.utils import timezone
 from django.views.decorators.http import require_http_methods
 
-from core.import_json import import_data_from_view
+from core.import_json import import_data
 from core.models import Benefit, DropCampaign, Game, TimeBasedDrop
 
 if TYPE_CHECKING:
@@ -134,7 +134,7 @@ def get_import(request: HttpRequest) -> HttpResponse:
         logger.info(data)
 
         # Import the data.
-        import_data_from_view(data)
+        import_data(data)
 
         return JsonResponse({"status": "success"}, status=200)
     except json.JSONDecodeError as e:

tests/models_test.py

@@ -0,0 +1,274 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from core.import_json import (
+    find_typename_in_json,
+    import_data,
+    import_drop_benefits,
+    import_drop_campaigns,
+    import_game_data,
+    import_owner_data,
+    import_time_based_drops,
+    type_names,
+)
+from core.models import Benefit, DropCampaign, Game, Owner, TimeBasedDrop
+
+
+def _validate_extraction(json: dict, typename: type_names, no_result_err_msg: str, id_err_msg: str) -> dict[str, Any]:
+    result: dict[str, Any] = find_typename_in_json(json, typename)[0]
+    assert result, no_result_err_msg
+    assert result.get("id"), id_err_msg
+
+    return result
+
+
+def test_find_typename_in_json() -> None:
+    """Test the find_typename_in_json function."""
+    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
+    json_file: dict = json.loads(json_file_raw)
+
+    result: list[dict[str, Any]] = find_typename_in_json(json_file, typename_to_find="DropCampaign")
+    assert len(result) == 20
+    assert result[0]["__typename"] == "DropCampaign"
+    assert result[0]["id"] == "5b5816c8-a533-11ef-9266-0a58a9feac02"
+
+
+@pytest.mark.django_db
+def test_import_game_data() -> None:
+    """Test the import_game_data function."""
+    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
+    json_file: dict = json.loads(json_file_raw)
+
+    game_json: dict[str, Any] = _validate_extraction(
+        json=json_file,
+        typename="Game",
+        no_result_err_msg="Game JSON not found",
+        id_err_msg="Game ID not found",
+    )
+    assert game_json.get("id") == "155409827", f"Game ID does not match expected value: {game_json.get('id')}"
+    assert game_json.get("slug") == "pokemon-trading-card-game-live", "Game slug does not match expected value"
+
+    assert_msg: str = f"Game display name does not match expected value: {game_json.get('displayName')}"
+    assert game_json.get("displayName") == "Pokémon Trading Card Game Live", assert_msg
+
+    assert_msg: str = f"Game URL does not match expected value: {game_json.get('gameUrl')}"
+    assert game_json.get("__typename") == "Game", assert_msg
+
+    owner_json: dict[str, Any] = _validate_extraction(
+        json=json_file,
+        typename="Organization",
+        no_result_err_msg="Owner JSON not found",
+        id_err_msg="Owner ID not found",
+    )
+
+    owner, created = Owner.objects.get_or_create(twitch_id=owner_json.get("id"))
+    assert_msg: str = f"Owner was not created: {owner=} != {owner_json.get('id')}. This means the old database was used instead of a new one."  # noqa: E501
+    assert created, assert_msg
+    assert owner
+
+    game: Game = import_game_data(drop_campaign=game_json, owner=owner)
+    assert game, f"Failed to import JSON data into Game model: {game_json=}"
+    assert game.org == owner, f"Owner was not set on the Game model: {game.org=} != {owner=}"
+    assert game.display_name == game_json.get("displayName"), "Game display name was not set on the Game model"
+
+    assert_msg: str = f"Game slug was not set on the Game model: {game.slug=} != {game_json.get('slug')}"
+    assert game.slug == game_json.get("slug"), assert_msg
+
+    assert_msg: str = f"Game ID was not set on the Game model: {game.twitch_id=} != {game_json.get('id')}"
+    assert game.twitch_id == game_json.get("id"), assert_msg
+
+    assert_msg: str = f"Game URL was not set on the Game model: {game.game_url} != https://www.twitch.tv/directory/category/{game.slug}"
+    assert game.game_url == f"https://www.twitch.tv/directory/category/{game.slug}", assert_msg
+
+    assert game.created_at, "Game created_at was not set on the Game model"
+    assert game.modified_at, "Game modified_at was not set on the Game model"
+
+
+@pytest.mark.django_db
+def test_import_owner_data() -> None:
+    """Test the import_owner_data function."""
+    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
+    json_file: dict = json.loads(json_file_raw)
+
+    owner_json: dict[str, Any] = _validate_extraction(
+        json=json_file,
+        typename="Organization",
+        no_result_err_msg="Owner JSON not found",
+        id_err_msg="Owner ID not found",
+    )
+
+    owner: Owner = import_owner_data(drop_campaign=owner_json)
+    assert owner, f"Failed to import JSON data into Owner model: {owner_json=}"
+
+    assert_msg: str = f"Owner ID was not set on the Owner model: {owner.twitch_id=} != {owner_json.get('id')}"
+    assert owner.twitch_id == owner_json.get("id"), assert_msg
+
+    assert_msg: str = f"Owner name was not set on the Owner model: {owner.name=} != {owner_json.get('name')}"
+    assert owner.name == owner_json.get("name"), assert_msg
+
+    assert owner.created_at, "Owner created_at was not set on the Owner model"
+    assert owner.modified_at, "Owner modified_at was not set on the Owner model"
+
+
+@pytest.mark.django_db
+def test_import_drop_benefits() -> None:
+    """Test the import_drop_benefits function."""
+    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
+    json_file: dict = json.loads(json_file_raw)
+
+    drop_campaign_json: dict[str, Any] = _validate_extraction(
+        json=json_file,
+        typename="DropCampaign",
+        no_result_err_msg="DropCampaign JSON not found",
+        id_err_msg="DropCampaign ID not found",
+    )
+    assert drop_campaign_json
+
+    drop_campaign: DropCampaign | None = import_drop_campaigns(drop_campaigns=drop_campaign_json)
+    assert drop_campaign, f"Failed to import JSON data into DropCampaign model: {drop_campaign_json=}"
+
+    assert find_typename_in_json(drop_campaign_json, "TimeBasedDrop"), "TimeBasedDrop JSON not found"
+    time_based_drop_json: dict[str, Any] = _validate_extraction(
+        json=drop_campaign_json,
+        typename="TimeBasedDrop",
+        no_result_err_msg="TimeBasedDrop JSON not found",
+        id_err_msg="TimeBasedDrop ID not found",
+    )
+    assert time_based_drop_json
+
+    time_based_drop: list[TimeBasedDrop] = import_time_based_drops(
+        drop_campaign_json=time_based_drop_json,
+        drop_campaign=drop_campaign,
+    )
+    assert time_based_drop, f"Failed to import JSON data into TimeBasedDrop model: {time_based_drop_json=}"
+
+    drop_benefit_json: dict[str, Any] = _validate_extraction(
+        json=drop_campaign_json,
+        typename="DropBenefit",
+        no_result_err_msg="DropBenefit JSON not found",
+        id_err_msg="DropBenefit ID not found",
+    )
+
+    drop_benefit: list[Benefit] = import_drop_benefits(drop_benefit_json, time_based_drop[0])
+    assert drop_benefit, f"Failed to import JSON data into DropBenefit model: {drop_benefit_json=}"
+    assert drop_benefit[0].twitch_id == drop_benefit_json.get("id"), "Benefit ID was not set on the Benefit model"
+
+    assert_msg: str = f"DropBenefit created_at was not set on the Benefit model: {drop_benefit[0].created_at=}"
+    assert drop_benefit[0].created_at, assert_msg
+
+    assert_msg = f"DropBenefit modified_at was not set on the Benefit model: {drop_benefit[0].modified_at=}"
+    assert drop_benefit[0].modified_at, assert_msg
+
+    assert drop_benefit[0].time_based_drop == time_based_drop[0], "TimeBasedDrop was not set on the Benefit model"
+
+
+@pytest.mark.django_db
+def test_import_time_based_drops() -> None:
+    """Test the import_time_based_drops function."""
+    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
+    json_file: dict = json.loads(json_file_raw)
+
+    drop_campaign_json: dict[str, Any] = _validate_extraction(
+        json=json_file,
+        typename="DropCampaign",
+        no_result_err_msg="DropCampaign JSON not found",
+        id_err_msg="DropCampaign ID not found",
+    )
+    assert drop_campaign_json
+
+    drop_campaign: DropCampaign | None = import_drop_campaigns(drop_campaigns=drop_campaign_json)
+    assert drop_campaign, f"Failed to import JSON data into DropCampaign model: {drop_campaign_json=}"
+
+    time_based_drop_json: dict[str, Any] = _validate_extraction(
+        json=drop_campaign_json,
+        typename="TimeBasedDrop",
+        no_result_err_msg="TimeBasedDrop JSON not found",
+        id_err_msg="TimeBasedDrop ID not found",
+    )
+    assert time_based_drop_json
+
+    time_based_drop: list[TimeBasedDrop] = import_time_based_drops(
+        drop_campaign_json=time_based_drop_json,
+        drop_campaign=drop_campaign,
+    )
+    assert time_based_drop, f"Failed to import JSON data into TimeBasedDrop model: {time_based_drop_json=}"
+
+    assert time_based_drop[0].twitch_id == time_based_drop_json.get(
+        "id",
+    ), "TimeBasedDrop ID was not set on the TimeBasedDrop model"
+
+    assert_msg: str = (
+        f"TimeBasedDrop created_at was not set on the TimeBasedDrop model: {time_based_drop[0].created_at=}"
+    )
+    assert time_based_drop[0].created_at, assert_msg
+
+    assert_msg = f"TimeBasedDrop modified_at was not set on the TimeBasedDrop model: {time_based_drop[0].modified_at=}"
+    assert time_based_drop[0].modified_at, assert_msg
+
+    assert time_based_drop[0].drop_campaign == drop_campaign, "DropCampaign was not set on the TimeBasedDrop model"
+
+
+@pytest.mark.django_db
+def test_import_drop_campaigns() -> None:
+    """Test the import_drop_campaigns function."""
+    json_file_raw: str = Path("tests/response.json").read_text(encoding="utf-8")
+    json_file: dict = json.loads(json_file_raw)
+
+    drop_campaign_json: dict[str, Any] = _validate_extraction(
+        json=json_file,
+        typename="DropCampaign",
+        no_result_err_msg="DropCampaign JSON not found",
+        id_err_msg="DropCampaign ID not found",
+    )
+    assert drop_campaign_json
+
+    drop_campaign: DropCampaign | None = import_drop_campaigns(drop_campaigns=drop_campaign_json)
+    assert drop_campaign, f"Failed to import JSON data into DropCampaign model: {drop_campaign_json=}"
+
+    assert drop_campaign.twitch_id == drop_campaign_json.get(
+        "id",
+    ), "DropCampaign ID was not set on the DropCampaign model"
+
+    assert_msg: str = f"DropCampaign created_at was not set on the DropCampaign model: {drop_campaign.created_at=}"
+    assert drop_campaign.created_at, assert_msg
+
+    assert_msg = f"DropCampaign modified_at was not set on the DropCampaign model: {drop_campaign.modified_at=}"
+    assert drop_campaign.modified_at, assert_msg
+
+
+@pytest.fixture
+def sample_data() -> dict[str, Any]:
+    """Sample data for testing import_data."""
+    return {
+        "__typename": "Root",
+        "data": [
+            {"__typename": "DropCampaign", "id": "campaign1", "name": "Campaign 1"},
+            {"__typename": "DropCampaign", "id": "campaign2", "name": "Campaign 2"},
+        ],
+    }
+
+
+@pytest.fixture
+def empty_data() -> dict[str, Any]:
+    """Empty data for testing import_data."""
+    return {"__typename": "Root", "data": []}
+
+
+@patch("core.import_json.import_drop_campaigns")
+def test_import_data(mock_import_drop_campaigns: MagicMock, sample_data: dict[str, Any]) -> None:
+    """Test the import_data function with valid data."""
+    import_data(sample_data)
+    assert mock_import_drop_campaigns.call_count == 2
+
+
+def test_import_data_no_campaigns(empty_data: dict[str, Any]) -> None:
+    """Test the import_data function with no drop campaigns."""
+    with patch("core.import_json.import_drop_campaigns") as mock_import_drop_campaigns:
+        import_data(empty_data)
+        mock_import_drop_campaigns.assert_not_called()