File: test_race.py

package info (click to toggle)
pyvirtualdisplay 3.0-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 388 kB
  • sloc: python: 1,396; sh: 62; ruby: 39; makefile: 4
file content (88 lines) | stat: -rw-r--r-- 2,052 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
import sys
from time import sleep

from easyprocess import EasyProcess
from entrypoint2 import entrypoint
from tutil import has_xvnc, worker

from pyvirtualdisplay import Display

# ubuntu 14.04 no displayfd
# ubuntu 16.04 displayfd

# Xephyr leaks shared memory segments until 18.04
# 		https://gitlab.freedesktop.org/xorg/xserver/-/issues/130
#       approximately 4MB/call


def test_race_100_xvfb():
    check_n(100, "xvfb")


# TODO: this fails with "Xephyr cannot open host display"
#   Xephyr bug?
# def test_race_500_100_xephyr():
#     for _ in range(500):
#         check_n(100, "xephyr")


if has_xvnc():

    def test_race_10_xvnc():
        check_n(10, "xvnc")


def check_n(n, backend):
    with Display():
        ls = []
        try:
            for i in range(n):
                cmd = [
                    sys.executable,
                    __file__.rsplit(".", 1)[0] + ".py",
                    str(i),
                    backend,
                    str(n),
                    "--debug",
                ]
                p = EasyProcess(cmd)
                p.start()
                ls += [p]

            sleep(3)

            good_count = 0
            rc_ls = []
            for p in ls:
                p.wait()
                if p.return_code == 0:
                    good_count += 1
                rc_ls += [p.return_code]
        finally:
            for p in ls:
                p.stop()
        print(rc_ls)
        print(good_count)
        assert good_count == n


@entrypoint
def main(i, backend, retries):
    retries = int(retries)
    kwargs = dict()
    if backend == "xvnc":
        kwargs["rfbport"] = 42000 + 100 * worker() + int(i)
    # print("$DISPLAY=%s" % (os.environ.get("DISPLAY")))
    d = Display(backend=backend, retries=retries, **kwargs).start()
    print(
        "my index:%s  backend:%s disp:%s retries:%s"
        % (
            i,
            backend,
            d.new_display_var,
            d._obj._retries_current,
        )
    )
    ok = d.is_alive()
    d.stop()
    assert ok