#!/usr/bin/env python3
"""
Test script to verify the Keet CLI agent functionality.
"""
import asyncio
import json
import logging
import os
import sys
import tempfile
from pathlib import Path
from ollamatools.plugins.integrations.keet.keet_cmd_agent import KeetCmdAgent
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def test_keet_cmd_agent():
"""Test the Keet CLI agent functionality."""
try:
# Create a temporary directory for the test
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# Create a mock room
room_key = "test-room-key"
rooms_file = temp_path / "rooms.json"
with open(rooms_file, "w") as f:
json.dump([], f)
# Create a mock identity
identity_file = temp_path / "identity.json"
with open(identity_file, "w") as f:
json.dump({
"publicKey": "mock-user",
"displayName": "Mock User"
}, f)
# Create a mock messages file
messages_file = temp_path / f"messages_{room_key}.json"
with open(messages_file, "w") as f:
json.dump([], f)
# Start the agent with mock mode
logger.info("Starting Keet CLI agent in mock mode...")
agent = KeetCmdAgent(
storage_path=temp_path,
room_key=room_key,
name="TestBot",
model="llama3.1:latest", # This won't actually be used in the test
ollama_host="localhost:11434",
prompt="You are a helpful assistant.",
mock=True,
verbose=True
)
# Start the agent
await agent.start()
# Wait for the agent to initialize
logger.info("Waiting for agent to initialize...")
await asyncio.sleep(2)
# Simulate a user message
logger.info("Simulating a user message...")
# Read the current messages
with open(messages_file, "r") as f:
messages = json.load(f)
# Add a test message from another user
test_message = {
"id": "test-message",
"timestamp": int(asyncio.get_event_loop().time() * 1000),
"author": "another-user",
"chat": {
"text": "Hello, bot! Can you help me?"
}
}
messages.append(test_message)
# Write the updated messages
with open(messages_file, "w") as f:
json.dump(messages, f)
# Wait for the agent to process the message
logger.info("Waiting for agent to process the message...")
await asyncio.sleep(5)
# Check if the agent responded
with open(messages_file, "r") as f:
updated_messages = json.load(f)
# Find the agent's response
agent_responses = [
msg for msg in updated_messages
if msg.get("author") == "mock-user" and msg.get("id") != "test-message"
]
if agent_responses:
logger.info(f"Agent responded with: {agent_responses[-1].get('chat', {}).get('text')}")
logger.info("Test successful!")
else:
logger.error("Agent did not respond to the message")
return False
# Stop the agent
logger.info("Stopping the agent...")
await agent.stop()
return True
except Exception as e:
logger.error(f"Error in Keet CLI agent test: {e}")
return False
if __name__ == "__main__":
success = asyncio.run(test_keet_cmd_agent())
if success:
logger.info("Keet CLI agent test successful!")
sys.exit(0)
else:
logger.error("Keet CLI agent test failed!")
sys.exit(1)