96 lines
2.4 KiB
Python
96 lines
2.4 KiB
Python
"""Tests for Protocol data structures"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timezone
|
|
|
|
from agentkit.core.protocol import (
|
|
AgentCapability,
|
|
HandoffMessage,
|
|
TaskMessage,
|
|
TaskResult,
|
|
TaskStatus,
|
|
EvolutionEvent,
|
|
)
|
|
|
|
|
|
def test_task_status_values():
|
|
assert TaskStatus.PENDING == "pending"
|
|
assert TaskStatus.RUNNING == "running"
|
|
assert TaskStatus.COMPLETED == "completed"
|
|
assert TaskStatus.FAILED == "failed"
|
|
assert TaskStatus.CANCELLED == "cancelled"
|
|
assert TaskStatus.HANDOFF == "handoff"
|
|
|
|
|
|
def test_agent_capability_with_schema():
|
|
cap = AgentCapability(
|
|
agent_name="test",
|
|
agent_type="test",
|
|
version="1.0.0",
|
|
supported_tasks=["echo"],
|
|
max_concurrency=2,
|
|
description="Test",
|
|
input_schema={"type": "object", "properties": {"x": {"type": "number"}}},
|
|
output_schema={"type": "object"},
|
|
)
|
|
|
|
d = cap.to_dict()
|
|
assert "input_schema" in d
|
|
assert "output_schema" in d
|
|
|
|
restored = AgentCapability.from_dict(d)
|
|
assert restored.agent_name == "test"
|
|
assert restored.input_schema is not None
|
|
|
|
|
|
def test_task_message_roundtrip():
|
|
msg = TaskMessage(
|
|
task_id="123",
|
|
agent_name="agent1",
|
|
task_type="echo",
|
|
priority=1,
|
|
input_data={"key": "value"},
|
|
callback_url=None,
|
|
created_at=datetime.now(timezone.utc),
|
|
conversation_id="conv-1",
|
|
)
|
|
|
|
d = msg.to_dict()
|
|
assert d["conversation_id"] == "conv-1"
|
|
|
|
restored = TaskMessage.from_dict(d)
|
|
assert restored.task_id == "123"
|
|
assert restored.conversation_id == "conv-1"
|
|
|
|
|
|
def test_handoff_message():
|
|
msg = HandoffMessage(
|
|
source_agent="agent_a",
|
|
target_agent="agent_b",
|
|
task_id="task-1",
|
|
task_type="analyze",
|
|
context={"data": "test"},
|
|
reason="Need expert analysis",
|
|
)
|
|
|
|
d = msg.to_dict()
|
|
assert d["source_agent"] == "agent_a"
|
|
assert d["target_agent"] == "agent_b"
|
|
|
|
restored = HandoffMessage.from_dict(d)
|
|
assert restored.reason == "Need expert analysis"
|
|
|
|
|
|
def test_evolution_event():
|
|
event = EvolutionEvent(
|
|
agent_name="optimizer",
|
|
change_type="prompt",
|
|
before={"instruction": "old"},
|
|
after={"instruction": "new"},
|
|
metrics={"quality_delta": 0.15},
|
|
)
|
|
|
|
d = event.to_dict()
|
|
assert d["change_type"] == "prompt"
|
|
assert d["metrics"]["quality_delta"] == 0.15
|