Resolve active season id from the season list

.test.env

@@ -90,9 +90,6 @@ ROLE__SEASON_SILVER=1099056313952125019
 ROLE__SEASON_BRONZE=1099056442281042022
 
 
-# Season ID, Updated at regular interval
-SEASON_ID=1
-
 #V4 Bearer Token
 HTB_API_KEY=CHANGE_ME
 

src/core/config.py

@@ -2,7 +2,7 @@
 import re
 from pathlib import Path
 
-from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -219,7 +219,6 @@ class Global(BaseSettings):
 
     ROOT: Path | None = None
     VERSION: str = "unknown"
-    SEASON_ID: int = Field(default=0, validation_alias=AliasChoices("SEASON_ID", "CURRENT_SEASON_ID"))
 
     @field_validator("guild_ids", "dev_guild_ids", mode="before")
     @classmethod

src/helpers/verification.py

@@ -1,5 +1,6 @@
 import logging
 import traceback
+from datetime import datetime, timezone
 from typing import Dict, List, Optional, Any, TypeVar
 
 import aiohttp
@@ -22,6 +23,44 @@
 
 logger = logging.getLogger(__name__)
 
+# (expires_at, season_id). expires_at is the active season's end_date.
+_active_season_cache: tuple[datetime, int] | None = None
+
+
+def clear_active_season_cache() -> None:
+    """Drop the cached active season id."""
+    global _active_season_cache
+    _active_season_cache = None
+
+
+def _parse_season_end(value: object) -> datetime | None:
+    if not isinstance(value, str) or not value:
+        return None
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
+def _cached_active_season_id() -> int | None:
+    if _active_season_cache is None:
+        return None
+    expires_at, season_id = _active_season_cache
+    if datetime.now(timezone.utc) >= expires_at:
+        return None
+    return season_id
+
+
+def _store_active_season_id(season_id: int, expires_at: datetime) -> None:
+    """Cache the season id until its end_date. A date already in the past is not cached."""
+    global _active_season_cache
+    if expires_at <= datetime.now(timezone.utc):
+        return
+    _active_season_cache = (expires_at, season_id)
+
 
 async def send_verification_instructions(
     ctx: ApplicationContext, member: Member
@@ -130,11 +169,40 @@ async def get_user_details(labs_id: int | str) -> dict:
     return profile
 
 
+async def get_active_season_id(session: aiohttp.ClientSession) -> int | None:
+    """Return the id of the currently active season from the season list."""
+    cached_season_id = _cached_active_season_id()
+    if cached_season_id is not None:
+        return cached_season_id
+
+    season_list_url = f"{settings.API_V4_URL}/season/list"
+    async with session.get(season_list_url) as r:
+        if r.status != 200:
+            logger.error(f"Non-OK HTTP status code returned from season list lookup: {r.status}.")
+            return None
+        response = await r.json()
+
+    for season in response.get("data") or []:
+        if season.get("active"):
+            season_id = season.get("id")
+            if isinstance(season_id, int):
+                expires_at = _parse_season_end(season.get("end_date"))
+                if expires_at is not None:
+                    _store_active_season_id(season_id, expires_at)
+                return season_id
+
+    logger.error("No active season found in season list.")
+    return None
+
+
 async def get_season_rank(htb_uid: int) -> str | None:
     """Get season rank from HTB."""
-    season_api_url = f"{settings.API_V4_URL}/user/achievement/season/{htb_uid}/{settings.SEASON_ID}"
-
     async with get_labs_session() as session:
+        season_id = await get_active_season_id(session)
+        if season_id is None:
+            return None
+
+        season_api_url = f"{settings.API_V4_URL}/user/achievement/season/{htb_uid}/{season_id}"
         async with session.get(season_api_url) as r:
             if r.status == 200:
                 response = await r.json()

tests/src/core/test_config.py

@@ -100,10 +100,6 @@ def test_dynamic_roles_are_optional(self):
         self.assertTrue(hasattr(settings.roles, "VIP"))
         self.assertTrue(hasattr(settings.roles, "BOX_CREATOR"))
 
-    def test_season_id_loads_from_nested_env_config(self):
-        """Test that SEASON_ID is loaded under the new nested env contract."""
-        self.assertEqual(settings.SEASON_ID, 1)
-
     def test_unknown_settings_are_ignored(self):
         """Test that leftover role and top-level env values do not stop settings from loading."""
         config = self.minimal_settings(
@@ -116,13 +112,6 @@ def test_unknown_settings_are_ignored(self):
         self.assertFalse(hasattr(config, "SLACK_WEBHOOK"))
         self.assertFalse(hasattr(config.roles, "RANK_FIVE"))
 
-    def test_current_season_id_fills_season_id(self):
-        """Test that the legacy CURRENT_SEASON_ID env var still sets the season."""
-        with patch.dict(os.environ, {"CURRENT_SEASON_ID": "4"}):
-            os.environ.pop("SEASON_ID", None)
-            config = self.minimal_settings()
-        self.assertEqual(config.SEASON_ID, 4)
-
     def test_assemble_db_connection_selects_driver(self):
         """Test that the caller chooses the async or sync MariaDB driver."""
         config = self.minimal_settings()

tests/src/helpers/test_verification.py

@@ -6,18 +6,51 @@
 import pytest
 
 from src.core import settings
-from src.helpers.verification import get_season_rank, get_user_details, process_labs_identification
+from src.helpers.verification import (
+    clear_active_season_cache,
+    get_season_rank,
+    get_user_details,
+    process_labs_identification,
+)
 from tests.helpers import MockRoleManager
 
+_ACTIVE_SEASON_ID = 21
+
+
+def _season_list_payload(
+    active_id: int = _ACTIVE_SEASON_ID,
+    end_date: str = "2099-01-01T00:00:00.000000Z",
+) -> dict:
+    return {
+        "data": [
+            {"id": 15, "name": "Season 11", "active": False, "state": "ended"},
+            {
+                "id": active_id,
+                "name": "Season 12",
+                "active": True,
+                "state": "active",
+                "end_date": end_date,
+            },
+        ]
+    }
+
 
 class TestGetSeasonRank(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        clear_active_season_cache()
+
     @pytest.mark.asyncio
     async def test_get_season_rank_success(self):
         htb_uid = 12345
 
         with aioresponses.aioresponses() as m:
             m.get(
-                f"{settings.API_V4_URL}/user/achievement/season/{htb_uid}/{settings.SEASON_ID}",
+                f"{settings.API_V4_URL}/season/list",
+                status=200,
+                payload=_season_list_payload(),
+            )
+            m.get(
+                f"{settings.API_V4_URL}/user/achievement/season/{htb_uid}/{_ACTIVE_SEASON_ID}",
                 status=200,
                 payload={"data": {"season": {"tier": "Gold"}}},
             )
@@ -31,14 +64,82 @@ async def test_get_season_rank_empty_data(self):
 
         with aioresponses.aioresponses() as m:
             m.get(
-                f"{settings.API_V4_URL}/user/achievement/season/{htb_uid}/{settings.SEASON_ID}",
+                f"{settings.API_V4_URL}/season/list",
+                status=200,
+                payload=_season_list_payload(),
+            )
+            m.get(
+                f"{settings.API_V4_URL}/user/achievement/season/{htb_uid}/{_ACTIVE_SEASON_ID}",
                 status=200,
                 payload={"data": None},
             )
 
             result = await get_season_rank(htb_uid)
             self.assertIsNone(result)
 
+    @pytest.mark.asyncio
+    async def test_get_season_rank_without_active_season(self):
+        htb_uid = 12345
+
+        with aioresponses.aioresponses() as m:
+            m.get(
+                f"{settings.API_V4_URL}/season/list",
+                status=200,
+                payload={"data": [{"id": 15, "name": "Season 11", "active": False, "state": "ended"}]},
+            )
+
+            result = await get_season_rank(htb_uid)
+            self.assertIsNone(result)
+
+    @pytest.mark.asyncio
+    async def test_get_season_rank_reuses_cached_season_id(self):
+        first_uid = 12345
+        second_uid = 67890
+
+        with aioresponses.aioresponses() as m:
+            m.get(
+                f"{settings.API_V4_URL}/season/list",
+                status=200,
+                payload=_season_list_payload(),
+            )
+            m.get(
+                f"{settings.API_V4_URL}/user/achievement/season/{first_uid}/{_ACTIVE_SEASON_ID}",
+                status=200,
+                payload={"data": {"season": {"tier": "Gold"}}},
+            )
+            m.get(
+                f"{settings.API_V4_URL}/user/achievement/season/{second_uid}/{_ACTIVE_SEASON_ID}",
+                status=200,
+                payload={"data": {"season": {"tier": "Silver"}}},
+            )
+
+            self.assertEqual(await get_season_rank(first_uid), "Gold")
+            self.assertEqual(await get_season_rank(second_uid), "Silver")
+
+    @pytest.mark.asyncio
+    async def test_expired_season_is_not_cached(self):
+        first_uid = 12345
+        second_uid = 67890
+        season_list_url = f"{settings.API_V4_URL}/season/list"
+        expired = _season_list_payload(end_date="2020-01-01T00:00:00.000000Z")
+
+        with aioresponses.aioresponses() as m:
+            m.get(season_list_url, status=200, payload=expired)
+            m.get(season_list_url, status=200, payload=expired)
+            m.get(
+                f"{settings.API_V4_URL}/user/achievement/season/{first_uid}/{_ACTIVE_SEASON_ID}",
+                status=200,
+                payload={"data": {"season": {"tier": "Gold"}}},
+            )
+            m.get(
+                f"{settings.API_V4_URL}/user/achievement/season/{second_uid}/{_ACTIVE_SEASON_ID}",
+                status=200,
+                payload={"data": {"season": {"tier": "Silver"}}},
+            )
+
+            self.assertEqual(await get_season_rank(first_uid), "Gold")
+            self.assertEqual(await get_season_rank(second_uid), "Silver")
+
 
 class TestGetUserDetails(unittest.IsolatedAsyncioTestCase):
     @pytest.mark.asyncio