Repository navigation
Expand file tree
/
Copy pathtest_autointerrupt.py
More file actions
72 lines (55 loc) · 2.13 KB
/
Copy pathtest_autointerrupt.py
File metadata and controls
72 lines (55 loc) · 2.13 KB
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
from subprocess import PIPE
import sys
import threading
import pytest
from git.cmd import Git
class _DummyProc:
"""Minimal stand-in for subprocess.Popen used to exercise AutoInterrupt.
We deliberately raise AttributeError from terminate() to simulate interpreter
shutdown on Windows where subprocess internals (e.g. subprocess._winapi) may
already be torn down.
"""
stdin = None
stdout = None
stderr = None
def poll(self):
return None
def terminate(self):
raise AttributeError("TerminateProcess")
def wait(self): # pragma: no cover - should not be reached in this test
raise AssertionError("wait() should not be called if terminate() fails")
def test_autointerrupt_terminate_ignores_attributeerror():
ai = Git.AutoInterrupt(_DummyProc(), args=["git", "rev-list"])
# Should not raise, even if terminate() triggers AttributeError.
ai._terminate()
# Ensure the reference is cleared to avoid repeated attempts.
assert ai.proc is None
def test_autointerrupt_terminate_closes_buffered_stdin():
ai = Git().hash_object("--stdin", as_process=True, istream=PIPE)
proc = ai.proc
proc.stdin.write(b"buffered input")
try:
ai._terminate()
assert proc.stdin.closed
assert proc.stdout.closed
assert proc.stderr.closed
finally:
proc.stdout.close()
proc.stderr.close()
@pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX SIGTERM handling")
def test_autointerrupt_terminate_closes_pipes_before_waiting():
script = (
"import signal, sys; signal.signal(signal.SIGTERM, signal.SIG_IGN); "
"print('ready', flush=True); sys.stdin.read(); sys.stdout.buffer.write(b'x' * 1000000)"
)
ai = Git().execute([sys.executable, "-c", script], as_process=True, istream=PIPE)
proc = ai.proc
assert proc.stdout.readline() == b"ready\n"
watchdog = threading.Timer(5, proc.kill)
watchdog.start()
try:
ai._terminate()
assert proc.returncode != -9, "cleanup waited for exit with undrained output pipes"
finally:
watchdog.cancel()
watchdog.join()