giers10/Heimgeist
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Enhance workflow router tests to cover web search scenarios and validate fast path permissions

Browse Source
This commit is contained in:
Victor Giers
2026-06-15 20:28:39 +02:00
parent 83cde1011a
commit 8b37a8ef58
1 changed files with 25 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
backend/tests/test_workflow_router_builtins.py
View File

@@ -33,7 +33,7 @@ class RouterAndBuiltinTests(unittest.IsolatedAsyncioTestCase):
db = self.Session()
web = db.query(WorkflowDefinition).filter_by(slug="web-answer").one()
with patch("backend.agent.router.chat_typed", new=AsyncMock(return_value=OllamaChatResult(content=json.dumps({"workflow_id": web.id, "confidence": 0.9, "reason": "current", "inputs": {}})))):
result = await select_workflow(db, message="latest news", recent_messages=[], attachments=[], library_slug=None, router_model=None, chat_model="model")
result = await select_workflow(db, message="what changed recently?", recent_messages=[], attachments=[], library_slug=None, router_model=None, chat_model="model", web_search_enabled=True)
self.assertEqual(result["workflow_slug"], "web-answer")
with patch("backend.agent.router.chat_typed", new=AsyncMock(return_value=OllamaChatResult(content="not json"))):
self.assertEqual((await select_workflow(db, message="x", recent_messages=[], attachments=[], library_slug=None, router_model=None, chat_model="model"))["workflow_slug"], "input-output")
@@ -42,7 +42,30 @@ class RouterAndBuiltinTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual((await select_workflow(db, message="x", recent_messages=[], attachments=[], library_slug=None, router_model=None, chat_model="model"))["workflow_slug"], "input-output")
web.enabled = True; db.commit()
with patch("backend.agent.router.chat_typed", new=AsyncMock(return_value=OllamaChatResult(content=json.dumps({"workflow_id": web.id, "confidence": 0.2, "reason": "x", "inputs": {}})))):
self.assertEqual((await select_workflow(db, message="x", recent_messages=[], attachments=[], library_slug=None, router_model=None, chat_model="model"))["workflow_slug"], "input-output")
self.assertEqual((await select_workflow(db, message="x", recent_messages=[], attachments=[], library_slug=None, router_model=None, chat_model="model", web_search_enabled=True))["workflow_slug"], "input-output")
db.close()
async def test_fast_paths_respect_permissions(self):
db = self.Session()
with patch("backend.agent.router.chat_typed", new=AsyncMock()) as mocked:
result = await select_workflow(
db, message="latest news today", recent_messages=[], attachments=[],
library_slug=None, router_model=None, chat_model="model", web_search_enabled=True,
)
self.assertEqual(result["workflow_slug"], "web-answer")
mocked.assert_not_awaited()
result = await select_workflow(
db, message="latest news today", recent_messages=[], attachments=[],
library_slug=None, router_model=None, chat_model="model", web_search_enabled=False,
)
self.assertEqual(result["workflow_slug"], "input-output")
result = await select_workflow(
db, message="remember this: I prefer short answers", recent_messages=[], attachments=[],
library_slug="memory", router_model=None, chat_model="model", web_search_enabled=False,
)
self.assertEqual(result["workflow_slug"], "remember-this")
db.close()
async def test_missing_router_model_falls_back_to_chat_model(self):