Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/contracts/process.py: 100%

1# SPDX-License-Identifier: MIT 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Defines the contract for a worker process pool service. 

5 

6This module specifies the `ProcessPoolProtocol`, a formal interface that any 

7service managing a pool of worker processes for command execution must 

8implement. This allows for abstracting the details of subprocess management. 

9""" 

10 

11from __future__ import annotations 

12 

13from typing import Any, Protocol, runtime_checkable 

14 

15 

16@runtime_checkable 

17class ProcessPoolProtocol(Protocol): 

18 """Defines the contract for a worker process pool. 

19 

20 This interface specifies the methods for running commands in isolated 

21 worker processes and managing the lifecycle of the pool. 

22 """ 

23 

24 def run(self, cmd: list[str], *, executor: str) -> tuple[int, bytes, bytes]: 

25 """Executes a command in a worker process. 

26 

27 Args: 

28 cmd (list[str]): A list of command arguments to execute. 

29 executor (str): The name or identifier of the executor to use. 

30 

31 Returns: 

32 tuple[int, bytes, bytes]: A tuple containing the return code, 

33 standard output (stdout), and standard error (stderr) as bytes. 

34 """ 

35 ... 

36 

37 def shutdown(self) -> None: 

38 """Shuts down the process pool and releases all associated resources.""" 

39 ... 

40 

41 def get_status(self) -> dict[str, Any]: 

42 """Returns the current status of the process pool. 

43 

44 Returns: 

45 dict[str, Any]: A dictionary containing status information, such as 

46 the number of active workers, queue size, etc. 

47 """ 

48 ...