tests/test_windows_input_driver.py

from __future__ import annotations

import ctypes
import importlib.util
import sys
from pathlib import Path
from types import ModuleType, SimpleNamespace

import pytest


ROOT = Path(__file__).resolve().parents[1]
WINDOWS_SANDBOX_ROOT = ROOT / "sandbox" / "windows"
DRIVER_PATH = WINDOWS_SANDBOX_ROOT / "broker" / "pyautogui_driver.py"
BROKER_SOURCE_PATH = WINDOWS_SANDBOX_ROOT / "broker" / "AutoCADBench.Broker.cs"
READINESS_PATH = WINDOWS_SANDBOX_ROOT / "bootstrap" / "verify-boot-readiness.ps1"
START_BROKER_PATH = WINDOWS_SANDBOX_ROOT / "bootstrap" / "start-broker.ps1"
RECOVERY_PATH = WINDOWS_SANDBOX_ROOT / "bootstrap" / "recover-boot-readiness.ps1"
DEPLOY_PATH = WINDOWS_SANDBOX_ROOT / "bootstrap" / "deploy-broker-update.ps1"
CLEAN_IMAGE_PATH = WINDOWS_SANDBOX_ROOT / "bootstrap" / "clean-for-image.ps1"
INSTALL_BOOT_PATH = WINDOWS_SANDBOX_ROOT / "bootstrap" / "install-boot-automation.ps1"
INTERACTIVE_EVALUATOR_PATH = (
    WINDOWS_SANDBOX_ROOT / "evaluator" / "run-interactive-evaluation.ps1"
)


class FakePyAutoGUI(ModuleType):
    def __init__(self) -> None:
        super().__init__("pyautogui")
        self.calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = []
        self.FAILSAFE = True
        self.PAUSE = 0.0
        self.cursor = (1200, 700)
        self.ignored_moves = 0

    def _record(self, name: str, *args: object, **kwargs: object) -> None:
        self.calls.append((name, args, kwargs))

    def size(self) -> tuple[int, int]:
        return 1920, 1080

    def position(self) -> tuple[int, int]:
        return self.cursor

    def click(self, *args: object, **kwargs: object) -> None:
        self._record("click", *args, **kwargs)

    def doubleClick(self, *args: object, **kwargs: object) -> None:
        self._record("doubleClick", *args, **kwargs)

    def moveTo(self, *args: object, **kwargs: object) -> None:
        self._record("moveTo", *args, **kwargs)
        if self.ignored_moves:
            self.ignored_moves -= 1
            return
        self.cursor = (int(args[0]), int(args[1]))

    def mouseDown(self, *args: object, **kwargs: object) -> None:
        self._record("mouseDown", *args, **kwargs)

    def mouseUp(self, *args: object, **kwargs: object) -> None:
        self._record("mouseUp", *args, **kwargs)

    def hscroll(self, *args: object, **kwargs: object) -> None:
        self._record("hscroll", *args, **kwargs)

    def scroll(self, *args: object, **kwargs: object) -> None:
        self._record("scroll", *args, **kwargs)

    def hotkey(self, *args: object, **kwargs: object) -> None:
        self._record("hotkey", *args, **kwargs)

    def press(self, *args: object, **kwargs: object) -> None:
        self._record("press", *args, **kwargs)

    def write(self, *args: object, **kwargs: object) -> None:
        self._record("write", *args, **kwargs)


@pytest.fixture
def driver(monkeypatch: pytest.MonkeyPatch):
    fake = FakePyAutoGUI()
    monkeypatch.setitem(sys.modules, "pyautogui", fake)
    monkeypatch.setattr(
        ctypes,
        "windll",
        SimpleNamespace(user32=SimpleNamespace(), kernel32=SimpleNamespace()),
        raising=False,
    )
    spec = importlib.util.spec_from_file_location("autocad_pyautogui_driver", DRIVER_PATH)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module, fake


