diff --git a/alembic/versions/0072_design_systems.py b/alembic/versions/0072_design_systems.py new file mode 100644 index 0000000..80e467c --- /dev/null +++ b/alembic/versions/0072_design_systems.py @@ -0,0 +1,138 @@ +"""design systems + tokens, and the project pointer + +Revision ID: 0072 +Revises: 0071 +Create Date: 2026-07-30 + +Makes the design system a first-class record instead of prose in a rulebook: a +named set of tokens with an optional parent, so a family system holds the house +style and an app system holds only what it changes. + +`design_systems.parent_id` is the whole model. It replaces both an `always_on` +flag (a family system is one with no parent) and a subscription join table (a +project points at ONE system, and the chain supplies the rest), which is less +schema than the rulebook shape it mirrors. + +Two deliberate choices worth stating here rather than leaving to be re-derived: + + - **`design_tokens.value_by_mode` is JSONB keyed by mode**, not `value_light` + + `value_dark` columns. In a child system an unset mode means "inherit"; in a + root it would mean "not mode-dependent", and as columns both are NULL and + indistinguishable. As a map, resolution is a dict merge at every level with + no special case for roots — and a third mode (high-contrast, print) is data + rather than a schema change. The cost is that a typo'd mode key is not + rejected by the database. Nothing filters tokens by value in SQL, so the + queryability the columns would have bought is for a query no caller makes. + (Named `value_by_mode` rather than `values`, which is reserved in SQL.) + - **`group_name` is free text, not a CHECK enum.** Groupings are each design + system's own vocabulary; a whitelist would bake one install's kit into the + schema. No CHECK is introduced anywhere in this migration. + +`parent_id` and `projects.design_system_id` are both ON DELETE SET NULL. Deleting +a family system must orphan its children into roots that still hold their own +overrides, not cascade away every app system that inherited from it; deleting a +system a project points at must unstyle that project, not delete it. + +Downgrade drops both tables and the column. Any design system defined this way +is lost — this is the migration that introduces the concept, so there is no +earlier representation to fall back to. +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB + + +revision = "0072" +down_revision = "0071" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "design_systems", + sa.Column("id", sa.BigInteger(), primary_key=True), + sa.Column( + "owner_user_id", sa.BigInteger(), + sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False, + ), + sa.Column("title", sa.Text(), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column( + "parent_id", sa.BigInteger(), + sa.ForeignKey("design_systems.id", ondelete="SET NULL"), nullable=True, + ), + sa.Column( + "created_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.text("now()"), + ), + sa.Column( + "updated_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.text("now()"), + ), + sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("deleted_batch_id", sa.Text(), nullable=True), + ) + op.create_index( + "ix_design_systems_owner_user_id", "design_systems", ["owner_user_id"] + ) + op.create_index("ix_design_systems_parent_id", "design_systems", ["parent_id"]) + + op.create_table( + "design_tokens", + sa.Column("id", sa.BigInteger(), primary_key=True), + sa.Column( + "design_system_id", sa.BigInteger(), + sa.ForeignKey("design_systems.id", ondelete="CASCADE"), nullable=False, + ), + sa.Column("name", sa.Text(), nullable=False), + # NOT NULL with a '{}' default: a nullable JSONB column has two empty + # states (SQL NULL and JSON null) and every reader has to test for both. + sa.Column( + "value_by_mode", JSONB, + nullable=False, server_default=sa.text("'{}'::jsonb"), + ), + sa.Column("group_name", sa.Text(), nullable=True), + sa.Column("purpose", sa.Text(), nullable=True), + sa.Column("order_index", sa.Integer(), nullable=False, server_default="0"), + sa.Column( + "created_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.text("now()"), + ), + sa.Column( + "updated_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.text("now()"), + ), + sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("deleted_batch_id", sa.Text(), nullable=True), + ) + op.create_index( + "ix_design_tokens_design_system_id", "design_tokens", ["design_system_id"] + ) + # Partial unique: a name is unique among LIVE tokens in a system. Two live + # rows with the same name are a duplicate definition and the cascade would + # pick between them arbitrarily; a trashed row must not block reusing its + # name. + op.create_index( + "uq_token_per_design_system", "design_tokens", ["design_system_id", "name"], + unique=True, postgresql_where=sa.text("deleted_at IS NULL"), + ) + + op.add_column( + "projects", sa.Column("design_system_id", sa.BigInteger(), nullable=True) + ) + op.create_foreign_key( + "fk_projects_design_system_id", "projects", "design_systems", + ["design_system_id"], ["id"], ondelete="SET NULL", + ) + + +def downgrade() -> None: + op.drop_constraint("fk_projects_design_system_id", "projects", type_="foreignkey") + op.drop_column("projects", "design_system_id") + op.drop_index("uq_token_per_design_system", table_name="design_tokens") + op.drop_index("ix_design_tokens_design_system_id", table_name="design_tokens") + op.drop_table("design_tokens") + op.drop_index("ix_design_systems_parent_id", table_name="design_systems") + op.drop_index("ix_design_systems_owner_user_id", table_name="design_systems") + op.drop_table("design_systems") diff --git a/src/scribe/models/__init__.py b/src/scribe/models/__init__.py index 2297cdd..f4528a8 100644 --- a/src/scribe/models/__init__.py +++ b/src/scribe/models/__init__.py @@ -43,3 +43,6 @@ from scribe.models.rulebook import ( # noqa: E402, F401 ) from scribe.models.repo_binding import RepoBinding # noqa: E402, F401 from scribe.models.system import System, RecordSystem # noqa: E402, F401 +from scribe.models.design_system import ( # noqa: E402, F401 + BASE_MODE, DesignSystem, DesignToken, +) diff --git a/src/scribe/models/design_system.py b/src/scribe/models/design_system.py new file mode 100644 index 0000000..7c91a02 --- /dev/null +++ b/src/scribe/models/design_system.py @@ -0,0 +1,130 @@ +"""Design systems — a stylesheet held as records, inherited family -> app. + +A DesignSystem is a named set of design tokens with an OPTIONAL parent, and that +single self-FK is the whole model. A system with no parent is a family system; a +system WITH one holds only what it changes. "What does this app alter?" is +therefore `list its tokens` — nothing to compute, nothing to diff — which is why +inheritance won over a flat family-plus-loose-overrides shape. + +Resolution walks the chain and lets the deepest system win by token name. That +is the CSS cascade rather than an analogy to it, which is why the storage model +and the stylesheet model come out the same shape. + +`parent_id` also replaces two things the rulebook model needs to express the same +idea: an `always_on` flag (a family system is simply one with no parent) and a +subscription join table (a project points at ONE system, and the chain supplies +the rest). Less schema for more structure. +""" +from sqlalchemy import BigInteger, ForeignKey, Index, Integer, Text, text +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from scribe.models import Base +from scribe.models.base import SoftDeleteMixin, TimestampMixin + +# The mode key that applies when no more specific one does. Emitted CSS puts it +# on the base selector and every other key on a mode selector, mirroring how a +# stylesheet is actually written: light on `:root`, dark layered over it. +BASE_MODE = "base" + + +class DesignSystem(Base, TimestampMixin, SoftDeleteMixin): + __tablename__ = "design_systems" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + owner_user_id: Mapped[int] = mapped_column( + BigInteger, ForeignKey("users.id", ondelete="CASCADE"), index=True + ) + title: Mapped[str] = mapped_column(Text) + description: Mapped[str | None] = mapped_column(Text, nullable=True) + # SET NULL, not CASCADE: deleting a family system must not delete every app + # system that inherited from it. Orphaning turns each child into a root that + # still holds its own overrides — recoverable. A cascade would destroy data + # the operator never asked to touch. + parent_id: Mapped[int | None] = mapped_column( + BigInteger, + ForeignKey("design_systems.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + + def to_dict(self) -> dict: + return { + "id": self.id, + "owner_user_id": self.owner_user_id, + "title": self.title, + "description": self.description or "", + "parent_id": self.parent_id, + "created_at": self.created_at.isoformat() if self.created_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + } + + +class DesignToken(Base, TimestampMixin, SoftDeleteMixin): + """One custom property in one system: its name, and its value per mode. + + `value_by_mode` is a JSONB map of mode -> value, e.g. + `{"base": "#f7f5ef", "dark": "#14171a"}`. Three reasons it beats a pair of + `value_light` / `value_dark` columns here: + + - **Absence means one thing.** In a child system an unset mode means + "inherit"; in a root it would have to mean "not mode-dependent". With + columns those are both NULL and the resolver cannot tell them apart. With + a map, resolution is `{**parent_map, **child_map}` at every level — + one rule, no special case for roots. + - **Per-mode overrides are already real.** A palette rule in this operator's + own kit deepens one accent on light backgrounds for contrast while + leaving the dark value alone. Mode is a second override axis, not a + second column. + - **Nothing filters tokens by value in SQL.** Drift comparison resolves the + set first and compares in the client; the importer diffs in Python. The + queryability columns would buy is for a query no caller makes. + + The cost is real — a third mode is data rather than schema, so the DB will + not reject a typo'd mode key. That is the trade taken. + + NOT NULL with a `{}` default deliberately: a JSONB column otherwise has two + empty states (SQL NULL and JSON null) and code has to test for both. + """ + __tablename__ = "design_tokens" + __table_args__ = ( + # Partial unique: a name is unique among LIVE tokens in a system, so a + # trashed token doesn't block recreating the same name. Two live rows + # named `--fs-obsidian` in one system is a duplicate definition, and the + # cascade would pick between them arbitrarily. + Index( + "uq_token_per_design_system", "design_system_id", "name", + unique=True, postgresql_where=text("deleted_at IS NULL"), + ), + ) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + design_system_id: Mapped[int] = mapped_column( + BigInteger, ForeignKey("design_systems.id", ondelete="CASCADE"), index=True + ) + name: Mapped[str] = mapped_column(Text) + # Named for what it holds rather than the bare word `values`, which is + # reserved in SQL — the same reason `group_name` is not `group`. + value_by_mode: Mapped[dict] = mapped_column( + JSONB, nullable=False, default=dict, server_default=text("'{}'::jsonb") + ) + # `group` is a reserved word in SQL; `group_name` throughout — model, column + # and payload — so no layer has to remember which spelling it is on. + # Free text, not a CHECK enum: groupings are the design system's own + # vocabulary, and a whitelist would bake one install's kit into the schema. + group_name: Mapped[str | None] = mapped_column(Text, nullable=True) + purpose: Mapped[str | None] = mapped_column(Text, nullable=True) + order_index: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + + def to_dict(self) -> dict: + return { + "id": self.id, + "design_system_id": self.design_system_id, + "name": self.name, + "value_by_mode": self.value_by_mode or {}, + "group_name": self.group_name, + "purpose": self.purpose, + "order_index": self.order_index, + "created_at": self.created_at.isoformat() if self.created_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + } diff --git a/src/scribe/models/project.py b/src/scribe/models/project.py index 8ccacc5..d05a165 100644 --- a/src/scribe/models/project.py +++ b/src/scribe/models/project.py @@ -1,5 +1,5 @@ import enum -from sqlalchemy import ForeignKey, Integer, Text +from sqlalchemy import BigInteger, ForeignKey, Integer, Text from sqlalchemy.orm import Mapped, mapped_column from scribe.models import Base from scribe.models.base import TimestampMixin, SoftDeleteMixin @@ -21,6 +21,12 @@ class Project(Base, TimestampMixin, SoftDeleteMixin): goal: Mapped[str] = mapped_column(Text, default="") status: Mapped[str] = mapped_column(Text, default="active") color: Mapped[str | None] = mapped_column(Text, nullable=True) # hex color + # The design system this project's UI is built from, or NULL. NULL is the + # ordinary state, not a degraded one — most installs have no design system + # at all and nothing may assume one exists. + design_system_id: Mapped[int | None] = mapped_column( + BigInteger, ForeignKey("design_systems.id", ondelete="SET NULL"), nullable=True + ) def to_dict(self) -> dict: return { @@ -31,6 +37,7 @@ class Project(Base, TimestampMixin, SoftDeleteMixin): "goal": self.goal, "status": self.status, "color": self.color, + "design_system_id": self.design_system_id, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), } diff --git a/src/scribe/routes/design.py b/src/scribe/routes/design.py index 33b3146..630e710 100644 --- a/src/scribe/routes/design.py +++ b/src/scribe/routes/design.py @@ -7,7 +7,7 @@ resolved. This endpoint supplies the claims to check them against. from quart import Blueprint, jsonify from scribe.auth import get_current_user_id, login_required -from scribe.services import design_system as design_svc +from scribe.services import design_rulebook_import as design_svc design_bp = Blueprint("design", __name__, url_prefix="/api/design") diff --git a/src/scribe/services/access.py b/src/scribe/services/access.py index bb39d61..2a43374 100644 --- a/src/scribe/services/access.py +++ b/src/scribe/services/access.py @@ -12,11 +12,13 @@ import logging from sqlalchemy import or_, select from scribe.models import async_session +from scribe.models.design_system import DesignSystem from scribe.models.group import GroupMembership from scribe.models.note import Note from scribe.models.project import Project from scribe.models.share import NoteShare, ProjectShare from scribe.models.user import User +from scribe.services.design_cascade import ancestry logger = logging.getLogger(__name__) @@ -164,6 +166,88 @@ async def can_write_note(user_id: int, note_id: int) -> bool: return perm in ("editor", "admin", "owner") +# --------------------------------------------------------------------------- +# Design-system permissions +# --------------------------------------------------------------------------- + +async def get_design_system_permission( + user_id: int, design_system_id: int +) -> str | None: + """Effective permission on a design system, or None. + + Two ways in, and the asymmetry between them is the point: + + - **Owning it** grants "owner" — full read and write. + - **Reaching it through a project you can see** grants "viewer", and only + ever "viewer". Being an editor on a shared project must NOT confer the + right to rewrite the family system that project inherits from: one + project's collaborator would be editing tokens every other project in + the family resolves through. Editing a design system stays the owner's + act, and it is the same reasoning that keeps a rulebook owner-scoped. + + Reachability follows the parent chain UPWARD. Rendering a project's UI means + resolving its whole chain, so read access to a system implies read access to + its ancestors — otherwise a shared project would resolve to a truncated + cascade and silently render with the wrong values. + """ + async with async_session() as session: + system = await session.get(DesignSystem, design_system_id) + if system is None or system.deleted_at is not None: + return None + if system.owner_user_id == user_id: + return "owner" + + shared_project_ids = select(ProjectShare.project_id).where( + or_( + ProjectShare.shared_with_user_id == user_id, + ProjectShare.shared_with_group_id.in_(_my_group_ids(user_id)), + ) + ) + entry_points = set( + ( + await session.execute( + select(Project.design_system_id).where( + Project.design_system_id.is_not(None), + Project.deleted_at.is_(None), + or_( + Project.user_id == user_id, + Project.id.in_(shared_project_ids), + ), + ) + ) + ).scalars().all() + ) + if not entry_points: + return None + + # One narrow query for the whole forest's shape. Design systems are a + # handful of rows per install — a family and one per app — so walking + # from each entry point in memory beats a recursive CTE per check. + parents = dict( + ( + await session.execute( + select(DesignSystem.id, DesignSystem.parent_id).where( + DesignSystem.deleted_at.is_(None) + ) + ) + ).all() + ) + + for entry in entry_points: + if design_system_id in ancestry(entry, parents): + return "viewer" + return None + + +async def can_read_design_system(user_id: int, design_system_id: int) -> bool: + return (await get_design_system_permission(user_id, design_system_id)) is not None + + +async def can_write_design_system(user_id: int, design_system_id: int) -> bool: + perm = await get_design_system_permission(user_id, design_system_id) + return perm in ("editor", "admin", "owner") + + # --------------------------------------------------------------------------- # Set-based visibility (for LIST queries) # diff --git a/src/scribe/services/design_cascade.py b/src/scribe/services/design_cascade.py new file mode 100644 index 0000000..8d7f187 --- /dev/null +++ b/src/scribe/services/design_cascade.py @@ -0,0 +1,61 @@ +"""The design-system cascade, as pure functions over already-loaded rows. + +Deliberately free of every database and service import — including +`services/access.py`, which needs `ancestry` to answer "can this caller read +this system?" and would otherwise form an import cycle with the service that +needs `access` back. A module that imports nothing can be imported by both. + +Being pure is also what makes the cascade rule testable without a database: +these take a plain `{id: parent_id}` map, so a test states the shape of a +hierarchy in one literal instead of building one. +""" +from collections.abc import Mapping + + +def ancestry(system_id: int, parents: Mapping[int, int | None]) -> list[int]: + """The inheritance chain from `system_id` up to its root, nearest first. + + Includes `system_id` itself at index 0, because resolution wants + deepest-to-shallowest and the system being resolved is the deepest link. + + A system missing from `parents` terminates the chain rather than raising: a + parent whose row was soft-deleted or filtered out is a truncated chain, not + a failed request. + + The visited-set is defensive, not the primary guard — writes already refuse + to create a cycle (`would_cycle`). It is here because a loop introduced by a + direct DB edit or a future bug must degrade to a truncated chain instead of + spinning forever. Truncation shows up in the result; a hang shows up as an + outage. + """ + chain: list[int] = [] + seen: set[int] = set() + current: int | None = system_id + while current is not None and current not in seen: + seen.add(current) + chain.append(current) + current = parents.get(current) + return chain + + +def would_cycle( + system_id: int, + proposed_parent_id: int | None, + parents: Mapping[int, int | None], +) -> bool: + """Would making `proposed_parent_id` the parent of `system_id` close a loop? + + True when the proposed parent IS the system, or already inherits from it. + + The walk goes UP from the proposed parent, which is the cheap direction — + each system has at most one parent, so the chain is a line. Asking the + equivalent downward question ("is the proposed parent among my + descendants?") would mean searching a whole forest for the same answer. + + `proposed_parent_id=None` clears the parent and can never cycle. + """ + if proposed_parent_id is None: + return False + if proposed_parent_id == system_id: + return True + return system_id in ancestry(proposed_parent_id, parents) diff --git a/src/scribe/services/design_system.py b/src/scribe/services/design_rulebook_import.py similarity index 100% rename from src/scribe/services/design_system.py rename to src/scribe/services/design_rulebook_import.py diff --git a/src/scribe/services/design_systems.py b/src/scribe/services/design_systems.py new file mode 100644 index 0000000..4588d5c --- /dev/null +++ b/src/scribe/services/design_systems.py @@ -0,0 +1,271 @@ +"""Design system + token persistence, and the guard on the parent chain. + +Access goes through `services/access.py` (rule #78), where reaching a system via +a project you can see grants READ but never write — see +`get_design_system_permission` for why that asymmetry exists. + +The cascade itself lives in `services/design_cascade.py` as pure functions. This +module is the part that needs a database: loading the shape of the hierarchy, +refusing writes that would break it, and storing tokens. + +Nothing here seeds or implies a default system. An install with no design +systems is an ordinary install, and every caller must handle an empty list as +the normal case rather than a missing prerequisite. +""" +import logging +from datetime import datetime, timezone + +from sqlalchemy import select + +from scribe.models import async_session +from scribe.models.design_system import DesignSystem, DesignToken +from scribe.models.project import Project +from scribe.services import access +from scribe.services.design_cascade import would_cycle + +logger = logging.getLogger(__name__) + + +class DesignSystemCycle(ValueError): + """A parent assignment that would close an inheritance loop. + + Raised rather than returned as None because the two outcomes need different + answers: None already means "not found, or not yours", and a caller that + conflated them would show "no such design system" for what is really "that + parent is one of its own descendants". Routes map this to a 400. + """ + + +async def _parent_map(session, user_id: int) -> dict[int, int | None]: + """`{id: parent_id}` for the caller's live systems — the hierarchy's shape. + + Owner-scoped, which is also the constraint on parenting: you can only build + a chain out of systems you own. A borrowed link would let someone else's + delete or re-parent silently restyle your app. + """ + rows = ( + await session.execute( + select(DesignSystem.id, DesignSystem.parent_id).where( + DesignSystem.owner_user_id == user_id, + DesignSystem.deleted_at.is_(None), + ) + ) + ).all() + return dict(rows) + + +# --- design systems --------------------------------------------------------- + +async def create_design_system( + user_id: int, + title: str, + description: str | None = None, + parent_id: int | None = None, +) -> DesignSystem | None: + """Create a system, with or without a parent. + + Returns None when `parent_id` names a system the caller may not write — + which, per the ACL, means one they do not own. + """ + if parent_id is not None and not await access.can_write_design_system( + user_id, parent_id + ): + return None + async with async_session() as session: + system = DesignSystem( + owner_user_id=user_id, + title=title.strip(), + description=description, + parent_id=parent_id, + ) + session.add(system) + await session.commit() + await session.refresh(system) + return system + + +async def get_design_system(user_id: int, design_system_id: int) -> DesignSystem | None: + async with async_session() as session: + system = await session.get(DesignSystem, design_system_id) + if system is None or system.deleted_at is not None: + return None + if not await access.can_read_design_system(user_id, design_system_id): + return None + return system + + +async def list_design_systems(user_id: int) -> list[DesignSystem]: + """The caller's own systems, ordered by title. Empty is normal.""" + async with async_session() as session: + rows = await session.execute( + select(DesignSystem) + .where( + DesignSystem.owner_user_id == user_id, + DesignSystem.deleted_at.is_(None), + ) + .order_by(DesignSystem.title) + ) + return list(rows.scalars().all()) + + +async def update_design_system( + user_id: int, design_system_id: int, **fields: object +) -> DesignSystem | None: + """Partial update. Raises DesignSystemCycle if `parent_id` would loop. + + `parent_id` is handled apart from the other fields because None is a + meaningful value for it — "make this a root" — where for every other field + None means "leave alone". Callers signal it by passing the key at all. + """ + if not await access.can_write_design_system(user_id, design_system_id): + return None + async with async_session() as session: + system = await session.get(DesignSystem, design_system_id) + if system is None or system.deleted_at is not None: + return None + + if "parent_id" in fields: + parent_id = fields.pop("parent_id") + if parent_id is not None: + parent_id = int(parent_id) + if not await access.can_write_design_system(user_id, parent_id): + return None + if would_cycle( + design_system_id, parent_id, await _parent_map(session, user_id) + ): + raise DesignSystemCycle( + f"Design system {design_system_id} cannot inherit from " + f"{parent_id}: that system already inherits from it." + ) + system.parent_id = parent_id + + for key, value in fields.items(): + if key in ("title", "description") and value is not None: + setattr(system, key, value) + system.updated_at = datetime.now(timezone.utc) + await session.commit() + await session.refresh(system) + return system + + +async def delete_design_system(user_id: int, design_system_id: int) -> bool: + """Soft-delete a system. Children survive as roots (the FK is SET NULL).""" + if not await access.can_write_design_system(user_id, design_system_id): + return False + async with async_session() as session: + system = await session.get(DesignSystem, design_system_id) + if system is None or system.deleted_at is not None: + return False + system.deleted_at = datetime.now(timezone.utc) + await session.commit() + return True + + +# --- tokens ----------------------------------------------------------------- + +async def create_token( + user_id: int, + design_system_id: int, + name: str, + value_by_mode: dict | None = None, + group_name: str | None = None, + purpose: str | None = None, + order_index: int = 0, +) -> DesignToken | None: + if not await access.can_write_design_system(user_id, design_system_id): + return None + async with async_session() as session: + token = DesignToken( + design_system_id=design_system_id, + name=name.strip(), + # `or {}` and not the argument as given: the column is NOT NULL so + # that absence has exactly one spelling. Passing None here would + # otherwise store JSON null and reintroduce the second empty state. + value_by_mode=value_by_mode or {}, + group_name=group_name, + purpose=purpose, + order_index=order_index, + ) + session.add(token) + await session.commit() + await session.refresh(token) + return token + + +async def list_tokens(user_id: int, design_system_id: int) -> list[DesignToken]: + """One system's OWN tokens — its override set, not its effective set. + + Resolving the chain is step 2's job; this deliberately answers the narrower + question ("what does this system change?") that the model exists to make + free. + """ + if not await access.can_read_design_system(user_id, design_system_id): + return [] + async with async_session() as session: + rows = await session.execute( + select(DesignToken) + .where( + DesignToken.design_system_id == design_system_id, + DesignToken.deleted_at.is_(None), + ) + .order_by(DesignToken.order_index.asc(), DesignToken.name.asc()) + ) + return list(rows.scalars().all()) + + +async def update_token( + user_id: int, token_id: int, **fields: object +) -> DesignToken | None: + allowed = {"name", "value_by_mode", "group_name", "purpose", "order_index"} + async with async_session() as session: + token = await session.get(DesignToken, token_id) + if token is None or token.deleted_at is not None: + return None + if not await access.can_write_design_system(user_id, token.design_system_id): + return None + for key, value in fields.items(): + if key in allowed and value is not None: + setattr(token, key, value) + token.updated_at = datetime.now(timezone.utc) + await session.commit() + await session.refresh(token) + return token + + +async def delete_token(user_id: int, token_id: int) -> bool: + async with async_session() as session: + token = await session.get(DesignToken, token_id) + if token is None or token.deleted_at is not None: + return False + if not await access.can_write_design_system(user_id, token.design_system_id): + return False + token.deleted_at = datetime.now(timezone.utc) + await session.commit() + return True + + +# --- the project pointer ---------------------------------------------------- + +async def set_project_design_system( + user_id: int, project_id: int, design_system_id: int | None +) -> bool: + """Point a project at a design system, or at nothing (None clears it). + + Needs write on the project and READ on the system: pointing at a system is + consuming it, not changing it, so a system shared with you through another + project is a legitimate choice here. + """ + if not await access.can_write_project(user_id, project_id): + return False + if design_system_id is not None and not await access.can_read_design_system( + user_id, design_system_id + ): + return False + async with async_session() as session: + project = await session.get(Project, project_id) + if project is None or project.deleted_at is not None: + return False + project.design_system_id = design_system_id + project.updated_at = datetime.now(timezone.utc) + await session.commit() + return True diff --git a/tests/test_design_cascade.py b/tests/test_design_cascade.py new file mode 100644 index 0000000..b00936c --- /dev/null +++ b/tests/test_design_cascade.py @@ -0,0 +1,94 @@ +"""The parent chain: walking it, and refusing to close it (milestone #254 step 1). + +Pure functions over a `{id: parent_id}` literal, so a whole hierarchy is one line +of setup and no database is involved. That is the reason the cascade lives in its +own import-free module — see services/design_cascade.py. +""" +from scribe.services.design_cascade import ancestry, would_cycle + + +# --- ancestry --------------------------------------------------------------- + +def test_ancestry_returns_the_chain_deepest_first(): + """Deepest first because that is the order resolution consumes it in: the + system being resolved wins over its parent, which wins over the root.""" + parents = {1: None, 2: 1, 3: 2} + assert ancestry(3, parents) == [3, 2, 1] + + +def test_a_root_is_its_own_whole_chain(): + """A family system has no parent, and that is an ordinary state — not an + incomplete one. It must resolve to exactly itself.""" + assert ancestry(1, {1: None}) == [1] + + +def test_a_missing_parent_truncates_rather_than_raising(): + """A parent that was soft-deleted (or filtered out of the caller's scope) is + a shorter chain, not a failed request. The alternative — raising — would make + one deleted system break every descendant's rendering.""" + assert ancestry(3, {3: 2}) == [3, 2] + + +def test_a_system_absent_from_the_map_still_yields_itself(): + assert ancestry(9, {}) == [9] + + +def test_ancestry_terminates_on_a_cycle_instead_of_hanging(): + """LOAD-BEARING, and the reason a visited-set exists even though writes are + guarded. A loop introduced by a direct DB edit or a future bug must degrade + to a truncated chain: truncation shows up in the result, a hang shows up as + an outage. Every id appears exactly once.""" + parents = {1: 3, 2: 1, 3: 2} + chain = ancestry(1, parents) + assert chain == [1, 3, 2] + assert len(chain) == len(set(chain)) + + +def test_ancestry_terminates_on_a_self_parent(): + assert ancestry(1, {1: 1}) == [1] + + +# --- would_cycle ------------------------------------------------------------ + +def test_clearing_the_parent_never_cycles(): + """None means "make this a root", which is always safe.""" + assert would_cycle(2, None, {1: None, 2: 1}) is False + + +def test_a_system_cannot_be_its_own_parent(): + assert would_cycle(1, 1, {1: None}) is True + + +def test_an_ordinary_reparent_is_allowed(): + """family <- app is the shape the whole model exists for; it must not trip + the guard.""" + assert would_cycle(2, 1, {1: None, 2: None}) is False + + +def test_a_direct_swap_is_refused(): + """A -> B, then B -> A. The two-system case, and the one a UI produces first + because both systems are on screen together.""" + parents = {1: None, 2: 1} + assert would_cycle(1, 2, parents) is True + + +def test_an_indirect_loop_is_refused(): + """Three deep: root <- mid <- leaf, then root's parent set to leaf. Catching + this is what makes the check a chain walk rather than a parent comparison.""" + parents = {1: None, 2: 1, 3: 2} + assert would_cycle(1, 3, parents) is True + + +def test_reparenting_onto_a_sibling_subtree_is_allowed(): + """Two branches off one root. Moving one under the other is legitimate — the + guard must refuse loops, not reorganisation.""" + parents = {1: None, 2: 1, 3: 1} + assert would_cycle(3, 2, parents) is False + + +def test_the_guard_survives_a_hierarchy_that_is_already_corrupt(): + """If a cycle somehow already exists, the guard still has to answer rather + than spin — the write path is exactly where such a hierarchy gets repaired.""" + parents = {1: 2, 2: 1, 3: None} + assert would_cycle(3, 1, parents) is False + assert would_cycle(1, 2, parents) is True diff --git a/tests/test_design_system.py b/tests/test_design_rulebook_import.py similarity index 99% rename from tests/test_design_system.py rename to tests/test_design_rulebook_import.py index 0b29af2..9a0c942 100644 --- a/tests/test_design_system.py +++ b/tests/test_design_rulebook_import.py @@ -11,7 +11,7 @@ be testing the instance, not the parser. """ from types import SimpleNamespace -from scribe.services.design_system import ( +from scribe.services.design_rulebook_import import ( expand_token_shorthand, extract_expectations, normalize_hex, diff --git a/tests/test_services_access_visibility.py b/tests/test_services_access_visibility.py index 1b9a6d0..e036fad 100644 --- a/tests/test_services_access_visibility.py +++ b/tests/test_services_access_visibility.py @@ -132,3 +132,40 @@ def test_clauses_are_pure_builders(clause_fn): import inspect assert not inspect.iscoroutinefunction(clause_fn) assert _sql(clause_fn(7)) # builds without touching a database + + +# --- design systems ---------------------------------------------------------- +# +# A design system is reachable two ways: you own it, or you can see a project +# that inherits from it. The gap between what those two grant is the invariant +# worth guarding — see get_design_system_permission. + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "permission, readable, writable", + [ + ("owner", True, True), + # Project-derived. Being an EDITOR on a shared project must not confer + # the right to rewrite the family system that project inherits from — + # that would let one project's collaborator restyle every other project + # in the family. + ("viewer", True, False), + (None, False, False), + ], +) +async def test_reaching_a_design_system_via_a_project_reads_but_never_writes( + permission, readable, writable +): + from unittest.mock import AsyncMock, patch + + from scribe.services.access import ( + can_read_design_system, + can_write_design_system, + ) + + with patch( + "scribe.services.access.get_design_system_permission", + AsyncMock(return_value=permission), + ): + assert await can_read_design_system(1, 3) is readable + assert await can_write_design_system(1, 3) is writable diff --git a/tests/test_services_design_systems.py b/tests/test_services_design_systems.py new file mode 100644 index 0000000..9d8bdc5 --- /dev/null +++ b/tests/test_services_design_systems.py @@ -0,0 +1,194 @@ +"""ACL gating and field handling for services/design_systems.py (unit, mocked). + +Mirrors tests/test_services_systems.py — same mocked-session shape, because the +question is the same one: does the service refuse before it touches a row? +""" +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from scribe.services.design_systems import DesignSystemCycle + + +def _make_mock_session(): + s = AsyncMock() + s.__aenter__ = AsyncMock(return_value=s) + s.__aexit__ = AsyncMock(return_value=False) + s.add = MagicMock() + s.commit = AsyncMock() + s.refresh = AsyncMock() + return s + + +# --- creating --------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_create_with_a_parent_is_denied_without_write_on_that_parent(): + """Parenting to someone else's system would let their delete or re-parent + silently restyle your app, so the chain may only be built from systems you + can write — which, per the ACL, means ones you own.""" + with patch("scribe.services.design_systems.access") as acc: + acc.can_write_design_system = AsyncMock(return_value=False) + from scribe.services.design_systems import create_design_system + result = await create_design_system(user_id=1, title="Scribe", parent_id=7) + assert result is None + + +@pytest.mark.asyncio +async def test_create_without_a_parent_never_consults_the_parent_acl(): + """A family system has no parent, and creating one must not be gated on a + permission check for a system that does not exist.""" + mock_session = _make_mock_session() + captured = {} + mock_session.add = MagicMock( + side_effect=lambda obj: captured.update( + title=obj.title, parent_id=obj.parent_id + ) + ) + with patch("scribe.services.design_systems.async_session") as mock_cls, \ + patch("scribe.services.design_systems.access") as acc: + acc.can_write_design_system = AsyncMock(return_value=False) + mock_cls.return_value = mock_session + from scribe.services.design_systems import create_design_system + await create_design_system(user_id=1, title=" FabledSword ") + assert captured["title"] == "FabledSword" # stripped + assert captured["parent_id"] is None + acc.can_write_design_system.assert_not_awaited() + + +# --- the cycle guard, through the service ----------------------------------- + +@pytest.mark.asyncio +async def test_reparenting_into_a_loop_raises_rather_than_returning_none(): + """None already means "not found, or not yours". A caller that conflated the + two would report "no such design system" for what is really "that parent is + one of its own descendants", so the cycle gets its own exception type.""" + mock_session = _make_mock_session() + mock_session.get = AsyncMock(return_value=MagicMock(deleted_at=None, parent_id=None)) + + with patch("scribe.services.design_systems.async_session") as mock_cls, \ + patch("scribe.services.design_systems.access") as acc, \ + patch("scribe.services.design_systems._parent_map", + AsyncMock(return_value={1: None, 2: 1})): + acc.can_write_design_system = AsyncMock(return_value=True) + mock_cls.return_value = mock_session + from scribe.services.design_systems import update_design_system + with pytest.raises(DesignSystemCycle): + await update_design_system(user_id=1, design_system_id=1, parent_id=2) + + mock_session.commit.assert_not_awaited() # refused BEFORE the write + + +@pytest.mark.asyncio +async def test_clearing_the_parent_is_allowed_and_is_not_read_as_no_change(): + """`parent_id=None` means "make this a root" — the one field where None is a + value rather than "leave alone", which is why it is handled apart from the + others.""" + system = MagicMock(deleted_at=None, parent_id=5) + mock_session = _make_mock_session() + mock_session.get = AsyncMock(return_value=system) + + with patch("scribe.services.design_systems.async_session") as mock_cls, \ + patch("scribe.services.design_systems.access") as acc, \ + patch("scribe.services.design_systems._parent_map", + AsyncMock(return_value={1: 5, 5: None})): + acc.can_write_design_system = AsyncMock(return_value=True) + mock_cls.return_value = mock_session + from scribe.services.design_systems import update_design_system + await update_design_system(user_id=1, design_system_id=1, parent_id=None) + + assert system.parent_id is None + + +# --- tokens ----------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_create_token_denied_without_write_on_the_system(): + with patch("scribe.services.design_systems.access") as acc: + acc.can_write_design_system = AsyncMock(return_value=False) + from scribe.services.design_systems import create_token + result = await create_token(user_id=1, design_system_id=3, name="--fs-obsidian") + assert result is None + + +@pytest.mark.asyncio +async def test_token_values_default_to_an_empty_map_not_json_null(): + """The column is NOT NULL so that absence has exactly ONE spelling. Passing + None straight through would store JSON null and hand every reader back the + second empty state the schema was shaped to remove.""" + mock_session = _make_mock_session() + captured = {} + mock_session.add = MagicMock( + side_effect=lambda obj: captured.update(value_by_mode=obj.value_by_mode) + ) + + with patch("scribe.services.design_systems.async_session") as mock_cls, \ + patch("scribe.services.design_systems.access") as acc: + acc.can_write_design_system = AsyncMock(return_value=True) + mock_cls.return_value = mock_session + from scribe.services.design_systems import create_token + await create_token( + user_id=1, design_system_id=3, name="--fs-radius-md", value_by_mode=None + ) + + assert captured["value_by_mode"] == {} + + +@pytest.mark.asyncio +async def test_list_tokens_denied_returns_empty(): + with patch("scribe.services.design_systems.access") as acc: + acc.can_read_design_system = AsyncMock(return_value=False) + from scribe.services.design_systems import list_tokens + assert await list_tokens(user_id=1, design_system_id=3) == [] + + +# --- the project pointer ---------------------------------------------------- + +@pytest.mark.asyncio +async def test_pointing_a_project_at_a_system_needs_write_on_the_project(): + with patch("scribe.services.design_systems.access") as acc: + acc.can_write_project = AsyncMock(return_value=False) + acc.can_read_design_system = AsyncMock(return_value=True) + from scribe.services.design_systems import set_project_design_system + assert await set_project_design_system(1, project_id=5, design_system_id=3) is False + + +@pytest.mark.asyncio +async def test_pointing_a_project_needs_only_READ_on_the_system(): + """Consuming a design system is not changing it, so a system reachable + through another project is a legitimate choice here. Requiring write would + make a shared family style unusable by the people it was shared with.""" + project = MagicMock(deleted_at=None, design_system_id=None) + mock_session = _make_mock_session() + mock_session.get = AsyncMock(return_value=project) + + with patch("scribe.services.design_systems.async_session") as mock_cls, \ + patch("scribe.services.design_systems.access") as acc: + acc.can_write_project = AsyncMock(return_value=True) + acc.can_read_design_system = AsyncMock(return_value=True) + acc.can_write_design_system = AsyncMock(return_value=False) + mock_cls.return_value = mock_session + from scribe.services.design_systems import set_project_design_system + assert await set_project_design_system(1, project_id=5, design_system_id=3) is True + + assert project.design_system_id == 3 + + +@pytest.mark.asyncio +async def test_clearing_a_projects_design_system_skips_the_system_acl(): + """Un-styling a project must not require permission on the system it is + letting go of — including one that has since been deleted.""" + project = MagicMock(deleted_at=None, design_system_id=3) + mock_session = _make_mock_session() + mock_session.get = AsyncMock(return_value=project) + + with patch("scribe.services.design_systems.async_session") as mock_cls, \ + patch("scribe.services.design_systems.access") as acc: + acc.can_write_project = AsyncMock(return_value=True) + acc.can_read_design_system = AsyncMock(return_value=False) + mock_cls.return_value = mock_session + from scribe.services.design_systems import set_project_design_system + assert await set_project_design_system(1, project_id=5, design_system_id=None) is True + + assert project.design_system_id is None + acc.can_read_design_system.assert_not_awaited()