openstack/oslo.concurrency
Code Issues Proposed changes

Remove six usage

Browse Source 
Remove the usage of six, the python2-python3 compatibility lib.

Change-Id: I56e479d0749504d65be1cf2b5235d454b0b90a1a
This commit is contained in:
Andreas Jaeger 2020-05-09 16:34:27 +02:00
parent c5e0dfedf9
commit dd5a141771
5 changed files with 50 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
oslo_concurrency/lockutils.py
View File

@ -28,7 +28,6 @@ import fasteners
from oslo_config import cfg
from oslo_utils import reflection
from oslo_utils import timeutils
import six
from oslo_concurrency._i18n import _
@ -342,7 +341,7 @@ def synchronized(name, lock_file_prefix=None, external=False, lock_path=None,
def wrap(f):
@six.wraps(f)
@functools.wraps(f)
def inner(*args, **kwargs):
t1 = timeutils.now()
t2 = None

39
oslo_concurrency/processutils.py
View File

@ -32,7 +32,6 @@ from oslo_utils import encodeutils
from oslo_utils import importutils
from oslo_utils import strutils
from oslo_utils import timeutils
import six
from oslo_concurrency._i18n import _
@ -329,7 +328,7 @@ def execute(*cmd, **kwargs):
if kwargs:
raise UnknownArgumentError(_('Got unknown keyword args: %r') % kwargs)
if isinstance(log_errors, six.integer_types):
if isinstance(log_errors, int):
log_errors = LogErrors(log_errors)
if not isinstance(log_errors, LogErrors):
raise InvalidArgumentError(_('Got invalid arg log_errors: %r') %
@ -413,16 +412,15 @@ def execute(*cmd, **kwargs):
if not ignore_exit_code and _returncode not in check_exit_code:
(stdout, stderr) = result
if six.PY3:
stdout = os.fsdecode(stdout)
stderr = os.fsdecode(stderr)
stdout = os.fsdecode(stdout)
stderr = os.fsdecode(stderr)
sanitized_stdout = strutils.mask_password(stdout)
sanitized_stderr = strutils.mask_password(stderr)
raise ProcessExecutionError(exit_code=_returncode,
stdout=sanitized_stdout,
stderr=sanitized_stderr,
cmd=sanitized_cmd)
if six.PY3 and not binary and result is not None:
if not binary and result is not None:
(stdout, stderr) = result
# Decode from the locale using using the surrogateescape error
# handler (decoding cannot fail)
@ -491,7 +489,7 @@ def trycmd(*args, **kwargs):
out, err = execute(*args, **kwargs)
failed = False
except ProcessExecutionError as exn:
out, err = '', six.text_type(exn)
out, err = '', str(exn)
failed = True
if not failed and discard_warnings and err:
@ -543,12 +541,11 @@ def ssh_execute(ssh, cmd, process_input=None,
exit_status = channel.recv_exit_status()
if six.PY3:
# Decode from the locale using using the surrogateescape error handler
# (decoding cannot fail). Decode even if binary is True because
# mask_password() requires Unicode on Python 3
stdout = os.fsdecode(stdout)
stderr = os.fsdecode(stderr)
# Decode from the locale using using the surrogateescape error handler
# (decoding cannot fail). Decode even if binary is True because
# mask_password() requires Unicode on Python 3
stdout = os.fsdecode(stdout)
stderr = os.fsdecode(stderr)
if sanitize_stdout:
stdout = strutils.mask_password(stdout)
@ -570,19 +567,9 @@ def ssh_execute(ssh, cmd, process_input=None,
cmd=sanitized_cmd)
if binary:
if six.PY2:
# On Python 2, stdout is a bytes string if mask_password() failed
# to decode it, or an Unicode string otherwise. Encode to the
# default encoding (ASCII) because mask_password() decodes from
# the same encoding.
if isinstance(stdout, six.text_type):
stdout = stdout.encode()
if isinstance(stderr, six.text_type):
stderr = stderr.encode()
else:
# fsencode() is the reverse operation of fsdecode()
stdout = os.fsencode(stdout)
stderr = os.fsencode(stderr)
# fsencode() is the reverse operation of fsdecode()
stdout = os.fsencode(stdout)
stderr = os.fsencode(stderr)
return (stdout, stderr)

11
oslo_concurrency/tests/unit/test_lockutils.py
View File

@ -25,7 +25,6 @@ from unittest import mock
from oslo_config import cfg
from oslotest import base as test_base
import six
from oslo_concurrency.fixture import lockutils as fixtures
from oslo_concurrency import lockutils
@ -278,10 +277,7 @@ class LockTestCase(test_base.BaseTestCase):
# Note(flaper87): Lock is not external, which means
# a semaphore will be yielded
with lockutils.lock("test") as sem:
if six.PY2:
self.assertIsInstance(sem, threading._Semaphore)
else:
self.assertIsInstance(sem, threading.Semaphore)
self.assertIsInstance(sem, threading.Semaphore)
# NOTE(flaper87): Lock is external so an InterProcessLock
# will be yielded.
@ -295,10 +291,7 @@ class LockTestCase(test_base.BaseTestCase):
self.config(lock_path=tempfile.mkdtemp(), group='oslo_concurrency')
with lockutils.lock("test") as sem:
if six.PY2:
self.assertIsInstance(sem, threading._Semaphore)
else:
self.assertIsInstance(sem, threading.Semaphore)
self.assertIsInstance(sem, threading.Semaphore)
with lockutils.lock("test2", external=True) as lock:
self.assertTrue(lock.exists())

98
oslo_concurrency/tests/unit/test_processutils.py
View File

@ -16,6 +16,7 @@
from __future__ import print_function
import errno
import io
import logging
import multiprocessing
import os
@ -30,7 +31,6 @@ from unittest import mock
import fixtures
from oslotest import base as test_base
import six
from oslo_concurrency import processutils
@ -112,17 +112,11 @@ class UtilsTest(test_base.BaseTestCase):
processutils.execute(TRUE_UTILITY)
if six.PY2:
self.assertRaises(processutils.InvalidArgumentError,
processutils.execute,
TRUE_UTILITY,
preexec_fn=preexec_fn)
else:
try:
processutils.execute(TRUE_UTILITY, preexec_fn=preexec_fn)
except Exception as e:
if type(e).__name__ != 'SubprocessError':
raise
try:
processutils.execute(TRUE_UTILITY, preexec_fn=preexec_fn)
except Exception as e:
if type(e).__name__ != 'SubprocessError':
raise
@mock.patch.object(os, 'name', 'nt')
@mock.patch.object(processutils.subprocess, "Popen")
@ -168,23 +162,23 @@ class ProcessExecutionErrorTest(test_base.BaseTestCase):
def test_defaults(self):
err = processutils.ProcessExecutionError()
self.assertIn('None\n', six.text_type(err))
self.assertIn('code: -\n', six.text_type(err))
self.assertIn('None\n', str(err))
self.assertIn('code: -\n', str(err))
def test_with_description(self):
description = 'The Narwhal Bacons at Midnight'
err = processutils.ProcessExecutionError(description=description)
self.assertIn(description, six.text_type(err))
self.assertIn(description, str(err))
def test_with_exit_code(self):
exit_code = 0
err = processutils.ProcessExecutionError(exit_code=exit_code)
self.assertIn(str(exit_code), six.text_type(err))
self.assertIn(str(exit_code), str(err))
def test_with_cmd(self):
cmd = 'telinit'
err = processutils.ProcessExecutionError(cmd=cmd)
self.assertIn(cmd, six.text_type(err))
self.assertIn(cmd, str(err))
def test_with_stdout(self):
stdout = """
@ -207,13 +201,13 @@ class ProcessExecutionErrorTest(test_base.BaseTestCase):
the Wielder of Wonder, with world's renown.
""".strip()
err = processutils.ProcessExecutionError(stdout=stdout)
print(six.text_type(err))
self.assertIn('people-kings', six.text_type(err))
print(str(err))
self.assertIn('people-kings', str(err))
def test_with_stderr(self):
stderr = 'Cottonian library'
err = processutils.ProcessExecutionError(stderr=stderr)
self.assertIn(stderr, six.text_type(err))
self.assertIn(stderr, str(err))
def test_retry_on_failure(self):
fd, tmpfilename = tempfile.mkstemp()
@ -446,8 +440,8 @@ grep foo
'something')
self.assertEqual(38, err.exit_code)
self.assertIsInstance(err.stdout, six.text_type)
self.assertIsInstance(err.stderr, six.text_type)
self.assertIsInstance(err.stdout, str)
self.assertIsInstance(err.stderr, str)
self.assertIn('onstdout --password="***"', err.stdout)
self.assertIn('onstderr --password="***"', err.stderr)
self.assertEqual(' '.join([tmpfilename,
@ -458,20 +452,12 @@ grep foo
def execute_undecodable_bytes(self, out_bytes, err_bytes,
exitcode=0, binary=False):
if six.PY3:
code = ';'.join(('import sys',
'sys.stdout.buffer.write(%a)' % out_bytes,
'sys.stdout.flush()',
'sys.stderr.buffer.write(%a)' % err_bytes,
'sys.stderr.flush()',
'sys.exit(%s)' % exitcode))
else:
code = ';'.join(('import sys',
'sys.stdout.write(%r)' % out_bytes,
'sys.stdout.flush()',
'sys.stderr.write(%r)' % err_bytes,
'sys.stderr.flush()',
'sys.exit(%s)' % exitcode))
code = ';'.join(('import sys',
'sys.stdout.buffer.write(%a)' % out_bytes,
'sys.stdout.flush()',
'sys.stderr.buffer.write(%a)' % err_bytes,
'sys.stderr.flush()',
'sys.exit(%s)' % exitcode))
return processutils.execute(sys.executable, '-c', code, binary=binary)
@ -480,7 +466,7 @@ grep foo
err_bytes = b'err: ' + UNDECODABLE_BYTES
out, err = self.execute_undecodable_bytes(out_bytes, err_bytes,
binary=binary)
if six.PY3 and not binary:
if not binary:
self.assertEqual(os.fsdecode(out_bytes), out)
self.assertEqual(os.fsdecode(err_bytes), err)
else:
@ -505,16 +491,8 @@ grep foo
err = exc.stderr
out_bytes = b'out: password="***" ' + UNDECODABLE_BYTES
err_bytes = b'err: password="***" ' + UNDECODABLE_BYTES
if six.PY3:
# On Python 3, stdout and stderr attributes of
# ProcessExecutionError must always be Unicode
self.assertEqual(os.fsdecode(out_bytes), out)
self.assertEqual(os.fsdecode(err_bytes), err)
else:
# On Python 2, stdout and stderr attributes of
# ProcessExecutionError must always be bytes
self.assertEqual(out_bytes, out)
self.assertEqual(err_bytes, err)
self.assertEqual(os.fsdecode(out_bytes), out)
self.assertEqual(os.fsdecode(err_bytes), err)
def test_undecodable_bytes_error(self):
self.check_undecodable_bytes_error(False)
@ -642,7 +620,7 @@ class FakeSshChannel(object):
return self.rc
class FakeSshStream(six.BytesIO):
class FakeSshStream(io.BytesIO):
def setup_channel(self, rc):
self.channel = FakeSshChannel(rc)
@ -658,9 +636,9 @@ class FakeSshConnection(object):
raise socket.timeout()
stdout = FakeSshStream(self.out)
stdout.setup_channel(self.rc)
return (six.BytesIO(),
return (io.BytesIO(),
stdout,
six.BytesIO(self.err))
io.BytesIO(self.err))
class SshExecuteTestCase(test_base.BaseTestCase):
@ -684,8 +662,8 @@ class SshExecuteTestCase(test_base.BaseTestCase):
out, err = processutils.ssh_execute(FakeSshConnection(0), 'ls')
self.assertEqual('stdout', out)
self.assertEqual('stderr', err)
self.assertIsInstance(out, six.text_type)
self.assertIsInstance(err, six.text_type)
self.assertIsInstance(out, str)
self.assertIsInstance(err, str)
def test_binary(self):
o, e = processutils.ssh_execute(FakeSshConnection(0), 'ls',
@ -701,7 +679,7 @@ class SshExecuteTestCase(test_base.BaseTestCase):
conn = FakeSshConnection(0, out=out_bytes, err=err_bytes)
out, err = processutils.ssh_execute(conn, 'ls', binary=binary)
if six.PY3 and not binary:
if not binary:
self.assertEqual(os.fsdecode(out_bytes), out)
self.assertEqual(os.fsdecode(err_bytes), err)
else:
@ -729,16 +707,8 @@ class SshExecuteTestCase(test_base.BaseTestCase):
out = exc.stdout
err = exc.stderr
if six.PY3:
# On Python 3, stdout and stderr attributes of
# ProcessExecutionError must always be Unicode
self.assertEqual(os.fsdecode(out_bytes), out)
self.assertEqual(os.fsdecode(err_bytes), err)
else:
# On Python 2, stdout and stderr attributes of
# ProcessExecutionError must always be bytes
self.assertEqual(out_bytes, out)
self.assertEqual(err_bytes, err)
self.assertEqual(os.fsdecode(out_bytes), out)
self.assertEqual(os.fsdecode(err_bytes), err)
def test_undecodable_bytes_error(self):
self.check_undecodable_bytes_error(False)
@ -752,7 +722,7 @@ class SshExecuteTestCase(test_base.BaseTestCase):
def _test_compromising_ssh(self, rc, check):
fixture = self.useFixture(fixtures.FakeLogger(level=logging.DEBUG))
fake_stdin = six.BytesIO()
fake_stdin = io.BytesIO()
fake_stdout = mock.Mock()
fake_stdout.channel.recv_exit_status.return_value = rc

1
requirements.txt
View File

@ -6,5 +6,4 @@ pbr!=2.1.0,>=2.0.0 # Apache-2.0
oslo.config>=5.2.0 # Apache-2.0
oslo.i18n>=3.15.3 # Apache-2.0
oslo.utils>=3.33.0 # Apache-2.0
six>=1.10.0 # MIT
fasteners>=0.7.0 # Apache-2.0