doubao-say/tests/e2e/gtk_runtime_fixture.py at d971577c4036ac82caebc3ef570fcc70c9ed5b28 · quanru/doubao-say · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
"""Synthetic post-setup dictation target for Midscene E2E."""

import gi

gi.require_version("Gtk", "4.0")
gi.require_version("Gdk", "4.0")
from gi.repository import Gdk, GLib, Gtk

from doubao_input.inject.delivery import Delivery
from doubao_input.ui.overlay import Overlay


class ImmediateWorker:
    """Run Delivery work on GTK's thread while retaining its real state machine."""

    def submit(self, work, completed):
        completed(work())

    def close(self, cleanup):
        cleanup()


def build_runtime_fixture():
    overlay = Overlay()
    window = Gtk.Window(title="Synthetic dictation target")
    window.set_default_size(760, 420)
    box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=18)
    for side in ("start", "end", "top", "bottom"):
        getattr(box, "set_margin_" + side)(32)
    window.set_child(box)

    heading = Gtk.Label(label="Synthetic dictation delivery", xalign=0)
    heading.add_css_class("title-1")
    box.append(heading)
    box.append(Gtk.Label(
        label=(
            "This CI-only target uses synthetic recognition text and never "
            "reads or changes the system clipboard. Press F8 to start or "
            "finish; press Escape to cancel."
        ),
        xalign=0,
        wrap=True,
    ))
    target = Gtk.Entry(placeholder_text="Synthetic text will be delivered here")
    box.append(target)
    status = Gtk.Label(label="Idle · 0 deliveries", xalign=0, wrap=True)
    status.add_css_class("title-3")
    box.append(status)

    state = {"recording": False, "deliveries": 0}

    def schedule(delay_ms, callback):
        def invoke():
            callback()
            return GLib.SOURCE_REMOVE

        GLib.timeout_add(delay_ms, invoke)

    def paste(text, _target, cancelled):
        if cancelled():
            return False
        target.set_text(text)
        state["deliveries"] += 1
        return True

    def delivery_changed(value):
        if value == "attempted":
            status.set_text("Delivered once · closing overlay…")

    delivery = Delivery(
        schedule=schedule,
        target=lambda: "synthetic-target",
        paste=paste,
        enter=lambda *_args: False,
        changed=delivery_changed,
        worker=ImmediateWorker(),
    )

    def restore_target_focus():
        window.present()
        window.set_focus(target)
        target.grab_focus()

    def key_pressed(_controller, keyval, _keycode, _modifiers):
        if keyval == Gdk.KEY_F8:
            if state["recording"]:
                state["recording"] = False
                overlay.set_status("Synthetic recognition complete")
                delivery.submit(
                    "Synthetic dictation delivered exactly once.",
                    "synthetic-target",
                )

                def close_overlay_and_restore_target():
                    overlay.hide()
                    restore_target_focus()
                    status.set_text(
                        f"Delivered once · {state['deliveries']} delivery"
                    )

                schedule(500, close_overlay_and_restore_target)
            else:
                state["recording"] = True
                overlay.show("Listening to synthetic speech…")
                overlay.set_text("Synthetic dictation is ready to deliver.")
                status.set_text("Recording synthetic dictation · press F8 to finish")
            return True
        if keyval == Gdk.KEY_Escape and state["recording"]:
            state["recording"] = False
            overlay.hide()
            restore_target_focus()
            status.set_text("Cancelled · no text delivered · ready to try again")
            return True
        return False

    def listen_for_shortcuts(owner):
        keys = Gtk.EventControllerKey()
        keys.set_propagation_phase(Gtk.PropagationPhase.CAPTURE)
        keys.connect("key-pressed", key_pressed)
        owner.add_controller(keys)

    # The production app receives F8/Escape from its global trigger monitor.
    # This isolated fixture has no monitor, so listen from both of its windows:
    # some X11 window managers activate the overlay while it is visible even
    # though GTK marks it non-focusable.
    overlay._ensure_window()
    listen_for_shortcuts(window)
    listen_for_shortcuts(overlay._window)
    window.present()
    GLib.idle_add(target.grab_focus)

    def cleanup():
        delivery.close(lambda: None)
        overlay.hide()
        if overlay._window:
            overlay._window.destroy()
        window.destroy()

    return cleanup