feat(rulebook): service layer — Rule CRUD with multi-filter list_rules
This commit is contained in:
@@ -156,3 +156,68 @@ async def test_list_topics_returns_topics_for_owned_rulebook():
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user