Upgrade pylint (#8855)

This is part of https://github.com/certbot/certbot/issues/8782. I took it on now because the currently pinned version of `pylint` doesn't work with newer versions of `poetry` which I wanted to upgrade as part of https://github.com/certbot/certbot/issues/8787.

To say a bit more about the specific changes in this PR:

* Newer versions of `pylint` complain if `Popen` isn't used as a context manager. Instead of making this change, I switched to using `subprocess.run` which is simpler and [recommended in the Python docs](https://docs.python.org/3/library/subprocess.html#using-the-subprocess-module). I also disabled this check in a few places where no longer using `Popen` would require significant refactoring.
* The deleted code in `certbot/certbot/_internal/renewal.py` is cruft since https://github.com/certbot/certbot/pull/8685.
* The unused argument to `enable_mod` in the Apache plugin is used in some over the override classes that subclass that class.

* unpin pylint and repin dependencies

* disable raise-missing-from

* disable wrong-input-order

* remove unused code

* misc lint fixes

* remove unused import

* various lint fixes
This commit is contained in:
Brad Warren
2021-05-24 10:02:55 -07:00
committed by GitHub
parent 2df279bc5b
commit 315ddb247f
26 changed files with 223 additions and 225 deletions
+1 -1
View File
@@ -1361,7 +1361,7 @@ def make_displayer(config: configuration.NamespaceConfig
if config.quiet:
config.noninteractive_mode = True
devnull = open(os.devnull, "w")
devnull = open(os.devnull, "w") # pylint: disable=consider-using-with
displayer = display_util.NoninteractiveDisplay(devnull)
elif config.noninteractive_mode:
displayer = display_util.NoninteractiveDisplay(sys.stdout)
+2 -2
View File
@@ -134,7 +134,7 @@ to serve all files under specified web root ({0})."""
"webroot when using the webroot plugin.")
return None if index == 0 else known_webroots[index - 1] # code == display_util.OK
def _prompt_for_new_webroot(self, domain, allowraise=False): # pylint: no-self-use
def _prompt_for_new_webroot(self, domain, allowraise=False):
code, webroot = ops.validated_directory(
_validate_webroot,
"Input the webroot for {0}:".format(domain),
@@ -192,7 +192,7 @@ to serve all files under specified web root ({0})."""
finally:
filesystem.umask(old_umask)
def _get_validation_path(self, root_path, achall): # pylint: no-self-use
def _get_validation_path(self, root_path, achall):
return os.path.join(root_path, achall.chall.encode("token"))
def _perform_single(self, achall):
-8
View File
@@ -14,7 +14,6 @@ from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import load_pem_private_key
import OpenSSL
import zope.component
from certbot import crypto_util
@@ -306,13 +305,6 @@ def should_renew(config, lineage):
def _avoid_invalidating_lineage(config, lineage, original_server):
"""Do not renew a valid cert with one from a staging server!"""
# Some lineages may have begun with --staging, but then had production
# certificates added to them
with open(lineage.cert) as the_file:
contents = the_file.read()
latest_cert = OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, contents)
if util.is_staging(config.server):
if not util.is_staging(original_server):
if not config.break_my_certs:
+13 -12
View File
@@ -115,35 +115,36 @@ def underscores_for_unsupported_characters_in_path(path: str) -> str:
def execute_command(cmd_name: str, shell_cmd: str, env: Optional[dict] = None) -> Tuple[str, str]:
"""
Run a command:
- on Linux command will be run by the standard shell selected with Popen(shell=True)
- on Linux command will be run by the standard shell selected with
subprocess.run(shell=True)
- on Windows command will be run in a Powershell shell
:param str cmd_name: the user facing name of the hook being run
:param str shell_cmd: shell command to execute
:param dict env: environ to pass into Popen
:param dict env: environ to pass into subprocess.run
:returns: `tuple` (`str` stderr, `str` stdout)
"""
logger.info("Running %s command: %s", cmd_name, shell_cmd)
if POSIX_MODE:
cmd = subprocess.Popen(shell_cmd, shell=True, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, universal_newlines=True,
env=env)
proc = subprocess.run(shell_cmd, shell=True, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, universal_newlines=True,
check=False, env=env)
else:
line = ['powershell.exe', '-Command', shell_cmd]
cmd = subprocess.Popen(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True, env=env)
proc = subprocess.run(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True, check=False, env=env)
# universal_newlines causes Popen.communicate()
# to return str objects instead of bytes in Python 3
out, err = cmd.communicate()
# universal_newlines causes stdout and stderr to be str objects instead of
# bytes in Python 3
out, err = proc.stdout, proc.stderr
base_cmd = os.path.basename(shell_cmd.split(None, 1)[0])
if out:
logger.info('Output from %s command %s:\n%s', cmd_name, base_cmd, out)
if cmd.returncode != 0:
if proc.returncode != 0:
logger.error('%s command "%s" returned error code %d',
cmd_name, shell_cmd, cmd.returncode)
cmd_name, shell_cmd, proc.returncode)
if err:
logger.error('Error output from %s command %s:\n%s', cmd_name, base_cmd, err)
return err, out
+4 -5
View File
@@ -3,8 +3,8 @@ from datetime import datetime
from datetime import timedelta
import logging
import re
import subprocess
from subprocess import PIPE
from subprocess import Popen
from typing import Optional
from typing import Tuple
@@ -50,11 +50,10 @@ class RevocationChecker:
return
# New versions of openssl want -header var=val, old ones want -header var val
test_host_format = Popen(["openssl", "ocsp", "-header", "var", "val"],
test_host_format = subprocess.run(["openssl", "ocsp", "-header", "var", "val"],
stdout=PIPE, stderr=PIPE, universal_newlines=True,
env=util.env_no_snap_for_external_calls())
_out, err = test_host_format.communicate()
if "Missing =" in err:
check=False, env=util.env_no_snap_for_external_calls())
if "Missing =" in test_host_format.stderr:
self.host_args = lambda host: ["Host=" + host]
else:
self.host_args = lambda host: ["Host", host]
+5 -12
View File
@@ -198,6 +198,7 @@ class Reverter:
Read the file returning the lines, and a pointer to the end of the file.
"""
# pylint: disable=consider-using-with
# Open up filepath differently depending on if it already exists
if os.path.isfile(filepath):
op_fd = open(filepath, "r+")
@@ -348,26 +349,18 @@ class Reverter:
"""
commands_fp = os.path.join(self._get_cp_dir(temporary), "COMMANDS")
command_file = None
# It is strongly advised to set newline = '' on Python 3 with CSV,
# and it fixes problems on Windows.
kwargs = {'newline': ''}
try:
if os.path.isfile(commands_fp):
command_file = open(commands_fp, "a", **kwargs) # type: ignore
else:
command_file = open(commands_fp, "w", **kwargs) # type: ignore
csvwriter = csv.writer(command_file)
csvwriter.writerow(command)
mode = "a" if os.path.isfile(commands_fp) else "w"
with open(commands_fp, mode, **kwargs) as f: # type: ignore
csvwriter = csv.writer(f)
csvwriter.writerow(command)
except (IOError, OSError):
logger.error("Unable to register undo command")
raise errors.ReverterError(
"Unable to register undo command.")
finally:
if command_file is not None:
command_file.close()
def _get_cp_dir(self, temporary):
"""Return the proper reverter directory."""
+2 -2
View File
@@ -259,7 +259,7 @@ class FreezableMock:
def _create_get_utility_mock():
display = FreezableMock()
# Use pylint code for disable to keep on single line under line length limit
for name in interfaces.IDisplay.names(): # pylint: E1120
for name in interfaces.IDisplay.names():
if name != 'notification':
frozen_mock = FreezableMock(frozen=True, func=_assert_valid_call)
setattr(display, name, frozen_mock)
@@ -284,7 +284,7 @@ def _create_get_utility_mock_with_stdout(stdout):
display = FreezableMock()
# Use pylint code for disable to keep on single line under line length limit
for name in interfaces.IDisplay.names(): # pylint: E1120
for name in interfaces.IDisplay.names():
if name == 'notification':
frozen_mock = FreezableMock(frozen=True,
func=_write_msg)
+14 -15
View File
@@ -89,32 +89,31 @@ def env_no_snap_for_external_calls():
def run_script(params, log=logger.error):
"""Run the script with the given params.
:param list params: List of parameters to pass to Popen
:param list params: List of parameters to pass to subprocess.run
:param callable log: Logger method to use for errors
"""
try:
proc = subprocess.Popen(params,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
env=env_no_snap_for_external_calls())
proc = subprocess.run(params,
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
env=env_no_snap_for_external_calls())
except (OSError, ValueError):
msg = "Unable to run the command: %s" % " ".join(params)
log(msg)
raise errors.SubprocessError(msg)
stdout, stderr = proc.communicate()
if proc.returncode != 0:
msg = "Error while running %s.\n%s\n%s" % (
" ".join(params), stdout, stderr)
" ".join(params), proc.stdout, proc.stderr)
# Enter recovery routine...
log(msg)
raise errors.SubprocessError(msg)
return stdout, stderr
return proc.stdout, proc.stderr
def exe_exists(exe):
@@ -399,20 +398,20 @@ def get_python_os_info(pretty=False):
os_ver = info[1]
elif os_type.startswith('darwin'):
try:
proc = subprocess.Popen(
proc = subprocess.run(
["/usr/bin/sw_vers", "-productVersion"],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True,
check=False, universal_newlines=True,
env=env_no_snap_for_external_calls(),
)
except OSError:
proc = subprocess.Popen(
proc = subprocess.run(
["sw_vers", "-productVersion"],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True,
check=False, universal_newlines=True,
env=env_no_snap_for_external_calls(),
)
os_ver = proc.communicate()[0].rstrip('\n')
os_ver = proc.stdout.rstrip('\n')
elif os_type.startswith('freebsd'):
# eg "9.3-RC3-p1"
os_ver = os_ver.partition("-")[0]
+6 -5
View File
@@ -25,14 +25,15 @@ class ExecuteTest(unittest.TestCase):
def _test_common(self, returncode, stdout, stderr):
given_command = "foo"
given_name = "foo-hook"
with mock.patch("certbot.compat.misc.subprocess.Popen") as mock_popen:
mock_popen.return_value.communicate.return_value = (stdout, stderr)
mock_popen.return_value.returncode = returncode
with mock.patch("certbot.compat.misc.subprocess.run") as mock_run:
mock_run.return_value.stdout = stdout
mock_run.return_value.stderr = stderr
mock_run.return_value.returncode = returncode
with mock.patch("certbot.compat.misc.logger") as mock_logger:
self.assertEqual(self._call(given_name, given_command), (stderr, stdout))
executed_command = mock_popen.call_args[1].get(
"args", mock_popen.call_args[0][0])
executed_command = mock_run.call_args[1].get(
"args", mock_run.call_args[0][0])
if os.name == 'nt':
expected_command = ['powershell.exe', '-Command', given_command]
else:
+19 -23
View File
@@ -1078,29 +1078,25 @@ class MainTest(test_util.ConfigTestCase):
with test_util.patch_get_utility() as mock_get_utility:
if not quiet_mode:
mock_get_utility().notification.side_effect = write_msg
with mock.patch('certbot._internal.main.renewal.OpenSSL') as mock_ssl:
mock_latest = mock.MagicMock()
mock_latest.get_issuer.return_value = "Artificial pretend"
mock_ssl.crypto.load_certificate.return_value = mock_latest
with mock.patch('certbot._internal.main.renewal.crypto_util') \
as mock_crypto_util:
mock_crypto_util.notAfter.return_value = expiry_date
with mock.patch('certbot._internal.eff.handle_subscription'):
if not args:
args = ['-d', 'isnot.org', '-a', 'standalone', 'certonly']
if extra_args:
args += extra_args
try:
ret, stdout, _, _ = self._call(args, stdout)
if ret:
print("Returned", ret)
raise AssertionError(ret)
assert not error_expected, "renewal should have errored"
except: # pylint: disable=bare-except
if not error_expected:
raise AssertionError(
"Unexpected renewal error:\n" +
traceback.format_exc())
with mock.patch('certbot._internal.main.renewal.crypto_util') \
as mock_crypto_util:
mock_crypto_util.notAfter.return_value = expiry_date
with mock.patch('certbot._internal.eff.handle_subscription'):
if not args:
args = ['-d', 'isnot.org', '-a', 'standalone', 'certonly']
if extra_args:
args += extra_args
try:
ret, stdout, _, _ = self._call(args, stdout)
if ret:
print("Returned", ret)
raise AssertionError(ret)
assert not error_expected, "renewal should have errored"
except: # pylint: disable=bare-except
if not error_expected:
raise AssertionError(
"Unexpected renewal error:\n" +
traceback.format_exc())
if should_renew:
if reuse_key:
+9 -13
View File
@@ -40,11 +40,9 @@ class OCSPTestOpenSSL(unittest.TestCase):
def setUp(self):
from certbot import ocsp
with mock.patch('certbot.ocsp.Popen') as mock_popen:
with mock.patch('certbot.ocsp.subprocess.run') as mock_run:
with mock.patch('certbot.util.exe_exists') as mock_exists:
mock_communicate = mock.MagicMock()
mock_communicate.communicate.return_value = (None, out)
mock_popen.return_value = mock_communicate
mock_run.stderr = out
mock_exists.return_value = True
self.checker = ocsp.RevocationChecker(enforce_openssl_binary_usage=True)
@@ -52,28 +50,26 @@ class OCSPTestOpenSSL(unittest.TestCase):
pass
@mock.patch('certbot.ocsp.logger.info')
@mock.patch('certbot.ocsp.Popen')
@mock.patch('certbot.ocsp.subprocess.run')
@mock.patch('certbot.util.exe_exists')
def test_init(self, mock_exists, mock_popen, mock_log):
mock_communicate = mock.MagicMock()
mock_communicate.communicate.return_value = (None, out)
mock_popen.return_value = mock_communicate
def test_init(self, mock_exists, mock_run, mock_log):
mock_run.return_value.stderr = out
mock_exists.return_value = True
from certbot import ocsp
checker = ocsp.RevocationChecker(enforce_openssl_binary_usage=True)
self.assertEqual(mock_popen.call_count, 1)
self.assertEqual(mock_run.call_count, 1)
self.assertEqual(checker.host_args("x"), ["Host=x"])
mock_communicate.communicate.return_value = (None, out.partition("\n")[2])
mock_run.return_value.stderr = out.partition("\n")[2]
checker = ocsp.RevocationChecker(enforce_openssl_binary_usage=True)
self.assertEqual(checker.host_args("x"), ["Host", "x"])
self.assertIs(checker.broken, False)
mock_exists.return_value = False
mock_popen.call_count = 0
mock_run.call_count = 0
checker = ocsp.RevocationChecker(enforce_openssl_binary_usage=True)
self.assertEqual(mock_popen.call_count, 0)
self.assertEqual(mock_run.call_count, 0)
self.assertEqual(mock_log.call_count, 1)
self.assertIs(checker.broken, True)
+13 -17
View File
@@ -59,26 +59,26 @@ class RunScriptTest(unittest.TestCase):
from certbot.util import run_script
return run_script(params)
@mock.patch("certbot.util.subprocess.Popen")
def test_default(self, mock_popen):
@mock.patch("certbot.util.subprocess.run")
def test_default(self, mock_run):
"""These will be changed soon enough with reload."""
mock_popen().returncode = 0
mock_popen().communicate.return_value = ("stdout", "stderr")
mock_run().returncode = 0
mock_run().stdout = "stdout"
mock_run().stderr = "stderr"
out, err = self._call(["test"])
self.assertEqual(out, "stdout")
self.assertEqual(err, "stderr")
@mock.patch("certbot.util.subprocess.Popen")
def test_bad_process(self, mock_popen):
mock_popen.side_effect = OSError
@mock.patch("certbot.util.subprocess.run")
def test_bad_process(self, mock_run):
mock_run.side_effect = OSError
self.assertRaises(errors.SubprocessError, self._call, ["test"])
@mock.patch("certbot.util.subprocess.Popen")
def test_failure(self, mock_popen):
mock_popen().communicate.return_value = ("", "")
mock_popen().returncode = 1
@mock.patch("certbot.util.subprocess.run")
def test_failure(self, mock_run):
mock_run().returncode = 1
self.assertRaises(errors.SubprocessError, self._call, ["test"])
@@ -557,12 +557,8 @@ class OsInfoTest(unittest.TestCase):
with mock.patch('platform.system_alias',
return_value=('darwin', '', '')):
with mock.patch("subprocess.Popen") as popen_mock:
comm_mock = mock.Mock()
comm_attrs = {'communicate.return_value':
('42.42.42', 'error')}
comm_mock.configure_mock(**comm_attrs)
popen_mock.return_value = comm_mock
with mock.patch("subprocess.run") as run_mock:
run_mock().stdout = '42.42.42'
self.assertEqual(cbutil.get_python_os_info()[0], 'darwin')
self.assertEqual(cbutil.get_python_os_info()[1], '42.42.42')