replicate_GUI_test.py

#!/usr/bin/env python3
"""
GUI Test Replication Script

This script replays GUI tests based on the logs recorded during the original test execution.
It restores repository state from backups, executes recorded actions, and collects user feedback
to calculate play@k metrics for GUI-based testing.

Features:
- Restore repository state from timestamped backups
- Replay recorded actions with precise timing
- Interactive user feedback for test success evaluation
- Calculate play@k metrics for GUI test success
- Support for multiple test cases from log files

Usage:
    python replicate_GUI_test.py --log-file GUI_snap/gui_test_log_20241201_120000.json
    python replicate_GUI_test.py --log-dir GUI_snap/
"""

import json
import os
import sys
import shutil
import time
import argparse
import platform
import subprocess
from pathlib import Path
from datetime import datetime
import tkinter as tk
from tkinter import messagebox
import numpy as np

# Optional GUI automation imports
try:
    import pyautogui
    PYAUTOGUI_AVAILABLE = True
    pyautogui.FAILSAFE = False
except ImportError:
    print("⚠️  pyautogui not available. GUI automation will be simulated.")
    PYAUTOGUI_AVAILABLE = False


class GUITestReplicator:
    """Replays GUI tests based on recorded logs"""

    def __init__(self, log_file: str = None, log_dir: str = None, allow_empty: bool = False):
        self.log_file = Path(log_file) if log_file else None
        self.log_dir = Path(log_dir) if log_dir else None
        self.test_results = []
        self.current_test_index = 0
        self.game_process = None
        self.game_window_title = None

        # Validate input parameters (skip validation if allow_empty is True)
        if not allow_empty:
            if not log_file and not log_dir:
                raise ValueError("Either --log-file or --log-dir must be specified")

            if log_file and not self.log_file.exists():
                raise FileNotFoundError(f"Log file not found: {log_file}")

            if log_dir and not self.log_dir.exists():
                raise FileNotFoundError(f"Log directory not found: {log_dir}")

    def load_test_logs(self):
        """Load test logs from file or directory"""
        test_entries = []

        if self.log_file:
            # Load single log file
            with open(self.log_file, 'r', encoding='utf-8') as f:
                data = json.load(f)
                if isinstance(data, list):
                    test_entries = data
                else:
                    test_entries = [data]
        else:
            # Load all log files from directory
            for log_file in self.log_dir.glob("gui_test_log_*.json"):
                try:
                    with open(log_file, 'r', encoding='utf-8') as f:
                        data = json.load(f)
                        if isinstance(data, list):
                            test_entries.extend(data)
                        else:
                            test_entries.append(data)
                except Exception as e:
                    print(f"Warning: Could not load {log_file}: {e}")

        print(f"Loaded {len(test_entries)} test entries from logs")
        return test_entries

    def restore_repository_state(self, backup_dir: str):
        """Restore repository to the state captured during original test"""
        if not backup_dir or not os.path.exists(backup_dir):
            print(f"Warning: Backup directory not found: {backup_dir}")
            return False

        backup_path = Path(backup_dir) / "repository"
        if not backup_path.exists():
            print(f"Warning: Repository backup not found: {backup_path}")
            return False

        # Find target directory (assuming it's a game directory that needs restoration)
        # This is a heuristic - adjust based on your specific directory structure
        target_dirs = []
        for item in os.listdir("."):
            if os.path.isdir(item) and ("repos_gui_temp" in item.lower()):
                target_dirs.append(item)

        if not target_dirs:
            print("Could not identify target directory for restoration")
            return False

        target_dir = target_dirs[0]  # Use first matching directory
        print(f"Restoring repository to: {target_dir}")

        try:
            # Clear target directory
            if os.path.exists(target_dir):
                shutil.rmtree(target_dir)
            # Restore from backup
            shutil.copytree(backup_path, target_dir)
            print(f"✅ Repository restored from {backup_path} to {target_dir}")
            return True
        except Exception as e:
            print(f"❌ Failed to restore repository: {e}")
            return False

    def focus_game_window(self) -> bool:
        """将焦点设置到游戏窗口

        Returns:
            bool: 是否成功设置焦点
        """
        try:
            if not self.game_process or self.game_process.poll() is not None:
                print("    ⚠️  游戏进程不存在或已退出")
                return False

            # 获取进程PID
            pid = self.game_process.pid
            print(f"    🔍 获取游戏进程PID: {pid}")

            # 根据不同操作系统使用不同的方法设置窗口焦点
            if os.name == 'posix':  # macOS 或 Linux
                if platform.system() == 'Darwin':  # macOS
                    # 方法1: 通过PID激活应用
                    script1 = f'tell application "System Events" to set frontmost of (first process whose unix id is {pid}) to true'
                    script2 = f'delay 0.2'
                    script3 = f'tell application "System Events" to tell (first process whose unix id is {pid}) to perform action "AXRaise" of window 1'

                    result = subprocess.run(['osascript', '-e', script1, '-e', script2, '-e', script3],
                                          capture_output=True, text=True)

                    if result.returncode != 0:
                        # 方法2: 尝试通过窗口标题查找并激活
                        if self.game_window_title:
                            script = f'''
                            tell application "System Events"
                                repeat with p in processes
                                    if name of p contains "{self.game_window_title}" then
                                        set frontmost of p to true
                                        exit repeat
                                    end if
                                end repeat
                            end tell
                            '''
                            result = subprocess.run(['osascript', '-e', script],
                                                  capture_output=True, text=True)

                    if result.returncode == 0:
                        print(f"    ✅ 已将焦点设置到游戏窗口 (PID: {pid})")
                        # 给窗口一点时间来获得焦点
                        time.sleep(0.5)
                        return True
                    else:
                        print(f"    ❌ 无法设置窗口焦点: {result.stderr}")
                        return False

                else:  # Linux
                    # 在Linux上可以使用xdotool
                    try:
                        # 首先尝试通过PID查找窗口
                        result = subprocess.run(['xdotool', 'search', '--pid', str(pid)],
                                              capture_output=True, text=True)
                        if result.returncode == 0 and result.stdout.strip():
                            window_id = result.stdout.strip().split()[0]
                            # 激活窗口
                            subprocess.run(['xdotool', 'windowactivate', window_id])
                            print(f"    ✅ 已将焦点设置到游戏窗口 (PID: {pid})")
                            time.sleep(0.5)
                            return True
                    except FileNotFoundError:
                        print("    ⚠️  xdotool未安装，无法设置窗口焦点")
                        return False

            elif os.name == 'nt':  # Windows
                try:
                    import win32gui
                    import win32process
                    import win32con

                    def enum_window_callback(hwnd, pid):
                        if win32gui.IsWindowVisible(hwnd):
                            _, found_pid = win32process.GetWindowThreadProcessId(hwnd)
                            if found_pid == pid:
                                win32gui.SetForegroundWindow(hwnd)
                                return False
                        return True

                    win32gui.EnumWindows(enum_window_callback, pid)
                    print(f"    ✅ 已将焦点设置到游戏窗口 (PID: {pid})")
                    time.sleep(0.5)
                    return True
                except ImportError:
                    print("    ⚠️  pywin32未安装，无法设置窗口焦点")
                    return False

            return False

        except Exception as e:
            print(f"    ❌ 设置窗口焦点失败: {e}")
            return False

    def execute_test_actions(self, test_entry: dict):
        """Execute the recorded test actions with real game process and GUI automation"""
        actions = test_entry.get('actions', [])
        game_path = test_entry.get('game_path', '')

        print(f"🎮 Executing test for: {game_path}")
        print(f"📋 Found {len(actions)} actions to replay")

        # Filter actual game actions (exclude start/end markers)
        game_actions = [action for action in actions
                       if action.get('action') not in ['game_start', 'game_end']]

        if not game_actions:
            print("No game actions found to execute")
            return False

        # Get game configuration for execution
        game_config = self._get_game_config(test_entry)
        if not game_config:
            print("❌ Could not get game configuration")
            return False

        # Start the actual game process
        game_process = self._start_game_process(game_path, game_config)
        self.game_process = game_process  # Store for window focusing
        if not game_process:
            print("❌ Failed to start game process")
            return False

        try:
            # Wait for game to initialize
            print("⏳ Waiting for game to initialize...")
            time.sleep(3)  # Give game time to start up

            print("🚀 Starting real action execution...")
            success_count = 0

            for i, action in enumerate(game_actions):
                action_type = action.get('action', 'unknown')
                timestamp = action.get('timestamp', 'unknown')
                step_info = action.get('step_info', {})

                print(f"  [{i+1}/{len(game_actions)}] Executing: {action_type} at {timestamp}")

                # Execute the actual action using GUI automation
                action_success = self._execute_single_action(step_info, game_config)

                if action_success:
                    success_count += 1
                    print(f"    ✅ Action executed successfully")
                else:
                    print(f"    ❌ Action execution failed")

                # Wait between actions (use recorded timing if available)
                if i < len(game_actions) - 1:
                    next_action = game_actions[i + 1]
                    wait_time = self._calculate_wait_time(action, next_action)
                    print(f"    ⏱️  Waiting {wait_time:.1f}s before next action...")
                    time.sleep(wait_time)

            # Check if game process is still running
            if game_process.poll() is None:
                print("🛑 Terminating game process...")
                game_process.terminate()
                try:
                    game_process.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    game_process.kill()

            execution_rate = success_count / len(game_actions)
            print(f"✅ Action execution completed. Success rate: {execution_rate:.2f}")
            return execution_rate >= 0.8  # Consider successful if 80%+ actions succeeded

        except Exception as e:
            print(f"❌ Error during action execution: {e}")
            if game_process.poll() is None:
                game_process.terminate()
            return False

    def _get_game_config(self, test_entry: dict):
        """Extract game configuration from test entry"""
        try:
            # Load game config from the repository
            game_config_path = Path("game_config.json")
            if game_config_path.exists():
                with open(game_config_path, 'r', encoding='utf-8') as f:
                    all_configs = json.load(f)

                # Find matching config based on game path
                game_path = test_entry.get('game_path', '')
                for config_path, config in all_configs.items():
                    if config_path in game_path:
                        return config

                # Try to match by game hint
                game_hint = test_entry.get('final_result', {}).get('game_type_hint')
                for config_path, config in all_configs.items():
                    if config.get('game_hint') == game_hint:
                        return config

            # Return default config if no match found
            return {
                "running_arguments": [],
                "pip_requirements": [],
                "timeout": 30,
                "python_version": "3.9"
            }

        except Exception as e:
            print(f"Warning: Could not load game config: {e}")
            return {
                "running_arguments": [],
                "pip_requirements": [],
                "timeout": 30,
                "python_version": "3.9"
            }

    def _start_game_process(self, game_path: str, game_config: dict):
        """Start the actual game process"""
        try:
            import sys
            import subprocess

            # Build execution command (similar to test_games.py)
            python_executable = sys.executable
            running_arguments = game_config.get("running_arguments", [])
            full_game_args = [python_executable, game_path] + running_arguments

            print(f"🎮 Starting game: {' '.join(full_game_args)}")

            # Start game process in background
            game_process = subprocess.Popen(
                full_game_args,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                cwd=Path.cwd()  # Run from current directory
            )

            # Give process time to start
            time.sleep(1)

            if game_process.poll() is None:
                print("✅ Game process started successfully")
                return game_process
            else:
                stdout, stderr = game_process.communicate()
                print(f"❌ Game process failed to start. Return code: {game_process.returncode}")
                if stderr:
                    print(f"Error: {stderr}")
                return None

        except Exception as e:
            print(f"❌ Failed to start game process: {e}")
            return None

    def _execute_single_action(self, step_info: dict, game_config: dict):
        """Execute a single action using GUI automation"""
        try:
            # The action_dict is nested inside test_step
            test_step = step_info.get('test_step', {})
            action_dict = test_step.get('action_dict', {})
            action_type = action_dict.get('action_type', '')
            action_inputs = action_dict.get('action_inputs', {})

            # Ensure game window is focused before executing action
            self.focus_game_window()

            if action_type == 'press' and 'key' in action_inputs:
                return self._execute_key_press(action_inputs['key'])
            elif action_type in ['click', 'move', 'drag']:
                return self._execute_mouse_action(action_type, action_inputs)
            else:
                print(f"    ⚠️  Unsupported action type: {action_type}")
                return False

        except Exception as e:
            print(f"    ❌ Error executing action: {e}")
            return False

    def _execute_key_press(self, key: str):
        """Execute a key press action"""
        try:
            import pyautogui
            import pyperclip

            # Map common game keys
            key_mapping = {
                'a': 'left',
                'd': 'right',
                'w': 'up',
                's': 'down',
                ' ': 'space',
                '\n': 'enter',
                '\t': 'tab'
            }

            # Use mapped key or original key
            pyautogui_key = key_mapping.get(key, key)

            print(f"    📝 Pressing key: {pyautogui_key}")
            pyautogui.press(pyautogui_key)

            # Small delay after key press
            time.sleep(0.1)
            return True

        except ImportError:
            print("    ⚠️  pyautogui not available, simulating key press")
            time.sleep(0.5)  # Simulate key press time
            return True
        except Exception as e:
            print(f"    ❌ Key press failed: {e}")
            return False

    def _execute_mouse_action(self, action_type: str, action_inputs: dict):
        """Execute mouse-related actions"""
        try:
            import pyautogui

            if action_type == 'click':
                x = action_inputs.get('x', 0)
                y = action_inputs.get('y', 0)
                button = action_inputs.get('button', 'left')

                print(f"    🖱️  Clicking at ({x}, {y}) with {button} button")
                pyautogui.click(x, y, button=button)
                return True

            elif action_type == 'move':
                x = action_inputs.get('x', 0)
                y = action_inputs.get('y', 0)

                print(f"    🖱️  Moving mouse to ({x}, {y})")
                pyautogui.moveTo(x, y)
                return True

            else:
                print(f"    ⚠️  Unsupported mouse action: {action_type}")
                return False

        except ImportError:
            print("    ⚠️  pyautogui not available, simulating mouse action")
            time.sleep(0.3)  # Simulate mouse action time
            return True
        except Exception as e:
            print(f"    ❌ Mouse action failed: {e}")
            return False

    def _calculate_wait_time(self, current_action: dict, next_action: dict):
        """Calculate wait time between actions based on timestamps"""
        try:
            current_timestamp = current_action.get('timestamp')
            next_timestamp = next_action.get('timestamp')

            if current_timestamp and next_timestamp:
                # Parse ISO format timestamps
                from datetime import datetime
                current_dt = datetime.fromisoformat(current_timestamp.replace('Z', '+00:00'))
                next_dt = datetime.fromisoformat(next_timestamp.replace('Z', '+00:00'))

                time_diff = (next_dt - current_dt).total_seconds()
                # Clamp to reasonable range (0.5 to 5 seconds)
                return max(0.5, min(5.0, time_diff))

        except Exception as e:
            print(f"    ⚠️  Could not calculate wait time: {e}")

        # Default wait time
        return 1.0

    def get_user_feedback(self, test_entry: dict):
        """Get user feedback on test success using a dialog box"""
        root = tk.Tk()
        root.withdraw()  # Hide the main window

        # Prepare test information for display
        game_path = test_entry.get('game_path', 'Unknown')
        actions_count = len([a for a in test_entry.get('actions', [])
                           if a.get('action') not in ['game_start', 'game_end']])

        message = f"Test Information:\n"
        message += f"Game: {Path(game_path).name}\n"
        message += f"Actions executed: {actions_count}\n\n"
        message += "Did the GUI test execute successfully?"

        result = messagebox.askyesno("GUI Test Feedback", message)
        root.destroy()

        return result

    def calculate_play_at_k(self, results: list, k_values: list = [1, 3, 5, 10]):
        """Calculate play@k metrics for GUI tests"""
        if not results:
            return {f"play@{k}": 0.0 for k in k_values}

        # Group results by function (assuming each test entry represents a function attempt)
        function_results = {}
        for result in results:
            # Use a simple grouping key - in practice, you'd want to group by function
            key = f"test_{len(function_results)}"
            if key not in function_results:
                function_results[key] = []
            function_results[key].append(result['success'])

        metrics = {}
        for k in k_values:
            pass_count = 0
            total_functions = 0

            for func_results in function_results.values():
                total_functions += 1
                # Check if any of the k attempts succeeded
                if any(func_results[:k]):
                    pass_count += 1

            metrics[f"play@{k}"] = pass_count / total_functions if total_functions > 0 else 0.0

        return metrics

    def save_results(self, results: list, metrics: dict, output_file: str = None):
        """Save test results and metrics to file"""
        if not output_file:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            output_file = f"GUI_snap/gui_test_replication_results_{timestamp}.json"

        result_data = {
            'execution_timestamp': datetime.now().isoformat(),
            'total_tests': len(results),
            'successful_tests': sum(1 for r in results if r['success']),
            'metrics': metrics,
            'individual_results': results
        }

        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(result_data, f, indent=2, ensure_ascii=False)

        print(f"📊 Results saved to: {output_file}")
        return output_file

    def run_replication(self):
        """Main method to run the GUI test replication"""
        print("🎭 Starting GUI Test Replication")
        print("=" * 50)

        # Load test logs
        test_entries = self.load_test_logs()
        if not test_entries:
            print("❌ No test entries found in logs")
            return

        # Process each test entry
        for i, test_entry in enumerate(test_entries):
            print(f"\n{'='*30} Test {i+1}/{len(test_entries)} {'='*30}")

            # Restore repository state
            backup_dir = test_entry.get('backup_dir')
            if backup_dir:
                restore_success = self.restore_repository_state(backup_dir)
                if not restore_success:
                    print("⚠️  Repository restoration failed, continuing anyway...")
            else:
                print("ℹ️  No backup directory specified, using current state")

            # Execute test actions
            execution_success = self.execute_test_actions(test_entry)

            # Get user feedback
            if execution_success:
                user_success = self.get_user_feedback(test_entry)
            else:
                print("⚠️  Action execution failed, marking as unsuccessful")
                user_success = False

            # Record result
            result = {
                'test_index': i,
                'game_path': test_entry.get('game_path', ''),
                'backup_dir': test_entry.get('backup_dir', ''),
                'execution_success': execution_success,
                'user_success': user_success,
                'success': execution_success and user_success,
                'timestamp': datetime.now().isoformat()
            }

            self.test_results.append(result)
            print(f"📝 Result: {'✅ Success' if result['success'] else '❌ Failed'}")

        # Calculate metrics
        print(f"\n{'='*50}")
        print("📊 CALCULATING PLAY@K METRICS")
        print('='*50)

        metrics = self.calculate_play_at_k(self.test_results)

        # Display results
        successful_tests = sum(1 for r in self.test_results if r['success'])
        total_tests = len(self.test_results)

        print(f"Total Tests: {total_tests}")
        print(f"Successful Tests: {successful_tests}")
        print(f"Success Rate: {successful_tests/total_tests*100:.1f}%")
        print("\nPlay@k Metrics:")
        for k, value in metrics.items():
            print(f"  {k}: {value:.3f}")

        # Save results
        result_file = self.save_results(self.test_results, metrics)

        print(f"\n🎉 GUI test replication completed!")
        print(f"📄 Detailed results saved to: {result_file}")

        return metrics


