bvandeusen
4dc57f8ab2
CI & Build / Python lint (push) Successful in 2s
CI & Build / Plugin hooks (push) Successful in 7s
CI & Build / integration (push) Successful in 16s
CI & Build / TypeScript typecheck (push) Successful in 21s
CI & Build / Python tests (push) Successful in 42s
CI & Build / Build & push image (push) Successful in 38s
Milestone #254 step 3 (#2288). Reuses #251's prose extractor as the reader and adds the part that makes it an import rather than a list of claims. **The join is the whole trick.** A rulebook states a design system in two places and neither half is a token: one rule names the colours ("Obsidian #14171A (page bg, deepest surface)"), another names the custom properties (`--fs-obsidian/iron/slate`). The import pairs them on the word — `--fs-obsidian` ends with `obsidian` — which is the only reason it produces something usable instead of seventy empty names. The parenthetical becomes the token's purpose, which is the field a bare hex could never carry. **Prohibitions arrive as replacements, per the operator's reframe.** Rule 52 declares Parchment and forbids pure white in one breath, so the import emits "write --fs-parchment instead of #ffffff" — the same fact stated forwards. It attaches to the FIRST token that rule supplied a value for, not to every token of that rule, because claiming Vellum is also the replacement for white would be putting words in the rulebook's mouth. **A token the rulebook names but states no readable value for is still proposed, with an empty value.** Radius steps and type sizes are prose ("Small 4px") and nothing here parses them; inventing a parse per shape would be guessing. The name is real and the value needs a human, so the proposal says exactly that — and the UI leads with the COUNT of those, because an import that hid them would look more complete than it is. Preview is the default on both surfaces and in the UI. An import is a proposal: rulebooks are written aspirationally and some of what they describe was never built, so every entry carries the rule id and the sentence it came from and a reviewer can check the claim rather than trust it. Existing token names are never overwritten. A value already in the record was put there deliberately — most likely correcting this importer — so a re-run fills gaps and lists the rest as skipped, which also makes it safe to repeat. Colours the rulebook names but never exposes as a custom property produce no token: it never asked for one, and inventing a name would put something in the record no rule sanctions.
376 lines
17 KiB
Python
376 lines
17 KiB
Python
"""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()
|
|
|
|
|
|
# --- resolution (step 2) ----------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_denied_returns_none_not_an_empty_set():
|
|
"""None and [] mean different things here: "you may not see this" versus
|
|
"this chain genuinely holds no tokens". A caller that got [] for a denial
|
|
would render an empty design system as though it were real."""
|
|
with patch("scribe.services.design_systems.access") as acc:
|
|
acc.can_read_design_system = AsyncMock(return_value=False)
|
|
from scribe.services.design_systems import resolve_design_system
|
|
assert await resolve_design_system(user_id=1, design_system_id=3) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_scopes_the_hierarchy_to_the_systems_OWNER_not_the_caller():
|
|
"""LOAD-BEARING for shared projects. A caller reading through a project
|
|
shared with them owns no link in the chain, so a caller-scoped parent map
|
|
returns an empty forest and the cascade truncates to one system — a page
|
|
that renders with plausible wrong values and no error anywhere.
|
|
|
|
The ACL already grants read on the whole chain (reachability propagates
|
|
upward); this is the loading side keeping that promise.
|
|
"""
|
|
owner, caller = 42, 7
|
|
system = MagicMock(deleted_at=None, owner_user_id=owner)
|
|
mock_session = _make_mock_session()
|
|
mock_session.get = AsyncMock(return_value=system)
|
|
mock_session.execute = AsyncMock(
|
|
return_value=MagicMock(scalars=MagicMock(return_value=MagicMock(all=lambda: [])))
|
|
)
|
|
parent_map = AsyncMock(return_value={3: None})
|
|
|
|
with patch("scribe.services.design_systems.async_session") as mock_cls, \
|
|
patch("scribe.services.design_systems._parent_map", parent_map), \
|
|
patch("scribe.services.design_systems.access") as acc:
|
|
acc.can_read_design_system = AsyncMock(return_value=True)
|
|
mock_cls.return_value = mock_session
|
|
from scribe.services.design_systems import resolve_design_system
|
|
result = await resolve_design_system(user_id=caller, design_system_id=3)
|
|
|
|
assert result == [] # empty chain, not None
|
|
assert parent_map.await_args.args[1] == owner # NOT `caller`
|
|
|
|
|
|
# --- supersedes (step 6's reframe) ------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_token_supersedes_defaults_to_an_empty_list_not_json_null():
|
|
"""Same NOT NULL reasoning as value_by_mode: absence gets one spelling."""
|
|
mock_session = _make_mock_session()
|
|
captured = {}
|
|
mock_session.add = MagicMock(
|
|
side_effect=lambda obj: captured.update(supersedes=obj.supersedes)
|
|
)
|
|
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-x", supersedes=None)
|
|
assert captured["supersedes"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_token_records_the_literals_it_replaces():
|
|
mock_session = _make_mock_session()
|
|
captured = {}
|
|
mock_session.add = MagicMock(
|
|
side_effect=lambda obj: captured.update(supersedes=obj.supersedes)
|
|
)
|
|
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-text-on-action",
|
|
value_by_mode={"base": "#e8e4d8"}, supersedes=["#fff", "#ffffff"],
|
|
)
|
|
assert captured["supersedes"] == ["#fff", "#ffffff"]
|
|
|
|
|
|
# --- import from a rulebook (step 3) ----------------------------------------
|
|
|
|
def _proposal(name, value=None):
|
|
from scribe.services.design_rulebook_import import ProposedToken
|
|
return ProposedToken(name=name, value_by_mode={"base": value} if value else {})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_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 import_from_rulebook
|
|
assert await import_from_rulebook(1, 3, 9) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preview_proposes_without_creating_anything():
|
|
"""apply=False is the default because an import is a PROPOSAL. A preview
|
|
that quietly wrote would make the review step decorative."""
|
|
created = AsyncMock()
|
|
with patch("scribe.services.design_systems.access") as acc, \
|
|
patch("scribe.services.design_systems.create_token", created), \
|
|
patch("scribe.services.design_systems.list_tokens", AsyncMock(return_value=[])), \
|
|
patch("scribe.services.rulebooks.list_rules", AsyncMock(return_value=[object()])), \
|
|
patch("scribe.services.design_rulebook_import.propose_tokens",
|
|
MagicMock(return_value=[_proposal("--fs-obsidian", "#14171a")])):
|
|
acc.can_write_design_system = AsyncMock(return_value=True)
|
|
from scribe.services.design_systems import import_from_rulebook
|
|
report = await import_from_rulebook(1, 3, 9, apply=False)
|
|
|
|
assert [p["name"] for p in report["proposed"]] == ["--fs-obsidian"]
|
|
assert report["created"] == []
|
|
created.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_never_overwrites_a_token_the_system_already_defines():
|
|
"""LOAD-BEARING for re-running it. A value already in the record was put
|
|
there deliberately — very likely correcting this importer — and a second run
|
|
must fill gaps rather than undo the correction."""
|
|
existing = MagicMock()
|
|
existing.name = "--fs-obsidian"
|
|
created_token = MagicMock()
|
|
created_token.to_dict.return_value = {"name": "--fs-iron"}
|
|
|
|
with patch("scribe.services.design_systems.access") as acc, \
|
|
patch("scribe.services.design_systems.create_token",
|
|
AsyncMock(return_value=created_token)) as create, \
|
|
patch("scribe.services.design_systems.list_tokens",
|
|
AsyncMock(return_value=[existing])), \
|
|
patch("scribe.services.rulebooks.list_rules", AsyncMock(return_value=[object()])), \
|
|
patch("scribe.services.design_rulebook_import.propose_tokens",
|
|
MagicMock(return_value=[
|
|
_proposal("--fs-obsidian", "#000000"),
|
|
_proposal("--fs-iron", "#1e2228"),
|
|
])):
|
|
acc.can_write_design_system = AsyncMock(return_value=True)
|
|
from scribe.services.design_systems import import_from_rulebook
|
|
report = await import_from_rulebook(1, 3, 9, apply=True)
|
|
|
|
assert report["skipped"] == ["--fs-obsidian"]
|
|
assert [c["name"] for c in report["created"]] == ["--fs-iron"]
|
|
assert create.await_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_valueless_proposal_is_still_created():
|
|
"""The rulebook names it, so its absence from the system is itself a
|
|
finding. A named token with a blank value says "this exists and needs
|
|
deciding"; silence says nothing at all."""
|
|
token = MagicMock()
|
|
token.to_dict.return_value = {"name": "--fs-radius-sm"}
|
|
with patch("scribe.services.design_systems.access") as acc, \
|
|
patch("scribe.services.design_systems.create_token",
|
|
AsyncMock(return_value=token)) as create, \
|
|
patch("scribe.services.design_systems.list_tokens", AsyncMock(return_value=[])), \
|
|
patch("scribe.services.rulebooks.list_rules", AsyncMock(return_value=[object()])), \
|
|
patch("scribe.services.design_rulebook_import.propose_tokens",
|
|
MagicMock(return_value=[_proposal("--fs-radius-sm")])):
|
|
acc.can_write_design_system = AsyncMock(return_value=True)
|
|
from scribe.services.design_systems import import_from_rulebook
|
|
report = await import_from_rulebook(1, 3, 9, apply=True)
|
|
|
|
assert len(report["created"]) == 1
|
|
assert create.await_args.kwargs["value_by_mode"] == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_an_unreadable_or_empty_rulebook_reports_nothing_rather_than_failing():
|
|
"""An install can point this at any rulebook, and most rulebooks are not
|
|
design rulebooks (rule #115)."""
|
|
with patch("scribe.services.design_systems.access") as acc, \
|
|
patch("scribe.services.rulebooks.list_rules", AsyncMock(return_value=[])):
|
|
acc.can_write_design_system = AsyncMock(return_value=True)
|
|
from scribe.services.design_systems import import_from_rulebook
|
|
report = await import_from_rulebook(1, 3, 9, apply=True)
|
|
assert report == {"rulebook_id": 9, "proposed": [], "created": [], "skipped": []}
|