test_kush.py•2.13 kB
#!/usr/bin/env python3
"""Simple test: Get email from kush.pw"""
import asyncio
import logging
from agents.action_generation import action_agent, input_agent, task_agent, State
from agents.browser_loop import browser_loop
async def main():
# Configure root logger so all module loggers inherit
logging.basicConfig(
level=logging.INFO,
format='%(message)s',
handlers=[
logging.FileHandler('logs/kush_test.log', mode='w'),
logging.StreamHandler()
],
force=True # Override any existing logging config
)
logger = logging.getLogger(__name__)
logger.info("="*80)
logger.info("TEST: Get email from kush.pw")
logger.info("="*80)
logger.info("")
# Create initial state
state = State(
action_description="get the email address from the website",
website_url="https://kush.pw"
)
logger.info("Step 1: action_agent")
logger.info(f" Input: action_description={state.action_description}, website_url={state.website_url}")
state = await action_agent(state)
logger.info(f" Output: {state.function_metadata}")
logger.info("")
logger.info("Step 2: input_agent")
logger.info(f" Input: function_metadata={state.function_metadata}")
state = await input_agent(state)
# Override with no inputs since this task needs none
state.test_inputs = [{}]
logger.info(f" Output: {state.test_inputs}")
logger.info("")
logger.info("Step 3: task_agent")
logger.info(f" Input: function_metadata={state.function_metadata}")
state = await task_agent(state)
logger.info(f" Output: {state.task_description}")
logger.info("")
logger.info("Step 4: browser_loop")
logger.info("")
state = await browser_loop(state)
logger.info("")
logger.info("="*80)
logger.info(f"RESULT: {state.result}")
logger.info(f"Test results: {state.test_results}")
logger.info(f"Winning task: {state.winning_task if hasattr(state, 'winning_task') and state.winning_task else 'None'}")
logger.info("="*80)
if __name__ == "__main__":
asyncio.run(main())