GeezerVison is showing github.com — GitPython/test/test_autointerrupt.py at main · gitpython-developers/GitPython

Expand file tree

test_autointerrupt.py

Copy path

More file actions

Latest commit

History

72 lines (55 loc) · 2.13 KB

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

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()