Cua Driver 接入核对与离线动作验收

driver_contract.py · Python 标准库

"""Offline teaching contract; no Cua imports, pixels, OS input, or network."""
from dataclasses import asdict, dataclass
import json
from itertools import count


SESSION_IDS = count(1)


class DriverError(RuntimeError):
    pass


@dataclass(frozen=True)
class Observation:
    session: str
    capture: int
    revision: int
    window: int
    origin: tuple
    size: tuple
    scale: int
    focus: str
    text: str
    saved: str
    pending: bool
    tick: int
    kind: str = "metadata_only"


class MockDriver:
    """Single-threaded deterministic fixture, not a vendor-compatible driver."""

    def __init__(self, *, authenticated=True, observe=True, input_allowed=True):
        self.authenticated = authenticated
        self.observe_allowed = observe
        self.input_allowed = input_allowed
        self.state = "new"
        self.session = f"mock-session-{next(SESSION_IDS)}"
        self.window = 7
        self.origin = (100, 80)
        self.size = (800, 600)
        self.scale = 2
        self.focus = "none"
        self.text = ""
        self.saved = ""
        self.pending = None
        self.tick = 0
        self.revision = 0
        self.capture = 0
        self.latest = None
        self.held_keys = set()
        self.events = []
        self.cleanup_count = 0

    def connect(self):
        if self.state != "new":
            raise DriverError("bad_lifecycle")
        if not self.authenticated:
            raise DriverError("unauthenticated")
        self.state = "open"
        self.events.append({"event": "connect"})
        return self

    def __enter__(self):
        return self.connect()

    def __exit__(self, *_):
        self.close()

    def _live(self):
        if self.state != "open":
            raise DriverError(self.state)

    def _changed(self):
        self.revision += 1
        self.latest = None

    def observe(self):
        self._live()
        if not self.observe_allowed:
            raise DriverError("observation_denied")
        self.capture += 1
        self.latest = Observation(
            self.session, self.capture, self.revision, self.window,
            self.origin, self.size, self.scale, self.focus, self.text,
            self.saved, self.pending is not None, self.tick,
        )
        return self.latest

    def _admit(self, observation, window):
        self._live()
        if not self.input_allowed:
            raise DriverError("input_denied")
        if type(window) is not int or window != self.window:
            raise DriverError("wrong_window")
        if not isinstance(observation, Observation):
            raise DriverError("invalid_observation")
        if observation != self.latest or observation.revision != self.revision:
            raise DriverError("stale_capture")
        if self.tick - observation.tick > 2:
            raise DriverError("expired_capture")

    def _consume(self, event):
        self._changed()
        self.events.append(event)
        return {"delivered": True, "effect": "unverifiable"}

    def click(self, observation, x, y, *, window=7):
        self._admit(observation, window)
        if type(x) is not int or type(y) is not int:
            raise DriverError("integer_pixels_required")
        w, h = self.size
        if not (0 <= x < w * self.scale and 0 <= y < h * self.scale):
            raise DriverError("outside_image")
        if self.pending is not None:
            raise DriverError("busy")
        lx, ly = x / self.scale, y / self.scale
        target = "none"
        if 20 <= lx < 300 and 20 <= ly < 70:
            target = "input"
        elif 320 <= lx < 420 and 20 <= ly < 70:
            target = "save"
        self.focus = target
        if target == "save":
            self.pending = (self.tick + 2, self.text)
        return self._consume({
            "event": "click", "target": target,
            "screen_point": [self.origin[0] + lx, self.origin[1] + ly],
        })

    def type_text(self, observation, text, *, window=7):
        self._admit(observation, window)
        if self.focus != "input":
            raise DriverError("wrong_focus")
        if type(text) is not str or len(text) > 100:
            raise DriverError("invalid_text")
        self.text += text
        return self._consume({"event": "type", "characters": len(text)})

    def key_down(self, observation, key, *, window=7):
        self._admit(observation, window)
        if type(key) is not str or key not in {"Shift", "Control"}:
            raise DriverError("unsupported_key")
        self.held_keys.add(key)
        return self._consume({"event": "key_down", "key": key})

    def external_focus(self, name):
        self._live()
        self.focus = name
        self._changed()

    def resize(self, width, height, *, scale=2):
        self._live()
        if any(type(v) is not int or v <= 0 for v in (width, height, scale)):
            raise DriverError("invalid_geometry")
        self.size, self.scale = (width, height), scale
        self._changed()

    def advance(self, ticks=1):
        self._live()
        if type(ticks) is not int or ticks < 0:
            raise DriverError("invalid_ticks")
        self.tick += ticks
        if self.pending is not None and self.tick >= self.pending[0]:
            self.saved = self.pending[1]
            self.pending = None
            self._changed()

    def wait_saved(self, expected, *, budget=4, on_tick=None):
        self._live()
        if type(budget) is not int or budget < 0:
            raise DriverError("invalid_budget")
        deadline = self.tick + budget
        while True:
            observed = self.observe()
            if observed.tick > deadline:
                raise DriverError("timeout")
            if observed.saved == expected and not observed.pending:
                return observed
            if self.tick >= deadline:
                raise DriverError("timeout")
            if on_tick:
                on_tick(self)
                self._live()
                if self.tick >= deadline:
                    continue
            self.advance()

    def cancel(self):
        self._live()
        self.state = "cancelled"
        self.pending = None
        self.latest = None
        self.events.append({"event": "cancel"})

    def close(self):
        if self.state == "closed":
            return
        self.held_keys.clear()
        self.pending = None
        self.latest = None
        self.state = "closed"
        self.cleanup_count += 1
        self.events.append({"event": "close"})


