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

fix test

parent 7a9f8b3a
No related branches found
No related tags found
No related merge requests found
import subprocess import subprocess
import typing as tp import typing as tp
import threading import threading
import logging
from .exceptions import ProcessFailed from .exceptions import ProcessFailed
logger = logging.getLogger(__name__)
def read_nowait(process: subprocess.Popen, output_list: tp.List[str]): def read_nowait(process: subprocess.Popen, output_list: tp.List[str]):
try: try:
...@@ -31,20 +34,24 @@ def call_and_return_stdout(args: tp.Union[str, tp.List[str]], ...@@ -31,20 +34,24 @@ def call_and_return_stdout(args: tp.Union[str, tp.List[str]],
""" """
if isinstance(args, str): if isinstance(args, str):
args = args.split(' ') args = args.split(' ')
logger.warning('Modifying kwargs')
kwargs['stdout'] = subprocess.PIPE kwargs['stdout'] = subprocess.PIPE
stdout_list = [] stdout_list = []
logger.warning('Starting popen')
proc = subprocess.Popen(args, **kwargs) proc = subprocess.Popen(args, **kwargs)
reader_thread = threading.Thread(target=read_nowait, args=(proc, stdout_list), daemon=True) reader_thread = threading.Thread(target=read_nowait, args=(proc, stdout_list), daemon=True)
logger.warning('Starting rt')
reader_thread.start() reader_thread.start()
logger.warning('Waiting for termination')
try: try:
proc.wait(timeout=timeout) proc.wait(timeout=timeout)
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
proc.kill() proc.kill()
proc.wait() proc.wait()
logger.warning('Terminated')
if proc.returncode != expected_return_code: if proc.returncode != expected_return_code:
raise ProcessFailed(proc.returncode) raise ProcessFailed(proc.returncode)
......
...@@ -6,10 +6,10 @@ from satella.processes import call_and_return_stdout ...@@ -6,10 +6,10 @@ from satella.processes import call_and_return_stdout
class TestProcesses(unittest.TestCase): class TestProcesses(unittest.TestCase):
@unittest.skipIf('win' in sys.platform, 'Running on Windows') @unittest.skipIf('win' in sys.platform, 'Running on Windows')
def test_return_stdout(self): def test_return_stdout(self):
output = call_and_return_stdout('cat /proc/meminfo', shell=True) output = call_and_return_stdout('cat /proc/meminfo', shell=True, timeout=3)
self.assertIn(b'MemTotal', output) self.assertIn(b'MemTotal', output)
@unittest.skipIf('win' in sys.platform or sys.version_info.minor < 6, 'Running on Windows or Python 3.5') @unittest.skipIf('win' in sys.platform or sys.version_info.minor < 6, 'Running on Windows or Python 3.5')
def test_return_encoding(self): def test_return_encoding(self):
output = call_and_return_stdout('cat /proc/meminfo', shell=True, encoding='utf8') output = call_and_return_stdout('cat /proc/meminfo', shell=True, encoding='utf8', timeout=3)
self.assertIn('MemTotal', output) 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