Viewing file:  process_recorder.py (1.92 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
from typing import Iterator
from typing import List
from typing import Optional
from typing import TYPE_CHECKING
from typing import Union
from .types import COMMAND
from .utils import Command
if TYPE_CHECKING:
from .fake_popen import FakePopen, AsyncFakePopen
class ProcessRecorder:
"""
Recorder class that holds all FakePopen (or AsyncFakePopen) instances that were
created by the subprocess. The class contains auxiliary
"""
calls: List[Union["FakePopen", "AsyncFakePopen"]]
def __init__(self) -> None:
self.calls = []
@property
def first_call(self) -> Optional[Union["FakePopen", "AsyncFakePopen"]]:
"""Get the first process call"""
if not self.calls:
return None
return self.calls[0]
@property
def last_call(self) -> Optional[Union["FakePopen", "AsyncFakePopen"]]:
"""Get the last (latest) process call"""
if not self.calls:
return None
return self.calls[-1]
def call_count(self, command: Optional[COMMAND] = None) -> int:
"""Get process call count - optionally match with arguments"""
if not command:
return len(self.calls)
return len(tuple(self.get_matching_calls(command)))
def was_called(self, command: Optional[COMMAND] = None) -> bool:
"""Check if process was called - optionally match with arguments"""
if not self.calls:
return False
if not command:
return True
return any(self.get_matching_calls(command))
def get_matching_calls(
self, command: COMMAND
) -> Iterator[Union["FakePopen", "AsyncFakePopen"]]:
"""Get calls that match arguments"""
if not isinstance(command, Command):
command = Command(command)
return (call for call in self.calls if self.calls if command == call.args)
def clear(self) -> None:
"""Clear records"""
self.calls.clear()
|