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
|
#!/usr/bin/env python3
"""Test public window behavior on isolated display servers."""
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import time
def wait_for(process, ready):
deadline = time.monotonic() + 10
while time.monotonic() < deadline:
if process.poll() is not None:
raise RuntimeError("The display server stopped.")
if ready():
return
time.sleep(0.02)
raise TimeoutError("The display server did not start.")
def main():
weston, xwayland, test, timers, tablet, hello, input_x11, interactive = sys.argv[1:]
servers = []
with tempfile.TemporaryDirectory(prefix="yasl-backends-") as directory:
root = Path(directory)
env = os.environ.copy()
env.pop("WAYLAND_SOCKET", None)
env.update(XDG_RUNTIME_DIR=directory, WAYLAND_DISPLAY="yasl-test",
XDG_SESSION_TYPE="wayland")
with (root / "weston.log").open("w+") as wl_log, \
(root / "xwayland.log").open("w+") as x_log:
try:
wl = subprocess.Popen([
weston, "--backend=headless", "--renderer=pixman",
"--socket=yasl-test", "--idle-time=0", "--no-config",
], env=env, stdout=wl_log, stderr=wl_log)
servers.append(wl)
wait_for(wl, lambda: (root / "yasl-test").is_socket())
display_file = root / "display"
with display_file.open("w") as display_fd:
x = subprocess.Popen([
xwayland, "-displayfd", str(display_fd.fileno()),
"-shm", "-noreset", "-nolisten", "tcp",
], env=env, pass_fds=(display_fd.fileno(),), stdout=x_log, stderr=x_log)
servers.append(x)
wait_for(x, lambda: bool(display_file.read_text().strip()))
env["DISPLAY"] = ":" + display_file.read_text().strip()
def run(mode, request=None, expected=None, **changes):
case_env = env.copy()
for key, value in changes.items():
if value is None:
case_env.pop(key, None)
else:
case_env[key] = value
args = [test, mode]
if request is not None:
args.append(str(request))
if expected is not None:
args.append(str(expected))
print("Test:", mode, request, expected, changes, flush=True)
subprocess.run(args, env=case_env, check=True, timeout=8)
run("open", 1, 1)
run("open", 2, 2)
run("open", 0, 1)
run("null", 0, 1)
run("open", 0, 2, WAYLAND_DISPLAY=None, XDG_SESSION_TYPE="x11")
run("open", 0, 2, WAYLAND_DISPLAY="missing-socket")
run("failure", 1, WAYLAND_DISPLAY="missing-socket")
run("failure", 2, DISPLAY=":59999")
run("failure", 0, WAYLAND_DISPLAY="missing-socket", DISPLAY=":59999")
run("mixed")
print("Test: Wayland surface timers", flush=True)
subprocess.run([timers], env=env, check=True, timeout=8)
x_env = env.copy()
x_env.pop("WAYLAND_DISPLAY", None)
x_env["XDG_SESSION_TYPE"] = "x11"
print("Test: X11 surface timers", flush=True)
subprocess.run([timers], env=x_env, check=True, timeout=8)
print("Test: X11 UTF-8 keyboard fallback and clipboard", flush=True)
subprocess.run([input_x11], env=x_env, check=True, timeout=8)
for backend in ("wayland", "x11"):
print("Test: tablet example on", backend, flush=True)
subprocess.run([tablet, "--backend", backend, "--smoke-test"],
env=env, check=True, timeout=8)
print("Test: hello example on", backend, flush=True)
subprocess.run([hello],
env=env if backend == "wayland" else x_env,
check=True, timeout=8)
print("Test: interactive example on", backend, flush=True)
subprocess.run([interactive, "--smoke-test"],
env=env if backend == "wayland" else x_env,
check=True, timeout=8)
except BaseException:
for log in (wl_log, x_log):
log.flush()
log.seek(0)
print(log.read(), file=sys.stderr)
raise
finally:
for server in reversed(servers):
server.terminate()
try:
server.wait(timeout=5)
except subprocess.TimeoutExpired:
server.kill()
server.wait()
if __name__ == "__main__":
main()
|