def main():
    parser = argparse.ArgumentParser(
        description="Replicate GUI tests from recorded logs",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python replicate_GUI_test.py --log-file GUI_snap/gui_test_log_20241201_120000.json
  python replicate_GUI_test.py --log-dir GUI_snap/
  python replicate_GUI_test.py --log-file GUI_snap/gui_test_log_20241201_120000.json --output results.json
        """
    )

    parser.add_argument("--log-file", type=str,
                       help="Path to specific GUI test log file")
    parser.add_argument("--log-dir", type=str,
                       help="Directory containing GUI test log files")
    parser.add_argument("--output", type=str,
                       help="Output file for results (default: auto-generated name)")

    args = parser.parse_args()

    try:
        replicator = GUITestReplicator(log_file=args.log_file, log_dir=args.log_dir)
        metrics = replicator.run_replication()

        # Exit with success/failure based on overall results
        success_rate = metrics.get('play@1', 0)
        sys.exit(0 if success_rate > 0 else 1)

    except Exception as e:
        print(f"❌ Replication failed: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


def test_implementation():
    """Test the GUI automation implementation"""
    print("🧪 Testing GUI automation implementation...")

    # Create a test instance
    replicator = GUITestReplicator(allow_empty=True)

    # Test with a mock test entry
    mock_test_entry = {
        "test_info": {
            "log_file": "GUI_snap/test_log.json",
            "snap_dir": "GUI_snap",
            "patch_info": {
                "function_name": "test_function",
                "file_path": "test_game.py",
                "timestamp": "2024-12-01T12:00:00.000000"
            }
        },
        "game_path": "test_game.py",
        "backup_dir": None,
        "actions": [
            {
                "action": "game_start",
                "timestamp": "2025-08-21T22:44:29.640409",
                "game_path": "test_game.py",
                "game_args": []
            },
            {
                "action": "unknown",
                "timestamp": "2025-08-21T22:44:29.640429",
                "step_info": {
                    "action_dict": {
                        "action_type": "press",
                        "action_inputs": {"key": "a"},
                        "raw_text": "press(key='a')"
                    }
                }
            },
            {
                "action": "game_end",
                "timestamp": "2025-08-21T22:44:29.640461",
                "result": {"status": "PASSED"}
            }
        ]
    }

    # Test game config extraction
    print("1. Testing game config extraction...")
    config = replicator._get_game_config(mock_test_entry)
    print(f"   ✅ Config loaded: {config is not None}")

    # Test action parsing
    print("2. Testing action parsing...")
    actions = mock_test_entry.get('actions', [])
    game_actions = [action for action in actions
                   if action.get('action') not in ['game_start', 'game_end']]
    print(f"   ✅ Found {len(game_actions)} game actions")

    # Test key press execution (simulation mode)
    print("3. Testing key press execution...")
    step_info = game_actions[0].get('step_info', {})
    action_success = replicator._execute_single_action(step_info, config)
    print(f"   ✅ Key press execution: {action_success}")

    # Test with real log structure (nested action_dict)
    print("4. Testing with real log structure...")
    real_step_info = {
        "test_step": {
            "action_dict": {
                "action_type": "press",
                "action_inputs": {"key": "w"},
                "raw_text": "press(key='w')"
            }
        }
    }
    action_success_real = replicator._execute_single_action(real_step_info, config)
    print(f"   ✅ Real structure key press execution: {action_success_real}")

    # Test wait time calculation
    print("5. Testing wait time calculation...")
    current_action = {"timestamp": "2024-12-01T12:00:00.000000"}
    next_action = {"timestamp": "2024-12-01T12:00:02.000000"}
    wait_time = replicator._calculate_wait_time(current_action, next_action)
    print(f"   ✅ Wait time calculated: {wait_time}s")

    print("✅ All implementation tests passed!")
    return True


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--test":
        test_implementation()
    else:
        main()