style(src): resolve simple standard Python typings (691ac201) · Commits · RadianDevCore / Tools / pexpect-executor

src/cli/main.py

+7 −3

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3

		# Standard libraries
		from argparse import ArgumentParser, RawTextHelpFormatter
		from argparse import (
		ArgumentParser,
		Namespace,
		RawTextHelpFormatter,
		)
		from shutil import get_terminal_size
		from sys import exit as sys_exit

		@@ -26,7 +30,7 @@ def main() -> None:
		result: Entrypoint.Result = Entrypoint.Result.ERROR

		# Arguments creation
		parser = ArgumentParser(
		parser: ArgumentParser = ArgumentParser(
		prog=Bundle.NAME,
		description=f'{Bundle.NAME}: Automate interactive CLI tools actions',
		add_help=False,
		@@ -224,7 +228,7 @@ def main() -> None:
		)

		# Arguments parser
		options = parser.parse_args()
		options: Namespace = parser.parse_args()

		# Help informations
		if options.help:

+21 −10

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3

		# Standard libraries
		from typing import Any, List, Optional, Union

		# Components
		from ..system.platform import Platform

		@@ -7,28 +10,29 @@ from ..system.platform import Platform
		class Base:

		# Members
		_child = None
		_child: Optional[Any] = None

		# Constructor
		def __init__(self, command):
		def __init__(self, command: str) -> None:

		# Virtual method
		raise NotImplementedError() # pragma: no cover

		# Read
		def _read(self):
		def _read(self) -> Union[bytes, List[Union[bytes, str]], str]:

		# Virtual method
		raise NotImplementedError() # pragma: no cover

		# Is alive
		def isalive(self):
		def isalive(self) -> bool:

		# Result
		return self._child.isalive()
		assert self._child is not None
		return bool(self._child.isalive())

		# Read
		def read(self, strips=None):
		def read(self, strips: Optional[List[str]] = None) -> None:

		# Read stream
		while True:
		@@ -64,19 +68,26 @@ class Base:
		Platform.flush()

		# Send
		def send(self, key):
		def send(self, key: Union[bytes, str]) -> None:

		# Send key
		assert self._child is not None
		self._child.send(key)

		# Status
		def status(self):
		def status(self) -> int:

		# Fallback status
		assert self._child is not None
		if self._child.exitstatus is None:
		return 1

		# Process status
		return self._child.exitstatus
		return int(self._child.exitstatus)

		# Terminate
		def terminate(self, force=False):
		def terminate(self, force: bool = False) -> None:

		# Terminate process
		assert self._child is not None
		self._child.terminate(force=force)

+2 −3

Original line number	Diff line number	Diff line
		@@ -20,7 +20,7 @@ if Platform.IS_WINDOWS:

		# Optional Wexpect engine
		elif environ[Bundle.ENV_ENGINE] == 'wexpect':
		from .wexpect import Wexpect as Engine # pylint: disable=unused-import
		from .wexpect import Wexpect as Engine # type: ignore[assignment] # pylint: disable=unused-import

		# Unknown engine
		else:
		@@ -28,6 +28,5 @@ if Platform.IS_WINDOWS:

		# Optional modules libraries (Linux)
		else:

		# Optional Pexpect engine
		from .pexpect import Pexpect as Engine # pylint: disable=unused-import
		from .pexpect import Pexpect as Engine # type: ignore[assignment] # pylint: disable=unused-import

+11 −5

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3

		# Standard libraries
		from typing import Union

		# Modules libraries
		from pexpect import EOF, spawn, TIMEOUT # pylint: disable=import-error

		@@ -10,19 +13,22 @@ from .base import Base
		class Pexpect(Base):

		# Constructor
		def __init__(self, command): # pylint: disable=super-init-not-called
		def __init__(self, command: str) -> None: # pylint: disable=super-init-not-called

		# Spawn command
		self._child = spawn('sh', ['-c', command])

		# Read
		def _read(self): # pylint: disable=duplicate-code
		def _read(self) -> Union[bytes, str]:

		# Acquire output
		# Acquire output, pylint: disable=duplicate-code
		assert self._child is not None
		try:
		return self._child.read_nonblocking(size=1024, timeout=1)
		result = self._child.read_nonblocking(size=1024, timeout=1)
		assert isinstance(result, (bytes, str))
		return result
		except (AttributeError, EOF, TIMEOUT):
		pass

		# Empty output
		return None
		return ''

+13 −7

Original line number	Diff line number	Diff line
		@@ -2,6 +2,7 @@

		# Standard libraries
		from signal import SIGTERM
		from typing import Union

		# Modules libraries
		from pexpect import EOF, TIMEOUT # pylint: disable=import-error
		@@ -14,33 +15,38 @@ from .base import Base
		class PexpectPopen(Base):

		# Constructor
		def __init__(self, command): # pylint: disable=super-init-not-called
		def __init__(self, command: str) -> None: # pylint: disable=super-init-not-called

		# Spawn command
		self._child = PopenSpawn(command)

		# Read
		def _read(self): # pylint: disable=duplicate-code
		def _read(self) -> Union[bytes, str]:

		# Acquire output
		# Acquire output, pylint: disable=duplicate-code
		assert self._child is not None
		try:
		return self._child.read_nonblocking(size=1024, timeout=1)
		result = self._child.read_nonblocking(size=1024, timeout=1)
		assert isinstance(result, (bytes, str))
		return result
		except (EOF, TIMEOUT):
		pass

		# Empty output
		return None
		return ''

		# Is alive
		def isalive(self):
		def isalive(self) -> bool:

		# Result
		assert self._child is not None
		return not self._child.terminated

		# Terminate
		def terminate(self, force=False):
		def terminate(self, force: bool = False) -> None:

		# Terminate process
		assert self._child is not None
		if force:
		try:
		self._child.kill(sig=SIGTERM)