def test_driver_maps_coordinate_and_keyboard_actions(driver) -> None:
    module, fake = driver

    module._execute({"action": "click", "x": 600, "y": 400, "button": "left"})
    module._execute({"action": "key", "keys": ["CTRL", "S"]})
    module._execute({"action": "key", "keys": ["ENTER"]})
    module._execute({"action": "type", "text": "_LINE 10,20"})
    module._execute({"action": "scroll", "dx": -120, "dy": 240})

    assert fake.calls == [
        ("moveTo", (600, 400), {"duration": 0.05}),
        ("click", (), {"button": "left"}),
        ("hotkey", ("ctrl", "s"), {"interval": 0.03}),
        ("press", ("enter",), {}),
        ("write", ("_LINE 10,20",), {"interval": 0.05}),
        ("hscroll", (-1,), {}),
        ("scroll", (2,), {}),
    ]


def test_driver_maps_full_drag_path(driver) -> None:
    module, fake = driver
    module._execute(
        {
            "action": "drag",
            "button": "left",
            "path": [
                {"x": 10, "y": 20},
                {"x": 30, "y": 40},
                {"x": 50, "y": 60},
            ],
        }
    )

    assert fake.calls == [
        ("moveTo", (10, 20), {"duration": 0.05}),
        ("mouseDown", (), {"button": "left"}),
        ("moveTo", (30, 40), {"duration": 0.03}),
        ("moveTo", (50, 60), {"duration": 0.03}),
        ("mouseUp", (), {"button": "left"}),
    ]


def test_driver_retries_cursor_move_before_emitting_click(driver) -> None:
    module, fake = driver
    fake.ignored_moves = 1

    module._execute({"action": "click", "x": 67, "y": 36, "button": "left"})

    assert fake.calls == [
        ("moveTo", (67, 36), {"duration": 0.05}),
        ("moveTo", (67, 36), {"duration": 0.05}),
        ("click", (), {"button": "left"}),
    ]


def test_driver_exhausts_ten_cursor_moves_without_clicking(driver) -> None:
    module, fake = driver
    fake.ignored_moves = module.CURSOR_MOVE_ATTEMPTS

    with pytest.raises(module.DeliveryError, match="after 10 moves"):
        module._execute({"action": "click", "x": 67, "y": 36, "button": "left"})

    assert [call[0] for call in fake.calls] == [
        "moveTo"
    ] * module.CURSOR_MOVE_ATTEMPTS


def test_driver_rejects_non_ascii_typing_and_unknown_keys(driver) -> None:
    module, _ = driver
    with pytest.raises(module.DeliveryError, match="printable ASCII"):
        module._execute({"action": "type", "text": "LINE\n"})
    with pytest.raises(module.DeliveryError, match="allowlist"):
        module._execute({"action": "key", "keys": ["WIN"]})


def test_driver_allows_literal_printable_space(driver) -> None:
    module, fake = driver
    module._execute({"action": "type", "text": " "})
    assert fake.calls == [("write", (" ",), {"interval": 0.05})]


def test_driver_follows_foreground_dialog_after_initial_handoff(
    driver, monkeypatch: pytest.MonkeyPatch
) -> None:
    module, _ = driver
    monkeypatch.setattr(
        module,
        "_desktop_identity",
        lambda: {
            "driver_version": module.DRIVER_VERSION,
            "python_pid": 10,
            "session_id": 1,
            "window_station": "WinSta0",
            "desktop": "Default",
            "screen_width": 1920,
            "screen_height": 1080,
        },
    )
    target = {
        "foreground_hwnd": 100,
        "foreground_thread_id": 11,
        "foreground_process_id": 456,
        "foreground_title": "Save Drawing As",
        "foreground_class": "#32770",
        "active_hwnd": 100,
        "focused_hwnd": 101,
        "focused_thread_id": 11,
        "focused_process_id": 456,
        "focused_title": "File name",
        "focused_class": "Edit",
        "focus_matches_foreground_process": True,
    }
    monkeypatch.setattr(module, "_input_focus_snapshot", lambda: target)

    receipt = module._handle(
        {
            "action": {"action": "key", "keys": ["ENTER"]},
            "expected_session_id": 1,
            "expected_width": 1920,
            "expected_height": 1080,
            "expected_foreground_pid": 123,
        }
    )

    assert receipt["ok"] is True
    assert receipt["foreground_before"] == 456
    assert receipt["foreground_was_autocad"] is False
    assert receipt["focus_was_autocad"] is False
    assert receipt["keyboard_target_verified"] is True
    assert receipt["target_before"]["focused_class"] == "Edit"


