test_cases.py•3.35 kB
#!/usr/bin/env python3
"""
Simple test cases for the complete workflow
Each test generates its own log file
"""
import asyncio
import logging
from datetime import datetime
from agents.action_generation import action_agent, input_agent, task_agent, State
from agents.browser_loop import browser_loop
async def run_test(test_name, action_description, website_url):
"""Run a single test case and log to its own file"""
log_file = f"logs/{test_name}.log"
# Configure logging for this test
logger = logging.getLogger(test_name)
logger.setLevel(logging.INFO)
# Clear any existing handlers
logger.handlers = []
# Add file and console handlers
file_handler = logging.FileHandler(log_file, mode='w')
console_handler = logging.StreamHandler()
formatter = logging.Formatter('%(message)s')
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
logger.info("="*80)
logger.info(f"TEST: {test_name}")
logger.info("="*80)
logger.info("")
# Create initial state
state = State(
action_description=action_description,
website_url=website_url
)
logger.info(f"Action: {action_description}")
logger.info(f"Website: {website_url}")
logger.info("")
# Run pipeline
logger.info("Running action_agent...")
state = await action_agent(state)
logger.info(f"✓ Function: {state.function_metadata.function_name}")
logger.info("")
logger.info("Running input_agent...")
state = await input_agent(state)
logger.info(f"✓ Generated {len(state.test_inputs)} test inputs")
logger.info("")
logger.info("Running task_agent...")
state = await task_agent(state)
logger.info(f"✓ Task description created")
logger.info("")
logger.info("Running browser_loop...")
logger.info("")
state = await browser_loop(state)
logger.info("")
logger.info("="*80)
logger.info(f"FINAL RESULT: {state.result}")
logger.info("="*80)
logger.info("")
logger.info(f"Log saved to: {log_file}")
return state
async def main():
"""Run all test cases"""
test_cases = [
{
"name": "github_search",
"action": "search for repositories on GitHub",
"website": "https://github.com"
},
{
"name": "google_search",
"action": "search for a query on Google",
"website": "https://google.com"
},
{
"name": "hackernews_top",
"action": "get the top stories from HackerNews",
"website": "https://news.ycombinator.com"
}
]
print(f"\nRunning {len(test_cases)} test cases...\n")
for test in test_cases:
print(f"\n{'='*80}")
print(f"Starting test: {test['name']}")
print(f"{'='*80}\n")
await run_test(
test_name=test['name'],
action_description=test['action'],
website_url=test['website']
)
print(f"\n✓ Test {test['name']} completed\n")
print(f"\n{'='*80}")
print("All tests completed!")
print(f"{'='*80}\n")
if __name__ == "__main__":
# Create logs directory
import os
os.makedirs("logs", exist_ok=True)
asyncio.run(main())