ghmkandClaude Opus 4.6
d5bd8521f3
New nodes (5):
- Relay: local message delivery between agents (outbox→inbox)
- ChatHistory: multi-turn conversation persistence across runs
- PromptTemplate: {a}/{b}/{c}/{d} variable substitution
- ConditionalRouter: keyword/regex routing with matched/unmatched outputs
- BatchSplitter: split text into numbered sections by delimiter
Fixed nodes (3):
- TextCombine: removed forceInput, text_a/text_b now typeable widgets
- Remember: removed forceInput on text, now typeable widget
- Send: removed forceInput on body, now typeable widget
Demo workflows (7):
1. Character Design - art director with memory for consistent characters
2. Novel Crafter - writer+editor dual-agent with Relay message delivery
3. Story to Video - director breaks story into scenes with BatchSplitter
4. Multi-Turn Chat - persistent conversation history across queue runs
5. Agent Debate - two agents argue via message passing (FOR vs AGAINST)
6. World Builder - game world with PromptTemplate for locations/characters
7. RAG Pipeline - store docs, search, answer with ConditionalRouter error handling
76 tests all passing (46 original + 30 new).
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
"""Tests for AgentAZAll ComfyUI messaging nodes (Send, Inbox, ReadMessage)."""
|
|
|
|
import pytest
|
|
|
|
from agentazall.helpers import agent_day, today_str, ensure_dirs
|
|
from agentazall.config import INBOX
|
|
from agentazall.messages import format_message
|
|
|
|
from nodes.messaging_nodes import AgentAZAll_Send, AgentAZAll_Inbox, AgentAZAll_ReadMessage
|
|
|
|
|
|
class TestSend:
|
|
|
|
def test_send_creates_outbox_file(self, agent_config):
|
|
"""Send queues a message in the outbox directory."""
|
|
node = AgentAZAll_Send()
|
|
(result,) = node.send(
|
|
agent_config,
|
|
"other-agent@localhost",
|
|
"Hello",
|
|
"Hi there!",
|
|
)
|
|
assert "Message queued" in result
|
|
assert "other-agent@localhost" in result
|
|
|
|
# Verify file exists in outbox
|
|
outbox = agent_day(agent_config, today_str()) / "outbox"
|
|
files = list(outbox.glob("*.txt"))
|
|
assert len(files) == 1
|
|
|
|
def test_send_format_correct(self, agent_config):
|
|
"""Send creates correctly formatted message file."""
|
|
node = AgentAZAll_Send()
|
|
node.send(agent_config, "bob@localhost", "Test Subject", "Test body content.")
|
|
|
|
outbox = agent_day(agent_config, today_str()) / "outbox"
|
|
msg_file = next(outbox.glob("*.txt"))
|
|
content = msg_file.read_text(encoding="utf-8")
|
|
|
|
assert "From: test-agent@localhost" in content
|
|
assert "To: bob@localhost" in content
|
|
assert "Subject: Test Subject" in content
|
|
assert "Test body content." in content
|
|
|
|
def test_send_requires_to(self, agent_config):
|
|
"""Send returns error when 'to' is empty."""
|
|
node = AgentAZAll_Send()
|
|
(result,) = node.send(agent_config, "", "Subject", "Body")
|
|
assert "Error" in result
|
|
|
|
def test_send_requires_subject(self, agent_config):
|
|
"""Send returns error when 'subject' is empty."""
|
|
node = AgentAZAll_Send()
|
|
(result,) = node.send(agent_config, "bob@localhost", "", "Body")
|
|
assert "Error" in result
|
|
|
|
|
|
class TestInbox:
|
|
|
|
def test_inbox_empty(self, agent_config):
|
|
"""Inbox returns empty message when no mail."""
|
|
node = AgentAZAll_Inbox()
|
|
(result,) = node.inbox(agent_config, False)
|
|
assert "empty" in result.lower()
|
|
|
|
def test_inbox_with_messages(self, agent_config):
|
|
"""Inbox lists received messages."""
|
|
# Manually place a message in the inbox
|
|
d = today_str()
|
|
ensure_dirs(agent_config, d)
|
|
inbox_dir = agent_day(agent_config, d) / INBOX
|
|
content, msg_id = format_message(
|
|
"sender@localhost", agent_config["agent_name"],
|
|
"Test Subject", "Test body",
|
|
)
|
|
(inbox_dir / f"{msg_id}.txt").write_text(content, encoding="utf-8")
|
|
|
|
node = AgentAZAll_Inbox()
|
|
(result,) = node.inbox(agent_config, False)
|
|
assert "sender@localhost" in result
|
|
assert "Test Subject" in result
|
|
assert "1 message" in result
|
|
|
|
|
|
class TestReadMessage:
|
|
|
|
def test_read_message_found(self, agent_config):
|
|
"""ReadMessage reads a message by ID."""
|
|
d = today_str()
|
|
ensure_dirs(agent_config, d)
|
|
inbox_dir = agent_day(agent_config, d) / INBOX
|
|
content, msg_id = format_message(
|
|
"alice@localhost", agent_config["agent_name"],
|
|
"Important", "Please read this.",
|
|
)
|
|
(inbox_dir / f"{msg_id}.txt").write_text(content, encoding="utf-8")
|
|
|
|
node = AgentAZAll_ReadMessage()
|
|
(result,) = node.read_message(agent_config, msg_id)
|
|
assert "alice@localhost" in result
|
|
assert "Important" in result
|
|
assert "Please read this." in result
|
|
|
|
def test_read_message_not_found(self, agent_config):
|
|
"""ReadMessage returns error for non-existent message."""
|
|
node = AgentAZAll_ReadMessage()
|
|
(result,) = node.read_message(agent_config, "nonexistent123")
|
|
assert "not found" in result.lower()
|
|
|
|
def test_read_message_marks_read(self, agent_config):
|
|
"""ReadMessage marks message status from 'new' to 'read'."""
|
|
d = today_str()
|
|
ensure_dirs(agent_config, d)
|
|
inbox_dir = agent_day(agent_config, d) / INBOX
|
|
content, msg_id = format_message(
|
|
"bob@localhost", agent_config["agent_name"],
|
|
"Check", "Body text.",
|
|
)
|
|
msg_path = inbox_dir / f"{msg_id}.txt"
|
|
msg_path.write_text(content, encoding="utf-8")
|
|
|
|
# Verify initial status is 'new'
|
|
assert "Status: new" in msg_path.read_text(encoding="utf-8")
|
|
|
|
node = AgentAZAll_ReadMessage()
|
|
node.read_message(agent_config, msg_id)
|
|
|
|
# Verify status changed to 'read'
|
|
assert "Status: read" in msg_path.read_text(encoding="utf-8")
|