blob: 58daf18654b1c9de2013c1e8696296ab904ad416 [file] [log] [blame]
# Copyright 2023 The Fuchsia Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
from typing import Protocol
class Runner(Protocol):
"""A command runner."""
def run(
self,
command: str | list[str],
timeout_sec: int | None = None,
ignore_status: bool = False,
) -> CompletedProcess:
"""Run command with arguments.
Args:
command: Command to execute
timeout: Seconds to wait for command to finish
ignore_status: Ignore the exit code of command. Non-zero exit codes
need to be handled manually.
Returns:
Result of the completed command.
Raises:
subprocess.CalledProcessError: when the process exits with a non-zero status
subprocess.TimeoutExpired: when the timeout expires while waiting
for a child process
"""
...
def run_async(self, command: str) -> CompletedProcess:
"""Run command asynchronously.
Args:
command: Command to execute
Returns:
Results of the dispatched command.
"""
...
class CompletedProcess(Protocol):
@property
def returncode(self) -> int:
"""Exit status."""
...
@property
def stdout(self) -> str:
"""Output stream."""
...
@property
def stderr(self) -> str:
"""Error output stream."""
...