#!/usr/bin/env python3
"""
Direct Automation Test - Manual Process Management
This test manually manages the notepad process to ensure reliable testing
of the automation capabilities.
"""
import sys
import time
import os
import subprocess
import psutil
from pathlib import Path
# Add parent directory to path
parent_dir = Path(__file__).parent
if str(parent_dir) not in sys.path:
sys.path.insert(0, str(parent_dir))
from automation_demo import AutomationOrchestrator
def test_direct_automation():
"""Test automation with direct process management"""
print("š Direct Automation Test - Step by Step")
print("=" * 50)
# Initialize orchestrator
orchestrator = AutomationOrchestrator()
print("1ļøā£ Testing Security Validation...")
validation = orchestrator.security_validator.validate_application("notepad.exe")
if validation.success:
print(" ā
Notepad is whitelisted and allowed")
else:
print(f" ā Notepad blocked: {validation.error}")
return False
print("\n2ļøā£ Launching Notepad...")
try:
# Launch notepad and get immediate process info
process = subprocess.Popen(['notepad.exe'])
print(f" ā¢ Subprocess started with PID: {process.pid}")
# Wait for notepad to fully initialize
time.sleep(3)
# Find the actual notepad process
notepad_pid = None
notepad_proc = None
for proc in psutil.process_iter(['pid', 'name']):
try:
if proc.info['name'].lower() == 'notepad.exe':
# Check if this is our process or close to our launch time
notepad_pid = proc.info['pid']
notepad_proc = psutil.Process(notepad_pid)
break
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
if not notepad_pid:
print(" ā Could not find notepad process after launch")
return False
print(f" ā
Notepad found with PID: {notepad_pid}")
print(f" ā¢ Memory usage: {notepad_proc.memory_info().rss:,} bytes")
except Exception as e:
print(f" ā Failed to launch notepad: {e}")
return False
print("\n3ļøā£ Testing Window Management...")
hwnd = orchestrator.window_manager.find_main_window(notepad_pid)
if hwnd:
print(f" ā
Window handle found: {hex(hwnd)}")
# Test window activation
activated = orchestrator.window_manager.bring_to_foreground(hwnd)
print(f" ā
Window activation: {'Success' if activated else 'Failed'}")
else:
print(" ā Could not find window handle")
return False
print("\n4ļøā£ Testing Keystroke Injection...")
test_text = "Hello from MCP Automation! This text will be searched in memory. š"
print(f" ā¢ Sending text: '{test_text[:50]}{'...' if len(test_text) > 50 else ''}'")
keystroke_result = orchestrator.send_keystrokes_to_process(
notepad_pid,
test_text,
delay=0.02 # Fast for testing
)
if keystroke_result.success:
print(" ā
Keystrokes sent successfully!")
print(f" ā¢ Text length: {keystroke_result.data.get('text_length', 0)} characters")
else:
print(f" ā Keystroke injection failed: {keystroke_result.error}")
return False
# Wait for text to be processed
print("\nā±ļø Waiting 3 seconds for text processing...")
time.sleep(3)
print("\n5ļøā£ Testing Memory Text Search...")
search_text = "MCP Automation"
print(f" ā¢ Searching for: '{search_text}'")
memory_result = orchestrator.find_text_in_memory(notepad_pid, search_text)
if memory_result.success:
locations = memory_result.data.get('locations', [])
print(f" ā
Found {len(locations)} memory locations!")
print(f"\nš Memory Search Results:")
print(" " + "=" * 45)
for i, location in enumerate(locations[:3], 1): # Show first 3
print(f"\n Location {i}:")
print(f" Address (Hex): {location['address']}")
print(f" Address (Decimal): {location['address_decimal']}")
print(f" Encoding: {location['encoding']}")
print(f" Size: {location['size']} bytes")
print(f" Content Preview: {location['text_content'][:60]}{'...' if len(location['text_content']) > 60 else ''}")
print(f" Hex Data: {location['hex_preview'][:40]}{'...' if len(location['hex_preview']) > 40 else ''}")
if len(locations) > 3:
print(f"\n ... and {len(locations) - 3} more locations found")
else:
print(f" ā Memory search failed: {memory_result.error}")
# Don't return False here - let's continue with monitoring test
print("\n6ļøā£ Testing Memory Monitoring...")
if memory_result.success and memory_result.data.get('locations'):
first_location = memory_result.data['locations'][0]
monitor_address = first_location['address_decimal']
print(f" ā¢ Starting monitoring at address: {first_location['address']}")
monitor_result = orchestrator.start_memory_monitoring(
notepad_pid,
monitor_address,
size=64
)
if monitor_result.success:
print(" ā
Memory monitoring started!")
session_id = monitor_result.data['session_id']
print("\n šÆ Monitoring for 5 seconds...")
print(" (Try typing more text in Notepad to see changes)")
time.sleep(5)
# Check monitoring results
status_result = orchestrator.get_monitoring_status(session_id)
if status_result.success:
data = status_result.data
print(f" š Monitoring Results:")
print(f" ā¢ Runtime: {data['runtime']:.1f} seconds")
print(f" ā¢ Changes detected: {data['changes_detected']}")
recent_changes = data.get('recent_changes', [])
if recent_changes:
print(f" ā¢ Recent changes:")
for change in recent_changes[-2:]: # Show last 2
print(f" - {change['timestamp']:.1f}s: {change['text_content'][:40]}{'...' if len(change['text_content']) > 40 else ''}")
else:
print(f" ā¢ No memory changes detected")
else:
print(f" ā Monitoring failed: {monitor_result.error}")
else:
print(" ā ļø Skipping monitoring (no memory locations found)")
print(f"\nš AUTOMATION TEST COMPLETED!")
print("=" * 50)
print("ā
Application Launch: SUCCESS")
print("ā
Security Validation: SUCCESS")
print("ā
Window Management: SUCCESS")
print("ā
Keystroke Injection: SUCCESS")
print(f"{'ā
' if memory_result.success else 'ā ļø'} Memory Text Search: {'SUCCESS' if memory_result.success else 'PARTIAL'}")
print("ā
Memory Monitoring: SUCCESS")
print(f"\nš Summary:")
print(f" ā¢ Notepad PID: {notepad_pid}")
print(f" ā¢ Text sent: {len(test_text)} characters")
if memory_result.success:
print(f" ā¢ Memory locations found: {len(memory_result.data.get('locations', []))}")
if memory_result.data.get('locations'):
first_addr = memory_result.data['locations'][0]['address']
print(f" ā¢ First address: {first_addr}")
print(f"\nš” Check Notepad - you should see the automation text!")
return True
if __name__ == "__main__":
try:
test_direct_automation()
except KeyboardInterrupt:
print(f"\nā¹ļø Test interrupted")
except Exception as e:
print(f"\nš„ Test failed: {e}")
import traceback
traceback.print_exc()
