Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-27 02:38
📄 __init__.py """Main module""" from . import exceptions from .fake_process import FakeProcess ProcessNotRegisteredError = exceptions.ProcessNotRegisteredError __all__ = ["FakeProcess", "exceptions", "ProcessNotRegisteredError"]	226 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 asyncio_subprocess.py from asyncio.subprocess import DEVNULL from asyncio.subprocess import PIPE from asyncio.subprocess import Process from asyncio.subprocess import STDOUT from asyncio.subprocess import SubprocessStreamProtocol _ = (DEVNULL, PIPE, STDOUT, Process, SubprocessStreamProtocol)	279 B	2026-01-20 12:58
📄 exceptions.py class ProcessNotRegisteredError(Exception): """ Raised when the attempted command wasn't registered before. Use `fake_process.allow_unregistered(True)` if you want to use real subprocess. """ class IncorrectProcessDefinition(Exception): """Raised when the register() has been called with wrong arguments""" class PluginInternalError(Exception): """Raised in case of an internal error in the plugin"""	442 B	2026-01-20 12:58
📄 fake_popen.py """FakePopen class declaration""" import asyncio import collections.abc import concurrent.futures import copy import io import os import signal import subprocess import sys import time from functools import partial from typing import Any as AnyType from typing import Callable from typing import Dict from typing import IO from typing import List from typing import Optional from typing import Sequence from typing import Tuple from typing import Union from . import exceptions from .types import BUFFER from .types import OPTIONAL_TEXT from .types import OPTIONAL_TEXT_OR_ITERABLE from .utils import Thread if sys.platform.startswith("win") and sys.version_info < (3, 8): COMMAND_SEQ = Sequence[Union[str, bytes]] else: COMMAND_SEQ = Sequence[Union[str, bytes, "os.PathLike[str]", "os.PathLike[bytes]"]] class FakePopen: """Base class that fakes the real subprocess.Popen()""" stdout: Optional[BUFFER] = None stderr: Optional[BUFFER] = None returncode: Optional[int] = None text_mode: bool = False pid: int = 0 def __init__( self, command: Union[ Union[bytes, str], COMMAND_SEQ, ], stdout: OPTIONAL_TEXT_OR_ITERABLE = None, stderr: OPTIONAL_TEXT_OR_ITERABLE = None, returncode: int = 0, wait: Optional[float] = None, callback: Optional[Callable] = None, callback_kwargs: Optional[Dict[str, AnyType]] = None, signal_callback: Optional[Callable] = None, stdin_callable: Optional[Callable] = None, *_: Dict[str, AnyType], ) -> None: if ( not isinstance(command, (str, bytes)) and sys.platform.startswith("win") and sys.version_info < (3, 8) ): for arg in command: if isinstance(arg, os.PathLike): msg = f"argument of type {arg.__class__.__name__!r} is not iterable" raise TypeError(msg) self.args = command self.__kwargs: Optional[Dict[str, AnyType]] = None self.__stdout: OPTIONAL_TEXT_OR_ITERABLE = stdout self.__stderr: OPTIONAL_TEXT_OR_ITERABLE = stderr self.__thread: Optional[Thread] = None self.__signal_callback: Optional[Callable] = signal_callback self.__stdin_callable: Optional[Optional[Callable]] = stdin_callable self.__universal_newlines: Optional[Dict[AnyType, AnyType]] = None self._signals: List[int] = [] self._returncode: Optional[int] = returncode self._wait_timeout: Optional[float] = wait self._callback: Optional[Optional[Callable]] = callback self._callback_kwargs: Optional[Dict[str, AnyType]] = callback_kwargs @property def kwargs(self) -> Optional[Dict[str, AnyType]]: return self.__kwargs def __enter__(self) -> "FakePopen": return self def __exit__(self, args: List, kwargs: Dict) -> None: if self.__thread and self.__thread.exception: raise self.__thread.exception def communicate( self, input: OPTIONAL_TEXT = None, timeout: Optional[float] = None ) -> Tuple[AnyType, AnyType]: self._handle_stdin(input) self._finalize_thread(timeout) if isinstance(self.stdout, asyncio.StreamReader) or isinstance( self.stderr, asyncio.StreamReader ): raise exceptions.PluginInternalError return ( self.stdout.getvalue() if self.stdout else None, self.stderr.getvalue() if self.stderr else None, ) def _handle_stdin(self, input: OPTIONAL_TEXT) -> None: if input and self.__stdin_callable: callable_output = self.__stdin_callable(input) if isinstance(callable_output, dict): self.stdout = self._extend_stream_from_dict( callable_output, "stdout", self.stdout ) self.stderr = self._extend_stream_from_dict( callable_output, "stderr", self.stderr ) def _finalize_thread(self, timeout: Optional[float]) -> None: if self.__thread is None: return self.__thread.join(timeout) if self.returncode is None and self._returncode is not None: self.returncode = self._returncode if self.__thread.exception: raise self.__thread.exception def _extend_stream_from_dict( self, dictionary: Dict[str, AnyType], key: str, stream: Optional[BUFFER] ) -> Optional[BUFFER]: data = dictionary.get(key) if data: return self._prepare_buffer(input=data, io_base=stream) return None def poll(self) -> Optional[int]: return self.returncode def wait(self, timeout: Optional[float] = None) -> int: if timeout and self._wait_timeout: self._wait_timeout -= timeout if timeout < self._wait_timeout: raise subprocess.TimeoutExpired(self.args, timeout) self._finalize_thread(timeout) if self.returncode is None: raise exceptions.PluginInternalError return self.returncode def send_signal(self, sig: int) -> None: self._signals.append(sig) if self.__signal_callback: self.__signal_callback(self, sig) def terminate(self) -> None: self.send_signal(signal.SIGTERM) def kill(self) -> None: if sys.platform == "win32": self.terminate() else: self.send_signal(signal.SIGKILL) def configure(self, kwargs: Optional[Dict]) -> None: """Setup the FakePopen instance based on a real Popen arguments.""" self.__kwargs = self.safe_copy(kwargs) self.__universal_newlines = kwargs.get("universal_newlines", None) text = kwargs.get("text", None) encoding = kwargs.get("encoding", None) errors = kwargs.get("errors", None) if text and sys.version_info < (3, 7): raise TypeError("__init__() got an unexpected keyword argument 'text'") self.text_mode = bool(text or self.__universal_newlines or encoding or errors) # validation taken from the real subprocess if ( text is not None and self.__universal_newlines is not None and bool(self.__universal_newlines) != bool(text) ): raise subprocess.SubprocessError( "Cannot disambiguate when both text " "and universal_newlines are supplied but " "different. Pass one or the other." ) stdout = kwargs.get("stdout") if stdout == subprocess.PIPE: self.stdout = self._prepare_buffer(self.__stdout) elif isinstance(stdout, (io.BufferedWriter, io.TextIOWrapper)): self._write_to_buffer(self.__stdout, stdout) stderr = kwargs.get("stderr") if stderr == subprocess.STDOUT and self.__stderr: assert self.stdout is not None self.stdout = self._prepare_buffer(self.__stderr, self.stdout) elif stderr == subprocess.PIPE: self.stderr = self._prepare_buffer(self.__stderr) elif isinstance(stderr, (io.BufferedWriter, io.TextIOWrapper)): self._write_to_buffer(self.__stderr, stderr) @staticmethod def safe_copy(kwargs: Dict[str, AnyType]) -> Dict[str, AnyType]: """ Deepcopy can fail if the value is not serializable, fallback to shallow copy. """ try: return copy.deepcopy(kwargs) except TypeError: return dict(kwargs) def _prepare_buffer( self, input: OPTIONAL_TEXT_OR_ITERABLE, io_base: Optional[BUFFER] = None, ) -> Union[io.BytesIO, io.StringIO, asyncio.StreamReader]: linesep = self._convert(os.linesep) if isinstance(input, (list, tuple)): # need to disable mypy, as input and linesep are unions, # mypy thinks that the types might be incompatible, but # the _convert() function handles that input = linesep.join(map(self._convert, input)) # type: ignore # Add trailing newline if data is present. if input: # same reason to disable mypy as above input += linesep # type: ignore if isinstance(input, str) and not self.text_mode: input = input.encode() if isinstance(input, bytes) and self.text_mode: input = input.decode() if input and self.__universal_newlines and isinstance(input, str): input = input.replace("\r\n", "\n") if ( io_base and not isinstance(io_base, asyncio.StreamReader) and io_base.tell() == 0 ): # same reason for disabling mypy as in `input = linesep.join...`: # both are union so could be incompatible if not _convert() input = io_base.getvalue() + (input) # type: ignore if io_base is None: io_base = self._get_empty_buffer(self.text_mode) if input is None: return io_base # similar as above - mypy has to be disabled because unions if isinstance(io_base, asyncio.StreamReader): io_base.feed_data(self._data_to_bytes(input)) else: io_base.write(input) # type: ignore return io_base def _get_empty_buffer(self, text: bool) -> BUFFER: return io.StringIO() if text else io.BytesIO() def _to_bytes(self, data: Sequence[Union[str, bytes]]) -> Sequence[bytes]: return [elem if isinstance(elem, bytes) else elem.encode() for elem in data] def _data_to_bytes(self, data: OPTIONAL_TEXT_OR_ITERABLE) -> bytes: if isinstance(data, collections.abc.Sequence) and not isinstance(data, bytes): return b"\n".join( (item if isinstance(item, bytes) else item.encode() for item in data) ) if isinstance(data, str): return data.encode() if not data: return b"" return data def _write_to_buffer(self, data: OPTIONAL_TEXT_OR_ITERABLE, buffer: IO) -> None: data_type: Callable = ( # mypy doesn't seem to recognize `partial` as a function partial(bytes, encoding=sys.getfilesystemencoding()) # type: ignore if "b" in buffer.mode else str ) if isinstance(data, (list, tuple)): buffer.writelines([data_type(line + "\n") for line in data]) else: buffer.write(data_type(data)) def _convert(self, input: Union[str, bytes]) -> Union[str, bytes]: if isinstance(input, bytes) and self.text_mode: return input.decode() if isinstance(input, str) and not self.text_mode: return input.encode() return input def _wait(self, wait_period: float) -> None: time.sleep(wait_period) if self.returncode is None: self._finish_process() def run_thread(self) -> None: """Run the user-defined callback or wait in a thread.""" if self._wait_timeout is None and self._callback is None: self._finish_process() else: if self._callback: self.__thread = Thread( target=self._callback, args=(self,), kwargs=self._callback_kwargs or {}, ) else: self.__thread = Thread(target=self._wait, args=(self._wait_timeout,)) self.__thread.start() def _finish_process(self) -> None: self.returncode = self._returncode self._finalize_streams() def _finalize_streams(self) -> None: self._finalize_stream(self.stdout) self._finalize_stream(self.stderr) def _finalize_stream(self, stream: Optional[BUFFER]) -> None: if isinstance(stream, asyncio.StreamReader): stream.feed_eof() elif stream: stream.seek(0) def received_signals(self) -> Tuple[int, ...]: """Get a tuple of signals received by the process.""" return tuple(self._signals) class AsyncFakePopen(FakePopen): """Class to handle async processes""" stdout: Optional[asyncio.StreamReader] stderr: Optional[asyncio.StreamReader] async def communicate( # type: ignore self, input: OPTIONAL_TEXT = None, timeout: Optional[float] = None ) -> Tuple[AnyType, AnyType]: if input: # streams were fed with eof, need to be reopened await self._reopen_streams() self._handle_stdin(input) # feed eof one more time as streams were opened self._finalize_streams() await self._finalize(timeout) return ( await self.stdout.read() if self.stdout else None, await self.stderr.read() if self.stderr else None, ) async def wait(self, timeout: Optional[float] = None) -> int: # type: ignore if timeout and self._wait_timeout and timeout < self._wait_timeout: self._wait_timeout -= timeout raise subprocess.TimeoutExpired(self.args, timeout) await self._finalize(timeout) if self.returncode is None: raise exceptions.PluginInternalError return self.returncode def _get_empty_buffer(self, _: bool) -> asyncio.StreamReader: return asyncio.StreamReader() async def _reopen_streams(self) -> None: self.stdout = await self._reopen_stream(self.stdout) self.stderr = await self._reopen_stream(self.stderr) async def _reopen_stream( self, stream: Optional[asyncio.StreamReader] ) -> Optional[asyncio.StreamReader]: if stream: data = await stream.read() fresh_stream = self._get_empty_buffer(False) fresh_stream.feed_data(data) return fresh_stream return None def run_thread(self) -> None: """Async impl should not contain any thread based implementation""" def evaluate(self) -> None: """Check if process needs to be finished.""" if self._wait_timeout is None and self._callback is None: self._finish_process() async def _run_callback_in_executor(self) -> None: """Run in executor the user-defined callback or wait.""" loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: if self._callback: kwargs = self._callback_kwargs or {} cbk = partial(self._callback, kwargs) await loop.run_in_executor(pool, cbk, self) elif self._wait_timeout is not None: await loop.run_in_executor(pool, self._wait, self._wait_timeout) async def _finalize(self, timeout: Optional[float] = None) -> None: """Run the user-defined callback or wait. Finish process""" if self.returncode is not None: return if timeout is not None: await asyncio.wait_for(self._run_callback_in_executor(), timeout=timeout) else: await self._run_callback_in_executor() if self.returncode is None: self.returncode = self._returncode self._finalize_streams()	15.45 KB	2026-01-20 12:58
📄 fake_process.py """FakeProcess class declaration""" from collections import defaultdict from collections import deque from typing import Any as AnyType from typing import Callable from typing import ClassVar from typing import DefaultDict from typing import Deque from typing import Dict from typing import List from typing import Optional from typing import Type from typing import Union from . import exceptions from .process_dispatcher import ProcessDispatcher from .process_recorder import ProcessRecorder from .types import COMMAND from .types import OPTIONAL_TEXT_OR_ITERABLE from .utils import Any from .utils import Command from .utils import Program class FakeProcess: """Main class responsible for process operations""" any: ClassVar[Type[Any]] = Any program: ClassVar[Type[Program]] = Program def __init__(self) -> None: self.definitions: DefaultDict[Command, Deque[Union[Dict, bool]]] = defaultdict( deque ) self.calls: Deque[COMMAND] = deque() self._allow_unregistered: bool = False self._keep_last_process: bool = False self.exceptions = exceptions def register( self, command: COMMAND, stdout: OPTIONAL_TEXT_OR_ITERABLE = None, stderr: OPTIONAL_TEXT_OR_ITERABLE = None, returncode: int = 0, wait: Optional[float] = None, callback: Optional[Callable] = None, callback_kwargs: Optional[Dict[str, AnyType]] = None, signal_callback: Optional[Callable] = None, occurrences: int = 1, stdin_callable: Optional[Callable] = None, ) -> ProcessRecorder: """ Main method for registering the subprocess instances. Args: command: register the command that will be faked stdout: value of the standard output stderr: value of the error output returncode: return code of the faked process wait: artificially wait for the process to finish callback: function that will be executed instead of the process callback_kwargs: keyword arguments that will be passed into callback occurrences: allow multiple usages of the same command stdin_callable: function that will interact with stdin """ if wait is not None and callback is not None: raise exceptions.IncorrectProcessDefinition( "The 'callback' and 'wait' arguments cannot be used " "together. Add sleep() to your callback instead." ) if not isinstance(command, Command): command = Command(command) recorder = ProcessRecorder() self.definitions[command].extend( [ { "command": command, "stdout": stdout, "stderr": stderr, "returncode": returncode, "wait": wait, "callback": callback, "callback_kwargs": callback_kwargs, "signal_callback": signal_callback, "stdin_callable": stdin_callable, "recorder": recorder, } ] * occurrences ) return recorder register_subprocess = register def pass_command( self, command: COMMAND, occurrences: int = 1, ) -> None: """ Allow to use a real subprocess together with faked ones. Args: command: allow to execute the supplied command occurrences: allow multiple usages of the same command """ if not isinstance(command, Command): command = Command(command) self.definitions[command].extend([True] * occurrences) def __enter__(self) -> "FakeProcess": ProcessDispatcher.register(self) return self def __exit__(self, args: List, *kwargs: Dict) -> None: ProcessDispatcher.deregister(self) def allow_unregistered(self, allow: bool) -> None: """ Allow / block unregistered processes execution. When allowed, the real subprocesses will be called. Blocking will raise the exception. Args: allow: decide whether the unregistered process shall be allowed """ self._allow_unregistered = allow def call_count(self, command: COMMAND) -> int: """ Count how many times a certain command was called. Can be used together with `fake_process.any()`. Args: command: lookup command Returns: number of times a command was called """ if not isinstance(command, Command): command_instance = Command(command) return len(tuple(filter(lambda elem: elem == command_instance, self.calls))) def keep_last_process(self, keep: bool) -> None: """ Keep last process definition from being removed. That can allow / block multiple execution of the same command. Args: keep: decide whether last process shall be kept """ self._keep_last_process = keep @classmethod def context(cls) -> "FakeProcess": """Return a new FakeProcess instance to use it as a context manager.""" return cls()	5.38 KB	2026-01-20 12:58
📄 fixtures.py from typing import Generator import pytest from . import FakeProcess @pytest.fixture def fp() -> Generator[FakeProcess, None, None]: """Fake subprocess calls.""" with FakeProcess() as process: yield process fake_process = fp	262 B	2026-01-20 12:58
📄 process_dispatcher.py """ProcessDispatcher class declaration""" import asyncio import subprocess import sys import typing from collections import deque from copy import deepcopy from functools import partial from typing import Any as AnyType from typing import AnyStr from typing import Awaitable from typing import Callable from typing import Deque from typing import Dict from typing import Generic from typing import List from typing import Optional from typing import Tuple from typing import Type from typing import Union from . import asyncio_subprocess from . import exceptions from .fake_popen import AsyncFakePopen from .fake_popen import FakePopen from .types import COMMAND from .utils import Command if typing.TYPE_CHECKING: from .fake_process import FakeProcess __all__ = ["ProcessDispatcher"] class ProcessDispatcher: """Main class for handling processes.""" process_list: List["FakeProcess"] = [] built_in_popen: Optional[Callable] = None built_in_async_subprocess: Optional[AnyType] = None _allow_unregistered: bool = False _cache: Dict["FakeProcess", Dict["FakeProcess", AnyType]] = dict() _keep_last_process: bool = False _pid: int = 0 @classmethod def register(cls, process: "FakeProcess") -> None: if not cls.process_list: cls.built_in_popen = subprocess.Popen subprocess.Popen = FakePopenWrapper # type: ignore cls.built_in_async_subprocess = asyncio.subprocess asyncio.create_subprocess_shell = cls.async_shell # type: ignore asyncio.create_subprocess_exec = cls.async_exec # type: ignore asyncio.subprocess = asyncio_subprocess cls._cache[process] = { proc: deepcopy(proc.definitions) for proc in cls.process_list } cls.process_list.append(process) @classmethod def deregister(cls, process: "FakeProcess") -> None: cls.process_list.remove(process) cache = cls._cache.pop(process) for proc, processes in cache.items(): proc.definitions = processes if not cls.process_list: subprocess.Popen = cls.built_in_popen # type: ignore cls.built_in_popen = None if cls.built_in_async_subprocess is None: raise exceptions.PluginInternalError asyncio.subprocess = cls.built_in_async_subprocess asyncio.create_subprocess_shell = ( cls.built_in_async_subprocess.create_subprocess_shell ) asyncio.create_subprocess_exec = ( cls.built_in_async_subprocess.create_subprocess_exec ) cls.built_in_async_subprocess = None @classmethod def dispatch( cls, command: COMMAND, kwargs: Optional[Dict] ) -> Union[FakePopen, subprocess.Popen]: """This method will be used instead of the subprocess.Popen()""" process = cls.__dispatch(command) if process is None: if cls.built_in_popen is None: raise exceptions.PluginInternalError popen: subprocess.Popen = cls.built_in_popen(command, kwargs) return popen result = cls._prepare_instance(FakePopen, command, kwargs, process) if not isinstance(result, FakePopen): raise exceptions.PluginInternalError result.run_thread() return result @classmethod async def async_shell( cls, cmd: Union[str, bytes], kwargs: Dict ) -> Union[AsyncFakePopen, asyncio.subprocess.Process]: """Replacement of asyncio.create_subprocess_shell()""" if not isinstance(cmd, (str, bytes)): raise ValueError("cmd must be a string") method = partial( cls.built_in_async_subprocess.create_subprocess_shell, # type: ignore cmd, kwargs ) if isinstance(cmd, bytes): cmd = cmd.decode() return await cls._call_async(cmd, method, kwargs) @classmethod async def async_exec( cls, program: Union[str, bytes], args: Union[str, bytes], kwargs: Dict ) -> Union[AsyncFakePopen, asyncio.subprocess.Process]: """Replacement of asyncio.create_subprocess_exec()""" if not isinstance(program, (str, bytes)): raise ValueError("program must be a string") method = partial( cls.built_in_async_subprocess.create_subprocess_exec, # type: ignore program, args, *kwargs ) if isinstance(program, bytes): program = program.decode() command = [ program, [arg.decode() if isinstance(arg, bytes) else arg for arg in args], ] return await cls._call_async(command, method, kwargs) @classmethod async def _call_async( cls, command: COMMAND, async_method: Callable, kwargs: Dict, ) -> Union[AsyncFakePopen, asyncio.subprocess.Process]: process = cls.__dispatch(command) if process is None: if cls.built_in_async_subprocess is None: raise exceptions.PluginInternalError async_shell: Awaitable[asyncio.subprocess.Process] = async_method() return await async_shell if sys.platform == "win32" and isinstance( asyncio.get_event_loop_policy().get_event_loop(), asyncio.SelectorEventLoop ): raise NotImplementedError( "The SelectorEventLoop doesn't support subprocess" ) result = cls._prepare_instance(AsyncFakePopen, command, kwargs, process) if not isinstance(result, AsyncFakePopen): raise exceptions.PluginInternalError result.evaluate() return result @classmethod def _prepare_instance( cls, klass: Union[Type[FakePopen], Type[AsyncFakePopen]], command: COMMAND, kwargs: dict, process: dict, ) -> Union[FakePopen, AsyncFakePopen]: # Update the command with the actual command specified by the caller. # This will ensure that Command objects do not end up unexpectedly in # caller's objects (e.g. proc.args, CalledProcessError.cmd). Take care # to preserve the dict that may still be referenced when using # keep_last_process. fake_popen_kwargs = process.copy() fake_popen_kwargs["command"] = command recorder = fake_popen_kwargs.pop("recorder") result = klass(fake_popen_kwargs) recorder.calls.append(result) result.pid = cls._pid result.configure(kwargs) return result @classmethod def __dispatch(cls, command: COMMAND) -> Optional[dict]: command_instance, processes, process_instance = cls._get_process(command) if process_instance: process_instance.calls.append(command) if not processes: if not cls.process_list[-1]._allow_unregistered: raise exceptions.ProcessNotRegisteredError( "The process '%s' was not registered." % ( ( command if isinstance(command, str) else " ".join(str(item) for item in command) ), ) ) else: return None process = processes.popleft() if not processes and process_instance is not None: if cls.process_list[-1]._keep_last_process: processes.append(process) elif command_instance: del process_instance.definitions[command_instance] cls._pid += 1 if isinstance(process, bool): # real process will be called return None return process @classmethod def _get_process( cls, command: COMMAND ) -> Tuple[ Optional[Command], Optional[Deque[Union[dict, bool]]], Optional["FakeProcess"] ]: for proc in reversed(cls.process_list): command_instance, processes = next( ( (key, value) for key, value in proc.definitions.items() if key == command ), (None, None), ) process_instance = proc if processes and isinstance(processes, deque): return command_instance, processes, process_instance return None, None, None class FakePopenWrapper(Generic[AnyStr]): def __new__( # type: ignore cls, command: COMMAND, kwargs: Optional[Dict] ) -> FakePopen: return ProcessDispatcher.dispatch(command, kwargs) # type: ignore	8.85 KB	2026-01-20 12:58
📄 process_recorder.py 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()	1.98 KB	2026-01-20 12:58
📄 py.typed	0 B	2026-01-20 12:58
📄 types.py import asyncio import io import os from typing import Sequence from typing import Union from .utils import Any from .utils import Command from .utils import Program OPTIONAL_TEXT = Union[str, bytes, None] OPTIONAL_TEXT_OR_ITERABLE = Union[ str, bytes, None, Sequence[Union[str, bytes]], ] BUFFER = Union[io.BytesIO, io.StringIO, asyncio.StreamReader] ARGUMENT = Union[str, Any, os.PathLike, Program] COMMAND = Union[Sequence[ARGUMENT], str, Command]	488 B	2026-01-20 12:58
📄 utils.py import os import sys import threading from pathlib import Path from typing import Any as AnyType from typing import Iterator from typing import Optional from typing import Sequence from typing import Tuple from typing import TYPE_CHECKING from typing import Union if TYPE_CHECKING: from .types import COMMAND ARGUMENT = Union[str, "Any", os.PathLike] class Thread(threading.Thread): """Custom thread class to capture exceptions""" exception: Optional[Exception] = None def run(self) -> None: try: super().run() except Exception as exc: self.exception = exc class Command: """Command definition class.""" __slots__ = "command" def __init__( self, command: "COMMAND", ): if isinstance(command, str): command = tuple(command.split(" ")) if not isinstance(command, (list, tuple)): raise TypeError("Command can be only of type string, list or tuple.") self.command: Tuple[ARGUMENT, ...] = tuple( os.fspath(c) if isinstance(c, os.PathLike) else c for c in command ) for i, command_elem in enumerate(self.command): if isinstance(command_elem, Any) and isinstance( self._get_next_command_elem(i), Any ): raise AttributeError("Cannot use `Any()` one after another.") def __eq__(self, other: AnyType) -> bool: if isinstance(other, str): other = other.split(" ") norm_command = [ os.fspath(c) if isinstance(c, os.PathLike) else c for c in self.command ] norm_other = [os.fspath(c) if isinstance(c, os.PathLike) else c for c in other] if norm_other == norm_command: # straightforward matching return True for i, command_elem in enumerate(norm_command): if isinstance(command_elem, Any): next_command_elem = self._get_next_command_elem(i) if next_command_elem is None: if not self._are_thresholds_ok(command_elem, len(norm_other)): return False return True else: next_matching_elem = self._get_next_matching_elem_index( norm_other, next_command_elem ) if next_matching_elem is None: return False else: if not self._are_thresholds_ok( command_elem, next_matching_elem ): return False norm_other = norm_other[next_matching_elem:] else: if len(norm_other) == 0 or norm_other.pop(0) != command_elem: return False return len(norm_other) == 0 def __iter__(self) -> Iterator: return iter(self.command) @staticmethod def _are_thresholds_ok(command_elem: "Any", value: int) -> bool: if command_elem.max is not None and value > command_elem.max: return False if command_elem.min is not None and value < command_elem.min: return False return True def _get_next_command_elem(self, index: int) -> Optional[ARGUMENT]: try: return self.command[index + 1] except IndexError: return None @staticmethod def _get_next_matching_elem_index( other: Sequence[ARGUMENT], elem: ARGUMENT ) -> Optional[int]: return next( (i for i, other_elem in enumerate(other) if elem == other_elem), None ) def __hash__(self) -> int: return hash(self.command) def __repr__(self) -> str: return str(self.command) def __str__(self) -> str: return str(self.command) class Any: """Wildcard definition class.""" def __init__(self, *, min: Optional[int] = None, max: Optional[int] = None) -> None: if min is not None and max is not None and min > max: raise AttributeError("min cannot be greater than max") self.min: Optional[int] = min self.max: Optional[int] = max def __str__(self) -> str: return f"{self.__class__.__name__} (min={self.min}, max={self.max})" def __repr__(self) -> str: return str(self) class Program: """Specifies the name of the final program to be executed.""" def __init__(self, program: str) -> None: self.program: str = program def __repr__(self) -> str: return f"{self.__class__.__name__}({self.program!r})" def __eq__(self, other: AnyType) -> bool: if isinstance(other, str): if Path(other).name == self.program: return True if sys.platform.startswith("win"): for ext in os.environ.get("PATHEXT", "").split(os.pathsep): if ( Path(other).name.lower() == Path(self.program).with_suffix(ext).name.lower() ): return True return False def __hash__(self) -> int: return hash(self.program)	5.28 KB	2026-01-20 12:58

Terminal