def test_driver_injects_and_reports_when_no_child_has_focus(
    driver, monkeypatch: pytest.MonkeyPatch
) -> None:
    module, _ = driver
    monkeypatch.setattr(
        module,
        "_desktop_identity",
        lambda: {
            "driver_version": module.DRIVER_VERSION,
            "python_pid": 10,
            "session_id": 1,
            "window_station": "WinSta0",
            "desktop": "Default",
            "screen_width": 1920,
            "screen_height": 1080,
        },
    )
    monkeypatch.setattr(
        module,
        "_input_focus_snapshot",
        lambda: {
            "foreground_hwnd": 100,
            "foreground_thread_id": 11,
            "foreground_process_id": 123,
            "foreground_title": "Autodesk AutoCAD 2019",
            "foreground_class": "Afx:00400000",
            "active_hwnd": 100,
            "focused_hwnd": 0,
            "focused_thread_id": 0,
            "focused_process_id": 0,
            "focused_title": "",
            "focused_class": "",
            "focus_matches_foreground_process": False,
        },
    )

    receipt = module._handle(
        {
            "action": {"action": "type", "text": "LINE"},
            "expected_session_id": 1,
            "expected_width": 1920,
            "expected_height": 1080,
            "expected_foreground_pid": 123,
        }
    )

    assert receipt["ok"] is True
    assert receipt["foreground_was_autocad"] is True
    assert receipt["keyboard_target_verified"] is False
    assert receipt["keyboard_target_mode"] == "no_focused_control"


def test_driver_injects_and_reports_keyboard_focused_by_another_process(
    driver, monkeypatch: pytest.MonkeyPatch
) -> None:
    module, fake = driver
    monkeypatch.setattr(
        module,
        "_desktop_identity",
        lambda: {
            "driver_version": module.DRIVER_VERSION,
            "python_pid": 10,
            "session_id": 1,
            "window_station": "WinSta0",
            "desktop": "Default",
            "screen_width": 1920,
            "screen_height": 1080,
        },
    )
    monkeypatch.setattr(
        module,
        "_input_focus_snapshot",
        lambda: {
            "foreground_hwnd": 100,
            "foreground_thread_id": 11,
            "foreground_process_id": 123,
            "foreground_title": "Autodesk AutoCAD 2019",
            "foreground_class": "AfxFrame",
            "active_hwnd": 100,
            "focused_hwnd": 200,
            "focused_thread_id": 12,
            "focused_process_id": 999,
            "focused_title": "Foreign",
            "focused_class": "Edit",
            "focus_matches_foreground_process": False,
        },
    )

    receipt = module._handle(
        {
            "action": {"action": "type", "text": "LINE"},
            "expected_session_id": 1,
            "expected_width": 1920,
            "expected_height": 1080,
            "expected_foreground_pid": 123,
        }
    )

    assert receipt["ok"] is True
    assert receipt["keyboard_target_verified"] is False
    assert receipt["keyboard_target_mode"] == "cross_process_focus"
    assert fake.calls == [("write", ("LINE",), {"interval": 0.05})]


