From 6decc719fcc130c063346448fe582556dfea0459 Mon Sep 17 00:00:00 2001 From: Camiel van Schoonhoven Date: Mon, 17 Aug 2026 17:21:07 -0700 Subject: [PATCH] feat: Add Banners Table and API --- cloud_pipelines_backend/api_router.py | 84 +++++ cloud_pipelines_backend/api_server_sql.py | 320 +++++++++++++++++ cloud_pipelines_backend/backend_types_sql.py | 59 +++ tests/test_banners_api.py | 357 +++++++++++++++++++ 4 files changed, 820 insertions(+) create mode 100644 tests/test_banners_api.py diff --git a/cloud_pipelines_backend/api_router.py b/cloud_pipelines_backend/api_router.py index 0e8fa41..b4cc7f9 100644 --- a/cloud_pipelines_backend/api_router.py +++ b/cloud_pipelines_backend/api_router.py @@ -588,6 +588,90 @@ def get_current_user( ) ) + ### Banner routes + + banner_service = api_server_sql.BannersApiService_Sql() + + # `default_config` is not used for the banner routes because it excludes the + # fields that are null, which the banner responses need to include. + + @router.get("/api/banners/active", tags=["banners"]) + def get_active_banners( + session: SessionDep, + response: fastapi.Response, + ) -> api_server_sql.ListBannersResponse: + # A banner can start or expire at any moment, so responses must not be cached. + response.headers["Cache-Control"] = "no-store" + return banner_service.list_active(session=session) + + admin_banner_write_dependencies = [ + ensure_admin_user_dependency, + fastapi.Depends(check_not_readonly), + ] + + @router.get( + "/api/admin/banners", + tags=["banners"], + dependencies=[ensure_admin_user_dependency], + ) + def admin_list_banners( + session: SessionDep, + include_deleted: bool = False, + ) -> api_server_sql.ListAdminBannersResponse: + return banner_service.list_all(session=session, include_deleted=include_deleted) + + @router.post( + "/api/admin/banners", + tags=["banners"], + dependencies=admin_banner_write_dependencies, + ) + def admin_create_banner( + session: SessionDep, + banner: api_server_sql.CreateBannerRequest, + user_name: typing.Annotated[str | None, get_user_name_dependency], + ) -> api_server_sql.AdminBannerResponse: + return banner_service.create( + session=session, banner=banner, user_name=user_name + ) + + @router.get( + "/api/admin/banners/{id}", + tags=["banners"], + dependencies=[ensure_admin_user_dependency], + ) + def admin_get_banner( + session: SessionDep, + id: backend_types_sql.IdType, + ) -> api_server_sql.AdminBannerResponse: + return banner_service.get(session=session, id=id) + + @router.patch( + "/api/admin/banners/{id}", + tags=["banners"], + dependencies=admin_banner_write_dependencies, + ) + def admin_update_banner( + session: SessionDep, + id: backend_types_sql.IdType, + banner: api_server_sql.UpdateBannerRequest, + user_name: typing.Annotated[str | None, get_user_name_dependency], + ) -> api_server_sql.AdminBannerResponse: + return banner_service.update( + session=session, id=id, banner=banner, user_name=user_name + ) + + @router.delete( + "/api/admin/banners/{id}", + tags=["banners"], + dependencies=admin_banner_write_dependencies, + ) + def admin_delete_banner( + session: SessionDep, + id: backend_types_sql.IdType, + user_name: typing.Annotated[str | None, get_user_name_dependency], + ) -> api_server_sql.AdminBannerResponse: + return banner_service.delete(session=session, id=id, user_name=user_name) + ### Admin routes @router.put( diff --git a/cloud_pipelines_backend/api_server_sql.py b/cloud_pipelines_backend/api_server_sql.py index cf71a16..a396fb9 100644 --- a/cloud_pipelines_backend/api_server_sql.py +++ b/cloud_pipelines_backend/api_server_sql.py @@ -2,6 +2,7 @@ import datetime import logging import typing +import urllib.parse from typing import Any, Final, Optional import sqlalchemy as sql @@ -1377,6 +1378,325 @@ def delete_settings( # endregion +# ==== BannersApiService + +MAX_BANNER_TITLE_LENGTH = 120 +MAX_BANNER_BODY_LENGTH = 2000 +MAX_BANNER_URL_TEXT_LENGTH = 80 +# Must not exceed the length of the `Banner.url` column. +MAX_BANNER_URL_LENGTH = 2048 + + +def _convert_datetime_to_utc( + value: datetime.datetime | None, +) -> datetime.datetime | None: + """Converts a client-provided datetime to UTC (like all datetimes in the DB). + + The DB stores "naive UTC" datetimes (see `bts.UtcDateTime`), so a datetime with + a non-UTC offset would otherwise be stored as if its wall clock time was UTC. + """ + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=datetime.timezone.utc) + return value.astimezone(datetime.timezone.utc) + + +def _validate_banner_title(title: str) -> str: + title = (title or "").strip() + if not title: + raise errors.ApiValidationError("Banner title must not be empty.") + if len(title) > MAX_BANNER_TITLE_LENGTH: + raise errors.ApiValidationError( + f"Banner title must be at most {MAX_BANNER_TITLE_LENGTH} characters, but got {len(title)}." + ) + return title + + +def _validate_banner_body(body: str) -> str: + body = (body or "").strip() + if not body: + raise errors.ApiValidationError("Banner body must not be empty.") + if len(body) > MAX_BANNER_BODY_LENGTH: + raise errors.ApiValidationError( + f"Banner body must be at most {MAX_BANNER_BODY_LENGTH} characters, but got {len(body)}." + ) + return body + + +def _validate_banner_url(url: str | None) -> str | None: + if url is None: + return None + url = url.strip() + if not url: + return None + if len(url) > MAX_BANNER_URL_LENGTH: + raise errors.ApiValidationError( + f"Banner URL must be at most {MAX_BANNER_URL_LENGTH} characters, but got {len(url)}." + ) + parsed_url = urllib.parse.urlparse(url) + if parsed_url.scheme not in ("http", "https") or not parsed_url.netloc: + raise errors.ApiValidationError( + f"Banner URL must be an absolute http or https URL, but got {url!r}." + ) + return url + + +def _validate_banner_url_text(url_text: str | None) -> str | None: + if url_text is None: + return None + url_text = url_text.strip() + if not url_text: + return None + if len(url_text) > MAX_BANNER_URL_TEXT_LENGTH: + raise errors.ApiValidationError( + f"Banner URL text must be at most {MAX_BANNER_URL_TEXT_LENGTH} characters, but got {len(url_text)}." + ) + return url_text + + +def _validate_banner_cross_field_constraints(banner: bts.Banner) -> None: + if banner.url_text and not banner.url: + raise errors.ApiValidationError( + "Banner URL text is only meaningful together with a banner URL." + ) + # The values loaded from the DB have no timezone, while the values coming from + # a request are timezone-aware, so both are normalized before being compared. + starts_at = _convert_datetime_to_utc(banner.starts_at) + ends_at = _convert_datetime_to_utc(banner.ends_at) + if starts_at is not None and ends_at is not None and ends_at <= starts_at: + raise errors.ApiValidationError( + f"Banner ends_at ({ends_at}) must be after starts_at ({starts_at})." + ) + + +def _get_public_banner_fields(banner: bts.Banner) -> dict[str, Any]: + return dict( + id=banner.id, + title=banner.title, + body=banner.body, + variant=banner.variant, + url=banner.url, + url_text=banner.url_text, + starts_at=banner.starts_at, + ends_at=banner.ends_at, + is_dismissible=banner.is_dismissible, + created_at=banner.created_at, + updated_at=banner.updated_at, + ) + + +@dataclasses.dataclass(kw_only=True) +class BannerResponse: + """A banner as exposed to all users.""" + + id: bts.IdType + title: str + body: str + variant: bts.BannerVariant + url: str | None = None + url_text: str | None = None + starts_at: datetime.datetime | None = None + ends_at: datetime.datetime | None = None + is_dismissible: bool = False + created_at: datetime.datetime | None = None + updated_at: datetime.datetime | None = None + + @staticmethod + def from_db(banner: bts.Banner) -> "BannerResponse": + return BannerResponse(**_get_public_banner_fields(banner)) + + +@dataclasses.dataclass(kw_only=True) +class AdminBannerResponse(BannerResponse): + """A banner including the fields that are only exposed to admins.""" + + is_enabled: bool = True + created_by: str | None = None + updated_by: str | None = None + deleted_at: datetime.datetime | None = None + + @staticmethod + def from_db(banner: bts.Banner) -> "AdminBannerResponse": + return AdminBannerResponse( + **_get_public_banner_fields(banner), + is_enabled=banner.is_enabled, + created_by=banner.created_by, + updated_by=banner.updated_by, + deleted_at=banner.deleted_at, + ) + + +@dataclasses.dataclass(kw_only=True) +class ListBannersResponse: + banners: list[BannerResponse] + + +@dataclasses.dataclass(kw_only=True) +class ListAdminBannersResponse: + banners: list[AdminBannerResponse] + + +@dataclasses.dataclass(kw_only=True) +class CreateBannerRequest: + title: str + body: str + variant: bts.BannerVariant + url: str | None = None + url_text: str | None = None + starts_at: datetime.datetime | None = None + ends_at: datetime.datetime | None = None + is_enabled: bool = True + is_dismissible: bool = False + + +@dataclasses.dataclass(kw_only=True) +class UpdateBannerRequest: + """A partial banner update. The fields that are not set (None) are not changed.""" + + title: str | None = None + body: str | None = None + variant: bts.BannerVariant | None = None + url: str | None = None + url_text: str | None = None + starts_at: datetime.datetime | None = None + ends_at: datetime.datetime | None = None + is_enabled: bool | None = None + is_dismissible: bool | None = None + + +class BannersApiService_Sql: + + def list_active(self, session: orm.Session) -> ListBannersResponse: + current_time = _get_current_time() + query = ( + sql.select(bts.Banner) + .where( + bts.Banner.deleted_at.is_(None), + bts.Banner.is_enabled == True, + sql.or_( + bts.Banner.starts_at.is_(None), + bts.Banner.starts_at <= current_time, + ), + sql.or_( + bts.Banner.ends_at.is_(None), + bts.Banner.ends_at > current_time, + ), + ) + .order_by( + # `starts_at DESC NULLS LAST`. The NULLS LAST modifier is not + # portable (MySQL does not support it), so the banners without + # `starts_at` are sorted last explicitly (False sorts before True). + bts.Banner.starts_at.is_(None), + bts.Banner.starts_at.desc(), + bts.Banner.created_at.desc(), + ) + ) + banners = session.scalars(query).all() + return ListBannersResponse( + banners=[BannerResponse.from_db(banner) for banner in banners] + ) + + def list_all( + self, session: orm.Session, include_deleted: bool = False + ) -> ListAdminBannersResponse: + query = sql.select(bts.Banner).order_by(bts.Banner.created_at.desc()) + if not include_deleted: + query = query.where(bts.Banner.deleted_at.is_(None)) + banners = session.scalars(query).all() + return ListAdminBannersResponse( + banners=[AdminBannerResponse.from_db(banner) for banner in banners] + ) + + def get(self, session: orm.Session, id: bts.IdType) -> AdminBannerResponse: + banner_row = session.get(bts.Banner, id) + if not banner_row: + raise errors.ItemNotFoundError(f"Banner with {id=} does not exist.") + return AdminBannerResponse.from_db(banner_row) + + def create( + self, + session: orm.Session, + banner: CreateBannerRequest, + user_name: str | None = None, + ) -> AdminBannerResponse: + current_time = _get_current_time() + banner_row = bts.Banner( + title=_validate_banner_title(banner.title), + body=_validate_banner_body(banner.body), + variant=banner.variant, + url=_validate_banner_url(banner.url), + url_text=_validate_banner_url_text(banner.url_text), + starts_at=_convert_datetime_to_utc(banner.starts_at), + ends_at=_convert_datetime_to_utc(banner.ends_at), + is_enabled=banner.is_enabled, + is_dismissible=banner.is_dismissible, + created_by=user_name, + updated_by=user_name, + created_at=current_time, + updated_at=current_time, + ) + _validate_banner_cross_field_constraints(banner_row) + session.add(banner_row) + session.commit() + session.refresh(banner_row) + return AdminBannerResponse.from_db(banner_row) + + def update( + self, + session: orm.Session, + id: bts.IdType, + banner: UpdateBannerRequest, + user_name: str | None = None, + ) -> AdminBannerResponse: + banner_row = session.get(bts.Banner, id) + if not banner_row: + raise errors.ItemNotFoundError(f"Banner with {id=} does not exist.") + if banner.title is not None: + banner_row.title = _validate_banner_title(banner.title) + if banner.body is not None: + banner_row.body = _validate_banner_body(banner.body) + if banner.variant is not None: + banner_row.variant = banner.variant + if banner.url is not None: + banner_row.url = _validate_banner_url(banner.url) + if banner.url_text is not None: + banner_row.url_text = _validate_banner_url_text(banner.url_text) + if banner.starts_at is not None: + banner_row.starts_at = _convert_datetime_to_utc(banner.starts_at) + if banner.ends_at is not None: + banner_row.ends_at = _convert_datetime_to_utc(banner.ends_at) + if banner.is_enabled is not None: + banner_row.is_enabled = banner.is_enabled + if banner.is_dismissible is not None: + banner_row.is_dismissible = banner.is_dismissible + _validate_banner_cross_field_constraints(banner_row) + banner_row.updated_by = user_name + banner_row.updated_at = _get_current_time() + session.commit() + session.refresh(banner_row) + return AdminBannerResponse.from_db(banner_row) + + def delete( + self, + session: orm.Session, + id: bts.IdType, + user_name: str | None = None, + ) -> AdminBannerResponse: + """Soft-deletes the banner. Banners are never removed from the DB.""" + banner_row = session.get(bts.Banner, id) + if not banner_row: + raise errors.ItemNotFoundError(f"Banner with {id=} does not exist.") + if banner_row.deleted_at is None: + current_time = _get_current_time() + banner_row.deleted_at = current_time + banner_row.updated_by = user_name + banner_row.updated_at = current_time + session.commit() + session.refresh(banner_row) + return AdminBannerResponse.from_db(banner_row) + + # ============ # Idea for how to add deep nested graph: diff --git a/cloud_pipelines_backend/backend_types_sql.py b/cloud_pipelines_backend/backend_types_sql.py index e061739..07430b6 100644 --- a/cloud_pipelines_backend/backend_types_sql.py +++ b/cloud_pipelines_backend/backend_types_sql.py @@ -502,6 +502,65 @@ class ContainerExecution(_TableBase): ) +class BannerVariant(str, enum.Enum): + INFO = "info" + WARNING = "warning" + SUCCESS = "success" + ERROR = "error" + + +class Banner(_TableBase): + """Site-wide announcement banner shown to all users. + + Banners are never hard-deleted: `deleted_at` marks a banner as removed. + """ + + __tablename__ = "banner" + id: orm.Mapped[IdType] = orm.mapped_column( + primary_key=True, init=False, insert_default=generate_unique_id + ) + title: orm.Mapped[str] + # `str` is mapped to VARCHAR(255) by default, which is too short for the body. + body: orm.Mapped[str] = orm.mapped_column(sql.Text()) + variant: orm.Mapped[BannerVariant] = orm.mapped_column( + # `values_callable` makes the DB store the enum values ("warning") + # instead of the default enum member names ("WARNING"). + sql.Enum( + BannerVariant, + values_callable=lambda enum_class: [member.value for member in enum_class], + ) + ) + # `mapped_column()` (despite having no arguments) is needed so that the columns + # can be referenced in `__table_args__` below. + created_at: orm.Mapped[datetime.datetime] = orm.mapped_column() + updated_at: orm.Mapped[datetime.datetime] = orm.mapped_column() + url: orm.Mapped[str | None] = orm.mapped_column(sql.String(2048), default=None) + url_text: orm.Mapped[str | None] = orm.mapped_column(default=None) + starts_at: orm.Mapped[datetime.datetime | None] = orm.mapped_column(default=None) + ends_at: orm.Mapped[datetime.datetime | None] = orm.mapped_column(default=None) + is_enabled: orm.Mapped[bool] = orm.mapped_column(default=True) + is_dismissible: orm.Mapped[bool] = orm.mapped_column(default=False) + created_by: orm.Mapped[str | None] = orm.mapped_column(default=None) + updated_by: orm.Mapped[str | None] = orm.mapped_column(default=None) + deleted_at: orm.Mapped[datetime.datetime | None] = orm.mapped_column(default=None) + + __table_args__ = ( + # Used by the "active banners" query. + sql.Index( + "ix_banner_is_enabled_deleted_at_starts_at_ends_at", + is_enabled, + deleted_at, + starts_at, + ends_at, + ), + # Used by the admin banner list. + sql.Index( + "ix_banner_created_at_desc", + created_at.desc(), + ), + ) + + class PipelineRunAnnotation(_TableBase): __tablename__ = "pipeline_run_annotation" _IX_ANNOTATION_RUN_ID_KEY_VALUE: Final[str] = ( diff --git a/tests/test_banners_api.py b/tests/test_banners_api.py new file mode 100644 index 0000000..b368048 --- /dev/null +++ b/tests/test_banners_api.py @@ -0,0 +1,357 @@ +import datetime + +import fastapi +from fastapi import testclient +import pytest + +from cloud_pipelines_backend import api_router +from cloud_pipelines_backend import database_ops + +ADMIN_USER_NAME = "admin user" +NON_ADMIN_USER_NAME = "regular user" + +ACTIVE_BANNERS_URL = "/api/banners/active" +ADMIN_BANNERS_URL = "/api/admin/banners" + + +def _make_user_details(name: str, *, is_admin: bool) -> api_router.UserDetails: + return api_router.UserDetails( + name=name, + permissions=api_router.Permissions(read=True, write=True, admin=is_admin), + ) + + +class _TestApi: + """A test API client that can switch between an admin and a non-admin user.""" + + def __init__(self, client: testclient.TestClient, current_user_details: dict): + self.client = client + self._current_user_details = current_user_details + + def become_non_admin(self): + self._current_user_details["user_details"] = _make_user_details( + NON_ADMIN_USER_NAME, is_admin=False + ) + + +@pytest.fixture(name="api") +def api_fixture(): + db_engine = database_ops.create_db_engine(database_uri="sqlite://") + app = fastapi.FastAPI() + current_user_details = { + "user_details": _make_user_details(ADMIN_USER_NAME, is_admin=True) + } + + def get_user_details(): + return current_user_details["user_details"] + + api_router.setup_routes( + app=app, + db_engine=db_engine, + user_details_getter=get_user_details, + ) + # The context manager triggers the lifespan event that creates the DB tables. + with testclient.TestClient(app) as client: + yield _TestApi(client=client, current_user_details=current_user_details) + + +def _parse_datetime(value: str) -> datetime.datetime: + # `datetime.fromisoformat` only supports the "Z" suffix since Python 3.11. + return datetime.datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def _get_current_time() -> datetime.datetime: + return datetime.datetime.now(tz=datetime.timezone.utc) + + +def _make_banner_request(**overrides) -> dict: + banner = { + "title": "Scheduled maintenance", + "body": "The service will be unavailable for 10 minutes.", + "variant": "warning", + } + banner.update(overrides) + return banner + + +def _create_banner(api: _TestApi, **overrides) -> dict: + response = api.client.post( + ADMIN_BANNERS_URL, json=_make_banner_request(**overrides) + ) + assert response.status_code == 200, response.text + return response.json() + + +def _get_active_banners(api: _TestApi) -> list[dict]: + response = api.client.get(ACTIVE_BANNERS_URL) + assert response.status_code == 200, response.text + return response.json()["banners"] + + +def test_active_banners_are_empty_by_default(api: _TestApi): + response = api.client.get(ACTIVE_BANNERS_URL) + assert response.status_code == 200, response.text + assert response.json() == {"banners": []} + assert response.headers["Cache-Control"] == "no-store" + + +def test_admin_can_create_banner(api: _TestApi): + starts_at = _get_current_time() - datetime.timedelta(hours=1) + banner = _create_banner( + api, + url="https://example.com/status", + url_text="View details", + starts_at=starts_at.isoformat(), + is_dismissible=True, + ) + assert banner["id"] + assert banner["title"] == "Scheduled maintenance" + assert banner["body"] == "The service will be unavailable for 10 minutes." + assert banner["variant"] == "warning" + assert banner["url"] == "https://example.com/status" + assert banner["url_text"] == "View details" + assert _parse_datetime(banner["starts_at"]) == starts_at + assert banner["ends_at"] is None + assert banner["is_enabled"] == True + assert banner["is_dismissible"] == True + assert banner["deleted_at"] is None + assert banner["created_by"] == ADMIN_USER_NAME + assert banner["updated_by"] == ADMIN_USER_NAME + assert _parse_datetime(banner["created_at"]) + assert _parse_datetime(banner["updated_at"]) + + get_response = api.client.get(f"{ADMIN_BANNERS_URL}/{banner['id']}") + assert get_response.status_code == 200, get_response.text + assert get_response.json() == banner + + list_response = api.client.get(ADMIN_BANNERS_URL) + assert list_response.status_code == 200, list_response.text + assert list_response.json() == {"banners": [banner]} + + +def test_active_banners_include_enabled_banner_in_window(api: _TestApi): + current_time = _get_current_time() + created_banner = _create_banner( + api, + starts_at=(current_time - datetime.timedelta(hours=1)).isoformat(), + ends_at=(current_time + datetime.timedelta(hours=1)).isoformat(), + ) + active_banners = _get_active_banners(api) + assert len(active_banners) == 1 + active_banner = active_banners[0] + assert active_banner["id"] == created_banner["id"] + # The public response must not expose the admin-only fields. + assert set(active_banner) == { + "id", + "title", + "body", + "variant", + "url", + "url_text", + "starts_at", + "ends_at", + "is_dismissible", + "created_at", + "updated_at", + } + + +def test_active_banners_exclude_disabled_banner(api: _TestApi): + _create_banner(api, is_enabled=False) + assert _get_active_banners(api) == [] + + +def test_active_banners_exclude_future_banner(api: _TestApi): + starts_at = _get_current_time() + datetime.timedelta(hours=1) + _create_banner(api, starts_at=starts_at.isoformat()) + assert _get_active_banners(api) == [] + + +def test_active_banners_exclude_expired_banner(api: _TestApi): + current_time = _get_current_time() + _create_banner( + api, + starts_at=(current_time - datetime.timedelta(hours=2)).isoformat(), + ends_at=(current_time - datetime.timedelta(hours=1)).isoformat(), + ) + assert _get_active_banners(api) == [] + + +def test_patch_updates_fields_and_updated_at(api: _TestApi): + banner = _create_banner(api, url="https://example.com/status", url_text="Details") + ends_at = _get_current_time() + datetime.timedelta(hours=1) + response = api.client.patch( + f"{ADMIN_BANNERS_URL}/{banner['id']}", + json={ + "title": " Updated title ", + "variant": "info", + "is_enabled": False, + "ends_at": ends_at.isoformat(), + }, + ) + assert response.status_code == 200, response.text + updated_banner = response.json() + assert updated_banner["title"] == "Updated title" + assert updated_banner["variant"] == "info" + assert updated_banner["is_enabled"] == False + assert _parse_datetime(updated_banner["ends_at"]) == ends_at + assert updated_banner["body"] == banner["body"] + assert updated_banner["url"] == banner["url"] + assert updated_banner["url_text"] == banner["url_text"] + assert updated_banner["is_dismissible"] == banner["is_dismissible"] + assert updated_banner["created_at"] == banner["created_at"] + assert _parse_datetime(updated_banner["updated_at"]) > _parse_datetime( + banner["updated_at"] + ) + + +def test_delete_soft_deletes_banner(api: _TestApi): + banner = _create_banner(api) + assert len(_get_active_banners(api)) == 1 + + response = api.client.delete(f"{ADMIN_BANNERS_URL}/{banner['id']}") + assert response.status_code == 200, response.text + deleted_banner = response.json() + assert _parse_datetime(deleted_banner["deleted_at"]) + assert deleted_banner["updated_by"] == ADMIN_USER_NAME + + assert _get_active_banners(api) == [] + list_response = api.client.get(ADMIN_BANNERS_URL) + assert list_response.json() == {"banners": []} + list_response_2 = api.client.get( + ADMIN_BANNERS_URL, params={"include_deleted": True} + ) + assert [b["id"] for b in list_response_2.json()["banners"]] == [banner["id"]] + get_response = api.client.get(f"{ADMIN_BANNERS_URL}/{banner['id']}") + assert get_response.status_code == 200, get_response.text + assert get_response.json()["deleted_at"] == deleted_banner["deleted_at"] + + +def test_non_admin_cannot_create_update_or_delete_banners(api: _TestApi): + banner = _create_banner(api) + api.become_non_admin() + + create_response = api.client.post(ADMIN_BANNERS_URL, json=_make_banner_request()) + assert create_response.status_code == 403, create_response.text + + update_response = api.client.patch( + f"{ADMIN_BANNERS_URL}/{banner['id']}", json={"title": "New title"} + ) + assert update_response.status_code == 403, update_response.text + + delete_response = api.client.delete(f"{ADMIN_BANNERS_URL}/{banner['id']}") + assert delete_response.status_code == 403, delete_response.text + + list_response = api.client.get(ADMIN_BANNERS_URL) + assert list_response.status_code == 403, list_response.text + + # Reading the active banners does not require admin permissions. + assert len(_get_active_banners(api)) == 1 + + +@pytest.mark.parametrize( + "banner_overrides", + [ + {"url": "example.com"}, + {"url": "javascript:alert(1)"}, + {"url": "ftp://example.com"}, + { + "starts_at": "2026-01-02T00:00:00+00:00", + "ends_at": "2026-01-01T00:00:00+00:00", + }, + { + "starts_at": "2026-01-01T00:00:00+00:00", + "ends_at": "2026-01-01T00:00:00+00:00", + }, + {"title": " "}, + {"title": "x" * 121}, + {"body": ""}, + {"body": "x" * 2001}, + # The URL text requires a URL. + {"url_text": "View details"}, + {"url_text": "x" * 81, "url": "https://example.com"}, + ], +) +def test_invalid_banner_is_rejected(api: _TestApi, banner_overrides: dict): + response = api.client.post( + ADMIN_BANNERS_URL, json=_make_banner_request(**banner_overrides) + ) + assert response.status_code == 422, response.text + assert _get_active_banners(api) == [] + + +@pytest.mark.parametrize("variant", ["critical", "", "WARNING", None]) +def test_invalid_banner_variant_is_rejected(api: _TestApi, variant): + response = api.client.post( + ADMIN_BANNERS_URL, json=_make_banner_request(variant=variant) + ) + assert response.status_code == 422, response.text + assert _get_active_banners(api) == [] + + +def test_invalid_banner_update_is_rejected(api: _TestApi): + banner = _create_banner( + api, + starts_at="2026-01-01T00:00:00+00:00", + url="https://example.com/status", + url_text="View details", + ) + banner_url = f"{ADMIN_BANNERS_URL}/{banner['id']}" + + for invalid_update in [ + {"variant": "critical"}, + {"url": "example.com"}, + {"title": " "}, + # Before the existing `starts_at`. + {"ends_at": "2025-01-01T00:00:00+00:00"}, + ]: + response = api.client.patch(banner_url, json=invalid_update) + assert response.status_code == 422, f"{invalid_update=}: {response.text}" + + assert api.client.get(banner_url).json() == banner + + +def test_banner_datetimes_are_converted_to_utc(api: _TestApi): + banner = _create_banner( + api, + starts_at="2026-01-01T12:00:00+02:00", + ends_at="2026-01-01T12:00:00-05:00", + ) + assert banner["starts_at"] == "2026-01-01T10:00:00Z" + assert banner["ends_at"] == "2026-01-01T17:00:00Z" + + +def test_banner_not_found(api: _TestApi): + assert api.client.get(f"{ADMIN_BANNERS_URL}/no-such-id").status_code == 404 + assert ( + api.client.patch( + f"{ADMIN_BANNERS_URL}/no-such-id", json={"title": "New title"} + ).status_code + == 404 + ) + assert api.client.delete(f"{ADMIN_BANNERS_URL}/no-such-id").status_code == 404 + + +def test_active_banners_are_sorted(api: _TestApi): + current_time = _get_current_time() + banner_without_start = _create_banner(api, title="No start time") + banner_older = _create_banner( + api, + title="Older", + starts_at=(current_time - datetime.timedelta(hours=2)).isoformat(), + ) + banner_newer = _create_banner( + api, + title="Newer", + starts_at=(current_time - datetime.timedelta(hours=1)).isoformat(), + ) + # `starts_at` descending, with the banners without a start time last. + assert [banner["id"] for banner in _get_active_banners(api)] == [ + banner_newer["id"], + banner_older["id"], + banner_without_start["id"], + ] + + +if __name__ == "__main__": + pytest.main()