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
|
import asyncio
import contextlib
import pytest
from libqtile.interactive.repl import (
REPL_PORT,
TERMINATOR,
QtileREPLServer,
get_completions,
)
def test_get_completions_top_level():
local_vars = {"qtile": "dummy", "qtiles": 123}
result = get_completions("qti", local_vars)
assert "qtile" in result
assert "qtiles" in result
def test_get_completions_attribute():
class Dummy:
def method(self):
pass
val = 42
local_vars = {"dummy": Dummy()}
result = get_completions("dummy.me", local_vars)
assert "dummy.method(" in result
result = get_completions("dummy.va", local_vars)
assert "dummy.val" in result
def test_get_completions_invalid_expr():
result = get_completions("invalid..expr", {})
assert result == []
@pytest.mark.anyio
async def test_repl_server_executes_code():
repl = QtileREPLServer()
locals_dict = {"x": 123}
# Start the REPL server in a background task
start_task = asyncio.create_task(repl.start(locals_dict=locals_dict))
# Wait for the server to bind the port
await asyncio.sleep(0.1)
reader, writer = await asyncio.open_connection("localhost", REPL_PORT)
try:
# Read welcome message
welcome = await reader.read(4096)
assert b"Connected to Qtile REPL" in welcome
# Send a simple expression to evaluate
writer.write(b"x\n" + f"{TERMINATOR}\n".encode())
await writer.drain()
# Read REPL result
result = await reader.readuntil(f"{TERMINATOR}\n".encode())
assert "123" in result.decode()
finally:
writer.close()
await writer.wait_closed()
# Stop the REPL server
await repl.stop()
# Cancel the server task
start_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await start_task
|