def test_driver_reads_gui_thread_focus_and_window_metadata(
    driver, monkeypatch: pytest.MonkeyPatch
) -> None:
    module, _ = driver

    class FakeUser32:
        @staticmethod
        def GetForegroundWindow() -> int:
            return 100

        @staticmethod
        def GetWindowThreadProcessId(window, process_id) -> int:
            handle = int(window.value)
            process_id._obj.value = 123 if handle in {100, 101} else 0
            return 11 if handle in {100, 101} else 0

        @staticmethod
        def GetGUIThreadInfo(thread_id, info) -> bool:
            assert thread_id == 11
            info._obj.hwndActive = 100
            info._obj.hwndFocus = 101
            return True

        @staticmethod
        def GetWindowTextLengthW(window) -> int:
            return len("AutoCAD" if int(window.value) == 100 else "Model space")

        @staticmethod
        def GetWindowTextW(window, buffer, size) -> int:
            value = "AutoCAD" if int(window.value) == 100 else "Model space"
            buffer.value = value[: size - 1]
            return len(buffer.value)

        @staticmethod
        def GetClassNameW(window, buffer, size) -> int:
            value = "AfxFrame" if int(window.value) == 100 else "AfxView"
            buffer.value = value[: size - 1]
            return len(buffer.value)

    monkeypatch.setattr(module, "_USER32", FakeUser32())

    target = module._input_focus_snapshot()

    assert target == {
        "foreground_hwnd": 100,
        "foreground_thread_id": 11,
        "foreground_process_id": 123,
        "foreground_title": "AutoCAD",
        "foreground_class": "AfxFrame",
        "active_hwnd": 100,
        "focused_hwnd": 101,
        "focused_thread_id": 11,
        "focused_process_id": 123,
        "focused_title": "Model space",
        "focused_class": "AfxView",
        "focus_matches_foreground_process": True,
    }