def demo():
    driver = MockDriver()
    with driver:
        initial = driver.observe()
        driver.click(initial, 80, 80)
        stale_error = None
        try:
            driver.type_text(initial, "not admitted")
        except DriverError as error:
            stale_error = str(error)
        driver.type_text(driver.observe(), "driver contract")
        receipt = driver.click(driver.observe(), 700, 80)
        result = driver.wait_saved("driver contract")
    return {
        "kind": "offline_metadata_simulation",
        "stale_error": stale_error,
        "receipt": receipt,
        "result": asdict(result),
        "closed": driver.state,
        "cleanup_count": driver.cleanup_count,
        "events": driver.events,
    }


if __name__ == "__main__":
    print(json.dumps(demo(), indent=2, ensure_ascii=False))

test_driver_contract.py · Python 标准库

"""Run with: python3 -m unittest -v. Standard library only."""
from dataclasses import replace
import unittest

from driver_contract import DriverError, MockDriver, demo


class DriverContractTests(unittest.TestCase):
    def setUp(self):
        self.d = MockDriver().connect()
        self.addCleanup(self.d.close)

    def reject(self, code, fn):
        with self.assertRaisesRegex(DriverError, "^" + code + "$"):
            fn()

    def focus_input(self):
        self.d.click(self.d.observe(), 80, 80)

    def start_save(self, text="note"):
        self.focus_input()
        self.d.type_text(self.d.observe(), text)
        return self.d.click(self.d.observe(), 700, 80)

    def test_demo(self):
        result = demo()
        self.assertEqual(result["result"]["saved"], "driver contract")
        self.assertEqual(result["cleanup_count"], 1)
        self.assertEqual(result["closed"], "closed")

    def test_requires_authentication(self):
        self.reject("unauthenticated", MockDriver(authenticated=False).connect)

    def test_auth_does_not_grant_input(self):
        with MockDriver(input_allowed=False) as d:
            self.reject("input_denied", lambda: d.click(d.observe(), 80, 80))

    def test_observation_permission(self):
        self.d.observe_allowed = False
        self.reject("observation_denied", self.d.observe)

    def test_observation_metadata_only(self):
        o = self.d.observe()
        self.assertEqual(o.kind, "metadata_only")
        self.assertFalse(hasattr(o, "png"))
        self.assertEqual((o.size, o.scale), ((800, 600), 2))

    def test_click_changes_focus_and_revision(self):
        o = self.d.observe()
        self.d.click(o, 80, 80)
        n = self.d.observe()
        self.assertEqual(n.focus, "input")
        self.assertGreater(n.revision, o.revision)

    def test_coordinate_conversion(self):
        self.focus_input()
        self.assertEqual(self.d.events[-1]["screen_point"], [140.0, 120.0])

    def test_old_capture_after_new_observation(self):
        old = self.d.observe()
        self.d.observe()
        self.reject("stale_capture", lambda: self.d.click(old, 80, 80))

    def test_consumed_capture(self):
        old = self.d.observe()
        self.d.click(old, 80, 80)
        self.reject("stale_capture", lambda: self.d.click(old, 80, 80))

    def test_other_session_capture(self):
        current = self.d.observe()
        with MockDriver() as other:
            foreign = other.observe()
            self.assertNotEqual(current.session, foreign.session)
            self.reject("stale_capture", lambda: self.d.click(foreign, 80, 80))

    def test_tampered_metadata(self):
        o = replace(self.d.observe(), scale=1)
        self.reject("stale_capture", lambda: self.d.click(o, 80, 80))

    def test_wrong_window(self):
        o = self.d.observe()
        self.reject("wrong_window", lambda: self.d.click(o, 80, 80, window=8))

    def test_bool_is_not_a_window_id(self):
        o = self.d.observe()
        self.reject("wrong_window", lambda: self.d.click(o, 80, 80, window=True))

    def test_outside_image(self):
        for x, y in [(-1, 80), (1600, 80), (80, 1200)]:
            o = self.d.observe()
            self.reject("outside_image", lambda: self.d.click(o, x, y))

    def test_reject_noninteger_pixels(self):
        for x in [True, 80.5, float("nan"), "80"]:
            o = self.d.observe()
            self.reject("integer_pixels_required", lambda: self.d.click(o, x, 80))

    def test_expired_capture(self):
        o = self.d.observe()
        self.d.advance(3)
        self.reject("expired_capture", lambda: self.d.click(o, 80, 80))

    def test_resize_invalidates(self):
        o = self.d.observe()
        self.d.resize(400, 300, scale=1)
        self.reject("stale_capture", lambda: self.d.click(o, 80, 80))

    def test_bad_geometry(self):
        self.reject("invalid_geometry", lambda: self.d.resize(0, 300))

    def test_focus_change_invalidates(self):
        self.focus_input()
        o = self.d.observe()
        self.d.external_focus("other-window")
        self.reject("stale_capture", lambda: self.d.type_text(o, "hello"))
        self.assertEqual(self.d.text, "")

    def test_fresh_observation_wrong_focus(self):
        self.reject("wrong_focus", lambda: self.d.type_text(self.d.observe(), "x"))

    def test_type_changes_visible_metadata(self):
        self.focus_input()
        self.d.type_text(self.d.observe(), "draft")
        self.assertEqual(self.d.observe().text, "draft")

    def test_invalid_text_no_mutation(self):
        self.focus_input()
        self.reject("invalid_text", lambda: self.d.type_text(self.d.observe(), 4))
        self.assertEqual(self.d.text, "")

    def test_receipt_does_not_mean_saved(self):
        receipt = self.start_save()
        self.assertTrue(receipt["delivered"])
        self.assertEqual(receipt["effect"], "unverifiable")
        self.assertEqual(self.d.observe().saved, "")

    def test_wait_completes_at_deadline(self):
        self.start_save()
        o = self.d.wait_saved("note", budget=2)
        self.assertEqual(o.saved, "note")
        self.assertEqual(o.tick, 2)

    def test_busy_save_not_replayed(self):
        self.start_save()
        self.reject("busy", lambda: self.d.click(self.d.observe(), 700, 80))
        self.assertEqual(sum(e["event"] == "click" for e in self.d.events), 2)

    def test_wait_timeout(self):
        self.start_save()
        self.reject("timeout", lambda: self.d.wait_saved("note", budget=1))
        self.assertEqual(self.d.tick, 1)

    def test_wrong_expected_result_times_out(self):
        self.start_save()
        self.reject("timeout", lambda: self.d.wait_saved("wrong", budget=3))

    def test_zero_budget_observes_once(self):
        self.reject("timeout", lambda: self.d.wait_saved("x", budget=0))
        self.assertEqual(self.d.tick, 0)

    def test_cancel_during_wait(self):
        self.start_save()
        self.reject("cancelled", lambda: self.d.wait_saved(
            "note", on_tick=lambda d: d.cancel()))
        self.assertIsNone(self.d.pending)
        self.reject("cancelled", self.d.observe)

    def test_cleanup_on_exception(self):
        d = MockDriver()
        with self.assertRaises(ValueError):
            with d:
                d.key_down(d.observe(), "Shift")
                raise ValueError("test")
        self.assertEqual(d.held_keys, set())
        self.assertEqual(d.cleanup_count, 1)

    def test_idempotent_close(self):
        self.d.close()
        self.d.close()
        self.assertEqual(self.d.cleanup_count, 1)
        self.reject("closed", self.d.observe)

    def test_closed_cannot_reconnect(self):
        self.d.close()
        self.reject("bad_lifecycle", self.d.connect)

    def test_permission_revoked_before_action(self):
        o = self.d.observe()
        self.d.input_allowed = False
        self.reject("input_denied", lambda: self.d.click(o, 80, 80))
        self.assertEqual(self.d.focus, "none")

    def test_bad_budget_and_ticks(self):
        self.reject("invalid_budget", lambda: self.d.wait_saved("", budget=True))
        self.reject("invalid_ticks", lambda: self.d.advance(-1))

    def test_unhashable_key_is_contract_error(self):
        for key in [[], {}]:
            with self.subTest(key=key):
                self.reject("unsupported_key", lambda: self.d.key_down(
                    self.d.observe(), key))

    def test_unsupported_key(self):
        self.reject("unsupported_key", lambda: self.d.key_down(
            self.d.observe(), "Escape"))


    def test_callback_consumes_last_tick_without_extra_advance(self):
        self.start_save()
        with self.assertRaisesRegex(DriverError, '^timeout$'):
            self.d.wait_saved('note', budget=1, on_tick=lambda d: d.advance(1))
        self.assertEqual(self.d.tick, 1)
        self.assertEqual(self.d.saved, '')
        self.assertIsNotNone(self.d.pending)
        self.d.advance(1)
        self.assertEqual(self.d.wait_saved('note', budget=0).saved, 'note')
        self.assertEqual(sum(e['event'] == 'click' for e in self.d.events), 2)

    def test_callback_completion_at_deadline_is_accepted(self):
        self.start_save()
        result = self.d.wait_saved('note', budget=2, on_tick=lambda d: d.advance(2))
        self.assertEqual(result.tick, 2)
        self.assertEqual(result.saved, 'note')

    def test_callback_completion_after_deadline_is_not_timely_success(self):
        self.start_save()
        with self.assertRaisesRegex(DriverError, '^timeout$'):
            self.d.wait_saved('note', budget=1, on_tick=lambda d: d.advance(3))
        self.assertEqual(self.d.tick, 3)
        self.assertEqual(self.d.observe().saved, 'note')
        self.assertIsNone(self.d.pending)

    def test_callback_cancel_at_deadline_is_not_success(self):
        self.start_save()
        def cancel_after_completion(d):
            d.advance(2)
            d.cancel()
        with self.assertRaisesRegex(DriverError, '^cancelled$'):
            self.d.wait_saved('note', budget=2, on_tick=cancel_after_completion)
        self.assertEqual(self.d.tick, 2)
        self.assertEqual(self.d.saved, 'note')

if __name__ == "__main__":
    unittest.main()