From cf260249a2ad018b751be162d3913046fab76545 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Thu, 17 Sep 2026 11:24:34 +0300 Subject: [PATCH] T_OS_EXEC_ENV is added. It replaces OsOperations.T_EXEC_ENV. --- src/local_ops.py | 5 +++-- src/os_ops.py | 7 +++---- src/remote_ops.py | 9 +++++---- src/types.py | 1 + tests/test_os_ops_common.py | 17 +++++++++-------- 5 files changed, 21 insertions(+), 18 deletions(-) diff --git a/src/local_ops.py b/src/local_ops.py index 9bddd57..e2845f6 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -33,6 +33,7 @@ from .os_ops import T_OS_IO from .os_ops import T_OS_IO_ID from .os_ops import T_OS_RUN_INPUT +from .os_ops import T_OS_EXEC_ENV from .raise_error import RaiseError from .helpers import Helpers @@ -509,7 +510,7 @@ def popen( stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, - exec_env: typing.Optional[OsOperations.T_EXEC_ENV] = None, + exec_env: typing.Optional[T_OS_EXEC_ENV] = None, cwd: typing.Optional[str] = None ) -> OsProcessController: assert text is None or type(text) is bool @@ -576,7 +577,7 @@ def run( stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, - exec_env: typing.Optional[OsOperations.T_EXEC_ENV] = None, + exec_env: typing.Optional[T_OS_EXEC_ENV] = None, cwd: typing.Optional[str] = None, timeout: typing.Optional[T_OS_TIMEOUT] = None, check: bool = True, diff --git a/src/os_ops.py b/src/os_ops.py index ffa1943..bf62ee8 100644 --- a/src/os_ops.py +++ b/src/os_ops.py @@ -6,6 +6,7 @@ from .types import T_OS_IO from .types import T_OS_IO_ID from .types import T_OS_RUN_INPUT +from .types import T_OS_EXEC_ENV from .raise_error import RaiseError import locale @@ -204,8 +205,6 @@ def exec_command( assert cwd is None or type(cwd) is str RaiseError.MethodIsNotImplemented(__class__, "exec_command") - T_EXEC_ENV = typing.Dict[str, typing.Optional[str]] - def popen( self, cmd: T_OS_CMD, @@ -215,7 +214,7 @@ def popen( stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, - exec_env: typing.Optional[T_EXEC_ENV] = None, + exec_env: typing.Optional[T_OS_EXEC_ENV] = None, cwd: typing.Optional[str] = None ) -> OsProcessController: assert type(cmd) is str or type(cmd) is list @@ -239,7 +238,7 @@ def run( stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, - exec_env: typing.Optional[T_EXEC_ENV] = None, + exec_env: typing.Optional[T_OS_EXEC_ENV] = None, cwd: typing.Optional[str] = None, timeout: typing.Optional[T_OS_TIMEOUT] = None, check: bool = True, diff --git a/src/remote_ops.py b/src/remote_ops.py index 8b1e652..dc56303 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -28,6 +28,7 @@ from .os_ops import T_OS_IO from .os_ops import T_OS_IO_ID from .os_ops import T_OS_RUN_INPUT +from .os_ops import T_OS_EXEC_ENV from .raise_error import RaiseError from .helpers import Helpers from .static_config import OsOperationStaticConfig @@ -544,7 +545,7 @@ def popen( stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, - exec_env: typing.Optional[OsOperations.T_EXEC_ENV] = None, + exec_env: typing.Optional[T_OS_EXEC_ENV] = None, cwd: typing.Optional[str] = None ) -> OsProcessController: assert type(cmd) is str or type(cmd) is list @@ -743,7 +744,7 @@ def run( stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, - exec_env: typing.Optional[OsOperations.T_EXEC_ENV] = None, + exec_env: typing.Optional[T_OS_EXEC_ENV] = None, cwd: typing.Optional[str] = None, timeout: typing.Optional[T_OS_TIMEOUT] = None, check: bool = True, @@ -1781,7 +1782,7 @@ def _transport_popen( stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, - exec_env: typing.Optional[OsOperations.T_EXEC_ENV] = None, + exec_env: typing.Optional[T_OS_EXEC_ENV] = None, cwd: typing.Optional[str] = None ) -> subprocess.Popen: assert type(cmd) in [str, list] @@ -1872,7 +1873,7 @@ def _transport_run( stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, - exec_env: typing.Optional[OsOperations.T_EXEC_ENV] = None, + exec_env: typing.Optional[T_OS_EXEC_ENV] = None, cwd: typing.Optional[str] = None, check: bool = True, ) -> tagTransportRunResult: diff --git a/src/types.py b/src/types.py index 8eee35d..050b364 100644 --- a/src/types.py +++ b/src/types.py @@ -9,3 +9,4 @@ T_OS_IO = typing.IO[typing.Any] T_OS_IO_ID = typing.Union[int, T_OS_IO] T_OS_RUN_INPUT = typing.Union[str, bytes] +T_OS_EXEC_ENV = typing.Dict[str, typing.Optional[str]] diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 7f983c9..6a6d7ef 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -10,6 +10,7 @@ from src.os_ops import OsProcessController from src.os_ops import OsCommandResult +from src.os_ops import T_OS_EXEC_ENV from src.exceptions import ExecTimeoutException import os @@ -4874,7 +4875,7 @@ def test_popen_set_env( os_ops = os_ops_descr.os_ops - envs: OsOperations.T_EXEC_ENV = { + envs: T_OS_EXEC_ENV = { "AAA": "abcdefg", } @@ -4957,7 +4958,7 @@ def test_popen_unset_env( s = controller.stdout.read() assert s != "" - envs: OsOperations.T_EXEC_ENV = { + envs: T_OS_EXEC_ENV = { "PATH": None, } @@ -5014,7 +5015,7 @@ def test_popen_unset_env_of_os_ops( s = controller.stderr.read() assert s == "" - envs: OsOperations.T_EXEC_ENV = { + envs: T_OS_EXEC_ENV = { "AAA": None, } @@ -5071,7 +5072,7 @@ def test_popen_replace_env_of_os_ops( s = controller.stderr.read() assert s == "" - envs: OsOperations.T_EXEC_ENV = { + envs: T_OS_EXEC_ENV = { "AAA": "xyz", } @@ -5912,7 +5913,7 @@ def test_run_set_env( os_ops = os_ops_descr.os_ops - envs: OsOperations.T_EXEC_ENV = { + envs: T_OS_EXEC_ENV = { "AAA": "abcdefg", } @@ -5985,7 +5986,7 @@ def test_run_unset_env( assert exec_r.stdout is not None assert exec_r.stdout != "" - envs: OsOperations.T_EXEC_ENV = { + envs: T_OS_EXEC_ENV = { "PATH": None, } @@ -6035,7 +6036,7 @@ def test_run_unset_env_of_os_ops( s = exec_r.stderr assert s == "" - envs: OsOperations.T_EXEC_ENV = { + envs: T_OS_EXEC_ENV = { "AAA": None, } @@ -6091,7 +6092,7 @@ def test_run_replace_env_of_os_ops( s = exec_r.stderr assert s == "" - envs: OsOperations.T_EXEC_ENV = { + envs: T_OS_EXEC_ENV = { "AAA": "xyz", }