File: test_message_pump.py

package info (click to toggle)
textual 2.1.2-1.1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 55,080 kB
  • sloc: python: 85,423; lisp: 1,669; makefile: 101
file content (171 lines) | stat: -rw-r--r-- 4,930 bytes parent folder | download | duplicates (2)
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
import pytest

from textual._dispatch_key import dispatch_key
from textual.app import App, ComposeResult
from textual.errors import DuplicateKeyHandlers
from textual.events import Key
from textual.message import Message
from textual.widget import Widget
from textual.widgets import Button, Input, Label


class ValidWidget(Widget):
    called_by = None

    def key_x(self):
        self.called_by = self.key_x

    def key_ctrl_i(self):
        self.called_by = self.key_ctrl_i


async def test_dispatch_key_valid_key():
    widget = ValidWidget()
    result = await dispatch_key(widget, Key(key="x", character="x"))
    assert result is True
    assert widget.called_by == widget.key_x


async def test_dispatch_key_valid_key_alias():
    """When you press tab or ctrl+i, it comes through as a tab key event, but handlers for
    tab and ctrl+i are both considered valid."""
    widget = ValidWidget()
    result = await dispatch_key(widget, Key(key="tab", character="\t"))
    assert result is True
    assert widget.called_by == widget.key_ctrl_i


class DuplicateHandlersWidget(Widget):
    called_by = None

    def key_x(self):
        self.called_by = self.key_x

    def _key_x(self):
        self.called_by = self._key_x

    def key_tab(self):
        self.called_by = self.key_tab

    def key_ctrl_i(self):
        self.called_by = self.key_ctrl_i


async def test_dispatch_key_raises_when_conflicting_handler_aliases():
    """If you've got a handler for e.g. ctrl+i and a handler for tab, that's probably a mistake.
    In the terminal, they're the same thing, so we fail fast via exception here."""
    widget = DuplicateHandlersWidget()
    with pytest.raises(DuplicateKeyHandlers):
        await dispatch_key(widget, Key(key="tab", character="\t"))
    assert widget.called_by == widget.key_tab


class PreventTestApp(App):
    def __init__(self) -> None:
        self.input_changed_events = []
        super().__init__()

    def compose(self) -> ComposeResult:
        yield Input()

    def on_input_changed(self, event: Input.Changed) -> None:
        self.input_changed_events.append(event)


async def test_message_queue_size():
    """Test message queue size property."""
    app = App()
    assert app.message_queue_size == 0

    class TestMessage(Message):
        pass

    async with app.run_test() as pilot:
        assert app.message_queue_size == 0
        app.post_message(TestMessage())
        assert app.message_queue_size == 1
        app.post_message(TestMessage())
        assert app.message_queue_size == 2
        # A pause will process all the messages
        await pilot.pause()
        assert app.message_queue_size == 0


async def test_prevent() -> None:
    app = PreventTestApp()

    async with app.run_test() as pilot:
        assert not app.input_changed_events
        input = app.query_one(Input)
        input.value = "foo"
        await pilot.pause()
        assert len(app.input_changed_events) == 1
        assert app.input_changed_events[0].value == "foo"

        with input.prevent(Input.Changed):
            input.value = "bar"

        await pilot.pause()
        assert len(app.input_changed_events) == 1
        assert app.input_changed_events[0].value == "foo"


async def test_prevent_with_call_next() -> None:
    """Test for https://github.com/Textualize/textual/issues/3166.

    Does a callback scheduled with `call_next` respect messages that
    were prevented when it was scheduled?
    """

    hits = 0

    class PreventTestApp(App[None]):
        def compose(self) -> ComposeResult:
            yield Input()

        def change_input(self) -> None:
            self.query_one(Input).value += "a"

        def on_input_changed(self) -> None:
            nonlocal hits
            hits += 1

    app = PreventTestApp()
    async with app.run_test() as pilot:
        app.call_next(app.change_input)
        await pilot.pause()
        assert hits == 1

        with app.prevent(Input.Changed):
            app.call_next(app.change_input)
        await pilot.pause()
        assert hits == 1

        app.call_next(app.change_input)
        await pilot.pause()
        assert hits == 2


async def test_prevent_default():
    """Test that prevent_default doesn't apply when a message is bubbled."""

    app_button_pressed = False

    class MyButton(Button):
        def _on_button_pressed(self, event: Button.Pressed) -> None:
            event.prevent_default()

    class PreventApp(App[None]):
        def compose(self) -> ComposeResult:
            yield MyButton("Press me")
            yield Label("No pressure")

        def on_button_pressed(self, event: Button.Pressed) -> None:
            nonlocal app_button_pressed
            app_button_pressed = True
            self.query_one(Label).update("Ouch!")

    app = PreventApp()
    async with app.run_test() as pilot:
        await pilot.click(MyButton)
        assert app_button_pressed