import pytest
import httpx
import os
import uuid
from conftest import BASE_URL
def _headers():
uid = os.getenv("SYNC_TEST_USER_ID", "")
return {"X-User-ID": uid}
def test_agent_lifecycle_and_api_coverage():
"""
Test suite covering Agent API endpoints:
1. Register Node
2. Register Template
3. Register Session
4. Register Instance
5. Register Trigger
6. Verify Agent Periodical Execution via Session Messages
7. List Agents
8. Stop Agent
9. Remove Agent
"""
node_id = f"test-agent-node-{uuid.uuid4().hex[:8]}"
admin_id = os.getenv("SYNC_TEST_USER_ID", "")
with httpx.Client(timeout=10.0) as client:
# 1. Register a test node specifically for this agent testing
node_payload = {
"node_id": node_id,
"display_name": "Agent Test Node",
"is_active": True,
"skill_config": {"shell": {"enabled": True}}
}
r_node = client.post(f"{BASE_URL}/nodes/admin", params={"admin_id": admin_id}, json=node_payload)
# If conflicts, clear first
if r_node.status_code in (400, 409):
client.delete(f"{BASE_URL}/nodes/admin/{node_id}", params={"admin_id": admin_id})
r_node = client.post(f"{BASE_URL}/nodes/admin", params={"admin_id": admin_id}, json=node_payload)
assert r_node.status_code == 200, f"Node registration failed: {r_node.text}"
# 2. Deploy Agent using the unified endpoint (matching UI behavior)
deploy_payload = {
"name": "Cron Print Agent",
"description": "Periodically prints to the node console",
"system_prompt": "You are a cron agent. Run shell tasks periodically.",
"max_loop_iterations": 1,
"mesh_node_id": node_id,
"provider_name": "gemini",
"trigger_type": "interval",
"interval_seconds": 5,
"default_prompt": "Hello test agent! Just reply the word 'Acknowledged' and nothing else.",
"initial_prompt": None
}
r_deploy = client.post(f"{BASE_URL}/agents/deploy", json=deploy_payload, headers=_headers())
assert r_deploy.status_code == 200, f"Deploy unified endpoint failed: {r_deploy.text}"
deploy_res = r_deploy.json()
instance_id = deploy_res["instance_id"]
session_id = deploy_res["session_id"]
template_id = deploy_res["template_id"]
# 3. VERIFY NODE BINDING (Fixing the exact edge case)
r_sess_check = client.get(f"{BASE_URL}/sessions/{session_id}", headers=_headers())
assert r_sess_check.status_code == 200, "Could not fetch agent session"
assert node_id in (r_sess_check.json().get("attached_node_ids") or []), "Node ID was NOT attached to the session during deployment!"
# We need to fetch the trigger ID for later checks
r_trig_get = client.get(f"{BASE_URL}/agents/{instance_id}/triggers", headers=_headers())
trigger_id = r_trig_get.json()[0]["id"]
# 6. Verify Agent Periodical Execution
print("\n[test] Waiting for background interval scheduler to wake the agent (timeout 60s)...")
import time
messages = []
for _ in range(30): # 30 * 2s = 60s
r_msgs = client.get(f"{BASE_URL}/sessions/{session_id}/messages", headers=_headers())
assert r_msgs.status_code == 200, f"Failed to fetch session messages: {r_msgs.text}"
messages = r_msgs.json()["messages"]
if any(m["sender"] == "assistant" for m in messages):
break
time.sleep(2)
print(f"\n[test] Agent Messages Count: {len(messages)}")
assert any(m["sender"] == "assistant" for m in messages), f"The agent failed to generate any response within 60s! History: {messages}"
# 7. Test if agent is in the active list
r_list = client.get(f"{BASE_URL}/agents", headers=_headers())
assert r_list.status_code == 200
agents = r_list.json()
assert any(a["id"] == instance_id for a in agents), "Instance not found in active list"
# Fetch triggers back
r_trig_get = client.get(f"{BASE_URL}/agents/{instance_id}/triggers", headers=_headers())
assert r_trig_get.status_code == 200
assert any(t["id"] == trigger_id for t in r_trig_get.json()), "Trigger not found on instance"
# 8. Test Stop Update / Config Update
r_stop = client.patch(f"{BASE_URL}/agents/{instance_id}/status", json={"status": "stopped"}, headers=_headers())
assert r_stop.status_code == 200
assert r_stop.json()["status"] == "stopped"
r_cfg = client.patch(f"{BASE_URL}/agents/{instance_id}/config", json={"name": "Updated Cron Agent", "mesh_node_id": node_id}, headers=_headers())
assert r_cfg.status_code == 200
# 9. Test Remove (delete agent directly, verifying cascading trigger deletion)
r_del_agent = client.delete(f"{BASE_URL}/agents/{instance_id}", headers=_headers())
assert r_del_agent.status_code == 200, f"Cascading delete failed: {r_del_agent.text}"
# Cleanup Node
client.delete(f"{BASE_URL}/nodes/admin/{node_id}", params={"admin_id": admin_id})
def test_agent_webhook_trigger():
"""
Test Agent Webhook Triggering:
1. Deploy agent with webhook trigger
2. Obtain secret token
3. Call webhook with custom prompt + token
4. Verify response contains custom prompt indicator
"""
node_id = f"test-webhook-node-{uuid.uuid4().hex[:8]}"
admin_id = os.getenv("SYNC_TEST_USER_ID", "")
with httpx.Client(timeout=10.0) as client:
# 1. Register a test node
node_payload = {
"node_id": node_id,
"display_name": "Webhook Test Node",
"is_active": True,
"skill_config": {"shell": {"enabled": True}}
}
client.post(f"{BASE_URL}/nodes/admin", params={"admin_id": admin_id}, json=node_payload)
# 2. Deploy Agent with Webhook Trigger
deploy_payload = {
"name": "Webhook Agent",
"system_prompt": "You are a helpful assistant. Just acknowledge the received webhook prompt.",
"max_loop_iterations": 1,
"mesh_node_id": node_id,
"provider_name": "gemini",
"trigger_type": "webhook",
"default_prompt": "Standard Webhook Prompt",
"initial_prompt": None
}
r_deploy = client.post(f"{BASE_URL}/agents/deploy", json=deploy_payload, headers=_headers())
assert r_deploy.status_code == 200, f"Deploy failed: {r_deploy.text}"
deploy_res = r_deploy.json()
instance_id = deploy_res["instance_id"]
session_id = deploy_res["session_id"]
# 3. Get the webhook secret
r_trig = client.get(f"{BASE_URL}/agents/{instance_id}/triggers", headers=_headers())
assert r_trig.status_code == 200
webhook_trigger = next(t for t in r_trig.json() if t["trigger_type"] == "webhook")
secret = webhook_trigger["webhook_secret"]
assert secret is not None
# 4. Trigger the Webhook with a custom prompt
custom_msg = "INTER-AGENT-SIGNAL-BEEP-BOOP"
r_hook = client.post(
f"{BASE_URL}/agents/{instance_id}/webhook",
params={"token": secret},
json={"prompt": f"Please respond exactly with: {custom_msg}"}
)
assert r_hook.status_code == 202, f"Webhook trigger failed: {r_hook.text}"
# 5. Wait for agent to process
print(f"\n[test] Waiting for agent to process webhook signal '{custom_msg}'...")
import time
found = False
for _ in range(30):
r_msgs = client.get(f"{BASE_URL}/sessions/{session_id}/messages", headers=_headers())
msgs = r_msgs.json()["messages"]
# Look for assistant response containing our custom signal
if any(custom_msg in (m.get("content") or "") for m in msgs if m["sender"] == "assistant"):
found = True
break
time.sleep(2)
assert found, "The agent did not process the custom webhook prompt correctly."
print("[test] Webhook custom prompt processed successfully!")
# 6. Cleanup
client.delete(f"{BASE_URL}/agents/{instance_id}", headers=_headers())
client.delete(f"{BASE_URL}/nodes/admin/{node_id}", params={"admin_id": admin_id})
