[PATCH 05/17] patman: Convert camel case in cros_subprocess.py
Simon Glass
sjg at chromium.org
Sat Jan 29 22:14:08 CET 2022
Convert this file to snake case and update all files which use it.
Signed-off-by: Simon Glass <sjg at chromium.org>
---
tools/patman/command.py | 2 +-
tools/patman/cros_subprocess.py | 59 +++++++++++++++++----------------
2 files changed, 31 insertions(+), 30 deletions(-)
diff --git a/tools/patman/command.py b/tools/patman/command.py
index c848aa19879..24358784f26 100644
--- a/tools/patman/command.py
+++ b/tools/patman/command.py
@@ -108,7 +108,7 @@ def run_pipe(pipe_list, infile=None, outfile=None,
if capture:
result.stdout, result.stderr, result.combined = (
- last_pipe.CommunicateFilter(output_func))
+ last_pipe.communicate_filter(output_func))
if result.stdout and oneline:
result.output = result.stdout.rstrip(b'\r\n')
result.return_code = last_pipe.wait()
diff --git a/tools/patman/cros_subprocess.py b/tools/patman/cros_subprocess.py
index 88a4693feff..f1b26087cfd 100644
--- a/tools/patman/cros_subprocess.py
+++ b/tools/patman/cros_subprocess.py
@@ -49,7 +49,7 @@ class Popen(subprocess.Popen):
to us as soon as it is produced, rather than waiting for the end of a
line.
- Use CommunicateFilter() to handle output from the subprocess.
+ Use communicate_filter() to handle output from the subprocess.
"""
@@ -100,7 +100,7 @@ class Popen(subprocess.Popen):
if kwargs:
raise ValueError("Unit tests do not test extra args - please add tests")
- def ConvertData(self, data):
+ def convert_data(self, data):
"""Convert stdout/stderr data to the correct format for output
Args:
@@ -113,7 +113,7 @@ class Popen(subprocess.Popen):
return b''
return data
- def CommunicateFilter(self, output):
+ def communicate_filter(self, output):
"""Interact with process: Read data from stdout and stderr.
This method runs until end-of-file is reached, then waits for the
@@ -122,7 +122,7 @@ class Popen(subprocess.Popen):
The output function is sent all output from the subprocess and must be
defined like this:
- def Output([self,] stream, data)
+ def output([self,] stream, data)
Args:
stream: the stream the output was received on, which will be
sys.stdout or sys.stderr.
@@ -236,9 +236,9 @@ class Popen(subprocess.Popen):
self.terminate()
# All data exchanged. Translate lists into strings.
- stdout = self.ConvertData(stdout)
- stderr = self.ConvertData(stderr)
- combined = self.ConvertData(combined)
+ stdout = self.convert_data(stdout)
+ stderr = self.convert_data(stderr)
+ combined = self.convert_data(combined)
# Translate newlines, if requested. We cannot let the file
# object do the translation: It is based on stdio, which is
@@ -281,7 +281,7 @@ class TestSubprocess(unittest.TestCase):
self.stdin_read_pipe = pipe[0]
self._stdin_write_pipe = os.fdopen(pipe[1], 'w')
- def Output(self, stream, data):
+ def output(self, stream, data):
"""Output handler for Popen. Stores the data for later comparison"""
if stream == sys.stdout:
self.stdout_data += data
@@ -294,7 +294,7 @@ class TestSubprocess(unittest.TestCase):
self._stdin_write_pipe.write(self._input_to_send + '\r\n')
self._stdin_write_pipe.flush()
- def _BasicCheck(self, plist, oper):
+ def _basic_check(self, plist, oper):
"""Basic checks that the output looks sane."""
self.assertEqual(plist[0], oper.stdout_data)
self.assertEqual(plist[1], oper.stderr_data)
@@ -306,15 +306,15 @@ class TestSubprocess(unittest.TestCase):
def test_simple(self):
"""Simple redirection: Get process list"""
oper = TestSubprocess.MyOperation()
- plist = Popen(['ps']).CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ plist = Popen(['ps']).communicate_filter(oper.output)
+ self._basic_check(plist, oper)
def test_stderr(self):
"""Check stdout and stderr"""
oper = TestSubprocess.MyOperation()
cmd = 'echo fred >/dev/stderr && false || echo bad'
- plist = Popen([cmd], shell=True).CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ plist = Popen([cmd], shell=True).communicate_filter(oper.output)
+ self._basic_check(plist, oper)
self.assertEqual(plist [0], 'bad\r\n')
self.assertEqual(plist [1], 'fred\r\n')
@@ -323,8 +323,8 @@ class TestSubprocess(unittest.TestCase):
oper = TestSubprocess.MyOperation()
cmd = 'echo test >/dev/stderr'
self.assertRaises(OSError, Popen, [cmd], shell=False)
- plist = Popen([cmd], shell=True).CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ plist = Popen([cmd], shell=True).communicate_filter(oper.output)
+ self._basic_check(plist, oper)
self.assertEqual(len(plist [0]), 0)
self.assertEqual(plist [1], 'test\r\n')
@@ -332,8 +332,8 @@ class TestSubprocess(unittest.TestCase):
"""Check with and without shell works using list arguments"""
oper = TestSubprocess.MyOperation()
cmd = ['echo', 'test', '>/dev/stderr']
- plist = Popen(cmd, shell=False).CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ plist = Popen(cmd, shell=False).communicate_filter(oper.output)
+ self._basic_check(plist, oper)
self.assertEqual(plist [0], ' '.join(cmd[1:]) + '\r\n')
self.assertEqual(len(plist [1]), 0)
@@ -341,16 +341,17 @@ class TestSubprocess(unittest.TestCase):
# this should be interpreted as 'echo' with the other args dropped
cmd = ['echo', 'test', '>/dev/stderr']
- plist = Popen(cmd, shell=True).CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ plist = Popen(cmd, shell=True).communicate_filter(oper.output)
+ self._basic_check(plist, oper)
self.assertEqual(plist [0], '\r\n')
def test_cwd(self):
"""Check we can change directory"""
for shell in (False, True):
oper = TestSubprocess.MyOperation()
- plist = Popen('pwd', shell=shell, cwd='/tmp').CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ plist = Popen('pwd', shell=shell, cwd='/tmp').communicate_filter(
+ oper.output)
+ self._basic_check(plist, oper)
self.assertEqual(plist [0], '/tmp\r\n')
def test_env(self):
@@ -361,8 +362,8 @@ class TestSubprocess(unittest.TestCase):
if add:
env ['FRED'] = 'fred'
cmd = 'echo $FRED'
- plist = Popen(cmd, shell=True, env=env).CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ plist = Popen(cmd, shell=True, env=env).communicate_filter(oper.output)
+ self._basic_check(plist, oper)
self.assertEqual(plist [0], add and 'fred\r\n' or '\r\n')
def test_extra_args(self):
@@ -380,8 +381,8 @@ class TestSubprocess(unittest.TestCase):
prompt = 'What is your name?: '
cmd = 'echo -n "%s"; read name; echo Hello $name' % prompt
plist = Popen([cmd], stdin=oper.stdin_read_pipe,
- shell=True).CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ shell=True).communicate_filter(oper.output)
+ self._basic_check(plist, oper)
self.assertEqual(len(plist [1]), 0)
self.assertEqual(plist [0], prompt + 'Hello Flash\r\r\n')
@@ -393,16 +394,16 @@ class TestSubprocess(unittest.TestCase):
both_cmds = ''
for fd in (1, 2):
both_cmds += cmd % (fd, fd, fd, fd, fd)
- plist = Popen(both_cmds, shell=True).CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ plist = Popen(both_cmds, shell=True).communicate_filter(oper.output)
+ self._basic_check(plist, oper)
self.assertEqual(plist [0], 'terminal 1\r\n')
self.assertEqual(plist [1], 'terminal 2\r\n')
# Now try with PIPE and make sure it is not a terminal
oper = TestSubprocess.MyOperation()
plist = Popen(both_cmds, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
- shell=True).CommunicateFilter(oper.Output)
- self._BasicCheck(plist, oper)
+ shell=True).communicate_filter(oper.output)
+ self._basic_check(plist, oper)
self.assertEqual(plist [0], 'not 1\n')
self.assertEqual(plist [1], 'not 2\n')
--
2.35.0.rc2.247.g8bbb082509-goog
More information about the U-Boot
mailing list