diff --git a/.specify/feature.json b/.specify/feature.json index f3119e3..ee62bbf 100644 --- a/.specify/feature.json +++ b/.specify/feature.json @@ -1 +1 @@ -{"feature_directory":"specs/005-signup-access"} +{"feature_directory":"specs/006-brand-kit"} diff --git a/backend/app/config.py b/backend/app/config.py index 7301570..639f885 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -80,6 +80,10 @@ def get_engine() -> Engine: AND has_table_privilege(current_user, 'public.brand_asset_operations', 'INSERT') AND has_table_privilege(current_user, 'public.brand_asset_operations', 'UPDATE') AND has_table_privilege(current_user, 'public.brand_asset_operations', 'DELETE') + AND has_table_privilege(current_user, 'public.brand_kits', 'SELECT') + AND has_table_privilege(current_user, 'public.brand_kits', 'INSERT') + AND has_table_privilege(current_user, 'public.brand_kits', 'UPDATE') + AND has_table_privilege(current_user, 'public.brand_kits', 'DELETE') AS application_dml, has_schema_privilege(current_user, 'vault', 'USAGE') AS vault_schema_usage, has_function_privilege( diff --git a/backend/app/main.py b/backend/app/main.py index f3c7857..74a302a 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -14,6 +14,7 @@ from .config import assert_database_role_privileges, load_settings from .routes.auth import router as auth_router +from .routes.brand_kits import router as brand_kits_router from .routes.brands import router as brands_router from .routes.health import router as health_router from .routes.me import router as me_router @@ -150,6 +151,7 @@ async def root() -> dict[str, str]: app.include_router(auth_router) +app.include_router(brand_kits_router) app.include_router(brands_router) app.include_router(health_router) app.include_router(me_router) diff --git a/backend/app/models/brand.py b/backend/app/models/brand.py index cce3e93..a2479bf 100644 --- a/backend/app/models/brand.py +++ b/backend/app/models/brand.py @@ -28,6 +28,7 @@ class Brand(BaseModel): name: str logo_url: str | None cleanup_state: Literal["normal", "cleanup_required"] = "normal" + kit_status: Literal["not_started", "in_progress", "complete"] = "not_started" created_at: datetime diff --git a/backend/app/models/brand_kit.py b/backend/app/models/brand_kit.py new file mode 100644 index 0000000..b99528c --- /dev/null +++ b/backend/app/models/brand_kit.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +import re +from datetime import datetime +from enum import Enum +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +_COLOR_PATTERN = re.compile(r"#[0-9A-Fa-f]{6}\Z") + + +class KitStatus(str, Enum): + NOT_STARTED = "not_started" + IN_PROGRESS = "in_progress" + COMPLETE = "complete" + + +class Tone(str, Enum): + FORMAL = "formal" + CASUAL = "casual" + PLAYFUL = "playful" + PROFESSIONAL = "professional" + FRIENDLY = "friendly" + + +class BrandKitAnswers(BaseModel): + model_config = ConfigDict(extra="forbid") + + tagline: str | None = None + tone: Tone | None = None + audience: str | None = None + colors: list[str] = Field(default_factory=list) + avoid_words: str | None = None + + @field_validator("tagline", "audience", "avoid_words", mode="before") + @classmethod + def trim_nullable_text(cls, value: object) -> object: + if not isinstance(value, str): + return value + normalized = value.strip() + return normalized or None + + @field_validator("tone", mode="before") + @classmethod + def clear_blank_tone(cls, value: object) -> object: + if isinstance(value, str): + normalized = value.strip() + return normalized or None + return value + + @field_validator("tagline") + @classmethod + def validate_tagline(cls, value: str | None) -> str | None: + if value is not None and len(value) > 160: + raise ValueError("tagline must be 160 characters or fewer.") + return value + + @field_validator("audience") + @classmethod + def validate_audience(cls, value: str | None) -> str | None: + if value is not None and not 2 <= len(value) <= 500: + raise ValueError("audience must be between 2 and 500 characters.") + return value + + @field_validator("colors", mode="before") + @classmethod + def clear_null_colors(cls, value: object) -> object: + return [] if value is None else value + + @field_validator("colors") + @classmethod + def normalize_colors(cls, value: list[str]) -> list[str]: + if len(value) > 3: + raise ValueError("colors must contain no more than 3 values.") + + normalized = [color.strip().upper() for color in value] + if any(_COLOR_PATTERN.fullmatch(color) is None for color in normalized): + raise ValueError("colors must use the #RRGGBB format.") + return normalized + + +class BrandKitUpsert(BaseModel): + model_config = ConfigDict(extra="forbid") + + name: str + answers: BrandKitAnswers = Field(default_factory=BrandKitAnswers) + + @field_validator("name") + @classmethod + def validate_name(cls, value: str) -> str: + normalized = value.strip() + if not 2 <= len(normalized) <= 120: + raise ValueError("name must be between 2 and 120 characters.") + return normalized + + +class BrandKit(BaseModel): + model_config = ConfigDict(extra="forbid") + + brand_id: UUID + brand_name: str + answers: BrandKitAnswers + summary: str | None + status: KitStatus + completed_at: datetime | None + updated_at: datetime | None diff --git a/backend/app/routes/brand_kits.py b/backend/app/routes/brand_kits.py new file mode 100644 index 0000000..96b0e79 --- /dev/null +++ b/backend/app/routes/brand_kits.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import logging +from typing import Annotated +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException, Request, status + +from ..auth import CurrentUserDep +from ..models.brand_kit import BrandKit, BrandKitUpsert +from ..services.brand_kit_store import ( + BrandKitStore, + get_brand_kit_store, +) +from ..services.brand_store import BrandCleanupRequiredError, BrandNameTakenError + + +router = APIRouter(prefix="/api/v1/brands", tags=["brand-kits"]) +logger = logging.getLogger(__name__) + +BrandKitStoreDep = Annotated[BrandKitStore, Depends(get_brand_kit_store)] + + +def _not_found() -> HTTPException: + return HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"code": "BRAND_NOT_FOUND", "message": "Brand not found."}, + ) + + +def _cleanup_required() -> HTTPException: + return HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={ + "code": "BRAND_CLEANUP_REQUIRED", + "message": "Brand cleanup is required. Retry deletion.", + }, + ) + + +def _name_taken() -> HTTPException: + return HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={ + "code": "BRAND_NAME_TAKEN", + "message": "You already have a brand with this name.", + }, + ) + + +def _map_access_error(exc: Exception) -> HTTPException: + if isinstance(exc, BrandCleanupRequiredError): + return _cleanup_required() + return _not_found() + + +@router.get("/{brand_id}/kit", response_model=BrandKit) +def get_brand_kit( + request: Request, + brand_id: UUID, + current_user: CurrentUserDep, + brand_kit_store: BrandKitStoreDep, +) -> BrandKit: + try: + kit = brand_kit_store.get_kit(current_user.user_id, brand_id) + except (LookupError, BrandCleanupRequiredError) as exc: + raise _map_access_error(exc) from exc + + logger.info( + "brand_kits.get_success", + extra={ + "event": "brand_kits.get_success", + "request_id": getattr(request.state, "request_id", "unknown"), + }, + ) + return kit + + +@router.put("/{brand_id}/kit", response_model=BrandKit) +def put_brand_kit( + request: Request, + brand_id: UUID, + payload: BrandKitUpsert, + current_user: CurrentUserDep, + brand_kit_store: BrandKitStoreDep, +) -> BrandKit: + try: + kit = brand_kit_store.upsert_kit(current_user.user_id, brand_id, payload) + except BrandNameTakenError as exc: + raise _name_taken() from exc + except (LookupError, BrandCleanupRequiredError) as exc: + raise _map_access_error(exc) from exc + + logger.info( + "brand_kits.put_success", + extra={ + "event": "brand_kits.put_success", + "request_id": getattr(request.state, "request_id", "unknown"), + }, + ) + return kit diff --git a/backend/app/services/brand_kit_store.py b/backend/app/services/brand_kit_store.py new file mode 100644 index 0000000..f50b8d7 --- /dev/null +++ b/backend/app/services/brand_kit_store.py @@ -0,0 +1,251 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from functools import lru_cache +from typing import Any +from uuid import UUID + +from sqlalchemy import text +from sqlalchemy.engine import Connection, Engine +from sqlalchemy.exc import IntegrityError + +from ..config import get_engine +from ..models.brand_kit import BrandKit, BrandKitAnswers, BrandKitUpsert, KitStatus +from .brand_store import BrandNameTakenError, BrandStore + + +def _collapse_whitespace(value: str) -> str: + return " ".join(value.split()) + + +def derive_summary( + brand_name: str, + answers: BrandKitAnswers | Mapping[str, Any], +) -> str: + normalized_answers = BrandKitAnswers.model_validate(answers) + normalized_name = _collapse_whitespace(brand_name) + audience = ( + _collapse_whitespace(normalized_answers.audience) + if normalized_answers.audience is not None + else "" + ) + if not normalized_name or normalized_answers.tone is None or not audience: + raise ValueError("A summary requires complete brand kit answers.") + if not normalized_answers.colors: + raise ValueError("A summary requires complete brand kit answers.") + + tagline = ( + _collapse_whitespace(normalized_answers.tagline) + if normalized_answers.tagline is not None + else "None specified" + ) + avoid_words = ( + _collapse_whitespace(normalized_answers.avoid_words) + if normalized_answers.avoid_words is not None + else "None specified" + ) + return "\n".join( + ( + f"Brand: {normalized_name}", + f"Tagline: {tagline}", + f"Tone: {normalized_answers.tone.value}", + f"Audience: {audience}", + f"Colors: {', '.join(normalized_answers.colors)}", + f"Avoid words: {avoid_words}", + ) + ) + + +@dataclass(frozen=True, slots=True) +class BrandKitStore: + engine: Engine + + @staticmethod + def _to_brand_kit(row: Mapping[str, Any]) -> BrandKit: + return BrandKit.model_validate( + { + "brand_id": row["brand_id"], + "brand_name": row["brand_name"], + "answers": { + "tagline": row["tagline"], + "tone": row["tone"], + "audience": row["audience"], + "colors": row["colors"], + "avoid_words": row["avoid_words"], + }, + "summary": row["summary"], + "status": row["status"], + "completed_at": row["completed_at"], + "updated_at": row["updated_at"], + } + ) + + @staticmethod + def _empty_kit(brand: Mapping[str, Any]) -> BrandKit: + return BrandKit( + brand_id=brand["id"], + brand_name=brand["name"], + answers=BrandKitAnswers(), + summary=None, + status=KitStatus.NOT_STARTED, + completed_at=None, + updated_at=None, + ) + + @staticmethod + def _kit_row( + connection: Connection, brand_id: UUID + ) -> Mapping[str, Any] | None: + return connection.execute( + text( + """ + SELECT brand_id, tagline, tone, audience, colors, avoid_words, + summary, status, completed_at, updated_at + FROM brand_kits + WHERE brand_id = :brand_id + """ + ), + {"brand_id": brand_id}, + ).mappings().one_or_none() + + @staticmethod + def _merged_answers( + existing: Mapping[str, Any] | None, + payload: BrandKitUpsert, + ) -> BrandKitAnswers: + current = { + "tagline": existing["tagline"] if existing else None, + "tone": existing["tone"] if existing else None, + "audience": existing["audience"] if existing else None, + "colors": list(existing["colors"] or []) if existing else [], + "avoid_words": existing["avoid_words"] if existing else None, + } + current.update(payload.answers.model_dump(exclude_unset=True)) + return BrandKitAnswers.model_validate(current) + + @staticmethod + def _has_saved_answer(answers: BrandKitAnswers) -> bool: + return any( + ( + answers.tagline, + answers.tone, + answers.audience, + answers.colors, + answers.avoid_words, + ) + ) + + @classmethod + def _status_for_answers(cls, answers: BrandKitAnswers) -> KitStatus: + if answers.tone is not None and answers.audience is not None and answers.colors: + return KitStatus.COMPLETE + if cls._has_saved_answer(answers): + return KitStatus.IN_PROGRESS + return KitStatus.NOT_STARTED + + def get_kit(self, user_id: str, brand_id: UUID) -> BrandKit: + with self.engine.begin() as connection: + brand = BrandStore.lock_owned_brand(connection, user_id, brand_id) + BrandStore.require_normal_brand(brand) + row = self._kit_row(connection, brand_id) + if row is None: + return self._empty_kit(brand) + return self._to_brand_kit({"brand_name": brand["name"], **row}) + + def upsert_kit( + self, + user_id: str, + brand_id: UUID, + payload: BrandKitUpsert, + ) -> BrandKit: + try: + with self.engine.begin() as connection: + brand = BrandStore.lock_owned_brand(connection, user_id, brand_id) + BrandStore.require_normal_brand(brand) + existing = self._kit_row(connection, brand_id) + answers = self._merged_answers(existing, payload) + status = self._status_for_answers(answers) + complete = status is KitStatus.COMPLETE + summary = derive_summary(payload.name, answers) if complete else None + + connection.execute( + text( + """ + UPDATE brands + SET name = :name + WHERE id = :brand_id + """ + ), + {"brand_id": brand_id, "name": payload.name}, + ) + if status is KitStatus.NOT_STARTED: + connection.execute( + text("DELETE FROM brand_kits WHERE brand_id = :brand_id"), + {"brand_id": brand_id}, + ) + return self._empty_kit( + {"id": brand_id, "name": payload.name} + ) + + connection.execute( + text( + """ + INSERT INTO brand_kits ( + brand_id, tagline, tone, audience, colors, avoid_words, + summary, status, completed_at + ) VALUES ( + :brand_id, :tagline, CAST(:tone AS public.tone_t), :audience, + :colors, :avoid_words, :summary, + CAST(:status AS public.kit_status_t), + CASE WHEN :status = 'complete' THEN now() ELSE NULL END + ) + ON CONFLICT (brand_id) DO UPDATE SET + tagline = EXCLUDED.tagline, + tone = EXCLUDED.tone, + audience = EXCLUDED.audience, + colors = EXCLUDED.colors, + avoid_words = EXCLUDED.avoid_words, + summary = EXCLUDED.summary, + status = EXCLUDED.status, + completed_at = CASE + WHEN brand_kits.status = 'complete' + AND EXCLUDED.status = 'complete' + THEN brand_kits.completed_at + ELSE EXCLUDED.completed_at + END + """ + ), + { + "brand_id": brand_id, + "tagline": answers.tagline, + "tone": answers.tone.value if answers.tone else None, + "audience": answers.audience, + "colors": answers.colors, + "avoid_words": answers.avoid_words, + "summary": summary, + "status": status.value, + }, + ) + row = self._kit_row(connection, brand_id) + assert row is not None + updated_brand_name = payload.name + return self._to_brand_kit( + {"brand_name": updated_brand_name, **row} + ) + except IntegrityError as exc: + original = exc.orig + constraint_name = getattr( + getattr(original, "diag", None), "constraint_name", None + ) + if ( + getattr(original, "pgcode", None) == "23505" + and constraint_name == "uq_brands_owner_name_ci" + ): + raise BrandNameTakenError from exc + raise + + +@lru_cache(maxsize=1) +def get_brand_kit_store() -> BrandKitStore: + return BrandKitStore(get_engine()) diff --git a/backend/app/services/brand_store.py b/backend/app/services/brand_store.py index 3e8b76d..5d77875 100644 --- a/backend/app/services/brand_store.py +++ b/backend/app/services/brand_store.py @@ -57,7 +57,14 @@ def _to_brand(row: Mapping[str, Any]) -> Brand: "cleanup_required": "cleanup_required", }[row["deletion_state"]] return Brand.model_validate( - {"logo_url": logo_url, "cleanup_state": cleanup_state, **row} + { + "id": row["id"], + "name": row["name"], + "logo_url": logo_url, + "cleanup_state": cleanup_state, + "kit_status": row.get("kit_status", "not_started"), + "created_at": row["created_at"], + } ) @staticmethod @@ -69,10 +76,12 @@ def lock_owned_brand( row = connection.execute( text( """ - SELECT id, name, logo_path, deletion_state, created_at - FROM brands - WHERE id = :brand_id AND owner_user_id = :owner_user_id - FOR UPDATE + SELECT b.id, b.name, b.logo_path, b.deletion_state, b.created_at, + COALESCE(k.status::TEXT, 'not_started') AS kit_status + FROM brands AS b + LEFT JOIN brand_kits AS k ON k.brand_id = b.id + WHERE b.id = :brand_id AND b.owner_user_id = :owner_user_id + FOR UPDATE OF b """ ), {"brand_id": brand_id, "owner_user_id": user_id}, @@ -155,10 +164,12 @@ def list_brands(self, user_id: str) -> list[Brand]: rows = connection.execute( text( """ - SELECT id, name, logo_path, deletion_state, created_at - FROM brands - WHERE owner_user_id = :owner_user_id - ORDER BY created_at DESC, id DESC + SELECT b.id, b.name, b.logo_path, b.deletion_state, b.created_at, + COALESCE(k.status::TEXT, 'not_started') AS kit_status + FROM brands AS b + LEFT JOIN brand_kits AS k ON k.brand_id = b.id + WHERE b.owner_user_id = :owner_user_id + ORDER BY b.created_at DESC, b.id DESC """ ), {"owner_user_id": user_id}, @@ -171,9 +182,11 @@ def get_brand(self, user_id: str, brand_id: UUID) -> Brand: row = connection.execute( text( """ - SELECT id, name, logo_path, deletion_state, created_at - FROM brands - WHERE id = :brand_id AND owner_user_id = :owner_user_id + SELECT b.id, b.name, b.logo_path, b.deletion_state, b.created_at, + COALESCE(k.status::TEXT, 'not_started') AS kit_status + FROM brands AS b + LEFT JOIN brand_kits AS k ON k.brand_id = b.id + WHERE b.id = :brand_id AND b.owner_user_id = :owner_user_id """ ), {"brand_id": brand_id, "owner_user_id": user_id}, @@ -380,6 +393,7 @@ def publish_uploaded_logo( ).mappings().one_or_none() if row is None: raise BrandAssetOperationStaleError + row = {**row, "kit_status": brand["kit_status"]} return self._to_brand(row) def complete_asset_operation( diff --git a/backend/tests/contract/test_brand_kits.py b/backend/tests/contract/test_brand_kits.py new file mode 100644 index 0000000..983d7ba --- /dev/null +++ b/backend/tests/contract/test_brand_kits.py @@ -0,0 +1,404 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime +from uuid import UUID + +import pytest +from fastapi.testclient import TestClient + +from backend.app.auth import CurrentUser, get_current_user +from backend.app.main import app +from backend.app.models.brand_kit import BrandKit, BrandKitUpsert +from backend.app.routes import brand_kits as brand_kit_routes +from backend.app.routes.brand_kits import get_brand_kit_store +from backend.app.services.brand_kit_store import derive_summary +from backend.app.services.brand_store import BrandNameTakenError + + +BRAND_ID = UUID("22222222-2222-2222-2222-222222222222") +OWNER_USER_ID = "11111111-1111-1111-1111-111111111111" +COMPLETE_ANSWERS = { + "tagline": "Innovation for everyone", + "tone": "professional", + "audience": "Small business owners aged 25-45", + "colors": ["#FF5733", "#3498DB"], + "avoid_words": "cheap, discount", +} +COMPLETE_SUMMARY = "\n".join( + ( + "Brand: My Brand", + "Tagline: Innovation for everyone", + "Tone: professional", + "Audience: Small business owners aged 25-45", + "Colors: #FF5733, #3498DB", + "Avoid words: cheap, discount", + ) +) +COMPLETED_AT = datetime(2026, 7, 29, tzinfo=UTC) + + +def _kit(*, answers=None, status="not_started", summary=None) -> BrandKit: + return BrandKit.model_validate( + { + "brand_id": BRAND_ID, + "brand_name": "My Brand", + "answers": answers or {}, + "summary": summary, + "status": status, + "completed_at": COMPLETED_AT if status == "complete" else None, + "updated_at": COMPLETED_AT if status == "complete" else None, + } + ) + + +@dataclass +class FakeBrandKitStore: + kit: BrandKit | None = None + saved_payloads: list[BrandKitUpsert] = None + + def __post_init__(self) -> None: + if self.saved_payloads is None: + self.saved_payloads = [] + + def get_kit(self, user_id: str, brand_id: UUID) -> BrandKit: + return self.kit or _kit() + + def upsert_kit( + self, user_id: str, brand_id: UUID, payload: BrandKitUpsert + ) -> BrandKit: + self.saved_payloads.append(payload) + current = self.kit.answers if self.kit else BrandKitUpsert.model_validate( + {"name": payload.name} + ).answers + answers = current.model_copy( + update=payload.answers.model_dump(exclude_unset=True) + ) + is_complete = ( + answers.tone is not None + and answers.audience is not None + and bool(answers.colors) + ) + kit_status = "complete" if is_complete else ( + "in_progress" + if any((answers.tagline, answers.tone, answers.audience, answers.colors, answers.avoid_words)) + else "not_started" + ) + self.kit = _kit( + answers=answers, + status=kit_status, + summary=derive_summary(payload.name, answers) if is_complete else None, + ) + return self.kit + + +class DuplicateNameStore(FakeBrandKitStore): + def upsert_kit( + self, user_id: str, brand_id: UUID, payload: BrandKitUpsert + ) -> BrandKit: + raise BrandNameTakenError + + +def _client(store: FakeBrandKitStore) -> TestClient: + app.dependency_overrides[get_current_user] = lambda: CurrentUser( + user_id=OWNER_USER_ID, + email="owner@example.com", + access_token="redacted", + ) + app.dependency_overrides[get_brand_kit_store] = lambda: store + return TestClient(app) + + +def test_get_without_row_returns_exact_not_started_shape(): + store = FakeBrandKitStore() + try: + with _client(store) as client: + response = client.get(f"/api/v1/brands/{BRAND_ID}/kit") + + assert response.status_code == 200 + assert response.json() == { + "brand_id": str(BRAND_ID), + "brand_name": "My Brand", + "answers": { + "tagline": None, + "tone": None, + "audience": None, + "colors": [], + "avoid_words": None, + }, + "summary": None, + "status": "not_started", + "completed_at": None, + "updated_at": None, + } + finally: + app.dependency_overrides.clear() + + +def test_put_complete_answers_returns_exact_public_shape(): + store = FakeBrandKitStore() + try: + with _client(store) as client: + response = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "My Brand", "answers": COMPLETE_ANSWERS}, + ) + + assert response.status_code == 200 + assert response.json() == { + "brand_id": str(BRAND_ID), + "brand_name": "My Brand", + "answers": COMPLETE_ANSWERS, + "summary": COMPLETE_SUMMARY, + "status": "complete", + "completed_at": "2026-07-29T00:00:00Z", + "updated_at": "2026-07-29T00:00:00Z", + } + assert store.saved_payloads[0].name == "My Brand" + finally: + app.dependency_overrides.clear() + + +def test_put_complete_answers_allows_optional_fields_to_be_omitted(): + store = FakeBrandKitStore() + answers = { + "tone": "professional", + "audience": COMPLETE_ANSWERS["audience"], + "colors": COMPLETE_ANSWERS["colors"], + } + try: + with _client(store) as client: + response = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "My Brand", "answers": answers}, + ) + + assert response.status_code == 200 + assert response.json()["answers"] == { + "tagline": None, + "tone": "professional", + "audience": COMPLETE_ANSWERS["audience"], + "colors": COMPLETE_ANSWERS["colors"], + "avoid_words": None, + } + assert response.json()["summary"] == "\n".join( + ( + "Brand: My Brand", + "Tagline: None specified", + "Tone: professional", + "Audience: Small business owners aged 25-45", + "Colors: #FF5733, #3498DB", + "Avoid words: None specified", + ) + ) + assert response.json()["status"] == "complete" + assert set(response.json()) == { + "brand_id", + "brand_name", + "answers", + "summary", + "status", + "completed_at", + "updated_at", + } + finally: + app.dependency_overrides.clear() + + +def test_put_maps_duplicate_brand_name_to_conflict(): + try: + with _client(DuplicateNameStore()) as client: + response = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "Existing Brand", "answers": COMPLETE_ANSWERS}, + ) + + assert response.status_code == 409 + assert response.json()["error"]["code"] == "BRAND_NAME_TAKEN" + assert response.json()["error"]["request_id"] + finally: + app.dependency_overrides.clear() + + +def test_put_zero_answers_returns_not_started_without_derived_fields(): + store = FakeBrandKitStore() + try: + with _client(store) as client: + response = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "My Brand", "answers": {}}, + ) + + assert response.status_code == 200 + assert response.json()["status"] == "not_started" + assert response.json()["summary"] is None + assert response.json()["completed_at"] is None + finally: + app.dependency_overrides.clear() + + +def test_answer_save_transitions_to_in_progress_and_preserves_omitted_values(): + store = FakeBrandKitStore() + try: + with _client(store) as client: + first = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "My Brand", "answers": {"tagline": "Hello"}}, + ) + second = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "My Brand", "answers": {"tone": "friendly"}}, + ) + + assert first.json()["status"] == "in_progress" + assert second.json()["answers"]["tagline"] == "Hello" + assert second.json()["answers"]["tone"] == "friendly" + finally: + app.dependency_overrides.clear() + + +def test_explicit_empty_values_clear_answers_and_derived_fields(): + store = FakeBrandKitStore() + try: + with _client(store) as client: + complete = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "My Brand", "answers": COMPLETE_ANSWERS}, + ) + cleared = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={ + "name": "My Brand", + "answers": {"tone": None, "colors": []}, + }, + ) + + assert complete.json()["status"] == "complete" + assert cleared.json()["status"] == "in_progress" + assert cleared.json()["answers"]["tone"] is None + assert cleared.json()["answers"]["colors"] == [] + assert cleared.json()["summary"] is None + assert cleared.json()["completed_at"] is None + finally: + app.dependency_overrides.clear() + + +def test_invalid_completion_is_rejected_without_replacing_saved_response(): + store = FakeBrandKitStore() + try: + with _client(store) as client: + saved = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "My Brand", "answers": {"tagline": "Keep me"}}, + ) + invalid = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={ + "name": "My Brand", + "answers": {"colors": ["not-a-color"]}, + }, + ) + + assert saved.status_code == 200 + assert invalid.status_code == 400 + assert store.saved_payloads[-1].answers.tagline == "Keep me" + finally: + app.dependency_overrides.clear() + + +def test_repeated_puts_keep_a_single_logical_kit_payload(): + store = FakeBrandKitStore() + try: + with _client(store) as client: + for _ in range(2): + response = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "My Brand", "answers": {"tagline": "Hello"}}, + ) + assert response.status_code == 200 + + assert len(store.saved_payloads) == 2 + finally: + app.dependency_overrides.clear() + + +def test_get_and_put_logs_only_safe_metadata(monkeypatch: pytest.MonkeyPatch): + private_answer = "Never log this private answer" + logged: list[tuple[str, dict[str, str]]] = [] + monkeypatch.setattr( + brand_kit_routes.logger, + "info", + lambda message, *, extra: logged.append((message, extra)), + ) + try: + with _client(FakeBrandKitStore()) as client: + get_response = client.get(f"/api/v1/brands/{BRAND_ID}/kit") + put_response = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={ + "name": "Private Brand", + "answers": {"tagline": private_answer}, + }, + ) + + assert get_response.status_code == 200 + assert put_response.status_code == 200 + assert [message for message, _ in logged] == [ + "brand_kits.get_success", + "brand_kits.put_success", + ] + assert all(set(extra) == {"event", "request_id"} for _, extra in logged) + assert all(extra["request_id"] for _, extra in logged) + for private_value in ( + private_answer, + "Private Brand", + OWNER_USER_ID, + "owner@example.com", + "redacted", + ): + assert private_value not in repr(logged) + finally: + app.dependency_overrides.clear() + + +@pytest.mark.parametrize("authorization", [None, "Basic malformed", "Bearer "]) +def test_kit_requires_a_valid_authorization_header(authorization: str | None): + headers = {} if authorization is None else {"Authorization": authorization} + try: + with TestClient(app) as client: + response = client.get(f"/api/v1/brands/{BRAND_ID}/kit", headers=headers) + + assert response.status_code == 401 + assert response.json()["error"]["code"] == "UNAUTHORIZED" + assert response.json()["error"]["message"] == "Sign in required." + assert set(response.json()["error"]) == {"code", "message", "request_id"} + finally: + app.dependency_overrides.clear() + + +def test_kit_store_errors_are_opaque_for_non_owner_and_missing_brands(): + class UnauthorizedStore(FakeBrandKitStore): + def get_kit(self, user_id: str, brand_id: UUID) -> BrandKit: + raise LookupError("Brand not found.") + + def upsert_kit( + self, user_id: str, brand_id: UUID, payload: BrandKitUpsert + ) -> BrandKit: + raise LookupError("Brand not found.") + + try: + with _client(UnauthorizedStore()) as client: + get_response = client.get(f"/api/v1/brands/{BRAND_ID}/kit") + put_response = client.put( + f"/api/v1/brands/{BRAND_ID}/kit", + json={"name": "Private Brand", "answers": {}}, + ) + + for response in (get_response, put_response): + assert response.status_code == 404 + error = response.json()["error"] + assert error["code"] == "BRAND_NOT_FOUND" + assert error["message"] == "Brand not found." + assert "Private Brand" not in response.text + finally: + app.dependency_overrides.clear() diff --git a/backend/tests/contract/test_brands.py b/backend/tests/contract/test_brands.py index cd6e665..902f3c0 100644 --- a/backend/tests/contract/test_brands.py +++ b/backend/tests/contract/test_brands.py @@ -355,6 +355,7 @@ def test_create_brand_returns_public_contract_shape(): "name": "Acme Coffee", "logo_url": None, "cleanup_state": "normal", + "kit_status": "not_started", "created_at": "2026-07-25T00:00:00Z", } finally: @@ -379,12 +380,14 @@ def test_list_brands_returns_empty_and_populated_contract_shapes(): id=UUID("33333333-3333-3333-3333-333333333333"), name="New Brand", logo_url=None, + kit_status="in_progress", created_at=datetime(2026, 7, 26, tzinfo=UTC), ), Brand( id=UUID("22222222-2222-2222-2222-222222222222"), name="First Brand", logo_url=None, + kit_status="complete", created_at=datetime(2026, 7, 25, tzinfo=UTC), ), ] @@ -400,6 +403,7 @@ def test_list_brands_returns_empty_and_populated_contract_shapes(): "name": "New Brand", "logo_url": None, "cleanup_state": "normal", + "kit_status": "in_progress", "created_at": "2026-07-26T00:00:00Z", }, { @@ -407,6 +411,7 @@ def test_list_brands_returns_empty_and_populated_contract_shapes(): "name": "First Brand", "logo_url": None, "cleanup_state": "normal", + "kit_status": "complete", "created_at": "2026-07-25T00:00:00Z", }, ] diff --git a/backend/tests/integration/benchmark_brand_kits.py b/backend/tests/integration/benchmark_brand_kits.py new file mode 100644 index 0000000..9f2036e --- /dev/null +++ b/backend/tests/integration/benchmark_brand_kits.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import logging +import math +import os +import sys +from time import perf_counter +from uuid import uuid4 + +import httpx +from fastapi.testclient import TestClient +from sqlalchemy import text + +from backend.tests.integration.test_brand_kits import ( + _delete_supabase_user, + _signup_and_login, +) + + +SAMPLE_COUNT = 30 +WARMUP_COUNT = 5 + + +def _required_env(name: str) -> str: + value = os.getenv(name) + if not value: + raise RuntimeError(f"{name} is required") + return value + + +def _p95(samples: list[float]) -> float: + ordered = sorted(samples) + return ordered[math.ceil(len(ordered) * 0.95) - 1] + + +def main() -> None: + logging.getLogger("backend.app.routes.brands").setLevel(logging.WARNING) + logging.getLogger("backend.app.routes.brand_kits").setLevel(logging.WARNING) + logging.getLogger("httpx").setLevel(logging.WARNING) + + supabase_url = _required_env("SUPABASE_URL") + supabase_key = _required_env("SUPABASE_SECRET_KEY") + _required_env("DATABASE_URL") + + from backend.app.config import get_engine + from backend.app.main import app + + user_id: str | None = None + brand_id: str | None = None + with httpx.Client(timeout=30.0) as supabase_client: + try: + user_id, access_token = _signup_and_login( + supabase_client, + supabase_url, + supabase_key, + f"brand-kit-benchmark-{uuid4().hex[:10]}@example.com", + ) + headers = {"Authorization": f"Bearer {access_token}"} + with TestClient(app) as client: + created = client.post( + "/api/v1/brands", + headers=headers, + json={"name": "Benchmark Brand"}, + ) + created.raise_for_status() + brand_id = created.json()["id"] + path = f"/api/v1/brands/{brand_id}/kit" + + for index in range(WARMUP_COUNT): + client.get(path, headers=headers).raise_for_status() + client.put( + path, + headers=headers, + json={ + "name": "Benchmark Brand", + "answers": {"tagline": f"Warmup {index}"}, + }, + ).raise_for_status() + + get_samples: list[float] = [] + put_samples: list[float] = [] + for index in range(SAMPLE_COUNT): + started = perf_counter() + client.get(path, headers=headers).raise_for_status() + get_samples.append((perf_counter() - started) * 1000) + + started = perf_counter() + client.put( + path, + headers=headers, + json={ + "name": "Benchmark Brand", + "answers": {"tagline": f"Sample {index}"}, + }, + ).raise_for_status() + put_samples.append((perf_counter() - started) * 1000) + + get_p95 = _p95(get_samples) + put_p95 = _p95(put_samples) + print( + f"Brand Kit local p95 ({SAMPLE_COUNT} samples): " + f"GET {get_p95:.2f}ms, PUT {put_p95:.2f}ms" + ) + if get_p95 >= 500 or put_p95 >= 500: + raise SystemExit("Brand Kit p95 exceeded the 500ms goal") + finally: + original_failure = sys.exc_info()[0] is not None + cleanup_errors: list[Exception] = [] + if brand_id: + try: + with get_engine().begin() as connection: + connection.execute( + text("DELETE FROM brands WHERE id = :brand_id"), + {"brand_id": brand_id}, + ) + except Exception as exc: + cleanup_errors.append(exc) + if user_id: + try: + _delete_supabase_user( + supabase_client, supabase_url, supabase_key, user_id + ) + except Exception as exc: + cleanup_errors.append(exc) + if cleanup_errors and not original_failure: + raise cleanup_errors[0] + + +if __name__ == "__main__": + main() diff --git a/backend/tests/integration/test_brand_kit_rls.py b/backend/tests/integration/test_brand_kit_rls.py new file mode 100644 index 0000000..b23fa75 --- /dev/null +++ b/backend/tests/integration/test_brand_kit_rls.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import json +import os +from uuid import uuid4 + +import httpx +import jwt +import pytest +from sqlalchemy import text +from sqlalchemy.exc import DBAPIError + + +def _required_env(name: str) -> str: + value = os.getenv(name) + if not value: + pytest.skip(f"{name} is required for integration tests") + return value + + +def test_brand_kit_rls_is_forced_and_owner_scoped(): + _required_env("DATABASE_URL") + + from backend.app.config import get_engine + + with get_engine().connect() as connection: + policy = connection.execute( + text( + """ + SELECT c.relrowsecurity, c.relforcerowsecurity, + EXISTS ( + SELECT 1 FROM pg_policies + WHERE schemaname = 'public' + AND tablename = 'brand_kits' + AND policyname = 'brand_kits_owner' + AND qual LIKE '%is_brand_owner%' + AND with_check LIKE '%is_brand_owner%' + ) AS owner_policy + FROM pg_class AS c + JOIN pg_namespace AS n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' AND c.relname = 'brand_kits' + """ + ) + ).mappings().one() + + assert policy["relrowsecurity"] is True + assert policy["relforcerowsecurity"] is True + assert policy["owner_policy"] is True + + +def test_brand_kit_rls_has_backend_dml_privileges(): + _required_env("DATABASE_URL") + + from backend.app.config import get_engine + + with get_engine().connect() as connection: + privileges = connection.execute( + text( + """ + SELECT has_table_privilege(current_user, 'public.brand_kits', 'SELECT') + AND has_table_privilege(current_user, 'public.brand_kits', 'INSERT') + AND has_table_privilege(current_user, 'public.brand_kits', 'UPDATE') + AND has_table_privilege(current_user, 'public.brand_kits', 'DELETE') + AS has_dml + """ + ) + ).scalar_one() + + assert privileges is True + + +def _signup_and_login( + client: httpx.Client, supabase_url: str, supabase_key: str, email: str +) -> tuple[str, str]: + signup = client.post( + f"{supabase_url}/auth/v1/signup", + headers={"apikey": supabase_key, "Content-Type": "application/json"}, + json={"email": email, "password": "12345678"}, + ) + assert signup.status_code in {200, 201} + user_id = signup.json()["user"]["id"] + token = client.post( + f"{supabase_url}/auth/v1/token?grant_type=password", + headers={"apikey": supabase_key, "Content-Type": "application/json"}, + json={"email": email, "password": "12345678"}, + ) + assert token.status_code == 200 + return user_id, token.json()["access_token"] + + +def _as_authenticated(connection, access_token: str) -> None: + claims = jwt.decode(access_token, options={"verify_signature": False}) + connection.execute(text("SET LOCAL ROLE authenticated")) + connection.execute( + text("SELECT set_config('request.jwt.claims', :claims, true)"), + {"claims": json.dumps(claims)}, + ) + + +def test_authenticated_roles_can_only_read_and_write_their_own_kit_rows(): + supabase_url = _required_env("SUPABASE_URL") + supabase_key = _required_env("SUPABASE_SECRET_KEY") + _required_env("DATABASE_URL") + + from backend.app.config import get_engine + + engine = get_engine() + user_a_id = user_b_id = None + brand_a = str(uuid4()) + brand_b = str(uuid4()) + with httpx.Client(timeout=30.0) as client: + try: + user_a_id, token_a = _signup_and_login( + client, supabase_url, supabase_key, f"kit-rls-a-{uuid4().hex[:10]}@example.com" + ) + user_b_id, token_b = _signup_and_login( + client, supabase_url, supabase_key, f"kit-rls-b-{uuid4().hex[:10]}@example.com" + ) + with engine.begin() as connection: + connection.execute( + text( + "INSERT INTO brands (id, owner_user_id, name) VALUES " + "(:brand_a, :user_a, 'Kit RLS A'), (:brand_b, :user_b, 'Kit RLS B')" + ), + {"brand_a": brand_a, "user_a": user_a_id, "brand_b": brand_b, "user_b": user_b_id}, + ) + _as_authenticated(connection, token_a) + connection.execute( + text( + "INSERT INTO brand_kits (brand_id, tagline) " + "VALUES (:brand_id, 'Owner answer')" + ), + {"brand_id": brand_a}, + ) + visible_to_a = connection.execute( + text("SELECT brand_id FROM brand_kits ORDER BY brand_id") + ).scalars().all() + assert [str(brand_id) for brand_id in visible_to_a] == [brand_a] + connection.execute( + text("UPDATE brand_kits SET tagline = 'Updated answer' WHERE brand_id = :brand_id"), + {"brand_id": brand_a}, + ) + + with pytest.raises(DBAPIError): + with engine.begin() as connection: + _as_authenticated(connection, token_a) + connection.execute( + text("INSERT INTO brand_kits (brand_id, tagline) VALUES (:brand_id, 'No access')"), + {"brand_id": brand_b}, + ) + + with engine.begin() as connection: + _as_authenticated(connection, token_b) + visible_to_b = connection.execute( + text("SELECT brand_id FROM brand_kits") + ).scalars().all() + assert visible_to_b == [] + update_result = connection.execute( + text( + "UPDATE brand_kits SET tagline = 'Cross-owner update' " + "WHERE brand_id = :brand_id" + ), + {"brand_id": brand_a}, + ) + delete_result = connection.execute( + text("DELETE FROM brand_kits WHERE brand_id = :brand_id"), + {"brand_id": brand_a}, + ) + assert update_result.rowcount == 0 + assert delete_result.rowcount == 0 + + with engine.begin() as connection: + _as_authenticated(connection, token_a) + owner_tagline = connection.execute( + text("SELECT tagline FROM brand_kits WHERE brand_id = :brand_id"), + {"brand_id": brand_a}, + ).scalar_one() + assert owner_tagline == "Updated answer" + finally: + with engine.begin() as connection: + connection.execute( + text("DELETE FROM brands WHERE id IN (:brand_a, :brand_b)"), + {"brand_a": brand_a, "brand_b": brand_b}, + ) + for user_id in (user_a_id, user_b_id): + if user_id: + client.delete( + f"{supabase_url}/auth/v1/admin/users/{user_id}", + headers={ + "apikey": supabase_key, + "Authorization": f"Bearer {supabase_key}", + }, + ) diff --git a/backend/tests/integration/test_brand_kits.py b/backend/tests/integration/test_brand_kits.py new file mode 100644 index 0000000..f157988 --- /dev/null +++ b/backend/tests/integration/test_brand_kits.py @@ -0,0 +1,364 @@ +from __future__ import annotations + +import os +import sys +from uuid import uuid4 + +import httpx +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import text + + +def _required_env(name: str) -> str: + value = os.getenv(name) + if not value: + pytest.skip(f"{name} is required for integration tests") + return value + + +def _signup_and_login( + client: httpx.Client, supabase_url: str, supabase_key: str, email: str +) -> tuple[str, str]: + signup = client.post( + f"{supabase_url}/auth/v1/signup", + headers={"apikey": supabase_key, "Content-Type": "application/json"}, + json={"email": email, "password": "12345678"}, + ) + assert signup.status_code in {200, 201} + user_id = signup.json()["user"]["id"] + try: + token = client.post( + f"{supabase_url}/auth/v1/token?grant_type=password", + headers={"apikey": supabase_key, "Content-Type": "application/json"}, + json={"email": email, "password": "12345678"}, + ) + assert token.status_code == 200 + return user_id, token.json()["access_token"] + except Exception: + try: + _delete_supabase_user(client, supabase_url, supabase_key, user_id) + except Exception: + pass + raise + + +def _delete_supabase_user( + client: httpx.Client, + supabase_url: str, + supabase_key: str, + user_id: str, +) -> None: + response = client.delete( + f"{supabase_url}/auth/v1/admin/users/{user_id}", + headers={ + "apikey": supabase_key, + "Authorization": f"Bearer {supabase_key}", + }, + ) + response.raise_for_status() + + +def test_real_supabase_create_save_read_edit_and_single_kit_row(): + supabase_url = _required_env("SUPABASE_URL") + supabase_key = _required_env("SUPABASE_SECRET_KEY") + _required_env("SUPABASE_JWT_SECRET") + _required_env("DATABASE_URL") + + from backend.app.config import get_engine + from backend.app.main import app + + user_id: str | None = None + brand_id: str | None = None + with httpx.Client(timeout=30.0) as supabase_client: + try: + user_id, access_token = _signup_and_login( + supabase_client, + supabase_url, + supabase_key, + f"brand-kit-{uuid4().hex[:12]}@example.com", + ) + headers = {"Authorization": f"Bearer {access_token}"} + with TestClient(app) as client: + create = client.post( + "/api/v1/brands", headers=headers, json={"name": "My Brand"} + ) + assert create.status_code == 201 + brand_id = create.json()["id"] + + initial = client.get( + f"/api/v1/brands/{brand_id}/kit", headers=headers + ) + empty = client.put( + f"/api/v1/brands/{brand_id}/kit", + headers=headers, + json={"name": "My Brand", "answers": {}}, + ) + assert initial.status_code == 200 + assert initial.json()["status"] == "not_started" + assert initial.json()["answers"] == { + "tagline": None, + "tone": None, + "audience": None, + "colors": [], + "avoid_words": None, + } + assert empty.status_code == 200 + assert empty.json()["status"] == "not_started" + with get_engine().connect() as connection: + assert connection.execute( + text( + "SELECT count(*) FROM brand_kits " + "WHERE brand_id = :brand_id" + ), + {"brand_id": brand_id}, + ).scalar_one() == 0 + + partial = client.put( + f"/api/v1/brands/{brand_id}/kit", + headers=headers, + json={ + "name": "My Brand", + "answers": {"tagline": "Saved before reload"}, + }, + ) + resumed = client.get( + f"/api/v1/brands/{brand_id}/kit", headers=headers + ) + assert partial.status_code == 200 + assert partial.json()["status"] == "in_progress" + assert resumed.json()["answers"]["tagline"] == "Saved before reload" + + cleared = client.put( + f"/api/v1/brands/{brand_id}/kit", + headers=headers, + json={ + "name": "My Brand", + "answers": {"tagline": None}, + }, + ) + assert cleared.status_code == 200 + assert cleared.json()["status"] == "not_started" + with get_engine().connect() as connection: + assert connection.execute( + text( + "SELECT count(*) FROM brand_kits " + "WHERE brand_id = :brand_id" + ), + {"brand_id": brand_id}, + ).scalar_one() == 0 + + saved = client.put( + f"/api/v1/brands/{brand_id}/kit", + headers=headers, + json={ + "name": "My Brand", + "answers": { + "tone": "professional", + "audience": "Small business owners aged 25-45", + "colors": ["#FF5733", "#3498DB"], + }, + }, + ) + read = client.get(f"/api/v1/brands/{brand_id}/kit", headers=headers) + edited = client.put( + f"/api/v1/brands/{brand_id}/kit", + headers=headers, + json={ + "name": "My Brand Updated", + "answers": { + "tagline": "Better work, every day", + "tone": "friendly", + "audience": "Growing teams", + "colors": ["#123456"], + "avoid_words": "cheap", + }, + }, + ) + invalid = client.put( + f"/api/v1/brands/{brand_id}/kit", + headers=headers, + json={ + "name": "My Brand Updated", + "answers": {"colors": ["invalid"]}, + }, + ) + after_invalid = client.get( + f"/api/v1/brands/{brand_id}/kit", headers=headers + ) + + assert saved.status_code == 200 + assert saved.json()["status"] == "complete" + assert read.status_code == 200 + assert read.json() == saved.json() + assert edited.status_code == 200 + assert edited.json()["brand_name"] == "My Brand Updated" + assert edited.json()["answers"]["tone"] == "friendly" + assert edited.json()["summary"].startswith("Brand: My Brand Updated\n") + assert edited.json()["completed_at"] == saved.json()["completed_at"] + assert invalid.status_code == 400 + assert after_invalid.json() == edited.json() + + with get_engine().connect() as connection: + assert connection.execute( + text("SELECT count(*) FROM brand_kits WHERE brand_id = :brand_id"), + {"brand_id": brand_id}, + ).scalar_one() == 1 + finally: + original_failure = sys.exc_info()[0] is not None + cleanup_errors: list[Exception] = [] + if brand_id: + try: + with get_engine().begin() as connection: + connection.execute( + text("DELETE FROM brands WHERE id = :brand_id"), + {"brand_id": brand_id}, + ) + except Exception as exc: + cleanup_errors.append(exc) + if user_id: + try: + _delete_supabase_user( + supabase_client, supabase_url, supabase_key, user_id + ) + except Exception as exc: + cleanup_errors.append(exc) + if cleanup_errors and not original_failure: + raise cleanup_errors[0] + + +def test_real_supabase_brand_delete_cascades_the_kit_row(): + supabase_url = _required_env("SUPABASE_URL") + supabase_key = _required_env("SUPABASE_SECRET_KEY") + _required_env("SUPABASE_JWT_SECRET") + _required_env("DATABASE_URL") + + from backend.app.config import get_engine + from backend.app.main import app + + user_id: str | None = None + brand_id: str | None = None + with httpx.Client(timeout=30.0) as supabase_client: + try: + user_id, access_token = _signup_and_login( + supabase_client, + supabase_url, + supabase_key, + f"brand-kit-delete-{uuid4().hex[:12]}@example.com", + ) + headers = {"Authorization": f"Bearer {access_token}"} + with TestClient(app) as client: + brand = client.post( + "/api/v1/brands", headers=headers, json={"name": "Delete Me"} + ) + assert brand.status_code == 201 + brand_id = brand.json()["id"] + saved = client.put( + f"/api/v1/brands/{brand_id}/kit", + headers=headers, + json={ + "name": "Delete Me", + "answers": {"tagline": "Private answer"}, + }, + ) + assert saved.status_code == 200 + deleted = client.request( + "DELETE", + f"/api/v1/brands/{brand_id}", + headers=headers, + json={"confirm_name": "Delete Me"}, + ) + + assert deleted.status_code == 204 + with get_engine().connect() as connection: + assert connection.execute( + text("SELECT count(*) FROM brand_kits WHERE brand_id = :brand_id"), + {"brand_id": brand_id}, + ).scalar_one() == 0 + finally: + original_failure = sys.exc_info()[0] is not None + if brand_id: + try: + with get_engine().begin() as connection: + connection.execute( + text("DELETE FROM brands WHERE id = :brand_id"), + {"brand_id": brand_id}, + ) + except Exception: + if not original_failure: + raise + if user_id: + _delete_supabase_user( + supabase_client, supabase_url, supabase_key, user_id + ) + + +def test_real_supabase_non_owner_cannot_read_or_update_kit(): + supabase_url = _required_env("SUPABASE_URL") + supabase_key = _required_env("SUPABASE_SECRET_KEY") + _required_env("SUPABASE_JWT_SECRET") + _required_env("DATABASE_URL") + + from backend.app.config import get_engine + from backend.app.main import app + + user_a_id: str | None = None + user_b_id: str | None = None + brand_id: str | None = None + with httpx.Client(timeout=30.0) as supabase_client: + try: + user_a_id, token_a = _signup_and_login( + supabase_client, + supabase_url, + supabase_key, + f"brand-kit-owner-{uuid4().hex[:12]}@example.com", + ) + user_b_id, token_b = _signup_and_login( + supabase_client, + supabase_url, + supabase_key, + f"brand-kit-other-{uuid4().hex[:12]}@example.com", + ) + with TestClient(app) as client: + created = client.post( + "/api/v1/brands", + headers={"Authorization": f"Bearer {token_a}"}, + json={"name": "Owner Only Kit"}, + ) + assert created.status_code == 201 + brand_id = created.json()["id"] + saved = client.put( + f"/api/v1/brands/{brand_id}/kit", + headers={"Authorization": f"Bearer {token_a}"}, + json={"name": "Owner Only Kit", "answers": {"tagline": "Secret"}}, + ) + unauthenticated = client.get(f"/api/v1/brands/{brand_id}/kit") + non_owner_get = client.get( + f"/api/v1/brands/{brand_id}/kit", + headers={"Authorization": f"Bearer {token_b}"}, + ) + non_owner_put = client.put( + f"/api/v1/brands/{brand_id}/kit", + headers={"Authorization": f"Bearer {token_b}"}, + json={"name": "Leaked Name", "answers": {"tagline": "Leaked"}}, + ) + + assert saved.status_code == 200 + assert unauthenticated.status_code == 401 + for response in (non_owner_get, non_owner_put): + assert response.status_code == 404 + assert response.json()["error"]["code"] == "BRAND_NOT_FOUND" + assert "Secret" not in response.text + assert "Owner Only Kit" not in response.text + finally: + if brand_id: + with get_engine().begin() as connection: + connection.execute( + text("DELETE FROM brands WHERE id = :brand_id"), + {"brand_id": brand_id}, + ) + for user_id in (user_a_id, user_b_id): + if user_id: + _delete_supabase_user( + supabase_client, supabase_url, supabase_key, user_id + ) diff --git a/backend/tests/unit/test_brand_kit_store.py b/backend/tests/unit/test_brand_kit_store.py new file mode 100644 index 0000000..fab1f1d --- /dev/null +++ b/backend/tests/unit/test_brand_kit_store.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +from backend.app.models.brand_kit import BrandKitUpsert +from backend.app.services.brand_kit_store import BrandKitStore, derive_summary + + +def test_derive_summary_is_deterministic_and_includes_all_answers(): + answers = { + "tagline": " Innovation for everyone ", + "tone": "professional", + "audience": " Small business owners aged 25-45 ", + "colors": ["#ff5733", "#3498db"], + "avoid_words": " cheap, discount ", + } + + assert derive_summary(" My Brand ", answers) == "\n".join( + ( + "Brand: My Brand", + "Tagline: Innovation for everyone", + "Tone: professional", + "Audience: Small business owners aged 25-45", + "Colors: #FF5733, #3498DB", + "Avoid words: cheap, discount", + ) + ) + + +def test_derive_summary_uses_none_specified_for_omitted_optional_answers(): + assert derive_summary( + "My Brand", + { + "tone": "professional", + "audience": "Small business owners aged 25-45", + "colors": ["#FF5733"], + }, + ) == "\n".join( + ( + "Brand: My Brand", + "Tagline: None specified", + "Tone: professional", + "Audience: Small business owners aged 25-45", + "Colors: #FF5733", + "Avoid words: None specified", + ) + ) + + +def test_merge_answers_preserves_omitted_values_and_clears_explicit_values(): + existing = { + "tagline": "Keep this", + "tone": "friendly", + "audience": "Small teams", + "colors": ["#123456"], + "avoid_words": "cheap", + } + + preserved = BrandKitStore._merged_answers( + existing, + BrandKitUpsert.model_validate({"name": "My Brand", "answers": {"tone": "formal"}}), + ) + cleared = BrandKitStore._merged_answers( + existing, + BrandKitUpsert.model_validate( + {"name": "My Brand", "answers": {"tagline": None, "colors": []}} + ), + ) + + assert preserved.tagline == "Keep this" + assert preserved.tone.value == "formal" + assert cleared.tagline is None + assert cleared.colors == [] + + +def test_saved_answer_detection_distinguishes_lifecycle_states(): + assert not BrandKitStore._has_saved_answer( + BrandKitUpsert.model_validate({"name": "My Brand"}).answers + ) + assert BrandKitStore._has_saved_answer( + BrandKitUpsert.model_validate( + {"name": "My Brand", "answers": {"tagline": "Saved"}} + ).answers + ) + + +def test_status_for_answers_covers_not_started_partial_and_complete(): + empty = BrandKitUpsert.model_validate({"name": "My Brand"}).answers + partial = BrandKitUpsert.model_validate( + {"name": "My Brand", "answers": {"audience": "Small teams"}} + ).answers + complete = BrandKitUpsert.model_validate( + { + "name": "My Brand", + "answers": { + "tone": "friendly", + "audience": "Small teams", + "colors": ["#123456"], + }, + } + ).answers + + assert BrandKitStore._status_for_answers(empty).value == "not_started" + assert BrandKitStore._status_for_answers(partial).value == "in_progress" + assert BrandKitStore._status_for_answers(complete).value == "complete" diff --git a/frontend/app/(dashboard)/brands/[brandId]/kit/page.tsx b/frontend/app/(dashboard)/brands/[brandId]/kit/page.tsx new file mode 100644 index 0000000..c404e6e --- /dev/null +++ b/frontend/app/(dashboard)/brands/[brandId]/kit/page.tsx @@ -0,0 +1,426 @@ +"use client"; + +import Link from "next/link"; +import { useParams, useRouter } from "next/navigation"; +import { useEffect, useRef, useState, type FormEvent } from "react"; + +import { getPublicEnv } from "@/lib/runtime-env"; +import { supabase } from "@/lib/supabase/client"; + +type Tone = "formal" | "casual" | "playful" | "professional" | "friendly"; +type KitStatus = "not_started" | "in_progress" | "complete"; + +type FormValues = { + name: string; + tagline: string; + tone: Tone | ""; + audience: string; + colors: string; + avoidWords: string; +}; + +type KitResponse = { + brand_id: string; + brand_name: string; + answers: { + tagline: string | null; + tone: Tone | null; + audience: string | null; + colors: string[]; + avoid_words: string | null; + }; + summary: string | null; + status: KitStatus; + completed_at: string | null; + updated_at: string | null; +}; + +type ApiError = { error?: { message?: string } }; + +const STEPS = ["Name", "Tagline", "Tone", "Audience", "Colors", "Avoid words"] as const; + +const EMPTY_FORM: FormValues = { + name: "", + tagline: "", + tone: "", + audience: "", + colors: "", + avoidWords: "", +}; + +function colorsFromInput(value: string) { + return value + .split(",") + .map((color) => color.trim()) + .filter(Boolean); +} + +function validateStep(step: number, values: FormValues): string | null { + switch (step) { + case 0: + return values.name.trim().length < 2 || values.name.trim().length > 120 + ? "Name must be between 2 and 120 characters." + : null; + case 1: + return values.tagline.length > 160 ? "Tagline must be 160 characters or fewer." : null; + case 2: + return values.tone ? null : "Choose a tone."; + case 3: + return values.audience.trim().length < 2 || values.audience.trim().length > 500 + ? "Audience must be between 2 and 500 characters." + : null; + case 4: { + const colors = colorsFromInput(values.colors); + if (colors.length < 1 || colors.length > 3) { + return "Enter between 1 and 3 colors, separated by commas."; + } + if (colors.some((color) => !/^#[0-9a-fA-F]{6}$/.test(color))) { + return "Colors must use the #RRGGBB format."; + } + return null; + } + case 5: + return null; + default: + return null; + } +} + +function apiErrorMessage(body: ApiError | null, fallback: string) { + return body?.error?.message ?? fallback; +} + +export default function BrandKitPage() { + const { brandId } = useParams<{ brandId: string }>(); + const router = useRouter(); + const apiBase = getPublicEnv("NEXT_PUBLIC_API_URL"); + const [values, setValues] = useState(EMPTY_FORM); + const [step, setStep] = useState(0); + const [status, setStatus] = useState("not_started"); + const [summary, setSummary] = useState(null); + const [showSummary, setShowSummary] = useState(false); + const [brandName, setBrandName] = useState(""); + const [isLoading, setIsLoading] = useState(true); + const [isSaving, setIsSaving] = useState(false); + const [error, setError] = useState(null); + const [validationError, setValidationError] = useState(null); + const [saveState, setSaveState] = useState<"idle" | "saved">("idle"); + const hasLocalEdits = useRef(false); + + useEffect(() => { + let active = true; + + async function loadKit() { + try { + const { data } = await supabase.auth.getSession(); + const session = data.session; + if (!session) { + router.push("/login"); + return; + } + + const response = await fetch( + `${apiBase}/v1/brands/${encodeURIComponent(brandId)}/kit`, + { headers: { Authorization: `Bearer ${session.access_token}` } } + ); + const body = (await response.json().catch(() => null)) as KitResponse | ApiError | null; + if (!response.ok) { + throw new Error(apiErrorMessage(body as ApiError | null, "Unable to load the Brand Kit.")); + } + + const kit = body as KitResponse; + if (active) { + setStatus(kit.status); + setBrandName(kit.brand_name); + if (!hasLocalEdits.current) { + setValues({ + name: kit.brand_name, + tagline: kit.answers.tagline ?? "", + tone: kit.answers.tone ?? "", + audience: kit.answers.audience ?? "", + colors: kit.answers.colors.join(", "), + avoidWords: kit.answers.avoid_words ?? "", + }); + setSummary(kit.summary); + setShowSummary(kit.status === "complete" && kit.summary !== null); + if (kit.status === "in_progress") { + const firstIncomplete = [0, 2, 3, 4].find((index) => { + return validateStep(index, { + name: kit.brand_name, + tagline: kit.answers.tagline ?? "", + tone: kit.answers.tone ?? "", + audience: kit.answers.audience ?? "", + colors: kit.answers.colors.join(", "), + avoidWords: kit.answers.avoid_words ?? "", + }); + }); + setStep(firstIncomplete ?? 0); + } + } + } + } catch (loadError) { + if (active) { + setError(loadError instanceof Error ? loadError.message : "Unable to load the Brand Kit."); + } + } finally { + if (active) setIsLoading(false); + } + } + + void loadKit(); + return () => { + active = false; + }; + }, [apiBase, brandId, router]); + + function updateValue(field: keyof FormValues, value: string) { + hasLocalEdits.current = true; + setValues((current) => ({ ...current, [field]: value })); + setValidationError(null); + setError(null); + setSaveState("idle"); + } + + async function nextStep() { + const message = validateStep(step, values); + if (message) { + setValidationError(message); + return; + } + setValidationError(null); + const saved = await saveKit(false); + if (saved) setStep((current) => Math.min(current + 1, STEPS.length - 1)); + } + + function previousStep() { + setValidationError(null); + setStep((current) => Math.max(current - 1, 0)); + } + + async function saveKit(complete: boolean, event?: FormEvent) { + event?.preventDefault(); + if (complete) { + for (let index = 0; index < STEPS.length; index += 1) { + const message = validateStep(index, values); + if (message) { + setStep(index); + setValidationError(message); + return false; + } + } + } else { + const message = validateStep(step, values); + if (message) { + setValidationError(message); + return false; + } + } + + setValidationError(null); + setError(null); + setIsSaving(true); + setSaveState("idle"); + try { + const { data } = await supabase.auth.getSession(); + const session = data.session; + if (!session) { + router.push("/login"); + return false; + } + + const response = await fetch( + `${apiBase}/v1/brands/${encodeURIComponent(brandId)}/kit`, + { + method: "PUT", + headers: { + Authorization: `Bearer ${session.access_token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + name: values.name.trim(), + answers: { + tagline: values.tagline.trim() || null, + tone: values.tone, + audience: values.audience.trim(), + colors: colorsFromInput(values.colors).map((color) => color.toUpperCase()), + avoid_words: values.avoidWords.trim() || null, + }, + }), + } + ); + const body = (await response.json().catch(() => null)) as KitResponse | ApiError | null; + if (!response.ok) { + setError(apiErrorMessage(body as ApiError | null, "Unable to save the Brand Kit.")); + return false; + } + + const kit = body as KitResponse; + setStatus(kit.status); + setBrandName(kit.brand_name); + setSummary(kit.summary); + setShowSummary(complete && kit.status === "complete" && kit.summary !== null); + setSaveState("saved"); + hasLocalEdits.current = false; + window.dispatchEvent(new Event("postforge:brands-changed")); + return true; + } catch { + setError("Unable to save the Brand Kit. Try again."); + return false; + } finally { + setIsSaving(false); + } + } + + if (isLoading) { + return

Loading Brand Kit...

; + } + + if (error && !brandName) { + return

{error}

; + } + + if (showSummary && summary) { + return ( +
+ + Back to {brandName} + +
+
+
+

Brand Kit

+

Complete

+
+ + Complete + +
+ 
+            {summary}
+
+ +
+
+ ); + } + + const fieldProps = { + value: values, + updateValue, + disabled: isSaving, + }; + + return ( +
+
+ + Back to brand + +

Brand Kit

+

Build your brand identity

+

Step {step + 1} of {STEPS.length}

+
+ +
+ {STEPS.map((stepName, index) => ( +
+ ))} +
+ +
void saveKit(true, event)} noValidate> +

{STEPS[step]}

+
+ {step === 0 ? ( + + ) : null} + {step === 1 ? ( + + ) : null} + {step === 2 ? ( + + ) : null} + {step === 3 ? ( +