36 lines
1.2 KiB
Python
36 lines
1.2 KiB
Python
import subprocess
|
|
from collections.abc import Mapping, Sequence
|
|
|
|
from .models import CompletedCommand # type: ignore[reportMissingImports]
|
|
|
|
|
|
class SubprocessRunner:
|
|
"""Small injectable subprocess seam used by every provider adapter."""
|
|
|
|
def run(self, argv: Sequence[str], *, cwd=None, env: Mapping[str, str] | None = None, timeout=None):
|
|
process = subprocess.run(
|
|
list(argv),
|
|
cwd=str(cwd) if cwd else None,
|
|
env=dict(env) if env is not None else None,
|
|
timeout=timeout,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
return CompletedCommand(process.stdout, process.stderr, process.returncode)
|
|
|
|
|
|
class RecordingRunner:
|
|
"""Useful public fake runner for consumers and contract tests."""
|
|
|
|
def __init__(self, responses=None):
|
|
self.calls = []
|
|
self.responses = list(responses or [])
|
|
|
|
def run(self, argv, **kwargs):
|
|
self.calls.append((list(argv), kwargs))
|
|
if self.responses:
|
|
response = self.responses.pop(0)
|
|
return response if isinstance(response, CompletedCommand) else CompletedCommand(*response)
|
|
return CompletedCommand("{}")
|