Remove dependency on six (#8650)

Fixes https://github.com/certbot/certbot/issues/8494.

I left the `six` dependency pinned in `tests/letstest/requirements.txt` and `tools/oldest_constraints.txt` because `six` is still a transitive dependency with our current pinnings.

The extra moving around of imports is due to me using `isort` to help me keep dependencies in sorted order after replacing imports of `six`.

* remove some six usage in acme

* remove six from acme

* remove six.add_metaclass usage

* fix six.moves.zip

* fix six.moves.builtins.open

* six.moves server fixes

* 's/six\.moves\.range/range/g'

* stop using six.moves.xrange

* fix urllib imports

* s/six\.binary_type/bytes/g

* s/six\.string_types/str/g

* 's/six\.text_type/str/g'

* fix six.iteritems usage

* fix itervalues usage

* switch from six.StringIO to io.StringIO

* remove six imports

* misc fixes

* stop using six.reload_module

* no six.PY2

* rip out six

* keep six pinned in oldest constraints

* fix log_test.py

* update changelog
This commit is contained in:
Brad Warren
2021-02-09 11:43:15 -08:00
committed by GitHub
parent edad9bd82b
commit 3d0dad8718
65 changed files with 196 additions and 255 deletions
+1 -1
View File
@@ -254,7 +254,7 @@ ignore-mixin-members=yes
# List of module names for which member attributes should not be checked
# (useful for modules/projects where namespaces are manipulated during runtime
# and thus existing member attributes cannot be deduced by static analysis
ignored-modules=pkg_resources,confargparse,argparse,six.moves,six.moves.urllib
ignored-modules=pkg_resources,confargparse,argparse
# import errors ignored only in 1.4.4
# https://bitbucket.org/logilab/pylint/commits/cd000904c9e2
+1 -3
View File
@@ -9,7 +9,6 @@ import socket
from cryptography.hazmat.primitives import hashes # type: ignore
import josepy as jose
import requests
import six
from OpenSSL import SSL # type: ignore # https://github.com/python/typeshed/issues/2052
from OpenSSL import crypto
@@ -145,8 +144,7 @@ class KeyAuthorizationChallengeResponse(ChallengeResponse):
return jobj
@six.add_metaclass(abc.ABCMeta)
class KeyAuthorizationChallenge(_TokenChallenge):
class KeyAuthorizationChallenge(_TokenChallenge, metaclass=abc.ABCMeta):
"""Challenge based on Key Authorization.
:param response_cls: Subclass of `KeyAuthorizationChallengeResponse`
+3 -4
View File
@@ -4,6 +4,7 @@ import collections
import datetime
from email.utils import parsedate_tz
import heapq
import http.client as http_client
import logging
import re
import time
@@ -14,8 +15,6 @@ import requests
from requests.adapters import HTTPAdapter
from requests.utils import parse_header_links
from requests_toolbelt.adapters.source import SourceAddressAdapter
import six
from six.moves import http_client
from acme import crypto_util
from acme import errors
@@ -248,7 +247,7 @@ class Client(ClientBase):
if net is None:
net = ClientNetwork(key, alg=alg, verify_ssl=verify_ssl)
if isinstance(directory, six.string_types):
if isinstance(directory, str):
directory = messages.Directory.from_json(
net.get(directory).json())
super(Client, self).__init__(directory=directory,
@@ -463,7 +462,7 @@ class Client(ClientBase):
exhausted.add(authzr)
if exhausted or any(authzr.body.status == messages.STATUS_INVALID
for authzr in six.itervalues(updated)):
for authzr in updated.values()):
raise errors.PollError(exhausted, updated)
updated_authzrs = tuple(updated[authzr] for authzr in authzrs)
-2
View File
@@ -2,7 +2,6 @@
import json
import josepy as jose
import six
from acme import challenges
from acme import errors
@@ -68,7 +67,6 @@ def is_acme_error(err):
return False
@six.python_2_unicode_compatible
class Error(jose.JSONObjectWithFields, errors.Error):
"""ACME error.
+3 -4
View File
@@ -1,14 +1,13 @@
"""Support for standalone client challenge solvers. """
import collections
import functools
import http.client as http_client
import http.server as BaseHTTPServer
import logging
import socket
import socketserver
import threading
from six.moves import BaseHTTPServer # type: ignore
from six.moves import http_client
from six.moves import socketserver # type: ignore
from acme import challenges
from acme import crypto_util
from acme.magic_typing import List
+1 -2
View File
@@ -1,7 +1,6 @@
"""ACME utilities."""
import six
def map_keys(dikt, func):
"""Map dictionary keys."""
return {func(key): value for key, value in six.iteritems(dikt)}
return {func(key): value for key, value in dikt.items()}
-1
View File
@@ -18,7 +18,6 @@ install_requires = [
'requests>=2.6.0',
'requests-toolbelt>=0.3.0',
'setuptools>=39.0.1',
'six>=1.11.0',
]
dev_extras = [
+1 -1
View File
@@ -1,11 +1,11 @@
"""Tests for acme.challenges."""
import urllib.parse as urllib_parse
import unittest
from unittest import mock
import josepy as jose
import OpenSSL
import requests
from six.moves.urllib import parse as urllib_parse
from acme import errors
+1 -1
View File
@@ -2,6 +2,7 @@
# pylint: disable=too-many-lines
import copy
import datetime
import http.client as http_client
import json
import unittest
from unittest import mock
@@ -9,7 +10,6 @@ from unittest import mock
import josepy as jose
import OpenSSL
import requests
from six.moves import http_client # pylint: disable=import-error
from acme import challenges
from acme import errors
+4 -8
View File
@@ -1,14 +1,13 @@
"""Tests for acme.crypto_util."""
import itertools
import socket
import socketserver
import threading
import time
import unittest
import josepy as jose
import OpenSSL
import six
from six.moves import socketserver # type: ignore # pylint: disable=import-error
from acme import errors
import test_util
@@ -27,8 +26,6 @@ class SSLSocketAndProbeSNITest(unittest.TestCase):
class _TestServer(socketserver.TCPServer):
# six.moves.* | pylint: disable=attribute-defined-outside-init,no-init
def server_bind(self): # pylint: disable=missing-docstring
self.socket = SSLSocket(socket.socket(),
certs)
@@ -62,7 +59,6 @@ class SSLSocketAndProbeSNITest(unittest.TestCase):
self.assertRaises(errors.Error, self._probe, b'bar')
def test_probe_connection_error(self):
# pylint has a hard time with six
self.server.server_close()
original_timeout = socket.getdefaulttimeout()
try:
@@ -121,9 +117,9 @@ class PyOpenSSLCertOrReqSANTest(unittest.TestCase):
@classmethod
def _get_idn_names(cls):
"""Returns expected names from '{cert,csr}-idnsans.pem'."""
chars = [six.unichr(i) for i in itertools.chain(range(0x3c3, 0x400),
range(0x641, 0x6fc),
range(0x1820, 0x1877))]
chars = [chr(i) for i in itertools.chain(range(0x3c3, 0x400),
range(0x641, 0x6fc),
range(0x1820, 0x1877))]
return [''.join(chars[i: i + 45]) + '.invalid'
for i in range(0, len(chars), 45)]
+2 -2
View File
@@ -1,13 +1,13 @@
"""Tests for acme.standalone."""
import http.client as http_client
import socket
import socketserver
import threading
import unittest
from unittest import mock
import josepy as jose
import requests
from six.moves import http_client # pylint: disable=import-error
from six.moves import socketserver # type: ignore # pylint: disable=import-error
from acme import challenges
from acme import crypto_util
@@ -100,12 +100,10 @@ For this reason the internal representation of data should not ignore the case.
"""
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class ParserNode(object):
class ParserNode(object, metaclass=abc.ABCMeta):
"""
ParserNode is the basic building block of the tree of such nodes,
representing the structure of the configuration. It is largely meant to keep
@@ -204,9 +202,7 @@ class ParserNode(object):
"""
# Linter rule exclusion done because of https://github.com/PyCQA/pylint/issues/179
@six.add_metaclass(abc.ABCMeta) # pylint: disable=abstract-method
class CommentNode(ParserNode):
class CommentNode(ParserNode, metaclass=abc.ABCMeta):
"""
CommentNode class is used for representation of comments within the parsed
configuration structure. Because of the nature of comments, it is not able
@@ -249,8 +245,7 @@ class CommentNode(ParserNode):
metadata=kwargs.get('metadata', {})) # pragma: no cover
@six.add_metaclass(abc.ABCMeta)
class DirectiveNode(ParserNode):
class DirectiveNode(ParserNode, metaclass=abc.ABCMeta):
"""
DirectiveNode class represents a configuration directive within the configuration.
It can have zero or more parameters attached to it. Because of the nature of
@@ -325,8 +320,7 @@ class DirectiveNode(ParserNode):
"""
@six.add_metaclass(abc.ABCMeta)
class BlockNode(DirectiveNode):
class BlockNode(DirectiveNode, metaclass=abc.ABCMeta):
"""
BlockNode class represents a block of nested configuration directives, comments
and other blocks as its children. A BlockNode can have zero or more parameters
@@ -5,7 +5,6 @@ import logging
import re
import sys
import six
from acme.magic_typing import Dict
from acme.magic_typing import List
@@ -275,7 +274,7 @@ class ApacheParser(object):
while len(mods) != prev_size:
prev_size = len(mods)
for match_name, match_filename in six.moves.zip(
for match_name, match_filename in zip(
iterator, iterator):
mod_name = self.get_arg(match_name)
mod_filename = self.get_arg(match_filename)
-1
View File
@@ -7,7 +7,6 @@ try:
import mock
except ImportError: # pragma: no cover
from unittest import mock # type: ignore
import six # pylint: disable=unused-import # six is used in mock.patch()
from certbot import errors
from certbot_apache._internal import constants
+2 -3
View File
@@ -10,7 +10,6 @@ try:
import mock
except ImportError: # pragma: no cover
from unittest import mock # type: ignore
import six # pylint: disable=unused-import # six is used in mock.patch()
from acme import challenges
from certbot import achallenges
@@ -726,7 +725,7 @@ class MultipleVhostsTest(util.ApacheTest):
# This calls open
self.config.reverter.register_file_creation = mock.Mock()
mock_open.side_effect = IOError
with mock.patch("six.moves.builtins.open", mock_open):
with mock.patch("builtins.open", mock_open):
self.assertRaises(
errors.PluginError,
self.config.make_vhost_ssl, self.vh_truth[0])
@@ -1834,7 +1833,7 @@ class InstallSslOptionsConfTest(util.ApacheTest):
def test_open_module_file(self):
mock_open = mock.mock_open(read_data="testing 12 3")
with mock.patch("six.moves.builtins.open", mock_open):
with mock.patch("builtins.open", mock_open):
self.assertEqual(self.config._open_module_file("/nonsense/"), "testing 12 3")
if __name__ == "__main__":
@@ -4,10 +4,12 @@ or outside during setup/teardown of the integration tests environment.
"""
import contextlib
import errno
import http.server as SimpleHTTPServer
import multiprocessing
import os
import re
import shutil
import socketserver
import stat
import sys
import tempfile
@@ -23,8 +25,6 @@ from cryptography.x509 import load_pem_x509_certificate
from OpenSSL import crypto
import pkg_resources
import requests
from six.moves import SimpleHTTPServer
from six.moves import socketserver
from certbot_integration_tests.utils.constants import \
PEBBLE_ALTERNATE_ROOTS, PEBBLE_MANAGEMENT_URL
@@ -4,6 +4,7 @@ This runnable module interfaces itself with the Pebble management interface in o
to serve a mock OCSP responder during integration tests against Pebble.
"""
import datetime
import http.server as BaseHTTPServer
import re
from cryptography import x509
@@ -13,7 +14,6 @@ from cryptography.hazmat.primitives import serialization
from cryptography.x509 import ocsp
from dateutil import parser
import requests
from six.moves import BaseHTTPServer
from certbot_integration_tests.utils.constants import MOCK_OCSP_SERVER_PORT
from certbot_integration_tests.utils.constants import PEBBLE_MANAGEMENT_URL
@@ -1,12 +1,12 @@
#!/usr/bin/env python
# pylint: disable=missing-module-docstring
import http.server as BaseHTTPServer
import json
import re
import sys
import requests
from six.moves import BaseHTTPServer
from certbot_integration_tests.utils.misc import GracefulTCPServer
-1
View File
@@ -18,7 +18,6 @@ install_requires = [
'python-dateutil',
'pyyaml',
'requests',
'six'
]
# Add pywin32 on Windows platforms to handle low-level system calls.
@@ -3,8 +3,6 @@ import logging
import socket
import requests
import six
from six.moves import xrange
from acme import crypto_util
from acme import errors as acme_errors
@@ -19,11 +17,11 @@ class Validator(object):
"""Verifies the certificate presented at name is cert"""
if alt_host is None:
host = socket.gethostbyname(name).encode()
elif isinstance(alt_host, six.binary_type):
elif isinstance(alt_host, bytes):
host = alt_host
else:
host = alt_host.encode()
name = name if isinstance(name, six.binary_type) else name.encode()
name = name if isinstance(name, bytes) else name.encode()
try:
presented_cert = crypto_util.probe_sni(name, host, port)
@@ -62,7 +60,7 @@ class Validator(object):
else:
response = requests.get(url, allow_redirects=False)
return response.status_code in xrange(300, 309)
return response.status_code in range(300, 309)
def hsts(self, name):
"""Test for HTTP Strict Transport Security header"""
-1
View File
@@ -8,7 +8,6 @@ version = '1.13.0.dev0'
install_requires = [
'certbot',
'certbot-apache',
'six',
'requests',
'zope.interface',
]
-1
View File
@@ -11,7 +11,6 @@ version = '1.13.0.dev0'
install_requires = [
'python-digitalocean>=1.11',
'setuptools>=39.0.1',
'six>=1.11.0',
'zope.interface',
]
@@ -15,7 +15,6 @@ from pyparsing import restOfLine
from pyparsing import stringEnd
from pyparsing import White
from pyparsing import ZeroOrMore
import six
from acme.magic_typing import IO, Any # pylint: disable=unused-import
logger = logging.getLogger(__name__)
@@ -79,7 +78,7 @@ class RawNginxDumper(object):
"""Iterates the dumped nginx content."""
blocks = blocks or self.blocks
for b0 in blocks:
if isinstance(b0, six.string_types):
if isinstance(b0, str):
yield b0
continue
item = copy.deepcopy(b0)
@@ -96,7 +95,7 @@ class RawNginxDumper(object):
yield '}'
else: # not a block - list of strings
semicolon = ";"
if isinstance(item[0], six.string_types) and item[0].strip() == '#': # comment
if isinstance(item[0], str) and item[0].strip() == '#': # comment
semicolon = ""
yield "".join(item) + semicolon
@@ -131,14 +130,14 @@ def load(_file):
def dumps(blocks):
# type: (UnspacedList) -> six.text_type
# type: (UnspacedList) -> str
"""Dump to a Unicode string.
:param UnspacedList block: The parsed tree
:rtype: six.text_type
:rtype: str
"""
return six.text_type(RawNginxDumper(blocks.spaced))
return str(RawNginxDumper(blocks.spaced))
def dump(blocks, _file):
@@ -154,7 +153,7 @@ def dump(blocks, _file):
_file.write(dumps(blocks))
spacey = lambda x: (isinstance(x, six.string_types) and x.isspace()) or x == ''
spacey = lambda x: (isinstance(x, str) and x.isspace()) or x == ''
class UnspacedList(list):
"""Wrap a list [of lists], making any whitespace entries magically invisible"""
+2 -3
View File
@@ -1,7 +1,6 @@
"""Module contains classes used by the Nginx Configurator."""
import re
import six
from certbot.plugins import common
@@ -211,7 +210,7 @@ class VirtualHost(object):
def contains_list(self, test):
"""Determine if raw server block contains test list at top level
"""
for i in six.moves.range(0, len(self.raw) - len(test) + 1):
for i in range(0, len(self.raw) - len(test) + 1):
if self.raw[i:i + len(test)] == test:
return True
return False
@@ -250,7 +249,7 @@ def _find_directive(directives, directive_name, match_content=None):
"""Find a directive of type directive_name in directives. If match_content is given,
Searches for `match_content` in the directive arguments.
"""
if not directives or isinstance(directives, six.string_types):
if not directives or isinstance(directives, str):
return None
# If match_content is None, just match on directive type. Otherwise, match on
@@ -7,7 +7,6 @@ import logging
import re
import pyparsing
import six
from acme.magic_typing import Dict
from acme.magic_typing import List
@@ -549,7 +548,7 @@ def _is_include_directive(entry):
"""
return (isinstance(entry, list) and
len(entry) == 2 and entry[0] == 'include' and
isinstance(entry[1], six.string_types))
isinstance(entry[1], str))
def _is_ssl_on_directive(entry):
"""Checks if an nginx parsed entry is an 'ssl on' directive.
@@ -654,7 +653,7 @@ def _add_directive(block, directive, insert_at_top):
directive_name = directive[0]
def can_append(loc, dir_name):
""" Can we append this directive to the block? """
return loc is None or (isinstance(dir_name, six.string_types)
return loc is None or (isinstance(dir_name, str)
and dir_name in REPEATABLE_DIRECTIVES)
err_fmt = 'tried to insert directive "{0}" but found conflicting "{1}".'
@@ -4,7 +4,6 @@ raw lists of tokens from pyparsing. """
import abc
import logging
import six
from acme.magic_typing import List
from certbot import errors
@@ -152,7 +151,7 @@ class Statements(Parsable):
if not isinstance(raw_list, list):
raise errors.MisconfigurationError("Statements parsing expects a list!")
# If there's a trailing whitespace in the list of statements, keep track of it.
if raw_list and isinstance(raw_list[-1], six.string_types) and raw_list[-1].isspace():
if raw_list and isinstance(raw_list[-1], str) and raw_list[-1].isspace():
self._trailing_whitespace = raw_list[-1]
raw_list = raw_list[:-1]
self._data = [parse_raw(elem, self, add_spaces) for elem in raw_list]
@@ -184,7 +183,7 @@ class Statements(Parsable):
def _space_list(list_):
""" Inserts whitespace between adjacent non-whitespace tokens. """
spaced_statement = [] # type: List[str]
for i in reversed(six.moves.xrange(len(list_))):
for i in reversed(range(len(list_))):
spaced_statement.insert(0, list_[i])
if i > 0 and not list_[i].isspace() and not list_[i-1].isspace():
spaced_statement.insert(0, " ")
@@ -206,7 +205,7 @@ class Sentence(Parsable):
:returns: whether this lists is parseable by `Sentence`.
"""
return isinstance(lists, list) and len(lists) > 0 and \
all(isinstance(elem, six.string_types) for elem in lists)
all(isinstance(elem, str) for elem in lists)
def parse(self, raw_list, add_spaces=False):
""" Parses a list of string types into this object.
@@ -214,7 +213,7 @@ class Sentence(Parsable):
if add_spaces:
raw_list = _space_list(raw_list)
if not isinstance(raw_list, list) or \
any(not isinstance(elem, six.string_types) for elem in raw_list):
any(not isinstance(elem, str) for elem in raw_list):
raise errors.MisconfigurationError("Sentence parsing expects a list of string types.")
self._data = raw_list
+1 -2
View File
@@ -6,7 +6,6 @@ try:
import mock
except ImportError: # pragma: no cover
from unittest import mock # type: ignore
import six
from acme import challenges
from certbot import achallenges
@@ -79,7 +78,7 @@ class HttpPerformTest(util.NginxTest):
http_responses = self.http01.perform()
self.assertEqual(len(http_responses), 4)
for i in six.moves.range(4):
for i in range(4):
self.assertEqual(http_responses[i], acme_responses[i])
def test_mod_config(self):
+1
View File
@@ -17,6 +17,7 @@ Certbot adheres to [Semantic Versioning](https://semver.org/).
* The acme library no longer depends on the `security` extras from `requests`
which was needed to support SNI in TLS requests when using old versions of
Python 2.
* Certbot and all of its components no longer depend on the library `six`.
### Fixed
+1 -2
View File
@@ -10,7 +10,6 @@ from cryptography.hazmat.primitives import serialization
import josepy as jose
import pyrfc3339
import pytz
import six
from acme import fields as acme_fields
from acme import messages
@@ -101,7 +100,7 @@ class AccountMemoryStorage(interfaces.AccountStorage):
self.accounts = initial_accounts if initial_accounts is not None else {}
def find_all(self):
return list(six.itervalues(self.accounts))
return list(self.accounts.values())
def save(self, account, client):
if account.id in self.accounts:
+2 -3
View File
@@ -7,7 +7,6 @@ import glob
import sys
import configargparse
import six
import zope.component
import zope.interface
@@ -99,7 +98,7 @@ class HelpfulArgumentParser(object):
if isinstance(help1, bool) and isinstance(help2, bool):
self.help_arg = help1 or help2
else:
self.help_arg = help1 if isinstance(help1, six.string_types) else help2
self.help_arg = help1 if isinstance(help1, str) else help2
short_usage = self._usage_string(plugins, self.help_arg)
@@ -470,7 +469,7 @@ class HelpfulArgumentParser(object):
may or may not be displayed as help topics.
"""
for name, plugin_ep in six.iteritems(plugins):
for name, plugin_ep in plugins.items():
parser_or_group = self.add_group(name,
description=plugin_ep.long_description)
plugin_ep.plugin_cls.inject_parser_options(parser_or_group, name)
+1 -1
View File
@@ -1,7 +1,7 @@
"""Certbot user-supplied configuration."""
import copy
from urllib import parse
from six.moves.urllib import parse
import zope.interface
from certbot import errors
+7 -8
View File
@@ -5,7 +5,6 @@ import logging
import sys
import pkg_resources
import six
import zope.interface
import zope.interface.verify
@@ -215,7 +214,7 @@ class PluginsRegistry(Mapping):
# This prevents deadlock caused by plugins acquiring a lock
# and ensures at least one concurrent Certbot instance will run
# successfully.
self._plugins = collections.OrderedDict(sorted(six.iteritems(plugins)))
self._plugins = collections.OrderedDict(sorted(plugins.items()))
@classmethod
def find_all(cls):
@@ -276,12 +275,12 @@ class PluginsRegistry(Mapping):
def init(self, config):
"""Initialize all plugins in the registry."""
return [plugin_ep.init(config) for plugin_ep
in six.itervalues(self._plugins)]
in self._plugins.values()]
def filter(self, pred):
"""Filter plugins based on predicate."""
return type(self)({name: plugin_ep for name, plugin_ep
in six.iteritems(self._plugins) if pred(plugin_ep)})
in self._plugins.items() if pred(plugin_ep)})
def visible(self):
"""Filter plugins based on visibility."""
@@ -297,7 +296,7 @@ class PluginsRegistry(Mapping):
def prepare(self):
"""Prepare all plugins in the registry."""
return [plugin_ep.prepare() for plugin_ep in six.itervalues(self._plugins)]
return [plugin_ep.prepare() for plugin_ep in self._plugins.values()]
def available(self):
"""Filter plugins based on availability."""
@@ -319,7 +318,7 @@ class PluginsRegistry(Mapping):
"""
# use list instead of set because PluginEntryPoint is not hashable
candidates = [plugin_ep for plugin_ep in six.itervalues(self._plugins)
candidates = [plugin_ep for plugin_ep in self._plugins.values()
if plugin_ep.initialized and plugin_ep.init() is plugin]
assert len(candidates) <= 1
if candidates:
@@ -329,9 +328,9 @@ class PluginsRegistry(Mapping):
def __repr__(self):
return "{0}({1})".format(
self.__class__.__name__, ','.join(
repr(p_ep) for p_ep in six.itervalues(self._plugins)))
repr(p_ep) for p_ep in self._plugins.values()))
def __str__(self):
if not self._plugins:
return "No plugins"
return "\n\n".join(str(p_ep) for p_ep in six.itervalues(self._plugins))
return "\n\n".join(str(p_ep) for p_ep in self._plugins.values())
@@ -3,7 +3,6 @@ from __future__ import print_function
import logging
import six
import zope.component
from certbot import errors
@@ -108,7 +107,7 @@ def pick_plugin(config, default, plugins, question, ifaces):
if len(prepared) > 1:
logger.debug("Multiple candidate plugins: %s", prepared)
plugin_ep = choose_plugin(list(six.itervalues(prepared)), question)
plugin_ep = choose_plugin(list(prepared.values()), question)
if plugin_ep is None:
return None
return plugin_ep.init()
@@ -6,7 +6,6 @@ import socket
from socket import errno as socket_errors # type: ignore
import OpenSSL # pylint: disable=unused-import
import six
import zope.interface
from acme import challenges
@@ -183,7 +182,7 @@ class Authenticator(common.Plugin):
for achall in achalls:
if achall in server_achalls:
server_achalls.remove(achall)
for port, servers in six.iteritems(self.servers.running()):
for port, servers in self.servers.running().items():
if not self.served[servers]:
self.servers.stop(port)
+2 -3
View File
@@ -4,7 +4,6 @@ import collections
import json
import logging
import six
import zope.component
import zope.interface
@@ -92,7 +91,7 @@ to serve all files under specified web root ({0})."""
for achall in achalls:
self.conf("map").setdefault(achall.domain, webroot_path)
else:
known_webroots = list(set(six.itervalues(self.conf("map"))))
known_webroots = list(set(self.conf("map").values()))
for achall in achalls:
if achall.domain not in self.conf("map"):
new_webroot = self._prompt_for_webroot(achall.domain,
@@ -242,7 +241,7 @@ class _WebrootMapAction(argparse.Action):
"""Action class for parsing webroot_map."""
def __call__(self, parser, namespace, webroot_map, option_string=None):
for domains, webroot_path in six.iteritems(json.loads(webroot_map)):
for domains, webroot_path in json.loads(webroot_map).items():
webroot_path = _validate_webroot(webroot_path)
namespace.webroot_map.update(
(d, webroot_path) for d in cli.add_domains(namespace, domains))
+6 -7
View File
@@ -13,7 +13,6 @@ from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import ec, rsa
from cryptography.hazmat.primitives.serialization import load_pem_private_key
import OpenSSL
import six
import zope.component
from acme.magic_typing import List
@@ -120,7 +119,7 @@ def _restore_webroot_config(config, renewalparams):
# see https://github.com/certbot/certbot/pull/7095
if "webroot_path" in renewalparams and not cli.set_by_cli("webroot_path"):
wp = renewalparams["webroot_path"]
if isinstance(wp, six.string_types): # prior to 0.1.0, webroot_path was a string
if isinstance(wp, str): # prior to 0.1.0, webroot_path was a string
wp = [wp]
config.webroot_path = wp
@@ -155,7 +154,7 @@ def _restore_plugin_configs(config, renewalparams):
for plugin_prefix in set(plugin_prefixes):
plugin_prefix = plugin_prefix.replace('-', '_')
for config_item, config_value in six.iteritems(renewalparams):
for config_item, config_value in renewalparams.items():
if config_item.startswith(plugin_prefix + "_") and not cli.set_by_cli(config_item):
# Values None, True, and False need to be treated specially,
# As their types aren't handled correctly by configobj
@@ -180,9 +179,9 @@ def restore_required_config_elements(config, renewalparams):
required_items = itertools.chain(
(("pref_challs", _restore_pref_challs),),
six.moves.zip(BOOL_CONFIG_ITEMS, itertools.repeat(_restore_bool)),
six.moves.zip(INT_CONFIG_ITEMS, itertools.repeat(_restore_int)),
six.moves.zip(STR_CONFIG_ITEMS, itertools.repeat(_restore_str)))
zip(BOOL_CONFIG_ITEMS, itertools.repeat(_restore_bool)),
zip(INT_CONFIG_ITEMS, itertools.repeat(_restore_int)),
zip(STR_CONFIG_ITEMS, itertools.repeat(_restore_str)))
for item_name, restore_func in required_items:
if item_name in renewalparams and not cli.set_by_cli(item_name):
value = restore_func(item_name, renewalparams[item_name])
@@ -220,7 +219,7 @@ def _restore_pref_challs(unused_name, value):
# If pref_challs has only one element, configobj saves the value
# with a trailing comma so it's parsed as a list. If this comma is
# removed by the user, the value is parsed as a str.
value = [value] if isinstance(value, six.string_types) else value
value = [value] if isinstance(value, str) else value
return cli.parse_preferred_challenges(value)
+1 -1
View File
@@ -3,10 +3,10 @@ from __future__ import print_function
import collections
import logging
import queue
import sys
import textwrap
from six.moves import queue # type: ignore
import zope.interface
from certbot import interfaces
+1 -2
View File
@@ -9,7 +9,6 @@ import stat
import configobj
import parsedatetime
import pytz
import six
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
from cryptography.hazmat.primitives.serialization import load_pem_private_key
@@ -275,7 +274,7 @@ def relevant_values(all_values):
rv = dict(
(option, value)
for option, value in six.iteritems(all_values)
for option, value in all_values.items()
if _relevant(namespaces, option) and cli.option_was_set(option, value))
# We always save the server value to help with forward compatibility
# and behavioral consistency when versions of Certbot with different
+3 -9
View File
@@ -22,7 +22,6 @@ from OpenSSL import crypto
from OpenSSL import SSL # type: ignore
import pyrfc3339
import six
import zope.component
from acme import crypto_util as acme_crypto_util
@@ -215,7 +214,7 @@ def make_key(bits=1024, key_type="rsa", elliptic_curve=None):
except TypeError:
raise errors.Error("Unsupported elliptic curve: {}".format(elliptic_curve))
except UnsupportedAlgorithm as e:
raise six.raise_from(e, errors.Error(str(e)))
raise e from errors.Error(str(e))
_key_pem = _key.private_bytes(
encoding=Encoding.PEM,
format=PrivateFormat.TraditionalOpenSSL,
@@ -492,14 +491,9 @@ def _notAfterBefore(cert_path, method):
reformatted_timestamp = [timestamp[0:4], b"-", timestamp[4:6], b"-",
timestamp[6:8], b"T", timestamp[8:10], b":",
timestamp[10:12], b":", timestamp[12:]]
# pyrfc3339 always uses the type `str`. This means that in Python 2, it
# expects str/bytes and in Python 3 it expects its str type or the Python 2
# equivalent of the type unicode.
# pyrfc3339 always uses the type `str`
timestamp_bytes = b"".join(reformatted_timestamp)
if six.PY3:
timestamp_str = timestamp_bytes.decode('ascii')
else:
timestamp_str = timestamp_bytes
timestamp_str = timestamp_bytes.decode('ascii')
return pyrfc3339.parse(timestamp_str)
+2 -2
View File
@@ -71,7 +71,7 @@ def _wrap_lines(msg):
def input_with_timeout(prompt=None, timeout=36000.0):
"""Get user input with a timeout.
Behaves the same as six.moves.input, however, an error is raised if
Behaves the same as the builtin input, however, an error is raised if
a user doesn't answer after timeout seconds. The default timeout
value was chosen to place it just under 12 hours for users following
our advice and running Certbot twice a day.
@@ -85,7 +85,7 @@ def input_with_timeout(prompt=None, timeout=36000.0):
:raises errors.Error if no answer is given before the timeout
"""
# use of sys.stdin and sys.stdout to mimic six.moves.input based on
# use of sys.stdin and sys.stdout to mimic the builtin input based on
# https://github.com/python/cpython/blob/baf7bb30a02aabde260143136bdf5b3738a1d409/Lib/getpass.py#L129
if prompt:
sys.stdout.write(prompt)
+4 -9
View File
@@ -1,14 +1,12 @@
"""Certbot client interfaces."""
import abc
import six
import zope.interface
# pylint: disable=no-self-argument,no-method-argument,inherit-non-class
@six.add_metaclass(abc.ABCMeta)
class AccountStorage(object):
class AccountStorage(object, metaclass=abc.ABCMeta):
"""Accounts storage interface."""
@abc.abstractmethod
@@ -547,8 +545,7 @@ class IReporter(zope.interface.Interface):
"""Prints messages to the user and clears the message queue."""
@six.add_metaclass(abc.ABCMeta)
class RenewableCert(object):
class RenewableCert(object, metaclass=abc.ABCMeta):
"""Interface to a certificate lineage."""
@abc.abstractproperty
@@ -613,8 +610,7 @@ class RenewableCert(object):
# an update during the run or install subcommand, it should do so when
# :func:`IInstaller.deploy_cert` is called.
@six.add_metaclass(abc.ABCMeta)
class GenericUpdater(object):
class GenericUpdater(object, metaclass=abc.ABCMeta):
"""Interface for update types not currently specified by Certbot.
This class allows plugins to perform types of updates that Certbot hasn't
@@ -646,8 +642,7 @@ class GenericUpdater(object):
"""
@six.add_metaclass(abc.ABCMeta)
class RenewDeployer(object):
class RenewDeployer(object, metaclass=abc.ABCMeta):
"""Interface for update types run when a lineage is renewed
This class allows plugins to perform types of updates that need to run at
+1 -2
View File
@@ -6,7 +6,6 @@ try:
import mock
except ImportError: # pragma: no cover
from unittest import mock # type: ignore
import six
from acme import challenges
from certbot import achallenges
@@ -31,7 +30,7 @@ class BaseAuthenticatorTest(object):
challb=acme_util.DNS01, domain=DOMAIN, account_key=KEY)
def test_more_info(self):
self.assertTrue(isinstance(self.auth.more_info(), six.string_types)) # pylint: disable=no-member
self.assertTrue(isinstance(self.auth.more_info(), str)) # pylint: disable=no-member
def test_get_chall_pref(self):
self.assertEqual(self.auth.get_chall_pref(None), [challenges.DNS01]) # pylint: disable=no-member
+1 -4
View File
@@ -1,8 +1,6 @@
"""New interface style Certbot enhancements"""
import abc
import six
from acme.magic_typing import Any
from acme.magic_typing import Dict
from acme.magic_typing import List
@@ -91,8 +89,7 @@ def populate_cli(add):
help=enh["cli_help"])
@six.add_metaclass(abc.ABCMeta)
class AutoHSTSEnhancement(object):
class AutoHSTSEnhancement(object, metaclass=abc.ABCMeta):
"""
Enhancement interface that installer plugins can implement in order to
provide functionality that configures the software to have a
+1 -2
View File
@@ -7,7 +7,6 @@ import sys
import time
import traceback
import six
from certbot import errors
from certbot import util
@@ -518,7 +517,7 @@ class Reverter(object):
# It is possible save checkpoints faster than 1 per second resulting in
# collisions in the naming convention.
for _ in six.moves.range(2):
for _ in range(2):
timestamp = self._checkpoint_timestamp()
final_dir = os.path.join(self.config.backup_dir, timestamp)
try:
+1 -2
View File
@@ -2,7 +2,6 @@
import datetime
import josepy as jose
import six
from acme import challenges
from acme import messages
@@ -69,7 +68,7 @@ def gen_authzr(authz_status, domain, challs, statuses, combos=True):
"""
challbs = tuple(
chall_to_challb(chall, status)
for chall, status in six.moves.zip(challs, statuses)
for chall, status in zip(challs, statuses)
)
authz_kwargs = {
"identifier": messages.Identifier(
+15 -13
View File
@@ -1,4 +1,6 @@
"""Test utilities."""
from importlib import reload as reload_module
import io
import logging
from multiprocessing import Event
from multiprocessing import Process
@@ -11,20 +13,8 @@ import warnings
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
import josepy as jose
try:
import mock
warnings.warn(
"The external mock module is being used for backwards compatibility "
"since it is available, however, future versions of Certbot's tests will "
"use unittest.mock. Be sure to update your code accordingly.",
PendingDeprecationWarning
)
except ImportError: # pragma: no cover
from unittest import mock # type: ignore
import OpenSSL
import pkg_resources
import six
from six.moves import reload_module
from certbot import interfaces
from certbot import util
@@ -36,6 +26,18 @@ from certbot.compat import filesystem
from certbot.compat import os
from certbot.display import util as display_util
try:
import mock
warnings.warn(
"The external mock module is being used for backwards compatibility "
"since it is available, however, future versions of Certbot's tests will "
"use unittest.mock. Be sure to update your code accordingly.",
PendingDeprecationWarning
)
except ImportError: # pragma: no cover
from unittest import mock # type: ignore
def vector_path(*names):
"""Path to a test vector."""
@@ -177,7 +179,7 @@ def patch_get_utility_with_stdout(target='zope.component.getUtility',
:rtype: mock.MagicMock
"""
stdout = stdout if stdout else six.StringIO()
stdout = stdout if stdout else io.StringIO()
freezable_mock = _create_get_utility_mock_with_stdout(stdout)
return mock.patch(target, new=freezable_mock)
+3 -4
View File
@@ -15,7 +15,6 @@ import subprocess
import sys
import configargparse
import six
from acme.magic_typing import Text
from acme.magic_typing import Tuple
@@ -153,7 +152,7 @@ def lock_dir_until_exit(dir_path):
def _release_locks():
for dir_lock in six.itervalues(_LOCKS):
for dir_lock in _LOCKS.values():
try:
dir_lock.release()
except: # pylint: disable=bare-except
@@ -517,7 +516,7 @@ def enforce_domain_sanity(domain):
"""
# Unicode
try:
if isinstance(domain, six.binary_type):
if isinstance(domain, bytes):
domain = domain.decode('utf-8')
domain.encode('ascii')
except UnicodeError:
@@ -579,7 +578,7 @@ def is_wildcard_domain(domain):
"""
wildcard_marker = b"*." # type: Union[Text, bytes]
if isinstance(domain, six.text_type):
if isinstance(domain, str):
wildcard_marker = u"*."
return domain.startswith(wildcard_marker)
+3 -3
View File
@@ -31,9 +31,9 @@ meta = dict(re.findall(r"""__([a-z]+)__ = '([^']+)""", read_file(init_fn)))
readme = read_file(os.path.join(here, 'README.rst'))
version = meta['version']
# This package relies on PyOpenSSL, requests, and six, however, it isn't
# specified here to avoid masking the more specific request requirements in
# acme. See https://github.com/pypa/pip/issues/988 for more info.
# This package relies on PyOpenSSL and requests, however, it isn't specified
# here to avoid masking the more specific request requirements in acme. See
# https://github.com/pypa/pip/issues/988 for more info.
install_requires = [
'acme>=1.8.0',
# We technically need ConfigArgParse 0.10.0 for Python 2.6 support, but
+2 -2
View File
@@ -276,14 +276,14 @@ class AccountFileStorageTest(test_util.ConfigTestCase):
self.storage.save(self.acc, self.mock_client)
mock_open = mock.mock_open()
mock_open.side_effect = IOError
with mock.patch("six.moves.builtins.open", mock_open):
with mock.patch("builtins.open", mock_open):
self.assertRaises(
errors.AccountStorageError, self.storage.load, self.acc.id)
def test_save_ioerrors(self):
mock_open = mock.mock_open()
mock_open.side_effect = IOError # TODO: [None, None, IOError]
with mock.patch("six.moves.builtins.open", mock_open):
with mock.patch("builtins.open", mock_open):
self.assertRaises(
errors.AccountStorageError, self.storage.save,
self.acc, self.mock_client)
+9 -8
View File
@@ -1,16 +1,11 @@
"""Tests for certbot._internal.cli."""
import argparse
import copy
from importlib import reload as reload_module
import io
import tempfile
import unittest
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import six
from six.moves import reload_module # pylint: disable=import-error
from acme import challenges
from certbot import errors
from certbot._internal import cli
@@ -21,6 +16,12 @@ from certbot.compat import os
import certbot.tests.util as test_util
from certbot.tests.util import TempDirTestCase
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
PLUGINS = disco.PluginsRegistry.find_all()
@@ -91,7 +92,7 @@ class ParseTest(unittest.TestCase):
def _help_output(self, args):
"Run a command, and return the output string for scrutiny"
output = six.StringIO()
output = io.StringIO()
def write_msg(message, *args, **kwargs): # pylint: disable=missing-docstring,unused-argument
output.write(message)
+5 -4
View File
@@ -3,19 +3,20 @@ try:
import readline # pylint: disable=import-error
except ImportError:
import certbot._internal.display.dummy_readline as readline # type: ignore
from importlib import reload as reload_module
import string
import sys
import unittest
from certbot.compat import filesystem # pylint: disable=ungrouped-imports
from certbot.compat import os # pylint: disable=ungrouped-imports
import certbot.tests.util as test_util # pylint: disable=ungrouped-imports
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
from six.moves import reload_module # pylint: disable=import-error
from certbot.compat import filesystem # pylint: disable=ungrouped-imports
from certbot.compat import os # pylint: disable=ungrouped-imports
import certbot.tests.util as test_util # pylint: disable=ungrouped-imports
class CompleterTest(test_util.TempDirTestCase):
+11 -17
View File
@@ -1,20 +1,22 @@
"""Test :mod:`certbot.display.util`."""
import inspect
import io
import socket
import tempfile
import unittest
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import six
from certbot import errors
from certbot import interfaces
from certbot.display import util as display_util
import certbot.tests.util as test_util
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
CHOICES = [("First", "Description1"), ("Second", "Description2")]
TAGS = ["tag1", "tag2", "tag3"]
TAGS_CHOICES = [("1", "tag1"), ("2", "tag2"), ("3", "tag3")]
@@ -34,7 +36,7 @@ class InputWithTimeoutTest(unittest.TestCase):
def test_input(self, prompt=None):
expected = "foo bar"
stdin = six.StringIO(expected + "\n")
stdin = io.StringIO(expected + "\n")
with mock.patch("certbot.compat.misc.select.select") as mock_select:
mock_select.return_value = ([stdin], [], [],)
self.assertEqual(self._call(prompt), expected)
@@ -328,11 +330,7 @@ class FileOutputDisplayTest(unittest.TestCase):
# Every IDisplay method implemented by FileDisplay must take
# force_interactive to prevent workflow regressions.
for name in interfaces.IDisplay.names():
if six.PY2:
getargspec = inspect.getargspec
else:
getargspec = inspect.getfullargspec
arg_spec = getargspec(getattr(self.displayer, name)) # pylint: disable=deprecated-method
arg_spec = inspect.getfullargspec(getattr(self.displayer, name))
self.assertTrue("force_interactive" in arg_spec.args)
@@ -402,12 +400,8 @@ class NoninteractiveDisplayTest(unittest.TestCase):
for name in interfaces.IDisplay.names(): # pylint: disable=E1120
method = getattr(self.displayer, name)
# asserts method accepts arbitrary keyword arguments
if six.PY2:
result = inspect.getargspec(method).keywords # pylint:deprecated-method
self.assertFalse(result is None)
else:
result = inspect.getfullargspec(method).varkw
self.assertFalse(result is None)
result = inspect.getfullargspec(method).varkw
self.assertFalse(result is None)
class SeparateListInputTest(unittest.TestCase):
+13 -11
View File
@@ -1,15 +1,11 @@
"""Tests for certbot._internal.log."""
import io
import logging
import logging.handlers
import sys
import time
import unittest
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import six
from acme import messages
from certbot import errors
@@ -19,6 +15,12 @@ from certbot.compat import filesystem
from certbot.compat import os
from certbot.tests import util as test_util
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
class PreArgParseSetupTest(unittest.TestCase):
"""Tests for certbot._internal.log.pre_arg_parse_setup."""
@@ -75,7 +77,7 @@ class PostArgParseSetupTest(test_util.ConfigTestCase):
self.devnull = open(os.devnull, 'w')
from certbot._internal.log import ColoredStreamHandler
self.stream_handler = ColoredStreamHandler(six.StringIO())
self.stream_handler = ColoredStreamHandler(io.StringIO())
from certbot._internal.log import MemoryHandler, TempHandler
self.temp_handler = TempHandler()
self.temp_path = self.temp_handler.path
@@ -179,7 +181,7 @@ class ColoredStreamHandlerTest(unittest.TestCase):
"""Tests for certbot._internal.log.ColoredStreamHandler"""
def setUp(self):
self.stream = six.StringIO()
self.stream = io.StringIO()
self.stream.isatty = lambda: True
self.logger = logging.getLogger()
self.logger.setLevel(logging.DEBUG)
@@ -213,7 +215,7 @@ class MemoryHandlerTest(unittest.TestCase):
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.DEBUG)
self.msg = 'hi there'
self.stream = six.StringIO()
self.stream = io.StringIO()
self.stream_handler = logging.StreamHandler(self.stream)
from certbot._internal.log import MemoryHandler
@@ -238,7 +240,7 @@ class MemoryHandlerTest(unittest.TestCase):
def test_target_reset(self):
self._test_log_debug()
new_stream = six.StringIO()
new_stream = io.StringIO()
new_stream_handler = logging.StreamHandler(new_stream)
self.handler.setTarget(new_stream_handler)
self.handler.flush(force=True)
@@ -325,7 +327,7 @@ class PostArgParseExceptHookTest(unittest.TestCase):
def test_acme_error(self):
# Get an arbitrary error code
acme_code = next(six.iterkeys(messages.ERROR_CODES))
acme_code = next(iter(messages.ERROR_CODES))
def get_acme_error(msg):
"""Wraps ACME errors so the constructor takes only a msg."""
@@ -349,7 +351,7 @@ class PostArgParseExceptHookTest(unittest.TestCase):
def _test_common(self, error_type, debug):
"""Returns the mocked logger and stderr output."""
mock_err = six.StringIO()
mock_err = io.StringIO()
def write_err(*args, **unused_kwargs):
"""Write error to mock_err."""
+17 -15
View File
@@ -4,6 +4,8 @@
from __future__ import print_function
import datetime
from importlib import reload as reload_module
import io
import itertools
import json
import shutil
@@ -13,13 +15,7 @@ import traceback
import unittest
import josepy as jose
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import pytz
import six
from six.moves import reload_module # pylint: disable=import-error
from certbot import crypto_util
from certbot import errors
@@ -39,6 +35,12 @@ from certbot.compat import os
from certbot.plugins import enhancements
import certbot.tests.util as test_util
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
CERT_PATH = test_util.vector_path('cert_512.pem')
CERT = test_util.vector_path('cert_512.pem')
@@ -598,7 +600,7 @@ class MainTest(test_util.ConfigTestCase):
"Run the client with output streams mocked out"
args = self.standard_args + args
toy_stdout = stdout if stdout else six.StringIO()
toy_stdout = stdout if stdout else io.StringIO()
with mock.patch('certbot._internal.main.sys.stdout', new=toy_stdout):
with mock.patch('certbot._internal.main.sys.stderr') as stderr:
with mock.patch("certbot.util.atexit"):
@@ -611,8 +613,8 @@ class MainTest(test_util.ConfigTestCase):
self.assertEqual(1, mock_run.call_count)
def test_version_string_program_name(self):
toy_out = six.StringIO()
toy_err = six.StringIO()
toy_out = io.StringIO()
toy_err = io.StringIO()
with mock.patch('certbot._internal.main.sys.stdout', new=toy_out):
with mock.patch('certbot._internal.main.sys.stderr', new=toy_err):
try:
@@ -820,7 +822,7 @@ class MainTest(test_util.ConfigTestCase):
flags = ['--init', '--prepare', '--authenticators', '--installers']
for args in itertools.chain(
*(itertools.combinations(flags, r)
for r in six.moves.range(len(flags)))):
for r in range(len(flags)))):
self._call(['plugins'] + list(args))
@mock.patch('certbot._internal.main.plugins_disco')
@@ -829,7 +831,7 @@ class MainTest(test_util.ConfigTestCase):
ifaces = [] # type: List[interfaces.IPlugin]
plugins = mock_disco.PluginsRegistry.find_all()
stdout = six.StringIO()
stdout = io.StringIO()
with test_util.patch_get_utility_with_stdout(stdout=stdout):
_, stdout, _, _ = self._call(['plugins'], stdout)
@@ -849,7 +851,7 @@ class MainTest(test_util.ConfigTestCase):
_, _, _ = directory, mode, strict
raise errors.Error()
stdout = six.StringIO()
stdout = io.StringIO()
with mock.patch('certbot.util.set_up_core_dir') as mock_set_up_core_dir:
with test_util.patch_get_utility_with_stdout(stdout=stdout):
mock_set_up_core_dir.side_effect = throw_error
@@ -866,7 +868,7 @@ class MainTest(test_util.ConfigTestCase):
ifaces = [] # type: List[interfaces.IPlugin]
plugins = mock_disco.PluginsRegistry.find_all()
stdout = six.StringIO()
stdout = io.StringIO()
with test_util.patch_get_utility_with_stdout(stdout=stdout):
_, stdout, _, _ = self._call(['plugins', '--init'], stdout)
@@ -884,7 +886,7 @@ class MainTest(test_util.ConfigTestCase):
ifaces = [] # type: List[interfaces.IPlugin]
plugins = mock_disco.PluginsRegistry.find_all()
stdout = six.StringIO()
stdout = io.StringIO()
with test_util.patch_get_utility_with_stdout(stdout=stdout):
_, stdout, _, _ = self._call(['plugins', '--init', '--prepare'], stdout)
@@ -1033,7 +1035,7 @@ class MainTest(test_util.ConfigTestCase):
mock_certr = mock.MagicMock()
mock_key = mock.MagicMock(pem='pem_key')
mock_client = mock.MagicMock()
stdout = six.StringIO()
stdout = io.StringIO()
mock_client.obtain_certificate.return_value = (mock_certr, 'chain',
mock_key, 'csr')
+2 -3
View File
@@ -8,7 +8,6 @@ try:
except ImportError: # pragma: no cover
from unittest import mock
import pkg_resources
import six
import zope.interface
from certbot import errors
@@ -56,7 +55,7 @@ class PluginEntryPointTest(unittest.TestCase):
EP_SA: "sa",
}
for entry_point, name in six.iteritems(names):
for entry_point, name in names.items():
self.assertEqual(
name, PluginEntryPoint.entry_point_to_plugin_name(entry_point, with_prefix=False))
@@ -70,7 +69,7 @@ class PluginEntryPointTest(unittest.TestCase):
self.ep3: "p3:ep3",
}
for entry_point, name in six.iteritems(names):
for entry_point, name in names.items():
self.assertEqual(
name, PluginEntryPoint.entry_point_to_plugin_name(entry_point, with_prefix=True))
+1 -2
View File
@@ -6,7 +6,6 @@ try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import six
from acme import challenges
from certbot import errors
@@ -53,7 +52,7 @@ class AuthenticatorTest(test_util.TempDirTestCase):
self.assertRaises(errors.HookCommandNotFound, self.auth.prepare)
def test_more_info(self):
self.assertTrue(isinstance(self.auth.more_info(), six.string_types))
self.assertTrue(isinstance(self.auth.more_info(), str))
def test_get_chall_pref(self):
self.assertEqual(self.auth.get_chall_pref('example.org'),
+1 -2
View File
@@ -5,7 +5,6 @@ try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import six
class InstallerTest(unittest.TestCase):
@@ -16,7 +15,7 @@ class InstallerTest(unittest.TestCase):
self.installer = Installer(config=mock.MagicMock(), name="null")
def test_it(self):
self.assertTrue(isinstance(self.installer.more_info(), six.string_types))
self.assertTrue(isinstance(self.installer.more_info(), str))
self.assertEqual([], self.installer.get_all_names())
self.assertEqual([], self.installer.supported_enhancements())
+1 -2
View File
@@ -10,7 +10,6 @@ try:
except ImportError: # pragma: no cover
from unittest import mock
import OpenSSL.crypto # pylint: disable=unused-import
import six
from acme import challenges
from acme import standalone as acme_standalone # pylint: disable=unused-import
@@ -91,7 +90,7 @@ class AuthenticatorTest(unittest.TestCase):
self.auth.servers = mock.MagicMock()
def test_more_info(self):
self.assertTrue(isinstance(self.auth.more_info(), six.string_types))
self.assertTrue(isinstance(self.auth.more_info(), str))
def test_get_chall_pref(self):
self.assertEqual(self.auth.get_chall_pref(domain=None),
+1 -1
View File
@@ -33,7 +33,7 @@ class PluginStorageTest(test_util.ConfigTestCase):
mock_open.side_effect = IOError
self.plugin.storage.storagepath = os.path.join(self.config.config_dir,
".pluginstorage.json")
with mock.patch("six.moves.builtins.open", mock_open):
with mock.patch("builtins.open", mock_open):
with mock.patch('certbot.compat.os.path.isfile', return_value=True):
with mock.patch("certbot.reverter.util"):
self.assertRaises(errors.PluginStorageError,
+3 -4
View File
@@ -14,7 +14,6 @@ try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import six
from acme import challenges
from certbot import achallenges
@@ -59,7 +58,7 @@ class AuthenticatorTest(unittest.TestCase):
def test_more_info(self):
more_info = self.auth.more_info()
self.assertTrue(isinstance(more_info, six.string_types))
self.assertTrue(isinstance(more_info, str))
self.assertTrue(self.path in more_info)
def test_add_parser_arguments(self):
@@ -83,7 +82,7 @@ class AuthenticatorTest(unittest.TestCase):
self.assertTrue(self.achall.domain in call[0][0])
self.assertTrue(all(
webroot in call[0][1]
for webroot in six.itervalues(self.config.webroot_map)))
for webroot in self.config.webroot_map.values()))
self.assertEqual(self.config.webroot_map[self.achall.domain],
self.path)
@@ -100,7 +99,7 @@ class AuthenticatorTest(unittest.TestCase):
self.assertTrue(self.achall.domain in call[0][0])
self.assertTrue(all(
webroot in call[0][1]
for webroot in six.itervalues(self.config.webroot_map)))
for webroot in self.config.webroot_map.values()))
@test_util.patch_get_utility()
def test_new_webroot(self, mock_get_utility):
+3 -2
View File
@@ -1,12 +1,13 @@
"""Tests for certbot._internal.reporter."""
import io
import sys
import unittest
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import six
class ReporterTest(unittest.TestCase):
@@ -16,7 +17,7 @@ class ReporterTest(unittest.TestCase):
self.reporter = reporter.Reporter(mock.MagicMock(quiet=False))
self.old_stdout = sys.stdout # type: ignore
sys.stdout = six.StringIO()
sys.stdout = io.StringIO()
def tearDown(self):
sys.stdout = self.old_stdout
+1 -2
View File
@@ -9,7 +9,6 @@ try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import six
from certbot import errors
from certbot.compat import os
@@ -156,7 +155,7 @@ class ReverterCheckpointLocalTest(test_util.ConfigTestCase):
act_coms = get_undo_commands(self.config.temp_checkpoint_dir)
for a_com, com in six.moves.zip(act_coms, coms):
for a_com, com in zip(act_coms, coms):
self.assertEqual(a_com, com)
def test_bad_register_undo_command(self):
+9 -10
View File
@@ -11,7 +11,6 @@ try:
except ImportError: # pragma: no cover
from unittest import mock
import pytz
import six
import certbot
from certbot import errors
@@ -286,7 +285,7 @@ class RenewableCertTests(BaseRenewableCertTest):
self.assertEqual(self.test_rc.current_version("cert"), None)
def test_latest_and_next_versions(self):
for ver in six.moves.range(1, 6):
for ver in range(1, 6):
for kind in ALL_FOUR:
self._write_out_kind(kind, ver)
self.assertEqual(self.test_rc.latest_common_version(), 5)
@@ -326,7 +325,7 @@ class RenewableCertTests(BaseRenewableCertTest):
def test_update_link_to(self):
for ver in six.moves.range(1, 6):
for ver in range(1, 6):
for kind in ALL_FOUR:
self._write_out_kind(kind, ver)
self.assertEqual(ver, self.test_rc.current_version(kind))
@@ -353,12 +352,12 @@ class RenewableCertTests(BaseRenewableCertTest):
os.path.basename(self.test_rc.version("cert", 8)))
def test_update_all_links_to_success(self):
for ver in six.moves.range(1, 6):
for ver in range(1, 6):
for kind in ALL_FOUR:
self._write_out_kind(kind, ver)
self.assertEqual(ver, self.test_rc.current_version(kind))
self.assertEqual(self.test_rc.latest_common_version(), 5)
for ver in six.moves.range(1, 6):
for ver in range(1, 6):
self.test_rc.update_all_links_to(ver)
for kind in ALL_FOUR:
self.assertEqual(ver, self.test_rc.current_version(kind))
@@ -396,11 +395,11 @@ class RenewableCertTests(BaseRenewableCertTest):
self.assertEqual(self.test_rc.current_version(kind), 11)
def test_has_pending_deployment(self):
for ver in six.moves.range(1, 6):
for ver in range(1, 6):
for kind in ALL_FOUR:
self._write_out_kind(kind, ver)
self.assertEqual(ver, self.test_rc.current_version(kind))
for ver in six.moves.range(1, 6):
for ver in range(1, 6):
self.test_rc.update_all_links_to(ver)
for kind in ALL_FOUR:
self.assertEqual(ver, self.test_rc.current_version(kind))
@@ -498,7 +497,7 @@ class RenewableCertTests(BaseRenewableCertTest):
# (to avoid instantiating parser)
mock_rv.side_effect = lambda x: x
for ver in six.moves.range(1, 6):
for ver in range(1, 6):
for kind in ALL_FOUR:
self._write_out_kind(kind, ver)
self.test_rc.update_all_links_to(3)
@@ -529,7 +528,7 @@ class RenewableCertTests(BaseRenewableCertTest):
self.test_rc.version("privkey", i))))
for kind in ALL_FOUR:
self.assertEqual(self.test_rc.available_versions(kind), list(six.moves.range(1, 9)))
self.assertEqual(self.test_rc.available_versions(kind), list(range(1, 9)))
self.assertEqual(self.test_rc.current_version(kind), 3)
# Test updating from latest version rather than old version
self.test_rc.update_all_links_to(8)
@@ -538,7 +537,7 @@ class RenewableCertTests(BaseRenewableCertTest):
b'attempt', self.config))
for kind in ALL_FOUR:
self.assertEqual(self.test_rc.available_versions(kind),
list(six.moves.range(1, 10)))
list(range(1, 10)))
self.assertEqual(self.test_rc.current_version(kind), 8)
with open(self.test_rc.version("fullchain", 9)) as f:
self.assertEqual(f.read(), "last" + "attempt")
+11 -10
View File
@@ -1,21 +1,22 @@
"""Tests for certbot.util."""
import argparse
import errno
from importlib import reload as reload_module
import io
import sys
import unittest
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
import six
from six.moves import reload_module # pylint: disable=import-error
from certbot import errors
from certbot.compat import filesystem
from certbot.compat import os
import certbot.tests.util as test_util
try:
import mock
except ImportError: # pragma: no cover
from unittest import mock
class EnvNoSnapForExternalCallsTest(unittest.TestCase):
"""Tests for certbot.util.env_no_snap_for_external_calls."""
@@ -265,11 +266,11 @@ class UniqueLineageNameTest(test_util.TempDirTestCase):
def test_multiple(self):
items = []
for _ in six.moves.range(10):
for _ in range(10):
items.append(self._call("wow"))
f, name = items[-1]
self.assertTrue(isinstance(f, file_type))
self.assertTrue(isinstance(name, six.string_types))
self.assertTrue(isinstance(name, str))
self.assertTrue("wow-0009.conf" in name)
for f, _ in items:
f.close()
@@ -361,7 +362,7 @@ class AddDeprecatedArgumentTest(unittest.TestCase):
def test_help(self):
self._call("--old-option", 2)
stdout = six.StringIO()
stdout = io.StringIO()
with mock.patch("sys.stdout", new=stdout):
try:
self.parser.parse_args(["-h"])
+2 -2
View File
@@ -39,11 +39,11 @@ import socket
import sys
import time
import traceback
import urllib.error as urllib_error
import urllib.request as urllib_request
import boto3
from botocore.exceptions import ClientError
from six.moves.urllib import error as urllib_error
from six.moves.urllib import request as urllib_request
import yaml
from fabric import Config