def test_broker_uses_dedicated_worker_reader_and_bounded_autocad_retry() -> None:
    source = BROKER_SOURCE_PATH.read_text()
    readiness = READINESS_PATH.read_text()
    startup = START_BROKER_PATH.read_text()

    assert "Task<string>.Factory.StartNew" not in source
    assert 'reader.Name = "AutoCADBench-PyAutoGUI-stdout"' in source
    assert 'input_driver_ready = true' in source
    assert '{ "action", "move" }' in source
    assert 'ReadLine(120000, "startup handshake")' in source
    assert 'ReadLine(60000, "action response")' in source
    assert 'broker_version = "windows-autocad-2019-v11"' in source
    assert 'RequiredIdentifier(action, "request_id")' in source
    assert 'RequiredInt(action, "expected_action_index")' in source
    assert 'string text = RequiredText(actionPayload, "text");' in source
    required_text = source[
        source.index("private static string RequiredText") : source.index(
            "private static string RequiredString"
        )
    ]
    assert "text.Length == 0" in required_text
    assert "IsNullOrWhiteSpace" not in required_text
    assert "session.ActionResults.TryGetValue" in source
    assert "WriteActionResponse(context, session, cachedResult, true);" in source
    assert 'driverAction.Remove("intent");' in source
    assert 'input_driver = "pyautogui-0.9.54-v3"' in source
    assert "AutoCADStartupTimeoutSeconds = 120" in source
    assert "AutoCADHandoffAttemptLimit = 2" in source
    assert "PrepareAutoCADHandoffWithRecovery(session);" in source
    assert "handoffAttempt <= AutoCADHandoffAttemptLimit" in source
    assert "StopTimedOutAutoCADProcess(session.AutoCADProcess);" in source
    assert "File.Copy(_template, session.ArtifactPath, true);" in source
    assert "autocad_launch_attempts = session.AutoCADLaunchAttempts" in source
    assert "handoff_attempts = session.HandoffAttempts" in source
    assert 'exception.Code == "desktop_unstable"' in source
    assert "EnsureCleanForegroundHandoff(session);" in source
    assert "TryGetVisiblePopup" in source
    assert "A non-AutoCAD window owns the foreground at handoff" in source
    assert '"autocad_restart_failed"' in source
    release = source[source.index("private static void Release"):source.index("private static object CaptureResponse")]
    assert "WaitForExit(AutoCADRelaunchStopTimeoutMs)" in release
    assert "StopAutoCADProcesses();" in release
    assert "WaitForVisualStability(" in source
    assert "WaitForRenderedRibbon(" in source
    assert "HandoffRenderedRibbonComparisons = 2" in source
    assert "ActionStableComparisons = 5" in source
    assert "IsRibbonRendered(" in source
    assert "VerifyKeyboardHandoff(session);" not in source
    checkpoint = source[
        source.index("private static void Checkpoint") : source.index(
            "private static void RetrieveArtifact"
        )
    ]
    assert "SaveActiveDocumentToControlledPath(session);" in checkpoint
    assert 'Marshal.GetActiveObject("AutoCAD.Application.23.0")' in checkpoint
    assert '"controller-final.dwg"' in checkpoint
    assert 'new object[] { "CTRL", "S" }' not in checkpoint
    assert "after 5 attempts" in checkpoint
    prepare = source[source.index("private static void PrepareInitialDesktop"):source.index("private static void FocusAutoCAD")]
    assert prepare.index("DeliverTrustedInput(session, canvasClick);") < prepare.index('{ "keys", new object[] { "ESC" } }')
    assert 'inputReceipt["post_action_visual_stable"] = visualStable;' in source
    assert "visualStable = false;" in source
    assert "captured = CaptureDesktopBitmap();" in source
    driver_source = DRIVER_PATH.read_text()
    assert "GetGUIThreadInfo" in driver_source
    assert '"keyboard_target_verified"' in driver_source
    assert '"cross_process_focus"' in driver_source
    assert '"no_focused_control"' in driver_source
    assert "keyboard target is unsafe" not in driver_source
    assert "time.sleep(0.2)" not in driver_source
    assert "-TimeoutSec 600" in readiness
    assert "autocad_launch_attempts = [int]$reset.autocad_launch_attempts" in readiness
    assert "handoff_attempts = [int]$reset.handoff_attempts" in readiness
    assert "handoff_probe" not in readiness
    assert '"windows-autocad-2019-v11"' in readiness
    assert '"pyautogui-0.9.54-v3"' in readiness
    assert '"windows-autocad-2019-v11"' in startup
    assert '"pyautogui-0.9.54-v3"' in startup
    assert "function Get-VerifiedDeployment" in startup
    assert 'Join-Path $BenchRoot "installed-manifest.json"' in startup
    assert "deployment_id = $deployment.deployment_id" in startup
    assert "manifest_sha256 = $deployment.manifest_sha256" in startup
    assert "function Set-DcvLayoutWithRetry" in startup
    assert "[int]$AttemptLimit = 6" in startup
    assert "WaitForExit(15000)" in startup
    assert "set-display-layout" in startup
    assert "dcv_layout_attempts = $dcvLayoutAttempts" in startup
    assert "dcv_layout_attempts = [int]$ready.dcv_layout_attempts" in readiness
    assert "$ready.broker_pid -ne $broker.Id" in readiness
    assert "$health.broker_pid -ne $broker.Id" in readiness
    assert "$ready.startup_id" in readiness
    assert "$ready.broker_sha256" in readiness
    assert "$liveBrokerSha256" in readiness
    assert "$liveManifestSha256" in readiness
    assert '$task.State.ToString() -ne "Running"' in readiness
    assert "$startupState.startup_id -ne $ready.startup_id" in readiness
    assert "Reset smoke passed but session release failed" in readiness


def test_boot_gate_requires_current_boot_readiness() -> None:
    source = READINESS_PATH.read_text()

    assert "$readyBootTime" in source
    assert "$bootTime" in source
    assert 'readiness files belong to an earlier boot' in source
    assert "$candidateHealth.input_driver_ready" in source


