#!/usr/bin/env python3
"""Integration test: twerk analysis + terminal execution.
This test proves the full trading integration works:
1. Use twerk to analyze game state and find optimal trade routes
2. Connect and login via terminal
3. Execute trade on best route
4. Verify via both terminal parsing AND twerk file read
Usage:
python scripts/test_trading_integration.py
Requirements:
- TW2002 server running on localhost:3003 (Docker container tw2002-dev)
- Docker volume tw2002-dev-data accessible
"""
from __future__ import annotations
import asyncio
import socket
import sys
from pathlib import Path
# Add src to path for imports
from bbsbot.games.tw2002.bot import TradingBot
from bbsbot.games.tw2002.admin import TW2002Admin
# Docker data volume path - TW2002 game data files
# Note: Docker volume requires sudo, so we copy files to /tmp/tw2002-data
# Use: docker cp tw2002-dev:/opt/tw2002/data/tw*.dat /tmp/tw2002-data/
DOCKER_DATA_PATH = Path("/tmp/tw2002-data")
# Default connection settings
DEFAULT_HOST = "localhost"
DEFAULT_PORT = 2002 # TW2002 telnet port
def is_port_open(host: str, port: int) -> bool:
"""Check if a port is open."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2)
try:
result = sock.connect_ex((host, port))
sock.close()
return result == 0
except socket.error:
return False
async def test_twerk_analysis(data_dir: Path) -> dict:
"""Test twerk file analysis capabilities.
Returns:
Dictionary with game state and trade routes
"""
print("\n" + "=" * 80)
print("PHASE 1: Twerk Analysis (Direct File Access)")
print("=" * 80)
if not data_dir.exists():
print(f" ERROR: Data directory not found: {data_dir}")
print(" Make sure Docker volume is accessible")
return {"error": "data_dir_not_found"}
# Initialize admin in direct mode
admin = TW2002Admin(mode="direct", data_dir=data_dir)
# Get game configuration
print("\n[Config]")
try:
config = await admin.get_config()
print(f" Game Title: {config.get('game_title', 'Unknown')}")
print(f" Turns per day: {config.get('turns_per_day', 'Unknown')}")
print(f" Initial credits: {config.get('initial_credits', 'Unknown')}")
except Exception as e:
print(f" ERROR reading config: {e}")
config = {}
# List players
print("\n[Players]")
try:
players = await admin.list_players()
print(f" Active players: {len(players)}")
for p in players[:5]: # First 5
print(f" - {p.name}: sector {p.sector}, credits {p.credits}")
except Exception as e:
print(f" ERROR listing players: {e}")
players = []
# List ports
print("\n[Ports]")
try:
ports = await admin.list_ports()
print(f" Active ports: {len(ports)}")
except Exception as e:
print(f" ERROR listing ports: {e}")
ports = []
# Find trade routes using twerk
print("\n[Trade Route Analysis]")
bot = TradingBot(twerk_data_dir=data_dir)
try:
routes = await bot.analyze_trade_routes(
data_dir=data_dir,
ship_holds=20, # Default ship capacity
max_hops=5,
)
print(f" Found {len(routes)} trade routes")
if routes:
print("\n Top 5 routes by efficiency:")
for i, route in enumerate(routes[:5]):
print(f" {i + 1}. {route.commodity}: "
f"sector {route.buy_sector} -> {route.sell_sector}")
print(f" Distance: {route.distance} hops, "
f"Profit/unit: {route.profit_per_unit:.1f}, "
f"Efficiency: {route.efficiency_score:.1f}")
except Exception as e:
print(f" ERROR analyzing routes: {e}")
routes = []
return {
"config": config,
"players": players,
"ports": ports,
"routes": routes,
}
async def test_terminal_login(host: str, port: int) -> TradingBot | None:
"""Test terminal login via MCP-BBS.
Returns:
Connected and logged-in TradingBot, or None on failure
"""
print("\n" + "=" * 80)
print("PHASE 2: Terminal Login (MCP-BBS)")
print("=" * 80)
if not is_port_open(host, port):
print(f" ERROR: Server not available at {host}:{port}")
return None
print(f"\n[Connecting to {host}:{port}]")
bot = TradingBot(character_name="claude_test")
try:
await bot.connect(host=host, port=port)
print(" Connected!")
print("\n[Login Sequence]")
await bot.login_sequence(
game_password="game",
character_password="tim",
username="claude",
)
print(" Login complete!")
print("\n[Orientation]")
game_state = await bot.orient()
print(f" Sector: {game_state.sector}")
credits_str = f"{game_state.credits:,}" if game_state.credits is not None else "None"
print(f" Credits: {credits_str}")
print(f" Turns left: {game_state.turns_left}")
print(f" Fighters: {game_state.fighters}")
return bot
except Exception as e:
print(f" ERROR during login: {e}")
import traceback
traceback.print_exc()
return None
async def test_execute_trade(bot: TradingBot, route) -> dict:
"""Execute a single trade cycle using a twerk-analyzed route.
Args:
bot: Connected TradingBot
route: TradeRoute from twerk analysis
Returns:
Trade result dictionary
"""
print("\n" + "=" * 80)
print(f"PHASE 3: Execute Trade Route")
print("=" * 80)
print(f"\n[Route Details]")
print(f" Commodity: {route.commodity}")
print(f" Buy at sector: {route.buy_sector}")
print(f" Sell at sector: {route.sell_sector}")
print(f" Path: {' -> '.join(str(s) for s in route.path)}")
print(f" Expected profit/unit: {route.profit_per_unit:.1f}")
initial_credits = bot.current_credits
initial_sector = bot.current_sector
print(f"\n[Starting State]")
print(f" Credits: {initial_credits:,}")
print(f" Sector: {initial_sector}")
try:
# Execute trading route via twerk pathing
print("\n[Executing Trade Route]")
await bot.execute_route(route, data_dir=DOCKER_DATA_PATH, max_retries=2)
# Get final state
final_credits = bot.current_credits
final_sector = bot.current_sector
profit = final_credits - initial_credits
print(f"\n[Final State]")
print(f" Credits: {final_credits:,}")
print(f" Sector: {final_sector}")
print(f" Profit: {profit:,}")
return {
"success": True,
"initial_credits": initial_credits,
"final_credits": final_credits,
"profit": profit,
"initial_sector": initial_sector,
"final_sector": final_sector,
}
except Exception as e:
print(f" ERROR during trade: {e}")
import traceback
traceback.print_exc()
return {
"success": False,
"error": str(e),
"initial_credits": initial_credits,
"final_credits": bot.current_credits,
}
async def test_verify_consistency(
data_dir: Path,
bot: TradingBot,
player_name: str = "claude",
) -> dict:
"""Verify terminal state matches file state.
Args:
data_dir: Path to TW2002 data files
bot: Connected TradingBot
player_name: Player name to look up
Returns:
Verification result dictionary
"""
print("\n" + "=" * 80)
print("PHASE 4: Consistency Verification")
print("=" * 80)
admin = TW2002Admin(mode="direct", data_dir=data_dir)
# Get terminal state
terminal_credits = bot.current_credits
terminal_sector = bot.current_sector
print(f"\n[Terminal State]")
print(f" Credits: {terminal_credits:,}")
print(f" Sector: {terminal_sector}")
# Get file state
try:
players = await admin.list_players()
player = next((p for p in players if p.name.lower() == player_name.lower()), None)
if player:
file_credits = player.credits
file_sector = player.sector
print(f"\n[File State (twerk)]")
print(f" Credits: {file_credits:,}")
print(f" Sector: {file_sector}")
# Compare
credits_match = terminal_credits == file_credits
sector_match = terminal_sector == file_sector
print(f"\n[Verification]")
print(f" Credits match: {'YES' if credits_match else 'NO'}")
print(f" Sector match: {'YES' if sector_match else 'NO'}")
return {
"success": credits_match and sector_match,
"terminal_credits": terminal_credits,
"file_credits": file_credits,
"credits_match": credits_match,
"terminal_sector": terminal_sector,
"file_sector": file_sector,
"sector_match": sector_match,
}
else:
print(f" WARNING: Player '{player_name}' not found in file data")
return {"success": False, "error": "player_not_found"}
except Exception as e:
print(f" ERROR reading file state: {e}")
return {"success": False, "error": str(e)}
async def main():
"""Run the full integration test."""
print("\n" + "=" * 80)
print("TW2002 TRADING INTEGRATION TEST")
print("=" * 80)
print("\nThis test verifies:")
print(" 1. Twerk can analyze game data files")
print(" 2. Bot can login via terminal")
print(" 3. Bot can execute trades")
print(" 4. Terminal state matches file state")
# Configuration
data_dir = DOCKER_DATA_PATH
host = DEFAULT_HOST
port = DEFAULT_PORT
print(f"\n[Configuration]")
print(f" Data directory: {data_dir}")
print(f" Server: {host}:{port}")
results = {
"analysis": None,
"login": None,
"trade": None,
"verification": None,
}
bot = None
try:
# Phase 1: Twerk Analysis
results["analysis"] = await test_twerk_analysis(data_dir)
if "error" in results["analysis"]:
print("\nABORTING: Twerk analysis failed")
return 1
# Phase 2: Terminal Login
bot = await test_terminal_login(host, port)
if bot is None:
print("\nABORTING: Login failed")
return 1
results["login"] = {"success": True}
# Phase 3: Execute Trade
routes = results["analysis"].get("routes", [])
if routes:
results["trade"] = await test_execute_trade(bot, routes[0])
else:
print("\nSKIPPING trade: No routes found")
results["trade"] = {"skipped": True}
# Phase 4: Verify Consistency
results["verification"] = await test_verify_consistency(
data_dir, bot, player_name="claude"
)
finally:
# Cleanup
if bot and bot.session_id:
print("\n[Cleanup]")
try:
await bot.session_manager.close_session(bot.session_id)
print(" Session closed")
except Exception as e:
print(f" Warning: {e}")
# Summary
print("\n" + "=" * 80)
print("SUMMARY")
print("=" * 80)
all_passed = True
# Analysis
analysis_ok = results["analysis"] and "error" not in results["analysis"]
print(f" Twerk Analysis: {'PASS' if analysis_ok else 'FAIL'}")
all_passed = all_passed and analysis_ok
# Login
login_ok = results["login"] and results["login"].get("success")
print(f" Terminal Login: {'PASS' if login_ok else 'FAIL'}")
all_passed = all_passed and login_ok
# Trade
trade_result = results.get("trade", {})
if trade_result.get("skipped"):
print(" Trade Execution: SKIPPED")
else:
trade_ok = trade_result.get("success", False)
print(f" Trade Execution: {'PASS' if trade_ok else 'FAIL'}")
all_passed = all_passed and trade_ok
# Verification
verify_ok = results["verification"] and results["verification"].get("success")
print(f" Consistency Check: {'PASS' if verify_ok else 'FAIL/SKIP'}")
# Note: Verification may fail if player doesn't exist yet, don't fail overall
print("\n" + "=" * 80)
if all_passed:
print("OVERALL: PASS")
return 0
else:
print("OVERALL: FAIL")
return 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)
