Abdelrahman Abdallah
c17c2768eb
New runtime/code: - src/ask_user_runtime.py - src/team_runtime.py New real tools in src/agent_tools.py: - ask_user_question - team_create - team_delete - team_list - team_get - send_message - team_messages - notebook_edit
85 lines
3.2 KiB
Python
85 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from src.agent_tools import build_tool_context, default_tool_registry, execute_tool
|
|
from src.agent_types import AgentPermissions, AgentRuntimeConfig
|
|
from src.remote_trigger_runtime import RemoteTriggerRuntime
|
|
|
|
|
|
class RemoteTriggerRuntimeTests(unittest.TestCase):
|
|
def test_remote_trigger_runtime_discovers_and_runs_trigger(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
workspace = Path(tmp_dir)
|
|
(workspace / '.claw-triggers.json').write_text(
|
|
(
|
|
'{"triggers":['
|
|
'{"trigger_id":"nightly","name":"Nightly","workflow":"review",'
|
|
'"schedule":"0 0 * * *","body":{"depth":"full"}}'
|
|
']}'
|
|
),
|
|
encoding='utf-8',
|
|
)
|
|
runtime = RemoteTriggerRuntime.from_workspace(workspace)
|
|
trigger_report = runtime.render_trigger('nightly')
|
|
run_report = runtime.render_run_report('nightly', body={'depth': 'quick'})
|
|
|
|
self.assertIn('trigger_id=nightly', trigger_report)
|
|
self.assertIn('workflow=review', run_report)
|
|
self.assertIn('"depth": "quick"', run_report)
|
|
|
|
def test_remote_trigger_tool_supports_create_update_run(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
workspace = Path(tmp_dir)
|
|
runtime = RemoteTriggerRuntime.from_workspace(workspace)
|
|
context = build_tool_context(
|
|
AgentRuntimeConfig(
|
|
cwd=workspace,
|
|
permissions=AgentPermissions(allow_file_write=True),
|
|
),
|
|
remote_trigger_runtime=runtime,
|
|
)
|
|
create_result = execute_tool(
|
|
default_tool_registry(),
|
|
'remote_trigger',
|
|
{
|
|
'action': 'create',
|
|
'body': {
|
|
'trigger_id': 'nightly',
|
|
'name': 'Nightly',
|
|
'workflow': 'review',
|
|
'body': {'depth': 'full'},
|
|
},
|
|
},
|
|
context,
|
|
)
|
|
update_result = execute_tool(
|
|
default_tool_registry(),
|
|
'remote_trigger',
|
|
{
|
|
'action': 'update',
|
|
'trigger_id': 'nightly',
|
|
'body': {'schedule': '0 0 * * *'},
|
|
},
|
|
context,
|
|
)
|
|
run_result = execute_tool(
|
|
default_tool_registry(),
|
|
'remote_trigger',
|
|
{
|
|
'action': 'run',
|
|
'trigger_id': 'nightly',
|
|
'body': {'depth': 'quick'},
|
|
},
|
|
context,
|
|
)
|
|
|
|
self.assertTrue(create_result.ok)
|
|
self.assertEqual(create_result.metadata.get('trigger_id'), 'nightly')
|
|
self.assertTrue(update_result.ok)
|
|
self.assertEqual(update_result.metadata.get('remote_trigger_action'), 'update')
|
|
self.assertTrue(run_result.ok)
|
|
self.assertIn('# Remote Trigger Run', run_result.content)
|