def test_boot_recovery_is_bounded_diagnostic_and_deployed_with_broker() -> None:
    recovery = RECOVERY_PATH.read_text()
    deploy = DEPLOY_PATH.read_text()

    assert "pre-recovery.json" in recovery
    assert '"startup-state.json", "startup-error.json", "ready.json"' in recovery
    assert '"broker.token"' not in recovery
    assert "function Stop-BenchmarkProcessTree" in recovery
    assert "taskkill.exe /PID $process.Id /T /F" in recovery
    assert "post-pass process census decide success" in recovery
    assert "Interactive broker task did not stop before recovery cleanup" in recovery
    assert "Interactive broker task did not enter Running state after recovery" in recovery
    assert "cleanup_passes = $cleanupPasses" in recovery
    assert "Start-ScheduledTask -TaskName $TaskName" in recovery
    assert "recover-boot-readiness.ps1" in deploy
    assert 'Join-Path $BenchRoot "diagnostics"' in CLEAN_IMAGE_PATH.read_text()


def test_broker_lease_and_reset_are_replay_safe() -> None:
    source = BROKER_SOURCE_PATH.read_text()
    readiness = READINESS_PATH.read_text()

    lease = source[
        source.index("private static void Lease") : source.index(
            "private static void Reset"
        )
    ]
    reset = source[
        source.index("private static void Reset") : source.index(
            "private static void ExecuteAction"
        )
    ]
    assert "active.TaskId == taskId && active.AttemptId == attemptId" in lease
    assert "WriteJson(context.Response, 200, SessionResponse(active));" in lease
    assert 'RequiredIdentifier(payload, "request_id")' in reset
    assert "session.ResetResponse != null" in reset
    assert "session.ResetRequestId != requestId" in reset
    assert "session.ResetResponse = CaptureResponse(session);" in reset
    assert 'request_id = "reset-$sessionId"' in readiness


def test_deployment_and_image_cleaning_fail_closed_on_mixed_v10_files() -> None:
    deploy = DEPLOY_PATH.read_text()
    startup = START_BROKER_PATH.read_text()
    clean = CLEAN_IMAGE_PATH.read_text()
    install = INSTALL_BOOT_PATH.read_text()

    assert 'Join-Path $BenchRoot "installed-manifest.json"' in deploy
    assert 'schema_version = "1"' in deploy
    assert 'broker_version = "windows-autocad-2019-v11"' in deploy
    assert 'relative="bootstrap\\clean-for-image.ps1"' in deploy
    assert 'relative="evaluator\\AutoCADBench.Plugin.dll"' in deploy
    assert 'relative="evaluator\\AutoCADBench.Plugin.cs"' in deploy
    assert 'relative="evaluator\\run-evaluation.ps1"' in deploy
    assert 'relative="evaluator\\run-interactive-evaluation.ps1"' in deploy
    assert '"evaluator_plugin"' in startup
    assert '"evaluator_plugin"' in clean
    assert '"evaluator_plugin"' in install
    assert ".autocad-bench-new-$deploymentId" in deploy
    assert "$swapped.Count - 1" in deploy
    assert "throw $deploymentFailure" in deploy
    assert '"clean_for_image"' in startup
    assert '"clean_for_image"' in clean
    assert "Installed deployment checksum mismatch" in clean
    assert "Benchmark processes remain after image cleanup" in clean
    assert 'Join-Path $BenchRoot "updates"' in clean
    assert "UpdateFiles" in clean
    assert 'Join-Path $BenchRoot "evaluations"' in clean
    assert 'Join-Path $BenchRoot "evaluator-validation"' in clean
    assert "EvaluationFiles" in clean
    assert "EvaluatorValidationFiles" in clean
    assert 'schema_version = "1"' in install
    assert 'broker_version = "windows-autocad-2019-v11"' in install


def test_evaluator_uses_the_logged_in_interactive_desktop() -> None:
    script = INTERACTIVE_EVALUATOR_PATH.read_text()

    assert "-LogonType Interactive" in script
    assert "run-evaluation.ps1" in script
    assert "interactive-status.json" in script
    assert "MetadataUploadUrl" in script
    assert "Invoke-WebRequest" in script
    assert "Unregister-ScheduledTask" in script