File: test_threads.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 (49 lines) | stat: -rw-r--r-- 962 bytes parent folder | download | duplicates (3)
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
from threading import Lock, Thread
from time import sleep

from easyprocess import EasyProcess

from pyvirtualdisplay import Display

disps = []
mutex = Lock()


def get_display(threadid, disp):
    dispnr = EasyProcess(["sh", "-c", "echo $DISPLAY"], env=disp.env()).call().stdout
    with mutex:
        disps.append((threadid, dispnr))


def func():
    with Display(manage_global_env=False) as disp:
        get_display(1, disp)
        sleep(2)
        get_display(1, disp)


def test_disp_var():
    t = Thread(target=func)
    t.start()
    sleep(1)

    with Display(manage_global_env=False) as disp:
        get_display(0, disp)
        sleep(2)
        get_display(0, disp)
    t.join()

    print(disps)

    assert disps[0][0] == 1
    assert disps[1][0] == 0
    assert disps[2][0] == 1
    assert disps[3][0] == 0

    # :1
    assert disps[0][1] == disps[2][1]

    # :0
    assert disps[1][1] == disps[3][1]

    assert disps[0][1] != disps[1][1]