FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
GitPython/test/test_autointerrupt.py at main · gitpython-developers/GitPython · GitHub
gitpython-developers
GitPython
Repository navigation
Code
Issues
6
(6)
Pull requests
1
(1)
Discussions
Actions
Security and quality
41
(41)
Insights
Expand file tree
Breadcrumbs
GitPython
/
test
/
test_autointerrupt.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
72 lines (55 loc) · 2.13 KB
Breadcrumbs
GitPython
/
test
/
test_autointerrupt.py
Copy path
File metadata and controls
72 lines (55 loc) · 2.13 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
()
Back
|
FazBrowse Home
|
New Git URL