bvandeusen
43a860c3ac
Rules can now belong to either a rulebook topic OR a single project,
enforced by a CHECK constraint (exactly-one of topic_id/project_id).
Adds the create_project_rule MCP tool + REST endpoint, surfaces
project-scoped rules in get_project/get_task/start_planning under a
new project_rules field, and adds a project Rules tab section with an
inline create form so the operator can author project rules from the
UI without rulebook ceremony.
- migration 0059: rules.project_id (FK projects ON DELETE CASCADE),
topic_id now nullable, CHECK ck_rule_topic_xor_project, index on
project_id
- model: Rule gains project_id; to_dict exposes it
- service: create_project_rule with project-ownership guard; list_rules
with project_id filter UNIONs subscription-derived + project-scoped;
get_applicable_rules adds a project_rules field; get_rule / update_rule
/ delete_rule fetch via a shared _fetch_owned_rule that handles both
rulebook and project ownership paths
- trash: project delete cascades to project-scoped rules
- MCP: create_project_rule tool registered; _INSTRUCTIONS mentions both
create_rule and create_project_rule paths
- REST: POST /api/projects/<id>/rules (statement required, title derived
if omitted)
- frontend: Rule type gains nullable topic_id + project_id; createProjectRule
client; ProjectRulesTab.vue gains a "Project rules" section with inline
create form and per-rule expand/delete
- tests: register count → 18; create_project_rule unit tests (required
fields, title derivation, explicit-title pass-through); applicable_rules
shape tests now include project_rules; trash cascade test updated to
expect 5 executions
S1+S2 (always_on flag + Scribe-first prompt) shipped in 658348f.
S4 (enter_project handshake) follows.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
321 lines
13 KiB
Python
321 lines
13 KiB
Python
"""Tests for services/rulebooks.py — mocks async_session, no real DB.
|
|
|
|
Mirrors the pattern in tests/test_events_service.py.
|
|
"""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
|
|
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
|
|
|
|
|
|
def _fake_rulebook(id=1, owner_user_id=7, title="FabledSword family", description=""):
|
|
rb = MagicMock()
|
|
rb.id = id
|
|
rb.owner_user_id = owner_user_id
|
|
rb.title = title
|
|
rb.description = description
|
|
rb.created_at = datetime.now(timezone.utc)
|
|
rb.updated_at = datetime.now(timezone.utc)
|
|
rb.to_dict.return_value = {
|
|
"id": id, "owner_user_id": owner_user_id,
|
|
"title": title, "description": description or "",
|
|
}
|
|
return rb
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_rulebook_stores_to_db():
|
|
mock_session = _make_mock_session()
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import create_rulebook
|
|
await create_rulebook(
|
|
user_id=7, title="FabledSword family", description="rules for the family",
|
|
)
|
|
assert mock_session.add.called
|
|
assert mock_session.commit.called
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_rulebooks_returns_owned_only():
|
|
rb = _fake_rulebook(id=1)
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [rb]
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import list_rulebooks
|
|
results = await list_rulebooks(user_id=7)
|
|
assert len(results) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rulebook_returns_none_when_not_owner():
|
|
"""get_rulebook scopes by owner_user_id — wrong user gets None."""
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import get_rulebook
|
|
result = await get_rulebook(rulebook_id=1, user_id=99)
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_rulebook_only_sets_provided_fields():
|
|
rb = _fake_rulebook(id=1, title="old")
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = rb
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import update_rulebook
|
|
await update_rulebook(rulebook_id=1, user_id=7, title="new")
|
|
assert rb.title == "new"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_rulebook_calls_delete():
|
|
rb = _fake_rulebook(id=1)
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = rb
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
mock_session.delete = AsyncMock()
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import delete_rulebook
|
|
await delete_rulebook(rulebook_id=1, user_id=7)
|
|
assert mock_session.delete.called
|
|
|
|
|
|
# ── Topic CRUD ───────────────────────────────────────────────────────────
|
|
|
|
def _fake_topic(id=1, rulebook_id=1, title="git-workflow", description="", order_index=0):
|
|
t = MagicMock()
|
|
t.id = id
|
|
t.rulebook_id = rulebook_id
|
|
t.title = title
|
|
t.description = description
|
|
t.order_index = order_index
|
|
t.created_at = datetime.now(timezone.utc)
|
|
t.updated_at = datetime.now(timezone.utc)
|
|
t.to_dict.return_value = {
|
|
"id": id, "rulebook_id": rulebook_id, "title": title,
|
|
"description": description or "", "order_index": order_index,
|
|
}
|
|
return t
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_topic_requires_owned_rulebook():
|
|
"""create_topic raises ValueError if the rulebook isn't owned by user."""
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import create_topic
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await create_topic(
|
|
rulebook_id=999, user_id=7, title="git-workflow",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_topics_returns_topics_for_owned_rulebook():
|
|
rb = _fake_rulebook(id=1)
|
|
topic = _fake_topic(id=10, rulebook_id=1, title="git-workflow")
|
|
|
|
# Two execute calls: ownership check, then topic select.
|
|
mock_session = _make_mock_session()
|
|
rb_result = MagicMock()
|
|
rb_result.scalar_one_or_none.return_value = rb
|
|
topic_result = MagicMock()
|
|
topic_result.scalars.return_value.all.return_value = [topic]
|
|
mock_session.execute = AsyncMock(side_effect=[rb_result, topic_result])
|
|
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import list_topics
|
|
results = await list_topics(rulebook_id=1, user_id=7)
|
|
assert len(results) == 1
|
|
assert results[0].title == "git-workflow"
|
|
|
|
|
|
# ── Rule CRUD ───────────────────────────────────────────────────────────
|
|
|
|
def _fake_rule(id=1, topic_id=10, title="dev is home",
|
|
statement="Work directly on dev", why="", how_to_apply=""):
|
|
r = MagicMock()
|
|
r.id = id
|
|
r.topic_id = topic_id
|
|
r.title = title
|
|
r.statement = statement
|
|
r.why = why
|
|
r.how_to_apply = how_to_apply
|
|
r.order_index = 0
|
|
r.created_at = datetime.now(timezone.utc)
|
|
r.updated_at = datetime.now(timezone.utc)
|
|
r.to_dict.return_value = {
|
|
"id": id, "topic_id": topic_id, "title": title,
|
|
"statement": statement, "why": why or "",
|
|
"how_to_apply": how_to_apply or "", "order_index": 0,
|
|
}
|
|
return r
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_rule_requires_owned_topic():
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None # topic not found
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import create_rule
|
|
with pytest.raises(ValueError, match="topic .* not found"):
|
|
await create_rule(
|
|
topic_id=999, user_id=7, title="x", statement="y",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_rules_filters_by_topic_id():
|
|
"""list_rules(topic_id=X) returns rules in that topic, ownership-scoped."""
|
|
rule = _fake_rule(id=1, topic_id=10)
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [rule]
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import list_rules
|
|
results = await list_rules(user_id=7, topic_id=10)
|
|
assert len(results) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rule_returns_none_when_not_owner():
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import get_rule
|
|
result = await get_rule(rule_id=1, user_id=99)
|
|
assert result is None
|
|
|
|
|
|
# ── Subscriptions + applicable_rules ────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_subscribe_project_requires_owned_rulebook():
|
|
"""subscribe_project raises if user doesn't own the rulebook."""
|
|
mock_session = _make_mock_session()
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import subscribe_project
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await subscribe_project(
|
|
project_id=1, rulebook_id=999, user_id=7,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_applicable_rules_returns_shape():
|
|
"""get_applicable_rules returns {rules, project_rules, truncated, subscribed_rulebooks}."""
|
|
mock_session = _make_mock_session()
|
|
sub_result = MagicMock()
|
|
sub_result.all.return_value = [(1, "FabledSword family")]
|
|
rules_result = MagicMock()
|
|
rules_result.all.return_value = [
|
|
(i, f"Rule {i}", f"Statement {i}", "git-workflow", "FabledSword family")
|
|
for i in range(50)
|
|
]
|
|
proj_rules_result = MagicMock()
|
|
proj_rules_result.all.return_value = []
|
|
mock_session.execute = AsyncMock(side_effect=[sub_result, rules_result, proj_rules_result])
|
|
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import get_applicable_rules
|
|
result = await get_applicable_rules(project_id=3, user_id=7, limit=50)
|
|
|
|
assert "rules" in result
|
|
assert "project_rules" in result
|
|
assert "truncated" in result
|
|
assert "subscribed_rulebooks" in result
|
|
assert result["subscribed_rulebooks"] == [{"id": 1, "title": "FabledSword family"}]
|
|
assert len(result["rules"]) == 50
|
|
assert result["project_rules"] == []
|
|
assert result["truncated"] is False # exactly 50, not over
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_applicable_rules_truncates_when_over_limit():
|
|
"""When limit+1 rows are returned, truncated=True and only `limit` returned."""
|
|
mock_session = _make_mock_session()
|
|
sub_result = MagicMock()
|
|
sub_result.all.return_value = []
|
|
rules_result = MagicMock()
|
|
# 51 rows means truncation detected
|
|
rules_result.all.return_value = [
|
|
(i, f"r{i}", "stmt", "topic", "rb") for i in range(51)
|
|
]
|
|
proj_rules_result = MagicMock()
|
|
proj_rules_result.all.return_value = []
|
|
mock_session.execute = AsyncMock(side_effect=[sub_result, rules_result, proj_rules_result])
|
|
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import get_applicable_rules
|
|
result = await get_applicable_rules(project_id=3, user_id=7, limit=50)
|
|
|
|
assert result["truncated"] is True
|
|
assert len(result["rules"]) == 50
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_applicable_rules_includes_project_scoped_rules():
|
|
"""Project-scoped rules surface in the project_rules field."""
|
|
mock_session = _make_mock_session()
|
|
sub_result = MagicMock()
|
|
sub_result.all.return_value = []
|
|
rules_result = MagicMock()
|
|
rules_result.all.return_value = []
|
|
proj_rules_result = MagicMock()
|
|
proj_rules_result.all.return_value = [
|
|
(100, "Use alembic", "Always run migrations via alembic, never raw SQL."),
|
|
(101, "PR-bound", "Land schema changes in their own PR."),
|
|
]
|
|
mock_session.execute = AsyncMock(side_effect=[sub_result, rules_result, proj_rules_result])
|
|
|
|
with patch("fabledassistant.services.rulebooks.async_session") as mock_cls:
|
|
mock_cls.return_value = mock_session
|
|
from fabledassistant.services.rulebooks import get_applicable_rules
|
|
result = await get_applicable_rules(project_id=3, user_id=7)
|
|
|
|
assert len(result["project_rules"]) == 2
|
|
assert result["project_rules"][0]["title"] == "Use alembic"
|
|
assert result["project_rules"][1]["id"] == 101
|