Design systems as records — the stylesheet Scribe holds, plus two live bug fixes #88
@@ -27,6 +27,7 @@ from scribe.routes.knowledge import knowledge_bp
|
||||
from scribe.routes.rulebooks import rulebooks_bp
|
||||
from scribe.routes.plugin import plugin_bp
|
||||
from scribe.routes.design import design_bp
|
||||
from scribe.routes.design_systems import design_systems_bp
|
||||
from scribe.routes.trash import trash_bp
|
||||
from scribe.routes.dashboard import dashboard_bp
|
||||
from scribe.routes.systems import systems_bp
|
||||
@@ -91,6 +92,7 @@ def create_app() -> Quart:
|
||||
app.register_blueprint(rulebooks_bp)
|
||||
app.register_blueprint(plugin_bp)
|
||||
app.register_blueprint(design_bp)
|
||||
app.register_blueprint(design_systems_bp)
|
||||
app.register_blueprint(trash_bp)
|
||||
app.register_blueprint(dashboard_bp)
|
||||
app.register_blueprint(systems_bp)
|
||||
|
||||
@@ -5,7 +5,8 @@ to a FastMCP instance. `register_all(mcp)` is the single entry point called
|
||||
from `mcp.server.build_mcp_server`.
|
||||
"""
|
||||
from scribe.mcp.tools import (
|
||||
milestones, notes, processes, projects, recent, repos, rulebooks, search, snippets, systems, tags, tasks, trash,
|
||||
design_systems, milestones, notes, processes, projects, recent, repos, rulebooks, search, snippets,
|
||||
systems, tags, tasks, trash,
|
||||
)
|
||||
|
||||
|
||||
@@ -17,6 +18,7 @@ def register_all(mcp) -> None:
|
||||
projects.register(mcp)
|
||||
milestones.register(mcp)
|
||||
systems.register(mcp)
|
||||
design_systems.register(mcp)
|
||||
tags.register(mcp)
|
||||
recent.register(mcp)
|
||||
repos.register(mcp)
|
||||
|
||||
@@ -0,0 +1,266 @@
|
||||
"""Design system + token MCP tools — wrappers over services/design_systems.py.
|
||||
|
||||
A design system is a stylesheet held as records: a named set of tokens with an
|
||||
optional parent, so a family system carries the house style and an app system
|
||||
carries only what it changes. Precedence by name along the parent chain IS the
|
||||
CSS cascade, which is why "what does this app alter?" is a plain list rather
|
||||
than a diff.
|
||||
|
||||
Parity with the REST surface is a rule, not a nicety (see
|
||||
`tests/test_routes_design_systems.py`): an agent and a browser are two callers
|
||||
of one service.
|
||||
|
||||
Sentinels, matching the milestone/task tool conventions:
|
||||
- title="" / description="" / etc. -> "leave unchanged" on update
|
||||
- parent_id / design_system_id: 0 = leave unchanged, -1 = clear, positive = set
|
||||
(three states, because clearing a parent is a real operation and not the
|
||||
same as omitting the argument)
|
||||
- order_index=-1 -> "leave unchanged" (0 is a valid order_index)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from scribe.mcp._context import current_user_id
|
||||
from scribe.services import design_systems as ds_svc
|
||||
from scribe.services.design_systems import DesignSystemCycle
|
||||
|
||||
|
||||
async def create_design_system(
|
||||
title: str,
|
||||
description: str = "",
|
||||
parent_id: int = 0,
|
||||
) -> dict:
|
||||
"""Create a design system, optionally inheriting from another.
|
||||
|
||||
Args:
|
||||
title: What this system is, e.g. "FabledSword" or "Scribe" (required).
|
||||
description: What it covers and when it applies.
|
||||
parent_id: Inherit from this system — it holds the defaults this one
|
||||
overrides. Omit (0) for a top-level "family" system, which is what
|
||||
a first design system usually is.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
system = await ds_svc.create_design_system(
|
||||
uid,
|
||||
title=title,
|
||||
description=description or None,
|
||||
parent_id=parent_id or None,
|
||||
)
|
||||
if system is None:
|
||||
raise ValueError(f"parent design system {parent_id} not found or not writable")
|
||||
return system.to_dict()
|
||||
|
||||
|
||||
async def list_design_systems() -> dict:
|
||||
"""List your design systems. An empty list is normal — most installs have none."""
|
||||
uid = current_user_id()
|
||||
rows = await ds_svc.list_design_systems(uid)
|
||||
return {"design_systems": [s.to_dict() for s in rows]}
|
||||
|
||||
|
||||
async def get_design_system(design_system_id: int) -> dict:
|
||||
"""Fetch a design system plus its OWN tokens — i.e. what it changes.
|
||||
|
||||
For what it actually resolves to once inheritance is applied, use
|
||||
`resolve_design_system`. The two answer different questions and a system
|
||||
that overrides nothing has an empty token list but a full resolved set.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
system = await ds_svc.get_design_system(uid, design_system_id)
|
||||
if system is None:
|
||||
raise ValueError(f"design system {design_system_id} not found")
|
||||
tokens = await ds_svc.list_tokens(uid, design_system_id)
|
||||
return {
|
||||
"design_system": system.to_dict(),
|
||||
"tokens": [t.to_dict() for t in tokens],
|
||||
}
|
||||
|
||||
|
||||
async def resolve_design_system(design_system_id: int) -> dict:
|
||||
"""The EFFECTIVE token set — everything inherited, with this system's on top.
|
||||
|
||||
Each token carries `origin_by_mode` (which system supplied each mode's
|
||||
value) and `contributions` (every system that offered one, deepest first —
|
||||
so entry 0 won and the rest were shadowed). Reach for this when you need to
|
||||
know what a value actually IS; reach for `get_design_system` when you need
|
||||
to know what this system CHANGES.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
resolved = await ds_svc.resolve_design_system(uid, design_system_id)
|
||||
if resolved is None:
|
||||
raise ValueError(f"design system {design_system_id} not found")
|
||||
return {
|
||||
"design_system_id": design_system_id,
|
||||
"tokens": [t.to_dict() for t in resolved],
|
||||
}
|
||||
|
||||
|
||||
async def update_design_system(
|
||||
design_system_id: int,
|
||||
title: str = "",
|
||||
description: str = "",
|
||||
parent_id: int = 0,
|
||||
) -> dict:
|
||||
"""Update a design system.
|
||||
|
||||
Args:
|
||||
design_system_id: The system to update.
|
||||
title: New title, or "" to leave unchanged.
|
||||
description: New description, or "" to leave unchanged.
|
||||
parent_id: 0 = leave unchanged, -1 = clear (make this a top-level
|
||||
family system), positive = inherit from that system. A parent that
|
||||
already inherits from this system is refused — that would be a loop.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
fields: dict = {}
|
||||
if title:
|
||||
fields["title"] = title
|
||||
if description:
|
||||
fields["description"] = description
|
||||
if parent_id:
|
||||
fields["parent_id"] = None if parent_id == -1 else parent_id
|
||||
try:
|
||||
system = await ds_svc.update_design_system(uid, design_system_id, **fields)
|
||||
except DesignSystemCycle as exc:
|
||||
raise ValueError(str(exc)) from exc
|
||||
if system is None:
|
||||
raise ValueError(f"design system {design_system_id} not found or not writable")
|
||||
return system.to_dict()
|
||||
|
||||
|
||||
async def delete_design_system(design_system_id: int) -> dict:
|
||||
"""Soft-delete a design system (recoverable).
|
||||
|
||||
Systems that inherited from it become top-level systems keeping their own
|
||||
tokens — deleting a family does not delete the apps under it.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
if not await ds_svc.delete_design_system(uid, design_system_id):
|
||||
raise ValueError(f"design system {design_system_id} not found or not writable")
|
||||
return {"message": f"Design system {design_system_id} deleted."}
|
||||
|
||||
|
||||
# ── Tokens ──────────────────────────────────────────────────────────────
|
||||
|
||||
async def create_design_token(
|
||||
design_system_id: int,
|
||||
name: str,
|
||||
value_by_mode: dict | None = None,
|
||||
group_name: str = "",
|
||||
purpose: str = "",
|
||||
order_index: int = 0,
|
||||
) -> dict:
|
||||
"""Add a token to a design system.
|
||||
|
||||
Args:
|
||||
design_system_id: The system that owns this token.
|
||||
name: The custom-property name, e.g. "--fs-obsidian" (required).
|
||||
value_by_mode: Values keyed by mode, e.g.
|
||||
{"base": "#f7f5ef", "dark": "#14171a"}. Use "base" for the value
|
||||
that applies when no mode is more specific; a token that is not
|
||||
mode-dependent needs only "base". In a system WITH a parent, an
|
||||
omitted mode is inherited rather than blanked.
|
||||
group_name: Free-text grouping — "surface", "text", "radius", whatever
|
||||
this system's own vocabulary is.
|
||||
purpose: What the token is for, e.g. "page bg, deepest surface".
|
||||
order_index: Display position within its group.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
token = await ds_svc.create_token(
|
||||
uid,
|
||||
design_system_id=design_system_id,
|
||||
name=name,
|
||||
value_by_mode=value_by_mode,
|
||||
group_name=group_name or None,
|
||||
purpose=purpose or None,
|
||||
order_index=order_index,
|
||||
)
|
||||
if token is None:
|
||||
raise ValueError(f"design system {design_system_id} not found or not writable")
|
||||
return token.to_dict()
|
||||
|
||||
|
||||
async def list_design_tokens(design_system_id: int) -> dict:
|
||||
"""A design system's OWN tokens — its override set, not its effective set."""
|
||||
uid = current_user_id()
|
||||
rows = await ds_svc.list_tokens(uid, design_system_id)
|
||||
return {"tokens": [t.to_dict() for t in rows]}
|
||||
|
||||
|
||||
async def update_design_token(
|
||||
token_id: int,
|
||||
name: str = "",
|
||||
value_by_mode: dict | None = None,
|
||||
group_name: str = "",
|
||||
purpose: str = "",
|
||||
order_index: int = -1,
|
||||
) -> dict:
|
||||
"""Update a token. Empty/None args leave a field unchanged.
|
||||
|
||||
`value_by_mode` REPLACES the whole map rather than merging into it, so send
|
||||
every mode you want the token to keep.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
fields: dict = {}
|
||||
if name:
|
||||
fields["name"] = name
|
||||
if value_by_mode is not None:
|
||||
fields["value_by_mode"] = value_by_mode
|
||||
if group_name:
|
||||
fields["group_name"] = group_name
|
||||
if purpose:
|
||||
fields["purpose"] = purpose
|
||||
if order_index >= 0:
|
||||
fields["order_index"] = order_index
|
||||
token = await ds_svc.update_token(uid, token_id, **fields)
|
||||
if token is None:
|
||||
raise ValueError(f"design token {token_id} not found or not writable")
|
||||
return token.to_dict()
|
||||
|
||||
|
||||
async def delete_design_token(token_id: int) -> dict:
|
||||
"""Soft-delete a token (recoverable).
|
||||
|
||||
In a system with a parent this restores inheritance: the token stops being
|
||||
overridden here and resolves to the parent's value again.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
if not await ds_svc.delete_token(uid, token_id):
|
||||
raise ValueError(f"design token {token_id} not found or not writable")
|
||||
return {"message": f"Design token {token_id} deleted."}
|
||||
|
||||
|
||||
async def set_project_design_system(project_id: int, design_system_id: int = 0) -> dict:
|
||||
"""Point a project at a design system.
|
||||
|
||||
Args:
|
||||
project_id: The project to style.
|
||||
design_system_id: The system it uses, or -1 to clear it. Pointing at a
|
||||
system only requires READ access to it — consuming a design system
|
||||
is not changing it.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
target = None if design_system_id == -1 else design_system_id
|
||||
ok = await ds_svc.set_project_design_system(uid, project_id, target)
|
||||
if not ok:
|
||||
raise ValueError(
|
||||
f"project {project_id} not writable, or design system "
|
||||
f"{design_system_id} not found"
|
||||
)
|
||||
return {"project_id": project_id, "design_system_id": target}
|
||||
|
||||
|
||||
def register(mcp) -> None:
|
||||
for fn in (
|
||||
create_design_system,
|
||||
list_design_systems,
|
||||
get_design_system,
|
||||
resolve_design_system,
|
||||
update_design_system,
|
||||
delete_design_system,
|
||||
create_design_token,
|
||||
list_design_tokens,
|
||||
update_design_token,
|
||||
delete_design_token,
|
||||
set_project_design_system,
|
||||
):
|
||||
mcp.tool(name=fn.__name__)(fn)
|
||||
@@ -0,0 +1,189 @@
|
||||
"""Design system + token REST endpoints (milestone #254 step 4).
|
||||
|
||||
Wraps `services/design_systems.py`, which owns the ACL and the cycle guard.
|
||||
Design systems are owner-scoped top-level records rather than project-scoped
|
||||
ones, so these do NOT nest under `/api/projects/` — `routes/rulebooks.py` is the
|
||||
closer shape.
|
||||
|
||||
Two failures this layer has to keep apart, which is why the service raises for
|
||||
one and returns None for the other:
|
||||
|
||||
- `DesignSystemCycle` -> 400 with its message. "That parent already inherits
|
||||
from this system" is a correctable mistake and the caller needs to be told
|
||||
which one they made.
|
||||
- None -> 404, covering both "no such system" and "not yours". Conflating
|
||||
those two IS the intent: distinguishing them would confirm the existence of
|
||||
records the caller may not see.
|
||||
"""
|
||||
from quart import Blueprint, g, jsonify, request
|
||||
|
||||
from scribe.auth import login_required
|
||||
from scribe.services import design_systems as ds_svc
|
||||
from scribe.services.design_systems import DesignSystemCycle
|
||||
|
||||
design_systems_bp = Blueprint("design_systems", __name__, url_prefix="/api")
|
||||
|
||||
|
||||
def _uid() -> int:
|
||||
return g.user.id
|
||||
|
||||
|
||||
def _not_found(what: str = "design system"):
|
||||
return jsonify({"error": f"{what} not found"}), 404
|
||||
|
||||
|
||||
# ── Design systems ──────────────────────────────────────────────────────
|
||||
|
||||
@design_systems_bp.get("/design-systems")
|
||||
@login_required
|
||||
async def list_design_systems():
|
||||
"""The caller's design systems. An empty list is the ordinary state for an
|
||||
install that has never made one, not an error."""
|
||||
rows = await ds_svc.list_design_systems(_uid())
|
||||
return jsonify({"design_systems": [s.to_dict() for s in rows]})
|
||||
|
||||
|
||||
@design_systems_bp.post("/design-systems")
|
||||
@login_required
|
||||
async def create_design_system():
|
||||
data = await request.get_json() or {}
|
||||
title = (data.get("title") or "").strip()
|
||||
if not title:
|
||||
return jsonify({"error": "title is required"}), 400
|
||||
system = await ds_svc.create_design_system(
|
||||
user_id=_uid(),
|
||||
title=title,
|
||||
description=data.get("description") or None,
|
||||
parent_id=data.get("parent_id"),
|
||||
)
|
||||
if system is None:
|
||||
return jsonify({"error": "parent design system not found"}), 404
|
||||
return jsonify(system.to_dict()), 201
|
||||
|
||||
|
||||
@design_systems_bp.get("/design-systems/<int:design_system_id>")
|
||||
@login_required
|
||||
async def get_design_system(design_system_id: int):
|
||||
system = await ds_svc.get_design_system(_uid(), design_system_id)
|
||||
if system is None:
|
||||
return _not_found()
|
||||
return jsonify(system.to_dict())
|
||||
|
||||
|
||||
@design_systems_bp.patch("/design-systems/<int:design_system_id>")
|
||||
@login_required
|
||||
async def update_design_system(design_system_id: int):
|
||||
data = await request.get_json() or {}
|
||||
fields = {k: v for k, v in data.items() if k in ("title", "description")}
|
||||
# Presence, not truthiness: `{"parent_id": null}` means "make this a root",
|
||||
# which a `if data.get("parent_id")` filter would silently drop.
|
||||
if "parent_id" in data:
|
||||
fields["parent_id"] = data["parent_id"]
|
||||
try:
|
||||
system = await ds_svc.update_design_system(_uid(), design_system_id, **fields)
|
||||
except DesignSystemCycle as exc:
|
||||
return jsonify({"error": str(exc)}), 400
|
||||
if system is None:
|
||||
return _not_found()
|
||||
return jsonify(system.to_dict())
|
||||
|
||||
|
||||
@design_systems_bp.delete("/design-systems/<int:design_system_id>")
|
||||
@login_required
|
||||
async def delete_design_system(design_system_id: int):
|
||||
if not await ds_svc.delete_design_system(_uid(), design_system_id):
|
||||
return _not_found()
|
||||
return "", 204
|
||||
|
||||
|
||||
@design_systems_bp.get("/design-systems/<int:design_system_id>/resolved")
|
||||
@login_required
|
||||
async def resolve_design_system(design_system_id: int):
|
||||
"""The EFFECTIVE token set — everything inherited, with this system's on top.
|
||||
|
||||
Distinct from `/tokens` on purpose: that returns what this system CHANGES,
|
||||
this returns what it ends up being. Both are real questions and answering
|
||||
only one would make the other a client-side computation.
|
||||
"""
|
||||
resolved = await ds_svc.resolve_design_system(_uid(), design_system_id)
|
||||
if resolved is None:
|
||||
return _not_found()
|
||||
return jsonify({
|
||||
"design_system_id": design_system_id,
|
||||
"tokens": [t.to_dict() for t in resolved],
|
||||
})
|
||||
|
||||
|
||||
# ── Tokens ──────────────────────────────────────────────────────────────
|
||||
|
||||
@design_systems_bp.get("/design-systems/<int:design_system_id>/tokens")
|
||||
@login_required
|
||||
async def list_design_tokens(design_system_id: int):
|
||||
"""This system's OWN tokens — its override set, not its effective set."""
|
||||
if await ds_svc.get_design_system(_uid(), design_system_id) is None:
|
||||
return _not_found()
|
||||
rows = await ds_svc.list_tokens(_uid(), design_system_id)
|
||||
return jsonify({"tokens": [t.to_dict() for t in rows]})
|
||||
|
||||
|
||||
@design_systems_bp.post("/design-systems/<int:design_system_id>/tokens")
|
||||
@login_required
|
||||
async def create_design_token(design_system_id: int):
|
||||
data = await request.get_json() or {}
|
||||
name = (data.get("name") or "").strip()
|
||||
if not name:
|
||||
return jsonify({"error": "name is required"}), 400
|
||||
token = await ds_svc.create_token(
|
||||
user_id=_uid(),
|
||||
design_system_id=design_system_id,
|
||||
name=name,
|
||||
value_by_mode=data.get("value_by_mode"),
|
||||
group_name=data.get("group_name") or None,
|
||||
purpose=data.get("purpose") or None,
|
||||
order_index=data.get("order_index") or 0,
|
||||
)
|
||||
if token is None:
|
||||
return _not_found()
|
||||
return jsonify(token.to_dict()), 201
|
||||
|
||||
|
||||
@design_systems_bp.patch("/design-tokens/<int:token_id>")
|
||||
@login_required
|
||||
async def update_design_token(token_id: int):
|
||||
data = await request.get_json() or {}
|
||||
fields = {
|
||||
k: v for k, v in data.items()
|
||||
if k in ("name", "value_by_mode", "group_name", "purpose", "order_index")
|
||||
}
|
||||
token = await ds_svc.update_token(_uid(), token_id, **fields)
|
||||
if token is None:
|
||||
return _not_found("design token")
|
||||
return jsonify(token.to_dict())
|
||||
|
||||
|
||||
@design_systems_bp.delete("/design-tokens/<int:token_id>")
|
||||
@login_required
|
||||
async def delete_design_token(token_id: int):
|
||||
if not await ds_svc.delete_token(_uid(), token_id):
|
||||
return _not_found("design token")
|
||||
return "", 204
|
||||
|
||||
|
||||
# ── The project pointer ─────────────────────────────────────────────────
|
||||
|
||||
@design_systems_bp.put("/projects/<int:project_id>/design-system")
|
||||
@login_required
|
||||
async def set_project_design_system(project_id: int):
|
||||
"""Point a project at a design system. `{"design_system_id": null}` clears it.
|
||||
|
||||
PUT rather than PATCH: this sets one field to exactly what is sent, and
|
||||
clearing it is a first-class outcome rather than an omission.
|
||||
"""
|
||||
data = await request.get_json() or {}
|
||||
ok = await ds_svc.set_project_design_system(
|
||||
_uid(), project_id, data.get("design_system_id")
|
||||
)
|
||||
if not ok:
|
||||
return _not_found("project or design system")
|
||||
return jsonify({"project_id": project_id,
|
||||
"design_system_id": data.get("design_system_id")})
|
||||
@@ -119,6 +119,30 @@ class ResolvedToken:
|
||||
entries = self.contributions.get(mode) or self.contributions.get(BASE_MODE)
|
||||
return entries[0].value if entries else None
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""Payload shape for both surfaces — and it carries the SHADOWED entries.
|
||||
|
||||
Serialising only the winner would throw away the provenance at the last
|
||||
step, which is the one thing this type exists to preserve. `contributions`
|
||||
is the audit trail; `value_by_mode` / `origin_by_mode` are alongside it so
|
||||
a client renders without re-deriving anything, and cannot derive it
|
||||
differently.
|
||||
"""
|
||||
return {
|
||||
"name": self.name,
|
||||
"group_name": self.group_name,
|
||||
"purpose": self.purpose,
|
||||
"order_index": self.order_index,
|
||||
"value_by_mode": self.value_by_mode,
|
||||
"origin_by_mode": self.origin_by_mode,
|
||||
"contributions": {
|
||||
mode: [
|
||||
{"system_id": c.system_id, "value": c.value} for c in entries
|
||||
]
|
||||
for mode, entries in self.contributions.items()
|
||||
},
|
||||
}
|
||||
|
||||
def is_overridden_in(self, system_id: int) -> bool:
|
||||
"""Does `system_id` win any mode of this token AND shadow something?
|
||||
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
"""MCP design-system tools — the sentinel translations, mostly.
|
||||
|
||||
The tools are thin wrappers, so the only logic worth testing is where the MCP
|
||||
calling convention meets the service's: an agent cannot omit an argument, so
|
||||
"leave unchanged", "clear" and "set" have to be encoded in the value. Getting
|
||||
that mapping wrong is silent — the call succeeds and changes the wrong thing.
|
||||
"""
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from scribe.mcp._context import _user_id_ctx
|
||||
from scribe.services.design_systems import DesignSystemCycle
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _bind_user():
|
||||
token = _user_id_ctx.set(7)
|
||||
yield
|
||||
_user_id_ctx.reset(token)
|
||||
|
||||
|
||||
def _fake_system():
|
||||
s = MagicMock()
|
||||
s.to_dict.return_value = {"id": 1, "title": "FabledSword", "parent_id": None}
|
||||
return s
|
||||
|
||||
|
||||
def _fake_token():
|
||||
t = MagicMock()
|
||||
t.to_dict.return_value = {"id": 9, "name": "--fs-obsidian"}
|
||||
return t
|
||||
|
||||
|
||||
# --- create -----------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_creating_without_a_parent_passes_none_not_zero():
|
||||
"""0 is the "omitted" sentinel, and it must not reach the service as a
|
||||
system id — there is no system 0, so the create would fail an ACL check for
|
||||
a record that cannot exist."""
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.create_design_system = AsyncMock(return_value=_fake_system())
|
||||
from scribe.mcp.tools.design_systems import create_design_system
|
||||
await create_design_system(title="FabledSword")
|
||||
assert svc.create_design_system.await_args.kwargs["parent_id"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_creating_with_a_parent_passes_it_through():
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.create_design_system = AsyncMock(return_value=_fake_system())
|
||||
from scribe.mcp.tools.design_systems import create_design_system
|
||||
await create_design_system(title="Scribe", parent_id=4)
|
||||
assert svc.create_design_system.await_args.kwargs["parent_id"] == 4
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_raises_when_the_parent_is_not_writable():
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.create_design_system = AsyncMock(return_value=None)
|
||||
from scribe.mcp.tools.design_systems import create_design_system
|
||||
with pytest.raises(ValueError):
|
||||
await create_design_system(title="Scribe", parent_id=4)
|
||||
|
||||
|
||||
# --- the three-state parent -------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_with_parent_id_zero_leaves_the_parent_alone():
|
||||
"""The common case — renaming a system must not silently re-root it."""
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.update_design_system = AsyncMock(return_value=_fake_system())
|
||||
from scribe.mcp.tools.design_systems import update_design_system
|
||||
await update_design_system(design_system_id=1, title="Renamed")
|
||||
fields = svc.update_design_system.await_args.kwargs
|
||||
assert "parent_id" not in fields
|
||||
assert fields["title"] == "Renamed"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_with_parent_id_minus_one_clears_it():
|
||||
"""-1 means "make this a family system". It has to arrive at the service as
|
||||
None, which is the value the service reads as "become a root" — where
|
||||
omitting the key means "leave alone"."""
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.update_design_system = AsyncMock(return_value=_fake_system())
|
||||
from scribe.mcp.tools.design_systems import update_design_system
|
||||
await update_design_system(design_system_id=1, parent_id=-1)
|
||||
assert svc.update_design_system.await_args.kwargs["parent_id"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_with_a_positive_parent_id_sets_it():
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.update_design_system = AsyncMock(return_value=_fake_system())
|
||||
from scribe.mcp.tools.design_systems import update_design_system
|
||||
await update_design_system(design_system_id=1, parent_id=4)
|
||||
assert svc.update_design_system.await_args.kwargs["parent_id"] == 4
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_cycle_surfaces_as_a_usable_error_not_a_not_found():
|
||||
"""The service raises so this layer can keep the two apart. An agent told
|
||||
"not found" would retry the same call; one told what the loop is can fix it."""
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.update_design_system = AsyncMock(
|
||||
side_effect=DesignSystemCycle("2 already inherits from 1")
|
||||
)
|
||||
from scribe.mcp.tools.design_systems import update_design_system
|
||||
with pytest.raises(ValueError, match="already inherits"):
|
||||
await update_design_system(design_system_id=1, parent_id=2)
|
||||
|
||||
|
||||
# --- tokens -----------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_token_treats_order_index_minus_one_as_unchanged():
|
||||
"""0 is a VALID order_index, so it cannot double as the omitted sentinel —
|
||||
the same reason the systems tools use -1."""
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.update_token = AsyncMock(return_value=_fake_token())
|
||||
from scribe.mcp.tools.design_systems import update_design_token
|
||||
await update_design_token(token_id=9, purpose="page bg")
|
||||
fields = svc.update_token.await_args.kwargs
|
||||
assert "order_index" not in fields
|
||||
assert fields["purpose"] == "page bg"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_token_accepts_order_index_zero():
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.update_token = AsyncMock(return_value=_fake_token())
|
||||
from scribe.mcp.tools.design_systems import update_design_token
|
||||
await update_design_token(token_id=9, order_index=0)
|
||||
assert svc.update_token.await_args.kwargs["order_index"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_token_can_set_an_empty_value_map():
|
||||
"""`value_by_mode={}` is meaningful — it strips every mode from a token. The
|
||||
guard is `is not None`, not truthiness, or that edit would be unreachable."""
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.update_token = AsyncMock(return_value=_fake_token())
|
||||
from scribe.mcp.tools.design_systems import update_design_token
|
||||
await update_design_token(token_id=9, value_by_mode={})
|
||||
assert svc.update_token.await_args.kwargs["value_by_mode"] == {}
|
||||
|
||||
|
||||
# --- the project pointer ----------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clearing_a_projects_design_system_passes_none():
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.set_project_design_system = AsyncMock(return_value=True)
|
||||
from scribe.mcp.tools.design_systems import set_project_design_system
|
||||
result = await set_project_design_system(project_id=2, design_system_id=-1)
|
||||
assert svc.set_project_design_system.await_args.args[2] is None
|
||||
assert result["design_system_id"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_returns_serialised_tokens_with_their_provenance():
|
||||
"""The payload has to carry the shadowed entries, not just the winner —
|
||||
dropping them at the serialisation boundary would discard the one thing
|
||||
resolution was built to preserve."""
|
||||
from scribe.services.design_cascade import resolve_tokens
|
||||
|
||||
class _T:
|
||||
def __init__(self, name, value_by_mode):
|
||||
self.name, self.value_by_mode = name, value_by_mode
|
||||
self.group_name = self.purpose = None
|
||||
self.order_index = 0
|
||||
|
||||
resolved = resolve_tokens(
|
||||
2, {1: None, 2: 1},
|
||||
{1: [_T("--fs-accent", {"base": "#6b2118"})],
|
||||
2: [_T("--fs-accent", {"base": "#5b4a8a"})]},
|
||||
)
|
||||
with patch("scribe.mcp.tools.design_systems.ds_svc") as svc:
|
||||
svc.resolve_design_system = AsyncMock(return_value=resolved)
|
||||
from scribe.mcp.tools.design_systems import resolve_design_system
|
||||
payload = await resolve_design_system(design_system_id=2)
|
||||
|
||||
token = payload["tokens"][0]
|
||||
assert token["value_by_mode"] == {"base": "#5b4a8a"}
|
||||
assert token["origin_by_mode"] == {"base": 2}
|
||||
assert token["contributions"]["base"] == [
|
||||
{"system_id": 2, "value": "#5b4a8a"},
|
||||
{"system_id": 1, "value": "#6b2118"},
|
||||
]
|
||||
@@ -0,0 +1,106 @@
|
||||
"""Structural tests for the design-systems blueprint + the parity guard.
|
||||
|
||||
Modelled on tests/test_routes_snippets.py. The enumerations below are
|
||||
deliberate: relaxing one to a pattern would stop it catching the thing it was
|
||||
written for, which is a capability landing on one surface and not the other.
|
||||
"""
|
||||
import inspect
|
||||
|
||||
|
||||
def test_design_systems_blueprint_registered():
|
||||
from scribe.routes.design_systems import design_systems_bp
|
||||
assert design_systems_bp.name == "design_systems"
|
||||
assert design_systems_bp.url_prefix == "/api"
|
||||
|
||||
|
||||
def test_design_systems_blueprint_registered_in_app():
|
||||
from scribe.app import create_app
|
||||
app = create_app()
|
||||
assert "design_systems" in app.blueprints
|
||||
|
||||
|
||||
def test_route_handlers_callable():
|
||||
from scribe.routes import design_systems as routes
|
||||
for name in (
|
||||
"list_design_systems", "create_design_system", "get_design_system",
|
||||
"update_design_system", "delete_design_system", "resolve_design_system",
|
||||
"list_design_tokens", "create_design_token", "update_design_token",
|
||||
"delete_design_token", "set_project_design_system",
|
||||
):
|
||||
assert callable(getattr(routes, name))
|
||||
|
||||
|
||||
def test_every_endpoint_is_reachable_on_the_app():
|
||||
"""The handlers existing is not the same as them being routed. This catches
|
||||
a decorator that was copied without its path changing — two handlers on one
|
||||
rule, where the second silently never runs."""
|
||||
from scribe.app import create_app
|
||||
app = create_app()
|
||||
rules = {
|
||||
str(r.rule) for r in app.url_map.iter_rules()
|
||||
if r.endpoint.startswith("design_systems.")
|
||||
}
|
||||
assert rules == {
|
||||
"/api/design-systems",
|
||||
"/api/design-systems/<int:design_system_id>",
|
||||
"/api/design-systems/<int:design_system_id>/resolved",
|
||||
"/api/design-systems/<int:design_system_id>/tokens",
|
||||
"/api/design-tokens/<int:token_id>",
|
||||
"/api/projects/<int:project_id>/design-system",
|
||||
}
|
||||
|
||||
|
||||
def test_service_functions_take_user_id():
|
||||
"""Routes must call the services with user_id — verify the contract (#33)."""
|
||||
from scribe.services import design_systems as svc
|
||||
for fn_name in (
|
||||
"create_design_system", "list_design_systems", "get_design_system",
|
||||
"update_design_system", "delete_design_system", "resolve_design_system",
|
||||
"create_token", "list_tokens", "update_token", "delete_token",
|
||||
"set_project_design_system",
|
||||
):
|
||||
fn = getattr(svc, fn_name)
|
||||
assert callable(fn)
|
||||
assert "user_id" in inspect.signature(fn).parameters
|
||||
|
||||
|
||||
def test_agent_and_web_surfaces_stay_at_parity():
|
||||
"""The MCP tools and the REST routes are two callers of one service; a
|
||||
capability on one has to exist on the other (rule #33).
|
||||
|
||||
This guard exists because the snippet surfaces drifted apart once — MCP had
|
||||
no delete, the web side had no near-duplicate gate — and neither failed
|
||||
anything until someone went looking.
|
||||
"""
|
||||
from scribe.mcp.tools import design_systems as tools
|
||||
from scribe.routes import design_systems as routes
|
||||
|
||||
for name in (
|
||||
"create_design_system", "list_design_systems", "get_design_system",
|
||||
"resolve_design_system", "update_design_system", "delete_design_system",
|
||||
"create_design_token", "list_design_tokens", "update_design_token",
|
||||
"delete_design_token", "set_project_design_system",
|
||||
):
|
||||
assert callable(getattr(tools, name)), f"MCP tool missing: {name}"
|
||||
assert callable(getattr(routes, name)), f"REST route missing: {name}"
|
||||
|
||||
|
||||
def test_every_mcp_tool_in_the_module_is_registered():
|
||||
"""A tool written but never registered is invisible to an agent, and nothing
|
||||
else in the codebase would notice."""
|
||||
from scribe.mcp.tools import design_systems as tools
|
||||
|
||||
registered = []
|
||||
|
||||
class _Recorder:
|
||||
def tool(self, name):
|
||||
registered.append(name)
|
||||
return lambda fn: fn
|
||||
|
||||
tools.register(_Recorder())
|
||||
|
||||
public = {
|
||||
name for name, obj in vars(tools).items()
|
||||
if inspect.iscoroutinefunction(obj) and not name.startswith("_")
|
||||
}
|
||||
assert set(registered) == public
|
||||
Reference in New Issue
Block a user