4c045e24a2
Some checks failed
Contributor Attribution Check / check-attribution (pull_request) Failing after 59s
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 48s
Tests / e2e (pull_request) Successful in 2m55s
Tests / test (pull_request) Failing after 56m44s
115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
"""Tests for Matrix Bridge — Issue #747."""
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from agent.matrix_bridge import MatrixBridge, AgentMessage, AgentRegistry
|
|
|
|
|
|
class TestMessageParsing:
|
|
"""Test message format parsing."""
|
|
|
|
def test_is_for_me_direct(self):
|
|
bridge = MatrixBridge(agent_name="Timmy")
|
|
assert bridge._is_for_me("[@Timmy] Hello") == True
|
|
|
|
def test_is_not_for_me(self):
|
|
bridge = MatrixBridge(agent_name="Timmy")
|
|
assert bridge._is_for_me("[@Allegro] Hello") == False
|
|
|
|
def test_is_broadcast(self):
|
|
bridge = MatrixBridge(agent_name="Timmy")
|
|
assert bridge._is_for_me("[@*] Broadcast") == True
|
|
|
|
def test_extract_content(self):
|
|
bridge = MatrixBridge(agent_name="Timmy")
|
|
assert bridge._extract_content("[@Timmy] Hello world") == "Hello world"
|
|
|
|
def test_extract_content_multiline(self):
|
|
bridge = MatrixBridge(agent_name="Timmy")
|
|
content = bridge._extract_content("[@Timmy] Line 1\nLine 2")
|
|
assert content == "Line 1\nLine 2"
|
|
|
|
|
|
class TestAgentMessage:
|
|
"""Test AgentMessage dataclass."""
|
|
|
|
def test_to_dict(self):
|
|
msg = AgentMessage(
|
|
sender="Timmy",
|
|
recipient="Allegro",
|
|
content="Hello",
|
|
timestamp=1234567890.0,
|
|
)
|
|
d = msg.to_dict()
|
|
assert d["sender"] == "Timmy"
|
|
assert d["recipient"] == "Allegro"
|
|
assert d["content"] == "Hello"
|
|
|
|
def test_from_dict(self):
|
|
d = {
|
|
"sender": "Timmy",
|
|
"recipient": "Allegro",
|
|
"content": "Hello",
|
|
"timestamp": 1234567890.0,
|
|
"message_id": "",
|
|
"room_id": "",
|
|
}
|
|
msg = AgentMessage.from_dict(d)
|
|
assert msg.sender == "Timmy"
|
|
assert msg.recipient == "Allegro"
|
|
|
|
|
|
class TestAgentRegistry:
|
|
"""Test AgentRegistry."""
|
|
|
|
def test_register(self):
|
|
registry = AgentRegistry()
|
|
registry.register("Timmy", capabilities=["code", "review"])
|
|
agent = registry.get_agent("Timmy")
|
|
assert agent["name"] == "Timmy"
|
|
assert "code" in agent["capabilities"]
|
|
|
|
def test_list_agents(self):
|
|
registry = AgentRegistry()
|
|
registry.register("Timmy")
|
|
registry.register("Allegro")
|
|
agents = registry.list_agents()
|
|
assert len(agents) == 2
|
|
|
|
def test_find_with_capability(self):
|
|
registry = AgentRegistry()
|
|
registry.register("Timmy", capabilities=["code"])
|
|
registry.register("Allegro", capabilities=["monitoring"])
|
|
coders = registry.find_agents_with_capability("code")
|
|
assert "Timmy" in coders
|
|
assert "Allegro" not in coders
|
|
|
|
def test_unregister(self):
|
|
registry = AgentRegistry()
|
|
registry.register("Timmy")
|
|
registry.unregister("Timmy")
|
|
agent = registry.get_agent("Timmy")
|
|
assert agent["status"] == "offline"
|
|
|
|
|
|
class TestBridgeInit:
|
|
"""Test bridge initialization."""
|
|
|
|
def test_default_agent_name(self):
|
|
bridge = MatrixBridge()
|
|
assert bridge.agent_name == "Hermes"
|
|
|
|
def test_custom_agent_name(self):
|
|
bridge = MatrixBridge(agent_name="Timmy")
|
|
assert bridge.agent_name == "Timmy"
|
|
|
|
def test_known_agents_empty(self):
|
|
bridge = MatrixBridge()
|
|
assert len(bridge.get_known_agents()) == 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import pytest
|
|
pytest.main([__file__, "-v"])
|