File: test_posix.py

package info (click to toggle)
pypy3 7.3.19%2Bdfsg-2
  • links: PTS, VCS
  • area: main
  • in suites: trixie
  • size: 212,236 kB
  • sloc: python: 2,098,316; ansic: 540,565; sh: 21,462; asm: 14,419; cpp: 4,451; makefile: 4,209; objc: 761; xml: 530; exp: 499; javascript: 314; pascal: 244; lisp: 45; csh: 12; awk: 4
file content (58 lines) | stat: -rw-r--r-- 1,663 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
import os
import sys

# Tests variant functions which also accept file descriptors,
# dir_fd and follow_symlinks.
def test_have_functions():
    assert os.stat in os.supports_fd  # fstat() is supported everywhere
    if os.name != 'nt':
        assert os.chdir in os.supports_fd  # fchdir()
    else:
        assert os.chdir not in os.supports_fd
    if os.name == 'posix':
        assert os.open in os.supports_dir_fd  # openat()

def test_popen():
    for i in range(5):
        stream = os.popen('echo 1')
        res = stream.read()
        assert res == '1\n'
        assert stream.close() is None

def test_popen_with():
    stream = os.popen('echo 1')
    with stream as fp:
        res = fp.read()
        assert res == '1\n'

def test_pickle():
    import pickle
    st = os.stat('.')
    # print(type(st).__module__)
    s = pickle.dumps(st)
    # print(repr(s))
    new = pickle.loads(s)
    assert new == st
    assert type(new) is type(st)

if hasattr(os, "fork") and not sys.platform == 'darwin':
    def test_fork_hook_creates_thread_bug():
        # Test fixed in 3.10, no longer errors
        import threading
        def daemon():
            while 1:
                time.sleep(10)

        daemon_thread = None
        def create_thread():
            nonlocal daemon_thread
            daemon_thread = threading.Thread(name="b", target=daemon, daemon=True)
            daemon_thread.start()

        os.register_at_fork(after_in_child=create_thread)
        pid = os.fork()
        if pid == 0:   # child
            os._exit(daemon_thread._ident in threading._active)

        pid1, status1 = os.waitpid(pid, 0)
        assert status1 == 0