diff --git a/.test.env b/.test.env index 7f34049..1ff5fa8 100644 --- a/.test.env +++ b/.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 diff --git a/src/core/config.py b/src/core/config.py index e98b1f0..9c55fc1 100644 --- a/src/core/config.py +++ b/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 diff --git a/src/helpers/verification.py b/src/helpers/verification.py index 6a730cf..40a2bf2 100644 --- a/src/helpers/verification.py +++ b/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() diff --git a/tests/src/core/test_config.py b/tests/src/core/test_config.py index f30f992..a732258 100644 --- a/tests/src/core/test_config.py +++ b/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() diff --git a/tests/src/helpers/test_verification.py b/tests/src/helpers/test_verification.py index 479d704..f4fa932 100644 --- a/tests/src/helpers/test_verification.py +++ b/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,7 +64,12 @@ 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}, ) @@ -39,6 +77,69 @@ async def test_get_season_rank_empty_data(self): 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