Skip to content
Snippets Groups Projects
Commit 86fd76fe authored by Piotr Maślanka's avatar Piotr Maślanka
Browse files

add processes

parent 1ad7d801
No related branches found
No related tags found
No related merge requests found
# v2.7.16
* extended `Proxy`
* made `Immutable` utilize `__slots__`
\ No newline at end of file
* made `Immutable` utilize `__slots__`
* added `satella.processes`
\ No newline at end of file
......@@ -26,6 +26,7 @@ Visit the project's page at GitHub_!
files
time
exceptions
processes
Indices and tables
......
processes
=========
.. autofunction:: satella.processes.call_and_return_stdout
__version__ = '2.7.16_a3'
__version__ = '2.7.16_a4'
......@@ -4,7 +4,8 @@ import typing as tp
__all__ = ['BaseSatellaError', 'ResourceLockingError', 'ResourceNotLocked', 'ResourceLocked',
'ConfigurationValidationError', 'ConfigurationError', 'ConfigurationSchemaError',
'PreconditionError', 'MetricAlreadyExists', 'BaseSatellaException', 'CustomException',
'CodedCustomException', 'CodedCustomExceptionMetaclass', 'WouldWaitMore', 'LockIsHeld']
'CodedCustomException', 'CodedCustomExceptionMetaclass', 'WouldWaitMore', 'LockIsHeld',
'ProcessFailed']
class CustomException(Exception):
......@@ -156,3 +157,10 @@ class LockIsHeld(ResourceLocked):
def __init__(self, pid):
self.pid = pid
class ProcessFailed(BaseSatellaError):
"""A process finished with other result code that it was requested"""
def __init__(self, rc: int):
self.rc = rc
import subprocess
import typing as tp
from .exceptions import ProcessFailed
def call_and_return_stdout(args: tp.Union[str, tp.List[str]],
expected_return_code: int = 0, **kwargs) -> tp.Union[bytes, str]:
"""
Call a process and return it's stdout.
:param args: arguments to run the program with. If passed a string, it will be split on space.
:param expected_return_code: an expected return code of this process. 0 is the default. If process
returns anything else, ProcessFailed will be raise
:param ProcessFailed: process' result code was different from the requested
"""
if isinstance(args, str):
args = args.split(' ')
kwargs['capture_output'] = True
proc = subprocess.run(args, **kwargs)
if proc.returncode != expected_return_code:
raise ProcessFailed(proc.returncode)
else:
return proc.stdout
import unittest
import sys
from satella.processes import call_and_return_stdout
class TestProcesses(unittest.TestCase):
@unittest.skipIf('win' in sys.platform, 'Running on Windows')
def test_return_stdout(self):
output = call_and_return_stdout('cat /proc/meminfo', shell=True, encoding='utf8')
self.assertIn('MemTotal', output)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment