ctfy.sdk.admin_resources.release_plan

  1"""``client.admin.release_plan`` — waves, per-challenge rules and hints of
  2a competition: drafts, publishing and releasing by hand (#2093 phase 2, #2101).
  3
  4A draft changes nothing a player sees; :meth:`AdminReleasePlanResource.
  5publish` and :meth:`~AdminReleasePlanResource.release_now` do, and a release
  6is never taken back.
  7
  8⚠️ Every request names its method and path as literals at the call site —
  9``test_sdk_openapi_parity`` reads them there, so a shared helper taking the
 10path as an argument would hide every route it serves.
 11"""
 12
 13from __future__ import annotations
 14
 15from collections.abc import Sequence
 16from datetime import datetime
 17from typing import Any
 18
 19import httpx
 20
 21from ctfy.core.release_conditions import PacePreset, ReleaseRule
 22from ctfy.core.state.models import HintTierDraft
 23from ctfy.sdk._helpers import _raise_for_status
 24from ctfy.sdk.base import BaseHttpClient
 25from ctfy.server.models import HintAttachmentUpload, HintEditorView, ReleasePlanView
 26
 27
 28def _plan(resp: httpx.Response) -> ReleasePlanView:
 29    _raise_for_status(resp)
 30    return ReleasePlanView.model_validate(resp.json())
 31
 32
 33def _editor(resp: httpx.Response) -> HintEditorView:
 34    _raise_for_status(resp)
 35    return HintEditorView.model_validate(resp.json())
 36
 37
 38class AdminReleasePlanResource:
 39    """For a competition's organisers (global admins and its own admins)."""
 40
 41    def __init__(self, http: BaseHttpClient) -> None:
 42        self._http = http
 43
 44    # -- waves and rules -------------------------------------------------------
 45
 46    def get(self, competition_id: str, *, pace: PacePreset = "normal") -> ReleasePlanView:
 47        """Every wave and challenge with its rule, draft, standing, forecast
 48        (under the assumed ``pace``) and the findings."""
 49        return _plan(
 50            self._http.request(
 51                "GET", f"/admin/competitions/{competition_id}/release-plan", params={"pace": pace}
 52            )
 53        )
 54
 55    def save_draft(self, competition_id: str, subject: str, rule: ReleaseRule) -> ReleasePlanView:
 56        """Save a draft for ``wave:<name>`` or ``challenge:<id>``."""
 57        return _plan(
 58            self._http.request(
 59                "PUT",
 60                f"/admin/competitions/{competition_id}/release-plan/drafts",
 61                json={"subject": subject, "rule": rule.model_dump(mode="json")},
 62            )
 63        )
 64
 65    def discard_draft(self, competition_id: str, subject: str) -> ReleasePlanView:
 66        return _plan(
 67            self._http.request(
 68                "DELETE",
 69                f"/admin/competitions/{competition_id}/release-plan/drafts",
 70                params={"subject": subject},
 71            )
 72        )
 73
 74    def publish(self, competition_id: str, subject: str) -> ReleasePlanView:
 75        """Publish ``subject``'s draft (refused with findings when invalid)."""
 76        return _plan(
 77            self._http.request(
 78                "POST",
 79                f"/admin/competitions/{competition_id}/release-plan/publish",
 80                json={"subject": subject},
 81            )
 82        )
 83
 84    def release_now(self, competition_id: str, subject: str) -> ReleasePlanView:
 85        """Release a wave or a challenge now, whatever its rule says."""
 86        return _plan(
 87            self._http.request(
 88                "POST",
 89                f"/admin/competitions/{competition_id}/release-plan/release",
 90                json={"subject": subject},
 91            )
 92        )
 93
 94    def follow_wave(self, competition_id: str, challenge_id: str) -> ReleasePlanView:
 95        """Drop a challenge's own rule; it follows its wave again."""
 96        return _plan(
 97            self._http.request(
 98                "DELETE",
 99                f"/admin/competitions/{competition_id}/release-plan/rules",
100                params={"subject": f"challenge:{challenge_id}"},
101            )
102        )
103
104    def create_wave(
105        self, competition_id: str, name: str, challenge_ids: Sequence[str] = ()
106    ) -> ReleasePlanView:
107        return _plan(
108            self._http.request(
109                "POST",
110                f"/admin/competitions/{competition_id}/release-plan/waves",
111                json={"name": name, "challenge_ids": list(challenge_ids)},
112            )
113        )
114
115    def delete_wave(self, competition_id: str, name: str) -> ReleasePlanView:
116        return _plan(
117            self._http.request(
118                "DELETE",
119                f"/admin/competitions/{competition_id}/release-plan/waves",
120                params={"name": name},
121            )
122        )
123
124    # -- hints -----------------------------------------------------------------
125
126    def hints(
127        self,
128        competition_id: str,
129        challenge_id: str,
130        *,
131        solved_at: datetime | None = None,
132        solved_teams: int = 1,
133    ) -> HintEditorView:
134        """The tiers, the draft, the findings and the two scenarios."""
135        params: dict[str, Any] = {"solved_teams": solved_teams}
136        if solved_at is not None:
137            params["solved_at"] = solved_at.isoformat()
138        return _editor(
139            self._http.request(
140                "GET",
141                f"/admin/competitions/{competition_id}/challenges/{challenge_id}/hints",
142                params=params,
143            )
144        )
145
146    def save_hint_draft(
147        self, competition_id: str, challenge_id: str, tiers: Sequence[HintTierDraft]
148    ) -> HintEditorView:
149        return _editor(
150            self._http.request(
151                "PUT",
152                f"/admin/competitions/{competition_id}/challenges/{challenge_id}/hints/draft",
153                json={"tiers": [t.model_dump(mode="json") for t in tiers]},
154            )
155        )
156
157    def discard_hint_draft(self, competition_id: str, challenge_id: str) -> HintEditorView:
158        return _editor(
159            self._http.request(
160                "DELETE",
161                f"/admin/competitions/{competition_id}/challenges/{challenge_id}/hints/draft",
162            )
163        )
164
165    def publish_hints(self, competition_id: str, challenge_id: str) -> HintEditorView:
166        return _editor(
167            self._http.request(
168                "POST",
169                f"/admin/competitions/{competition_id}/challenges/{challenge_id}/hints/publish",
170            )
171        )
172
173    def release_hint(self, competition_id: str, challenge_id: str, hint_id: str) -> HintEditorView:
174        return _editor(
175            self._http.request(
176                "POST",
177                f"/admin/competitions/{competition_id}/challenges/{challenge_id}"
178                f"/hints/{hint_id}/release",
179            )
180        )
181
182    def upload_hint_attachment(
183        self, competition_id: str, filename: str, content: bytes
184    ) -> HintAttachmentUpload:
185        """Store a file for a tier; put the answer's fields into a draft tier."""
186        resp = self._http.request(
187            "POST",
188            f"/admin/competitions/{competition_id}/hint-attachments",
189            files={"file": (filename, content, "application/octet-stream")},
190        )
191        _raise_for_status(resp)
192        return HintAttachmentUpload.model_validate(resp.json())