Update to Pylint 1.9.4 and corrections

This commit is contained in:
Adrien Ferrand
2019-04-09 09:22:19 +02:00
parent 4515a52d3f
commit 04152c21b5
64 changed files with 193 additions and 216 deletions
+2 -5
View File
@@ -189,9 +189,6 @@ init-import=no
# not used). # not used).
dummy-variables-rgx=(unused)?_.*|dummy dummy-variables-rgx=(unused)?_.*|dummy
# A regular expression matching ignored argument variable names
ignored-argument-names=(unused)?_.*|dummy
# List of additional names supposed to be defined in builtins. Remember that # List of additional names supposed to be defined in builtins. Remember that
# you should avoid to define new builtins when possible. # you should avoid to define new builtins when possible.
additional-builtins= additional-builtins=
@@ -254,7 +251,7 @@ ignored-modules=pkg_resources,confargparse,argparse,six.moves,six.moves.urllib
# List of classes names for which member attributes should not be checked # List of classes names for which member attributes should not be checked
# (useful for classes with attributes dynamically set). # (useful for classes with attributes dynamically set).
ignored-classes=SQLObject ignored-classes=Field,Header,JWS,closing
# When zope mode is activated, add a predefined set of Zope acquired attributes # When zope mode is activated, add a predefined set of Zope acquired attributes
# to generated-members. # to generated-members.
@@ -307,7 +304,7 @@ max-args=6
# Argument names that match this expression will be ignored. Default to name # Argument names that match this expression will be ignored. Default to name
# with leading underscore # with leading underscore
ignored-argument-names=_.* ignored-argument-names=(unused)?_.*|dummy
# Maximum number of locals for function / method body # Maximum number of locals for function / method body
max-locals=15 max-locals=15
+4 -13
View File
@@ -20,11 +20,8 @@ from acme import _TLSSNI01DeprecationModule
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# pylint: disable=too-few-public-methods
class Challenge(jose.TypedJSONObjectWithFields): class Challenge(jose.TypedJSONObjectWithFields):
# _fields_to_partial_json | pylint: disable=abstract-method # _fields_to_partial_json
"""ACME challenge.""" """ACME challenge."""
TYPES = {} # type: dict TYPES = {} # type: dict
@@ -38,7 +35,7 @@ class Challenge(jose.TypedJSONObjectWithFields):
class ChallengeResponse(jose.TypedJSONObjectWithFields): class ChallengeResponse(jose.TypedJSONObjectWithFields):
# _fields_to_partial_json | pylint: disable=abstract-method # _fields_to_partial_json
"""ACME challenge response.""" """ACME challenge response."""
TYPES = {} # type: dict TYPES = {} # type: dict
resource_type = 'challenge' resource_type = 'challenge'
@@ -165,7 +162,6 @@ class KeyAuthorizationChallengeResponse(ChallengeResponse):
@six.add_metaclass(abc.ABCMeta) @six.add_metaclass(abc.ABCMeta)
class KeyAuthorizationChallenge(_TokenChallenge): class KeyAuthorizationChallenge(_TokenChallenge):
# pylint: disable=too-many-ancestors
"""Challenge based on Key Authorization. """Challenge based on Key Authorization.
:param response_cls: Subclass of `KeyAuthorizationChallengeResponse` :param response_cls: Subclass of `KeyAuthorizationChallengeResponse`
@@ -234,7 +230,7 @@ class DNS01Response(KeyAuthorizationChallengeResponse):
"""ACME dns-01 challenge response.""" """ACME dns-01 challenge response."""
typ = "dns-01" typ = "dns-01"
def simple_verify(self, chall, domain, account_public_key): def simple_verify(self, chall, domain, account_public_key): # pylint: disable=unused-argument
"""Simple verify. """Simple verify.
This method no longer checks DNS records and is a simple wrapper This method no longer checks DNS records and is a simple wrapper
@@ -250,7 +246,6 @@ class DNS01Response(KeyAuthorizationChallengeResponse):
:rtype: bool :rtype: bool
""" """
# pylint: disable=unused-argument
verified = self.verify(chall, account_public_key) verified = self.verify(chall, account_public_key)
if not verified: if not verified:
logger.debug("Verification of key authorization in response failed") logger.debug("Verification of key authorization in response failed")
@@ -455,7 +450,6 @@ class TLSSNI01Response(KeyAuthorizationChallengeResponse):
kwargs.setdefault("port", self.PORT) kwargs.setdefault("port", self.PORT)
kwargs["name"] = self.z_domain kwargs["name"] = self.z_domain
# TODO: try different methods? # TODO: try different methods?
# pylint: disable=protected-access
return crypto_util.probe_sni(**kwargs) return crypto_util.probe_sni(**kwargs)
def verify_cert(self, cert): def verify_cert(self, cert):
@@ -516,9 +510,6 @@ class TLSSNI01(KeyAuthorizationChallenge):
# boulder#962, ietf-wg-acme#22 # boulder#962, ietf-wg-acme#22
#n = jose.Field("n", encoder=int, decoder=int) #n = jose.Field("n", encoder=int, decoder=int)
def __init__(self, *args, **kwargs):
super(TLSSNI01, self).__init__(*args, **kwargs)
def validation(self, account_key, **kwargs): def validation(self, account_key, **kwargs):
"""Generate validation. """Generate validation.
@@ -560,7 +551,7 @@ class TLSALPN01(KeyAuthorizationChallenge):
raise NotImplementedError() raise NotImplementedError()
@Challenge.register # pylint: disable=too-many-ancestors @Challenge.register
class DNS(_TokenChallenge): class DNS(_TokenChallenge):
"""ACME "dns" challenge.""" """ACME "dns" challenge."""
typ = "dns" typ = "dns"
+1 -1
View File
@@ -6,7 +6,7 @@ import mock
import OpenSSL import OpenSSL
import requests import requests
from six.moves.urllib import parse as urllib_parse # pylint: disable=import-error,relative-import from six.moves.urllib import parse as urllib_parse # pylint: disable=relative-import
from acme import errors from acme import errors
from acme import test_util from acme import test_util
+7 -11
View File
@@ -137,7 +137,7 @@ class ClientBase(object): # pylint: disable=too-many-instance-attributes
authzr = messages.AuthorizationResource( authzr = messages.AuthorizationResource(
body=messages.Authorization.from_json(response.json()), body=messages.Authorization.from_json(response.json()),
uri=response.headers.get('Location', uri)) uri=response.headers.get('Location', uri))
if identifier is not None and authzr.body.identifier != identifier: # pylint: disable=no-member if identifier is not None and authzr.body.identifier != identifier:
raise errors.UnexpectedUpdate(authzr) raise errors.UnexpectedUpdate(authzr)
return authzr return authzr
@@ -669,7 +669,7 @@ class ClientV2(ClientBase):
response = self._post(self.directory['newOrder'], order) response = self._post(self.directory['newOrder'], order)
body = messages.Order.from_json(response.json()) body = messages.Order.from_json(response.json())
authorizations = [] authorizations = []
for url in body.authorizations: for url in body.authorizations: # pylint: disable=not-an-iterable
authorizations.append(self._authzr_from_response(self._post_as_get(url), uri=url)) authorizations.append(self._authzr_from_response(self._post_as_get(url), uri=url))
return messages.OrderResource( return messages.OrderResource(
body=body, body=body,
@@ -775,10 +775,7 @@ class ClientV2(ClientBase):
def external_account_required(self): def external_account_required(self):
"""Checks if ACME server requires External Account Binding authentication.""" """Checks if ACME server requires External Account Binding authentication."""
if hasattr(self.directory, 'meta') and self.directory.meta.external_account_required: return hasattr(self.directory, 'meta') and self.directory.meta.external_account_required
return True
else:
return False
def _post_as_get(self, *args, **kwargs): def _post_as_get(self, *args, **kwargs):
""" """
@@ -794,7 +791,7 @@ class ClientV2(ClientBase):
# We add an empty payload for POST-as-GET requests # We add an empty payload for POST-as-GET requests
new_args = args[:1] + (None,) + args[1:] new_args = args[:1] + (None,) + args[1:]
try: try:
return self._post(*new_args, **kwargs) # pylint: disable=star-args return self._post(*new_args, **kwargs)
except messages.Error as error: except messages.Error as error:
if error.code == 'malformed': if error.code == 'malformed':
logger.debug('Error during a POST-as-GET request, ' logger.debug('Error during a POST-as-GET request, '
@@ -915,7 +912,7 @@ class BackwardsCompatibleClientV2(object):
'certificate, please rerun the command for a new one.') 'certificate, please rerun the command for a new one.')
cert = OpenSSL.crypto.dump_certificate( cert = OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_PEM, certr.body.wrapped).decode() # pylint: disable=no-member OpenSSL.crypto.FILETYPE_PEM, certr.body.wrapped).decode()
chain = crypto_util.dump_pyopenssl_chain(chain).decode() chain = crypto_util.dump_pyopenssl_chain(chain).decode()
return orderr.update(fullchain_pem=(cert + chain)) return orderr.update(fullchain_pem=(cert + chain))
@@ -945,8 +942,7 @@ class BackwardsCompatibleClientV2(object):
Always return False for ACMEv1 servers, as it doesn't use External Account Binding.""" Always return False for ACMEv1 servers, as it doesn't use External Account Binding."""
if self.acme_version == 1: if self.acme_version == 1:
return False return False
else: return self.client.external_account_required()
return self.client.external_account_required()
class ClientNetwork(object): # pylint: disable=too-many-instance-attributes class ClientNetwork(object): # pylint: disable=too-many-instance-attributes
@@ -1167,7 +1163,7 @@ class ClientNetwork(object): # pylint: disable=too-many-instance-attributes
if self.REPLAY_NONCE_HEADER in response.headers: if self.REPLAY_NONCE_HEADER in response.headers:
nonce = response.headers[self.REPLAY_NONCE_HEADER] nonce = response.headers[self.REPLAY_NONCE_HEADER]
try: try:
decoded_nonce = jws.Header._fields['nonce'].decode(nonce) # pylint: disable=no-member decoded_nonce = jws.Header._fields['nonce'].decode(nonce)
except jose.DeserializationError as error: except jose.DeserializationError as error:
raise errors.BadNonce(nonce, error) raise errors.BadNonce(nonce, error)
logger.debug('Storing nonce: %s', nonce) logger.debug('Storing nonce: %s', nonce)
+7 -10
View File
@@ -358,7 +358,6 @@ class ClientTest(ClientTestBase):
def test_register(self): def test_register(self):
# "Instance of 'Field' has no to_json/update member" bug: # "Instance of 'Field' has no to_json/update member" bug:
# pylint: disable=no-member
self.response.status_code = http_client.CREATED self.response.status_code = http_client.CREATED
self.response.json.return_value = self.regr.body.to_json() self.response.json.return_value = self.regr.body.to_json()
self.response.headers['Location'] = self.regr.uri self.response.headers['Location'] = self.regr.uri
@@ -371,7 +370,6 @@ class ClientTest(ClientTestBase):
def test_update_registration(self): def test_update_registration(self):
# "Instance of 'Field' has no to_json/update member" bug: # "Instance of 'Field' has no to_json/update member" bug:
# pylint: disable=no-member
self.response.headers['Location'] = self.regr.uri self.response.headers['Location'] = self.regr.uri
self.response.json.return_value = self.regr.body.to_json() self.response.json.return_value = self.regr.body.to_json()
self.assertEqual(self.regr, self.client.update_registration(self.regr)) self.assertEqual(self.regr, self.client.update_registration(self.regr))
@@ -448,7 +446,7 @@ class ClientTest(ClientTestBase):
def test_answer_challenge(self): def test_answer_challenge(self):
self.response.links['up'] = {'url': self.challr.authzr_uri} self.response.links['up'] = {'url': self.challr.authzr_uri}
self.response.json.return_value = self.challr.body.to_json() # pylint: disable=no-member self.response.json.return_value = self.challr.body.to_json()
chall_response = challenges.DNSResponse(validation=None) chall_response = challenges.DNSResponse(validation=None)
@@ -456,7 +454,7 @@ class ClientTest(ClientTestBase):
# TODO: split here and separate test # TODO: split here and separate test
self.assertRaises(errors.UnexpectedUpdate, self.client.answer_challenge, self.assertRaises(errors.UnexpectedUpdate, self.client.answer_challenge,
self.challr.body.update(uri='foo'), chall_response) # pylint: disable=no-member self.challr.body.update(uri='foo'), chall_response)
def test_answer_challenge_missing_next(self): def test_answer_challenge_missing_next(self):
self.assertRaises( self.assertRaises(
@@ -538,7 +536,7 @@ class ClientTest(ClientTestBase):
self.client.retry_after(response=self.response, default=10)) self.client.retry_after(response=self.response, default=10))
def test_poll(self): def test_poll(self):
self.response.json.return_value = self.authzr.body.to_json() # pylint: disable=no-member self.response.json.return_value = self.authzr.body.to_json()
self.assertEqual((self.authzr, self.response), self.assertEqual((self.authzr, self.response),
self.client.poll(self.authzr)) self.client.poll(self.authzr))
@@ -768,7 +766,7 @@ class ClientV2Test(ClientTestBase):
def test_new_account(self): def test_new_account(self):
self.response.status_code = http_client.CREATED self.response.status_code = http_client.CREATED
self.response.json.return_value = self.regr.body.to_json() # pylint: disable=no-member self.response.json.return_value = self.regr.body.to_json()
self.response.headers['Location'] = self.regr.uri self.response.headers['Location'] = self.regr.uri
self.assertEqual(self.regr, self.client.new_account(self.new_reg)) self.assertEqual(self.regr, self.client.new_account(self.new_reg))
@@ -823,7 +821,7 @@ class ClientV2Test(ClientTestBase):
def test_poll_authorizations_failure(self): def test_poll_authorizations_failure(self):
deadline = datetime.datetime(9999, 9, 9) deadline = datetime.datetime(9999, 9, 9)
challb = self.challr.body.update(status=messages.STATUS_INVALID, # pylint: disable=no-member challb = self.challr.body.update(status=messages.STATUS_INVALID,
error=messages.Error.with_code('unauthorized')) error=messages.Error.with_code('unauthorized'))
authz = self.authz.update(status=messages.STATUS_INVALID, challenges=(challb,)) authz = self.authz.update(status=messages.STATUS_INVALID, challenges=(challb,))
self.response.json.return_value = authz.to_json() self.response.json.return_value = authz.to_json()
@@ -872,7 +870,6 @@ class ClientV2Test(ClientTestBase):
def test_update_registration(self): def test_update_registration(self):
# "Instance of 'Field' has no to_json/update member" bug: # "Instance of 'Field' has no to_json/update member" bug:
# pylint: disable=no-member
self.response.headers['Location'] = self.regr.uri self.response.headers['Location'] = self.regr.uri
self.response.json.return_value = self.regr.body.to_json() self.response.json.return_value = self.regr.body.to_json()
self.assertEqual(self.regr, self.client.update_registration(self.regr)) self.assertEqual(self.regr, self.client.update_registration(self.regr))
@@ -964,7 +961,7 @@ class ClientNetworkTest(unittest.TestCase):
MockJSONDeSerializable('foo'), nonce=b'Tg', url="url", MockJSONDeSerializable('foo'), nonce=b'Tg', url="url",
acme_version=1) acme_version=1)
jws = acme_jws.JWS.json_loads(jws_dump) jws = acme_jws.JWS.json_loads(jws_dump)
self.assertEqual(json.loads(jws.payload.decode()), {'foo': 'foo'}) # pylint: disable=no-member self.assertEqual(json.loads(jws.payload.decode()), {'foo': 'foo'})
self.assertEqual(jws.signature.combined.nonce, b'Tg') self.assertEqual(jws.signature.combined.nonce, b'Tg')
def test_wrap_in_jws_v2(self): def test_wrap_in_jws_v2(self):
@@ -974,7 +971,7 @@ class ClientNetworkTest(unittest.TestCase):
MockJSONDeSerializable('foo'), nonce=b'Tg', url="url", MockJSONDeSerializable('foo'), nonce=b'Tg', url="url",
acme_version=2) acme_version=2)
jws = acme_jws.JWS.json_loads(jws_dump) jws = acme_jws.JWS.json_loads(jws_dump)
self.assertEqual(json.loads(jws.payload.decode()), {'foo': 'foo'}) # pylint: disable=no-member self.assertEqual(json.loads(jws.payload.decode()), {'foo': 'foo'})
self.assertEqual(jws.signature.combined.nonce, b'Tg') self.assertEqual(jws.signature.combined.nonce, b'Tg')
self.assertEqual(jws.signature.combined.kid, u'acct-uri') self.assertEqual(jws.signature.combined.kid, u'acct-uri')
self.assertEqual(jws.signature.combined.url, u'url') self.assertEqual(jws.signature.combined.url, u'url')
-1
View File
@@ -134,7 +134,6 @@ def probe_sni(name, host, port=443, timeout=300,
socket_kwargs = {'source_address': source_address} socket_kwargs = {'source_address': source_address}
try: try:
# pylint: disable=star-args
logger.debug( logger.debug(
"Attempting to connect to %s:%d%s.", host, port, "Attempting to connect to %s:%d%s.", host, port,
" from {0}:{1}".format( " from {0}:{1}".format(
+1 -1
View File
@@ -24,7 +24,7 @@ class HeaderTest(unittest.TestCase):
def test_nonce_decoder(self): def test_nonce_decoder(self):
from acme.jws import Header from acme.jws import Header
nonce_field = Header._fields['nonce'] # pylint: disable=no-member nonce_field = Header._fields['nonce']
self.assertRaises( self.assertRaises(
jose.DeserializationError, nonce_field.decode, self.wrong_nonce) jose.DeserializationError, nonce_field.decode, self.wrong_nonce)
+1 -1
View File
@@ -1,6 +1,6 @@
"""ACME protocol messages.""" """ACME protocol messages."""
import six
import json import json
import six
try: try:
from collections.abc import Hashable # pylint: disable=no-name-in-module from collections.abc import Hashable # pylint: disable=no-name-in-module
except ImportError: # pragma: no cover except ImportError: # pragma: no cover
+3 -3
View File
@@ -83,7 +83,7 @@ class BaseDualNetworkedServers(object):
kwargs["ipv6"] = ip_version kwargs["ipv6"] = ip_version
new_address = (server_address[0],) + (port,) + server_address[2:] new_address = (server_address[0],) + (port,) + server_address[2:]
new_args = (new_address,) + remaining_args new_args = (new_address,) + remaining_args
server = ServerClass(*new_args, **kwargs) # pylint: disable=star-args server = ServerClass(*new_args, **kwargs)
logger.debug( logger.debug(
"Successfully bound to %s:%s using %s", new_address[0], "Successfully bound to %s:%s using %s", new_address[0],
new_address[1], "IPv6" if ip_version else "IPv4") new_address[1], "IPv6" if ip_version else "IPv4")
@@ -91,8 +91,8 @@ class BaseDualNetworkedServers(object):
if self.servers: if self.servers:
# Already bound using IPv6. # Already bound using IPv6.
logger.debug( logger.debug(
"Certbot wasn't able to bind to %s:%s using %s, this " + "Certbot wasn't able to bind to %s:%s using %s, this "
"is often expected due to the dual stack nature of " + "is often expected due to the dual stack nature of "
"IPv6 socket implementations.", "IPv6 socket implementations.",
new_address[0], new_address[1], new_address[0], new_address[1],
"IPv6" if ip_version else "IPv4") "IPv6" if ip_version else "IPv4")
+4 -8
View File
@@ -31,14 +31,14 @@ class TLSServerTest(unittest.TestCase):
from acme.standalone import TLSServer from acme.standalone import TLSServer
server = TLSServer( server = TLSServer(
('', 0), socketserver.BaseRequestHandler, bind_and_activate=True) ('', 0), socketserver.BaseRequestHandler, bind_and_activate=True)
server.server_close() # pylint: disable=no-member server.server_close()
def test_ipv6(self): def test_ipv6(self):
if socket.has_ipv6: if socket.has_ipv6:
from acme.standalone import TLSServer from acme.standalone import TLSServer
server = TLSServer( server = TLSServer(
('', 0), socketserver.BaseRequestHandler, bind_and_activate=True, ipv6=True) ('', 0), socketserver.BaseRequestHandler, bind_and_activate=True, ipv6=True)
server.server_close() # pylint: disable=no-member server.server_close()
class TLSSNI01ServerTest(unittest.TestCase): class TLSSNI01ServerTest(unittest.TestCase):
@@ -52,12 +52,11 @@ class TLSSNI01ServerTest(unittest.TestCase):
)} )}
from acme.standalone import TLSSNI01Server from acme.standalone import TLSSNI01Server
self.server = TLSSNI01Server(('localhost', 0), certs=self.certs) self.server = TLSSNI01Server(('localhost', 0), certs=self.certs)
# pylint: disable=no-member
self.thread = threading.Thread(target=self.server.serve_forever) self.thread = threading.Thread(target=self.server.serve_forever)
self.thread.start() self.thread.start()
def tearDown(self): def tearDown(self):
self.server.shutdown() # pylint: disable=no-member self.server.shutdown()
self.thread.join() self.thread.join()
def test_it(self): def test_it(self):
@@ -80,13 +79,12 @@ class HTTP01ServerTest(unittest.TestCase):
from acme.standalone import HTTP01Server from acme.standalone import HTTP01Server
self.server = HTTP01Server(('', 0), resources=self.resources) self.server = HTTP01Server(('', 0), resources=self.resources)
# pylint: disable=no-member
self.port = self.server.socket.getsockname()[1] self.port = self.server.socket.getsockname()[1]
self.thread = threading.Thread(target=self.server.serve_forever) self.thread = threading.Thread(target=self.server.serve_forever)
self.thread.start() self.thread.start()
def tearDown(self): def tearDown(self):
self.server.shutdown() # pylint: disable=no-member self.server.shutdown()
self.thread.join() self.thread.join()
def test_index(self): def test_index(self):
@@ -139,7 +137,6 @@ class BaseDualNetworkedServersTest(unittest.TestCase):
# NB: On Windows, socket.IPPROTO_IPV6 constant may be missing. # NB: On Windows, socket.IPPROTO_IPV6 constant may be missing.
# We use the corresponding value (41) instead. # We use the corresponding value (41) instead.
level = getattr(socket, "IPPROTO_IPV6", 41) level = getattr(socket, "IPPROTO_IPV6", 41)
# pylint: disable=no-member
self.socket.setsockopt(level, socket.IPV6_V6ONLY, 1) self.socket.setsockopt(level, socket.IPV6_V6ONLY, 1)
try: try:
self.server_bind() self.server_bind()
@@ -212,7 +209,6 @@ class HTTP01DualNetworkedServersTest(unittest.TestCase):
from acme.standalone import HTTP01DualNetworkedServers from acme.standalone import HTTP01DualNetworkedServers
self.servers = HTTP01DualNetworkedServers(('', 0), resources=self.resources) self.servers = HTTP01DualNetworkedServers(('', 0), resources=self.resources)
# pylint: disable=no-member
self.port = self.servers.getsocknames()[0][1] self.port = self.servers.getsocknames()[0][1]
self.servers.serve_forever() self.servers.serve_forever()
@@ -276,8 +276,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
util.lock_dir_until_exit(self.option("server_root")) util.lock_dir_until_exit(self.option("server_root"))
except (OSError, errors.LockError): except (OSError, errors.LockError):
logger.debug("Encountered error:", exc_info=True) logger.debug("Encountered error:", exc_info=True)
raise errors.PluginError( raise errors.PluginError("Unable to lock {0}".format(self.option("server_root")))
"Unable to lock %s", self.option("server_root"))
self._prepared = True self._prepared = True
def _verify_exe_availability(self, exe): def _verify_exe_availability(self, exe):
@@ -1191,8 +1190,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
if fp.endswith(".conf"): if fp.endswith(".conf"):
return fp[:-(len(".conf"))] + self.option("le_vhost_ext") return fp[:-(len(".conf"))] + self.option("le_vhost_ext")
else: return fp + self.option("le_vhost_ext")
return fp + self.option("le_vhost_ext")
def _sift_rewrite_rule(self, line): def _sift_rewrite_rule(self, line):
"""Decides whether a line should be copied to a SSL vhost. """Decides whether a line should be copied to a SSL vhost.
@@ -2133,7 +2131,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
vhost.enabled = True vhost.enabled = True
return return
def enable_mod(self, mod_name, temp=False): # pylint: disable=unused-argument def enable_mod(self, mod_name, temp=False): # pylint: disable=unused-argument
"""Enables module in Apache. """Enables module in Apache.
Both enables and reloads Apache so module is active. Both enables and reloads Apache so module is active.
@@ -1,17 +1,16 @@
""" Distribution specific override class for CentOS family (RHEL, Fedora) """ """ Distribution specific override class for CentOS family (RHEL, Fedora) """
import logging import logging
import pkg_resources import pkg_resources
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
import zope.interface import zope.interface
from certbot import interfaces from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
from certbot_apache import apache_util from certbot_apache import apache_util
from certbot_apache import configurator from certbot_apache import configurator
from certbot_apache import parser from certbot_apache import parser
from certbot import interfaces
from certbot.errors import MisconfigurationError from certbot.errors import MisconfigurationError
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -55,7 +54,7 @@ class CentOSConfigurator(configurator.ApacheConfigurator):
self.aug, self.option("server_root"), self.option("vhost_root"), self.aug, self.option("server_root"), self.option("vhost_root"),
self.version, configurator=self) self.version, configurator=self)
def _deploy_cert(self, *args, **kwargs): def _deploy_cert(self, *args, **kwargs): # pylint: disable=arguments-differ
""" """
Override _deploy_cert in order to ensure that the Apache configuration Override _deploy_cert in order to ensure that the Apache configuration
has "LoadModule ssl_module..." before parsing the VirtualHost configuration has "LoadModule ssl_module..." before parsing the VirtualHost configuration
@@ -65,7 +64,6 @@ class CentOSConfigurator(configurator.ApacheConfigurator):
if self.version < (2, 4, 0): if self.version < (2, 4, 0):
self._deploy_loadmodule_ssl_if_needed() self._deploy_loadmodule_ssl_if_needed()
def _deploy_loadmodule_ssl_if_needed(self): def _deploy_loadmodule_ssl_if_needed(self):
""" """
Add "LoadModule ssl_module <pre-existing path>" to main httpd.conf if Add "LoadModule ssl_module <pre-existing path>" to main httpd.conf if
@@ -84,7 +84,6 @@ class DebianConfigurator(configurator.ApacheConfigurator):
return None return None
def enable_mod(self, mod_name, temp=False): def enable_mod(self, mod_name, temp=False):
# pylint: disable=unused-argument
"""Enables module in Apache. """Enables module in Apache.
Both enables and reloads Apache so module is active. Both enables and reloads Apache so module is active.
@@ -35,8 +35,9 @@ class AutoHSTSTest(util.ApacheTest):
pat = '(?:[ "]|^)(strict-transport-security)(?:[ "]|$)' pat = '(?:[ "]|^)(strict-transport-security)(?:[ "]|$)'
for head in header_path: for head in header_path:
if re.search(pat, self.config.parser.aug.get(head).lower()): if re.search(pat, self.config.parser.aug.get(head).lower()):
return self.config.parser.aug.get(head.replace("arg[3]", return self.config.parser.aug.get(
"arg[4]")) head.replace("arg[3]", "arg[4]"))
return None # pragma: no cover
@mock.patch("certbot_apache.configurator.ApacheConfigurator.restart") @mock.patch("certbot_apache.configurator.ApacheConfigurator.restart")
@mock.patch("certbot_apache.configurator.ApacheConfigurator.enable_mod") @mock.patch("certbot_apache.configurator.ApacheConfigurator.enable_mod")
@@ -31,7 +31,6 @@ from certbot_apache.tests import util
class MultipleVhostsTest(util.ApacheTest): class MultipleVhostsTest(util.ApacheTest):
"""Test two standard well-configured HTTP vhosts.""" """Test two standard well-configured HTTP vhosts."""
def setUp(self): # pylint: disable=arguments-differ def setUp(self): # pylint: disable=arguments-differ
super(MultipleVhostsTest, self).setUp() super(MultipleVhostsTest, self).setUp()
@@ -5,6 +5,8 @@ import subprocess
import zope.interface import zope.interface
from acme.magic_typing import Set # pylint: disable=unused-import, no-name-in-module
from certbot import configuration from certbot import configuration
from certbot_nginx import configurator from certbot_nginx import configurator
from certbot_nginx import constants from certbot_nginx import constants
@@ -68,7 +70,7 @@ def _get_server_root(config):
def _get_names(config): def _get_names(config):
"""Returns all and testable domain names in config""" """Returns all and testable domain names in config"""
all_names = set() all_names = set() # type: Set[str]
for root, _dirs, files in os.walk(config): for root, _dirs, files in os.walk(config):
for this_file in files: for this_file in files:
update_names = _get_server_names(root, this_file) update_names = _get_server_names(root, this_file)
@@ -76,6 +78,7 @@ def _get_names(config):
non_ip_names = set(n for n in all_names if not util.IP_REGEX.match(n)) non_ip_names = set(n for n in all_names if not util.IP_REGEX.match(n))
return all_names, non_ip_names return all_names, non_ip_names
def _get_server_names(root, filename): def _get_server_names(root, filename):
"""Returns all names in a config file path""" """Returns all names in a config file path"""
all_names = set() all_names = set()
@@ -83,5 +83,5 @@ class _GehirnLexiconClient(dns_common_lexicon.LexiconClient):
def _handle_http_error(self, e, domain_name): def _handle_http_error(self, e, domain_name):
if domain_name in str(e) and (str(e).startswith('404 Client Error: Not Found for url:')): if domain_name in str(e) and (str(e).startswith('404 Client Error: Not Found for url:')):
return # Expected errors when zone name guess is wrong return None # Expected errors when zone name guess is wrong
return super(_GehirnLexiconClient, self)._handle_http_error(e, domain_name) return super(_GehirnLexiconClient, self)._handle_http_error(e, domain_name)
@@ -73,4 +73,4 @@ class _LinodeLexiconClient(dns_common_lexicon.LexiconClient):
if not str(e).startswith('Domain not found'): if not str(e).startswith('Domain not found'):
return errors.PluginError('Unexpected error determining zone identifier for {0}: {1}' return errors.PluginError('Unexpected error determining zone identifier for {0}: {1}'
.format(domain_name, e)) .format(domain_name, e))
return None
@@ -86,5 +86,5 @@ class _SakuraCloudLexiconClient(dns_common_lexicon.LexiconClient):
def _handle_http_error(self, e, domain_name): def _handle_http_error(self, e, domain_name):
if domain_name in str(e) and (str(e).startswith('404 Client Error: Not Found for url:')): if domain_name in str(e) and (str(e).startswith('404 Client Error: Not Found for url:')):
return # Expected errors when zone name guess is wrong return None # Expected errors when zone name guess is wrong
return super(_SakuraCloudLexiconClient, self)._handle_http_error(e, domain_name) return super(_SakuraCloudLexiconClient, self)._handle_http_error(e, domain_name)
+1 -3
View File
@@ -166,7 +166,6 @@ class NginxConfigurator(common.Installer):
# Entry point in main.py for installing cert # Entry point in main.py for installing cert
def deploy_cert(self, domain, cert_path, key_path, def deploy_cert(self, domain, cert_path, key_path,
chain_path=None, fullchain_path=None): chain_path=None, fullchain_path=None):
# pylint: disable=unused-argument
"""Deploys certificate to specified virtual host. """Deploys certificate to specified virtual host.
.. note:: Aborts if the vhost is missing ssl_certificate or .. note:: Aborts if the vhost is missing ssl_certificate or
@@ -187,8 +186,7 @@ class NginxConfigurator(common.Installer):
for vhost in vhosts: for vhost in vhosts:
self._deploy_cert(vhost, cert_path, key_path, chain_path, fullchain_path) self._deploy_cert(vhost, cert_path, key_path, chain_path, fullchain_path)
def _deploy_cert(self, vhost, cert_path, key_path, chain_path, fullchain_path): def _deploy_cert(self, vhost, cert_path, key_path, chain_path, fullchain_path): # pylint: disable=unused-argument
# pylint: disable=unused-argument
""" """
Helper function for deploy_cert() that handles the actual deployment Helper function for deploy_cert() that handles the actual deployment
this exists because we might want to do multiple deployments per this exists because we might want to do multiple deployments per
+2 -1
View File
@@ -1,7 +1,8 @@
"""nginx plugin constants.""" """nginx plugin constants."""
import pkg_resources
import platform import platform
import pkg_resources
FREEBSD_DARWIN_SERVER_ROOT = "/usr/local/etc/nginx" FREEBSD_DARWIN_SERVER_ROOT = "/usr/local/etc/nginx"
LINUX_SERVER_ROOT = "/etc/nginx" LINUX_SERVER_ROOT = "/etc/nginx"
+2 -3
View File
@@ -83,8 +83,7 @@ class NginxParser(object):
""" """
if not os.path.isabs(path): if not os.path.isabs(path):
return os.path.normpath(os.path.join(self.root, path)) return os.path.normpath(os.path.join(self.root, path))
else: return os.path.normpath(path)
return os.path.normpath(path)
def _build_addr_to_ssl(self): def _build_addr_to_ssl(self):
"""Builds a map from address to whether it listens on ssl in any server block """Builds a map from address to whether it listens on ssl in any server block
@@ -395,7 +394,7 @@ class NginxParser(object):
addr.default = False addr.default = False
addr.ipv6only = False addr.ipv6only = False
for directive in enclosing_block[new_vhost.path[-1]][1]: for directive in enclosing_block[new_vhost.path[-1]][1]:
if len(directive) > 0 and directive[0] == 'listen': if directive and directive[0] == 'listen':
# Exclude one-time use parameters which will cause an error if repeated. # Exclude one-time use parameters which will cause an error if repeated.
# https://nginx.org/en/docs/http/ngx_http_core_module.html#listen # https://nginx.org/en/docs/http/ngx_http_core_module.html#listen
exclude = set(('default_server', 'default', 'setfib', 'fastopen', 'backlog', exclude = set(('default_server', 'default', 'setfib', 'fastopen', 'backlog',
+26 -22
View File
@@ -13,6 +13,7 @@ logger = logging.getLogger(__name__)
COMMENT = " managed by Certbot" COMMENT = " managed by Certbot"
COMMENT_BLOCK = ["#", COMMENT] COMMENT_BLOCK = ["#", COMMENT]
class Parsable(object): class Parsable(object):
""" Abstract base class for "Parsable" objects whose underlying representation """ Abstract base class for "Parsable" objects whose underlying representation
is a tree of lists. is a tree of lists.
@@ -112,6 +113,7 @@ class Parsable(object):
""" """
return [elem.dump(include_spaces) for elem in self._data] return [elem.dump(include_spaces) for elem in self._data]
class Statements(Parsable): class Statements(Parsable):
""" A group or list of "Statements". A Statement is either a Block or a Sentence. """ A group or list of "Statements". A Statement is either a Block or a Sentence.
@@ -142,24 +144,23 @@ class Statements(Parsable):
if self.parent is not None: if self.parent is not None:
self._trailing_whitespace = "\n" + self.parent.get_tabs() self._trailing_whitespace = "\n" + self.parent.get_tabs()
def parse(self, parse_this, add_spaces=False): def parse(self, raw_list, add_spaces=False):
""" Parses a list of statements. """ Parses a list of statements.
Expects all elements in `parse_this` to be parseable by `type(self).parsing_hooks`, Expects all elements in `raw_list` to be parseable by `type(self).parsing_hooks`,
with an optional whitespace string at the last index of `parse_this`. with an optional whitespace string at the last index of `raw_list`.
""" """
if not isinstance(parse_this, list): if not isinstance(raw_list, list):
raise errors.MisconfigurationError("Statements parsing expects a 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 there's a trailing whitespace in the list of statements, keep track of it.
if len(parse_this) > 0 and isinstance(parse_this[-1], six.string_types) \ if raw_list and isinstance(raw_list[-1], six.string_types) and raw_list[-1].isspace():
and parse_this[-1].isspace(): self._trailing_whitespace = raw_list[-1]
self._trailing_whitespace = parse_this[-1] raw_list = raw_list[:-1]
parse_this = parse_this[:-1] self._data = [parse_raw(elem, self, add_spaces) for elem in raw_list]
self._data = [parse_raw(elem, self, add_spaces) for elem in parse_this]
def get_tabs(self): def get_tabs(self):
""" Takes a guess at the tabbing of all contained Statements by retrieving the """ Takes a guess at the tabbing of all contained Statements by retrieving the
tabbing of the first Statement.""" tabbing of the first Statement."""
if len(self._data) > 0: if self._data:
return self._data[0].get_tabs() return self._data[0].get_tabs()
return "" return ""
@@ -179,6 +180,7 @@ class Statements(Parsable):
# ======== End overridden functions # ======== End overridden functions
def _space_list(list_): def _space_list(list_):
""" Inserts whitespace between adjacent non-whitespace tokens. """ """ Inserts whitespace between adjacent non-whitespace tokens. """
spaced_statement = [] # type: List[str] spaced_statement = [] # type: List[str]
@@ -188,6 +190,7 @@ def _space_list(list_):
spaced_statement.insert(0, " ") spaced_statement.insert(0, " ")
return spaced_statement return spaced_statement
class Sentence(Parsable): class Sentence(Parsable):
""" A list of words. Non-whitespace words are typically separated with whitespace tokens. """ """ A list of words. Non-whitespace words are typically separated with whitespace tokens. """
@@ -205,15 +208,15 @@ class Sentence(Parsable):
return isinstance(lists, list) and len(lists) > 0 and \ return isinstance(lists, list) and len(lists) > 0 and \
all([isinstance(elem, six.string_types) for elem in lists]) all([isinstance(elem, six.string_types) for elem in lists])
def parse(self, parse_this, add_spaces=False): def parse(self, raw_list, add_spaces=False):
""" Parses a list of string types into this object. """ Parses a list of string types into this object.
If add_spaces is set, adds whitespace tokens between adjacent non-whitespace tokens.""" If add_spaces is set, adds whitespace tokens between adjacent non-whitespace tokens."""
if add_spaces: if add_spaces:
parse_this = _space_list(parse_this) raw_list = _space_list(raw_list)
if not isinstance(parse_this, list) or \ if not isinstance(raw_list, list) or \
any([not isinstance(elem, six.string_types) for elem in parse_this]): any([not isinstance(elem, six.string_types) for elem in raw_list]):
raise errors.MisconfigurationError("Sentence parsing expects a list of string types.") raise errors.MisconfigurationError("Sentence parsing expects a list of string types.")
self._data = parse_this self._data = raw_list
def iterate(self, expanded=False, match=None): def iterate(self, expanded=False, match=None):
""" Simply yields itself. """ """ Simply yields itself. """
@@ -255,6 +258,7 @@ class Sentence(Parsable):
def __contains__(self, word): def __contains__(self, word):
return word in self.words return word in self.words
class Block(Parsable): class Block(Parsable):
""" Any sort of bloc, denoted by a block name and curly braces, like so: """ Any sort of bloc, denoted by a block name and curly braces, like so:
The parsed block: The parsed block:
@@ -297,26 +301,26 @@ class Block(Parsable):
for elem in self.contents.iterate(expanded, match): for elem in self.contents.iterate(expanded, match):
yield elem yield elem
def parse(self, parse_this, add_spaces=False): def parse(self, raw_list, add_spaces=False):
""" Parses a list that resembles a block. """ Parses a list that resembles a block.
The assumptions that this routine makes are: The assumptions that this routine makes are:
1. the first element of `parse_this` is a valid Sentence. 1. the first element of `raw_list` is a valid Sentence.
2. the second element of `parse_this` is a valid Statement. 2. the second element of `raw_list` is a valid Statement.
If add_spaces is set, we call it recursively on `names` and `contents`, and If add_spaces is set, we call it recursively on `names` and `contents`, and
add an extra trailing space to `names` (to separate the block's opening bracket add an extra trailing space to `names` (to separate the block's opening bracket
and the block name). and the block name).
""" """
if not Block.should_parse(parse_this): if not Block.should_parse(raw_list):
raise errors.MisconfigurationError("Block parsing expects a list of length 2. " raise errors.MisconfigurationError("Block parsing expects a list of length 2. "
"First element should be a list of string types (the bloc names), " "First element should be a list of string types (the bloc names), "
"and second should be another list of statements (the bloc content).") "and second should be another list of statements (the bloc content).")
self.names = Sentence(self) self.names = Sentence(self)
if add_spaces: if add_spaces:
parse_this[0].append(" ") raw_list[0].append(" ")
self.names.parse(parse_this[0], add_spaces) self.names.parse(raw_list[0], add_spaces)
self.contents = Statements(self) self.contents = Statements(self)
self.contents.parse(parse_this[1], add_spaces) self.contents.parse(raw_list[1], add_spaces)
self._data = [self.names, self.contents] self._data = [self.names, self.contents]
def get_tabs(self): def get_tabs(self):
+3 -5
View File
@@ -1,21 +1,19 @@
"""Common utilities for certbot_nginx.""" """Common utilities for certbot_nginx."""
import copy import copy
import os import os
import pkg_resources import shutil
import tempfile import tempfile
import unittest import unittest
import shutil
import warnings import warnings
import josepy as jose import josepy as jose
import mock import mock
import pkg_resources
import zope.component import zope.component
from certbot import configuration from certbot import configuration
from certbot.tests import util as test_util
from certbot.plugins import common from certbot.plugins import common
from certbot.tests import util as test_util
from certbot_nginx import configurator from certbot_nginx import configurator
from certbot_nginx import nginxparser from certbot_nginx import nginxparser
+2 -5
View File
@@ -191,7 +191,7 @@ class Installer(plugins_common.Installer):
"subset of configuration parameters.") "subset of configuration parameters.")
def deploy_cert(self, domain, cert_path, def deploy_cert(self, domain, cert_path,
key_path, chain_path, fullchain_path): key_path, chain_path, fullchain_path): # pylint: disable=unused-argument
"""Configure the Postfix SMTP server to use the given TLS cert. """Configure the Postfix SMTP server to use the given TLS cert.
:param str domain: domain to deploy certificate file :param str domain: domain to deploy certificate file
@@ -204,7 +204,6 @@ class Installer(plugins_common.Installer):
:raises .PluginError: when cert cannot be deployed :raises .PluginError: when cert cannot be deployed
""" """
# pylint: disable=unused-argument
if self._tls_enabled: if self._tls_enabled:
return return
self._tls_enabled = True self._tls_enabled = True
@@ -223,10 +222,9 @@ class Installer(plugins_common.Installer):
self.postconf.set("smtpd_tls_dh1024_param_file", self.ssl_dhparams) self.postconf.set("smtpd_tls_dh1024_param_file", self.ssl_dhparams)
self._confirm_changes() self._confirm_changes()
def enhance(self, domain, enhancement, options=None): def enhance(self, domain, enhancement, options=None): # pylint: disable=unused-argument
"""Raises an exception since this installer doesn't support any enhancements. """Raises an exception since this installer doesn't support any enhancements.
""" """
# pylint: disable=unused-argument
raise errors.PluginError( raise errors.PluginError(
"Unsupported enhancement: {0}".format(enhancement)) "Unsupported enhancement: {0}".format(enhancement))
@@ -285,4 +283,3 @@ class Installer(plugins_common.Installer):
:raises .PluginError: when server cannot be restarted :raises .PluginError: when server cannot be restarted
""" """
self.postfix.restart() self.postfix.restart()
+9 -9
View File
@@ -8,6 +8,7 @@ from certbot_postfix import util
from acme.magic_typing import Dict, List, Tuple from acme.magic_typing import Dict, List, Tuple
# pylint: enable=unused-import, no-name-in-module # pylint: enable=unused-import, no-name-in-module
class ConfigMain(util.PostfixUtilBase): class ConfigMain(util.PostfixUtilBase):
"""A parser for Postfix's main.cf file.""" """A parser for Postfix's main.cf file."""
@@ -91,18 +92,18 @@ class ConfigMain(util.PostfixUtilBase):
with a value in acceptable_overrides. with a value in acceptable_overrides.
""" """
if name not in self._db: if name not in self._db:
raise KeyError("Parameter name %s is not a valid Postfix parameter name.", name) raise KeyError("Parameter name {0} is not a valid Postfix parameter name.".format(name))
# Check to see if this parameter is overridden by master. # Check to see if this parameter is overridden by master.
overrides = self.get_master_overrides(name) overrides = self.get_master_overrides(name)
if not self._ignore_master_overrides and overrides is not None: if not self._ignore_master_overrides and overrides is not None:
util.report_master_overrides(name, overrides, acceptable_overrides) util.report_master_overrides(name, overrides, acceptable_overrides)
if value != self._db[name]: if value != self._db[name]:
# _db contains the "original" state of parameters. We only care about # _db contains the "original" state of parameters. We only care about
# writes if they cause a delta from the original state. # writes if they cause a delta from the original state.
self._updated[name] = value self._updated[name] = value
elif name in self._updated: elif name in self._updated:
# If this write reverts a previously updated parameter back to the # If this write reverts a previously updated parameter back to the
# original DB's state, we don't have to keep track of it in _updated. # original DB's state, we don't have to keep track of it in _updated.
del self._updated[name] del self._updated[name]
def flush(self): def flush(self):
@@ -110,7 +111,7 @@ class ConfigMain(util.PostfixUtilBase):
:raises error.PluginError: When flush to main.cf fails for some reason. :raises error.PluginError: When flush to main.cf fails for some reason.
""" """
if len(self._updated) == 0: if not self._updated:
return return
args = ['-e'] args = ['-e']
for name, value in six.iteritems(self._updated): for name, value in six.iteritems(self._updated):
@@ -118,7 +119,7 @@ class ConfigMain(util.PostfixUtilBase):
try: try:
self._get_output(args) self._get_output(args)
except IOError as e: except IOError as e:
raise errors.PluginError("Unable to save to Postfix config: %v", e) raise errors.PluginError("Unable to save to Postfix config: {0}".format(e))
for name, value in six.iteritems(self._updated): for name, value in six.iteritems(self._updated):
self._db[name] = value self._db[name] = value
self._updated = {} self._updated = {}
@@ -130,6 +131,7 @@ class ConfigMain(util.PostfixUtilBase):
""" """
return self._updated return self._updated
def _parse_main_output(output): def _parse_main_output(output):
"""Parses the raw output from Postconf about main.cf. """Parses the raw output from Postconf about main.cf.
@@ -148,5 +150,3 @@ def _parse_main_output(output):
for line in output.splitlines(): for line in output.splitlines():
name, _, value = line.partition(" =") name, _, value = line.partition(" =")
yield name, value.strip() yield name, value.strip()
@@ -1,20 +1,18 @@
"""Tests for certbot_postfix.installer.""" """Tests for certbot_postfix.installer."""
from contextlib import contextmanager
import copy import copy
import functools import functools
import os import os
import pkg_resources
import six
import unittest import unittest
from contextlib import contextmanager
import mock import mock
import pkg_resources
import six
from acme.magic_typing import Dict, Tuple # pylint: disable=unused-import, no-name-in-module
from certbot import errors from certbot import errors
from certbot.tests import util as certbot_test_util from certbot.tests import util as certbot_test_util
# pylint: disable=unused-import, no-name-in-module
from acme.magic_typing import Dict, Tuple, Union
# pylint: enable=unused-import, no-name-in-module
DEFAULT_MAIN_CF = { DEFAULT_MAIN_CF = {
"smtpd_tls_cert_file": "", "smtpd_tls_cert_file": "",
@@ -127,7 +125,7 @@ class InstallerTest(certbot_test_util.ConfigTestCase):
with create_installer(self.config) as installer: with create_installer(self.config) as installer:
installer.prepare() installer.prepare()
installer.restart() installer.restart()
self.assertEqual(installer.postfix.restart.call_count, 1) self.assertEqual(installer.postfix.restart.call_count, 1) # pylint: disable=no-member
def test_add_parser_arguments(self): def test_add_parser_arguments(self):
options = set(("ctl", "config-dir", "config-utility", options = set(("ctl", "config-dir", "config-utility",
@@ -269,7 +267,7 @@ class InstallerTest(certbot_test_util.ConfigTestCase):
installer.prepare() installer.prepare()
installer.deploy_cert("example.com", "cert_path", "key_path", installer.deploy_cert("example.com", "cert_path", "key_path",
"chain_path", "fullchain_path") "chain_path", "fullchain_path")
for param in more_secure.keys(): for param in more_secure:
self.assertFalse(param in installer.postconf.get_changes()) self.assertFalse(param in installer.postconf.get_changes())
def test_enhance(self): def test_enhance(self):
@@ -284,16 +282,20 @@ class InstallerTest(certbot_test_util.ConfigTestCase):
installer.prepare() installer.prepare()
self.assertEqual(installer.supported_enhancements(), []) self.assertEqual(installer.supported_enhancements(), [])
@contextmanager @contextmanager
def create_installer(config, main_cf=DEFAULT_MAIN_CF): def create_installer(config, main_cf=None):
# pylint: disable=dangerous-default-value
"""Creates a Postfix installer with calls to `postconf` and `postfix` mocked out. """Creates a Postfix installer with calls to `postconf` and `postfix` mocked out.
In particular, creates a ConfigMain object that does regular things, but seeds it In particular, creates a ConfigMain object that does regular things, but seeds it
with values from `main_cf` and `master_cf` dicts. with values from `main_cf` and `master_cf` dicts.
""" """
if main_cf is None:
main_cf = DEFAULT_MAIN_CF
from certbot_postfix.postconf import ConfigMain from certbot_postfix.postconf import ConfigMain
from certbot_postfix import installer from certbot_postfix import installer
def _mock_init_postconf(postconf, executable, ignore_master_overrides=False, config_dir=None): def _mock_init_postconf(postconf, executable, ignore_master_overrides=False, config_dir=None):
# pylint: disable=protected-access,unused-argument # pylint: disable=protected-access,unused-argument
postconf._ignore_master_overrides = ignore_master_overrides postconf._ignore_master_overrides = ignore_master_overrides
@@ -309,6 +311,6 @@ def create_installer(config, main_cf=DEFAULT_MAIN_CF):
return_value=mock.Mock()): return_value=mock.Mock()):
yield installer.Installer(config, "postfix") yield installer.Installer(config, "postfix")
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() # pragma: no cover unittest.main() # pragma: no cover
@@ -1,10 +1,12 @@
"""Tests for certbot_postfix.postconf.""" """Tests for certbot_postfix.postconf."""
import mock
import unittest import unittest
import mock
from certbot import errors from certbot import errors
class PostConfTest(unittest.TestCase): class PostConfTest(unittest.TestCase):
"""Tests for certbot_postfix.util.PostConf.""" """Tests for certbot_postfix.util.PostConf."""
def setUp(self): def setUp(self):
+1 -1
View File
@@ -12,6 +12,7 @@ logger = logging.getLogger(__name__)
COMMAND = "postfix" COMMAND = "postfix"
class PostfixUtilBase(object): class PostfixUtilBase(object):
"""A base class for wrapping Postfix command line utilities.""" """A base class for wrapping Postfix command line utilities."""
@@ -289,4 +290,3 @@ def _has_acceptable_tls_versions(parameter_string):
if "!" + bad_version not in supported_version_list: if "!" + bad_version not in supported_version_list:
return False return False
return True return True
+2 -3
View File
@@ -110,7 +110,6 @@ class AccountMemoryStorage(interfaces.AccountStorage):
return list(six.itervalues(self.accounts)) return list(six.itervalues(self.accounts))
def save(self, account, client): def save(self, account, client):
# pylint: disable=unused-argument
if account.id in self.accounts: if account.id in self.accounts:
logger.debug("Overwriting account: %s", account.id) logger.debug("Overwriting account: %s", account.id)
self.accounts[account.id] = account self.accounts[account.id] = account
@@ -244,8 +243,8 @@ class AccountFileStorage(interfaces.AccountStorage):
def load(self, account_id): def load(self, account_id):
return self._load_for_server_path(account_id, self.config.server_path) return self._load_for_server_path(account_id, self.config.server_path)
def save(self, account, acme): def save(self, account, client):
self._save(account, acme, regr_only=False) self._save(account, client, regr_only=False)
def save_regr(self, account, acme): def save_regr(self, account, acme):
"""Save the registration resource. """Save the registration resource.
+3 -4
View File
@@ -312,9 +312,8 @@ def config_help(name, hidden=False):
# pylint: disable=no-member # pylint: disable=no-member
if hidden: if hidden:
return argparse.SUPPRESS return argparse.SUPPRESS
else: field = interfaces.IConfig.__getitem__(name) # type: zope.interface.interface.Attribute # pylint: disable=no-value-for-parameter
field = interfaces.IConfig.__getitem__(name) # type: zope.interface.interface.Attribute return field.__doc__
return field.__doc__
class HelpfulArgumentGroup(object): class HelpfulArgumentGroup(object):
@@ -568,7 +567,7 @@ class HelpfulArgumentParser(object):
apache_doc = "(the certbot apache plugin is not installed)" apache_doc = "(the certbot apache plugin is not installed)"
usage = SHORT_USAGE usage = SHORT_USAGE
if help_arg == True: if help_arg is True:
self.notify(usage + COMMAND_OVERVIEW % (apache_doc, nginx_doc) + HELP_AND_VERSION_USAGE) self.notify(usage + COMMAND_OVERVIEW % (apache_doc, nginx_doc) + HELP_AND_VERSION_USAGE)
sys.exit(0) sys.exit(0)
elif help_arg in self.COMMANDS_TOPICS: elif help_arg in self.COMMANDS_TOPICS:
+1 -1
View File
@@ -3,7 +3,7 @@ import copy
import os import os
import zope.interface import zope.interface
from six.moves.urllib import parse # pylint: disable=import-error from six.moves.urllib import parse # pylint: disable=relative-import
from certbot import constants from certbot import constants
from certbot import errors from certbot import errors
+4 -8
View File
@@ -122,10 +122,9 @@ class FileDisplay(object):
else: else:
logger.debug("Not pausing for user confirmation") logger.debug("Not pausing for user confirmation")
def menu(self, message, choices, ok_label=None, cancel_label=None, def menu(self, message, choices, ok_label=None, cancel_label=None, # pylint: disable=unused-argument
help_label=None, default=None, help_label=None, default=None, # pylint: disable=unused-argument
cli_flag=None, force_interactive=False, **unused_kwargs): cli_flag=None, force_interactive=False, **unused_kwargs):
# pylint: disable=unused-argument
"""Display a menu. """Display a menu.
.. todo:: This doesn't enable the help label/button (I wasn't sold on .. todo:: This doesn't enable the help label/button (I wasn't sold on
@@ -227,7 +226,6 @@ class FileDisplay(object):
def checklist(self, message, tags, default=None, def checklist(self, message, tags, default=None,
cli_flag=None, force_interactive=False, **unused_kwargs): cli_flag=None, force_interactive=False, **unused_kwargs):
# pylint: disable=unused-argument
"""Display a checklist. """Display a checklist.
:param str message: Message to display to user :param str message: Message to display to user
@@ -467,8 +465,7 @@ class NoninteractiveDisplay(object):
msg += "\n\n(You can set this with the {0} flag)".format(cli_flag) msg += "\n\n(You can set this with the {0} flag)".format(cli_flag)
raise errors.MissingCommandlineFlag(msg) raise errors.MissingCommandlineFlag(msg)
def notification(self, message, pause=False, wrap=True, **unused_kwargs): def notification(self, message, pause=False, wrap=True, **unused_kwargs): # pylint: disable=unused-argument
# pylint: disable=unused-argument
"""Displays a notification without waiting for user acceptance. """Displays a notification without waiting for user acceptance.
:param str message: Message to display to stdout :param str message: Message to display to stdout
@@ -523,9 +520,8 @@ class NoninteractiveDisplay(object):
self._interaction_fail(message, cli_flag) self._interaction_fail(message, cli_flag)
return OK, default return OK, default
def yesno(self, message, yes_label=None, no_label=None, def yesno(self, message, yes_label=None, no_label=None, # pylint: disable=unused-argument
default=None, cli_flag=None, **unused_kwargs): default=None, cli_flag=None, **unused_kwargs):
# pylint: disable=unused-argument
"""Decide Yes or No, without asking anybody """Decide Yes or No, without asking anybody
:param str message: question for the user :param str message: question for the user
+16 -7
View File
@@ -176,7 +176,6 @@ def _handle_subset_cert_request(config, domains, cert):
raise errors.Error(USER_CANCELLED) raise errors.Error(USER_CANCELLED)
# pylint: disable=inconsistent-return-statements
def _handle_identical_cert_request(config, lineage): def _handle_identical_cert_request(config, lineage):
"""Figure out what to do if a lineage has the same names as a previously obtained one """Figure out what to do if a lineage has the same names as a previously obtained one
@@ -224,7 +223,8 @@ def _handle_identical_cert_request(config, lineage):
return "reinstall", lineage return "reinstall", lineage
elif response[1] == 1: elif response[1] == 1:
return "renew", lineage return "renew", lineage
assert False, "This is imporssible" raise AssertionError('This is impossible')
def _find_lineage_for_domains(config, domains): def _find_lineage_for_domains(config, domains):
"""Determine whether there are duplicated names and how to handle """Determine whether there are duplicated names and how to handle
@@ -501,6 +501,7 @@ def _determine_account(config):
raise errors.Error( raise errors.Error(
"Registration cannot proceed without accepting " "Registration cannot proceed without accepting "
"Terms of Service.") "Terms of Service.")
return None
account_storage = account.AccountFileStorage(config) account_storage = account.AccountFileStorage(config)
acme = None acme = None
@@ -679,7 +680,7 @@ def register(config, unused_plugins):
account_storage = account.AccountFileStorage(config) account_storage = account.AccountFileStorage(config)
accounts = account_storage.find_all() accounts = account_storage.find_all()
if len(accounts) > 0: if accounts:
# TODO: add a flag to register a duplicate account (this will # TODO: add a flag to register a duplicate account (this will
# also require extending _determine_account's behavior # also require extending _determine_account's behavior
# or else extracting the registration code from there) # or else extracting the registration code from there)
@@ -688,7 +689,7 @@ def register(config, unused_plugins):
"unsupported.") "unsupported.")
# _determine_account will register an account # _determine_account will register an account
_determine_account(config) _determine_account(config)
return return None
def update_account(config, unused_plugins): def update_account(config, unused_plugins):
@@ -711,7 +712,7 @@ def update_account(config, unused_plugins):
reporter_util = zope.component.getUtility(interfaces.IReporter) reporter_util = zope.component.getUtility(interfaces.IReporter)
add_msg = lambda m: reporter_util.add_message(m, reporter_util.MEDIUM_PRIORITY) add_msg = lambda m: reporter_util.add_message(m, reporter_util.MEDIUM_PRIORITY)
if len(accounts) == 0: if not accounts:
return "Could not find an existing account to update." return "Could not find an existing account to update."
if config.email is None: if config.email is None:
if config.register_unsafely_without_email: if config.register_unsafely_without_email:
@@ -762,6 +763,7 @@ def _install_cert(config, le_client, domains, lineage=None):
path_provider.cert_path, path_provider.chain_path, path_provider.fullchain_path) path_provider.cert_path, path_provider.chain_path, path_provider.fullchain_path)
le_client.enhance_config(domains, path_provider.chain_path) le_client.enhance_config(domains, path_provider.chain_path)
def install(config, plugins): def install(config, plugins):
"""Install a previously obtained cert in a server. """Install a previously obtained cert in a server.
@@ -813,13 +815,14 @@ def install(config, plugins):
raise errors.ConfigurationError("Path to certificate or key was not defined. " raise errors.ConfigurationError("Path to certificate or key was not defined. "
"If your certificate is managed by Certbot, please use --cert-name " "If your certificate is managed by Certbot, please use --cert-name "
"to define which certificate you would like to install.") "to define which certificate you would like to install.")
return None
if enhancements.are_requested(config): if enhancements.are_requested(config):
# In the case where we don't have certname, we have errored out already # In the case where we don't have certname, we have errored out already
lineage = cert_manager.lineage_for_certname(config, config.certname) lineage = cert_manager.lineage_for_certname(config, config.certname)
enhancements.enable(lineage, domains, installer, config) enhancements.enable(lineage, domains, installer, config)
return None
def _populate_from_certname(config): def _populate_from_certname(config):
"""Helper function for install to populate missing config values from lineage """Helper function for install to populate missing config values from lineage
defined by --cert-name.""" defined by --cert-name."""
@@ -882,6 +885,7 @@ def plugins_cmd(config, plugins):
logger.debug("Prepared plugins: %s", available) logger.debug("Prepared plugins: %s", available)
notify(str(available)) notify(str(available))
def enhance(config, plugins): def enhance(config, plugins):
"""Add security enhancements to existing configuration """Add security enhancements to existing configuration
@@ -938,6 +942,8 @@ def enhance(config, plugins):
if enhancements.are_requested(config): if enhancements.are_requested(config):
enhancements.enable(lineage, domains, installer, config) enhancements.enable(lineage, domains, installer, config)
return None
def rollback(config, plugins): def rollback(config, plugins):
"""Rollback server configuration changes made during install. """Rollback server configuration changes made during install.
@@ -1041,7 +1047,8 @@ def certificates(config, unused_plugins):
""" """
cert_manager.certificates(config) cert_manager.certificates(config)
def revoke(config, unused_plugins): # TODO: coop with renewal config # TODO: coop with renewal config
def revoke(config, unused_plugins):
"""Revoke a previously obtained certificate. """Revoke a previously obtained certificate.
:param config: Configuration object :param config: Configuration object
@@ -1334,6 +1341,8 @@ def main(cli_args=None):
:raises errors.Error: error if plugin command is not supported :raises errors.Error: error if plugin command is not supported
""" """
if not cli_args:
cli_args = sys.argv[1:]
log.pre_arg_parse_setup() log.pre_arg_parse_setup()
+3 -5
View File
@@ -69,8 +69,7 @@ class RevocationChecker(object):
if self.use_openssl_binary: if self.use_openssl_binary:
return self._check_ocsp_openssl_bin(cert_path, chain_path, host, url) return self._check_ocsp_openssl_bin(cert_path, chain_path, host, url)
else: return _check_ocsp_cryptography(cert_path, chain_path, url)
return _check_ocsp_cryptography(cert_path, chain_path, url)
def _check_ocsp_openssl_bin(self, cert_path, chain_path, host, url): def _check_ocsp_openssl_bin(self, cert_path, chain_path, host, url):
# type: (str, str, str, str) -> bool # type: (str, str, str, str) -> bool
@@ -121,9 +120,8 @@ def _determine_ocsp_server(cert_path):
if host: if host:
return url, host return url, host
else: logger.info("Cannot process OCSP host from URL (%s) in cert at %s", url, cert_path)
logger.info("Cannot process OCSP host from URL (%s) in cert at %s", url, cert_path) return None, None
return None, None
def _check_ocsp_cryptography(cert_path, chain_path, url): def _check_ocsp_cryptography(cert_path, chain_path, url):
+1 -1
View File
@@ -37,7 +37,7 @@ class DNSAuthenticator(common.Plugin):
help='The number of seconds to wait for DNS to propagate before asking the ACME server ' help='The number of seconds to wait for DNS to propagate before asking the ACME server '
'to verify the DNS record.') 'to verify the DNS record.')
def get_chall_pref(self, unused_domain): # pylint: disable=missing-docstring,no-self-use def get_chall_pref(self, unused_domain): # pylint: disable=missing-docstring,no-self-use
return [challenges.DNS01] return [challenges.DNS01]
def prepare(self): # pylint: disable=missing-docstring def prepare(self): # pylint: disable=missing-docstring
+1
View File
@@ -110,6 +110,7 @@ class LexiconClient(object):
if not str(e).startswith('No domain found'): if not str(e).startswith('No domain found'):
return errors.PluginError('Unexpected error determining zone identifier for {0}: {1}' return errors.PluginError('Unexpected error determining zone identifier for {0}: {1}'
.format(domain_name, e)) .format(domain_name, e))
return None
def build_lexicon_config(lexicon_provider_name, lexicon_options, provider_options): def build_lexicon_config(lexicon_provider_name, lexicon_options, provider_options):
+1 -1
View File
@@ -1,10 +1,10 @@
"""Tests for certbot.plugins.manual""" """Tests for certbot.plugins.manual"""
import os import os
import unittest import unittest
import sys
import six import six
import mock import mock
import sys
from acme import challenges from acme import challenges
+2 -2
View File
@@ -6,10 +6,10 @@ import unittest
import mock import mock
import zope.component import zope.component
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
from certbot import errors from certbot import errors
from certbot import interfaces from certbot import interfaces
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
from certbot.display import util as display_util from certbot.display import util as display_util
from certbot.plugins.disco import PluginsRegistry from certbot.plugins.disco import PluginsRegistry
from certbot.tests import util as test_util from certbot.tests import util as test_util
+1 -1
View File
@@ -1,8 +1,8 @@
"""Tests for certbot.plugins.standalone.""" """Tests for certbot.plugins.standalone."""
import socket import socket
import unittest
# https://github.com/python/typeshed/blob/master/stdlib/2and3/socket.pyi # https://github.com/python/typeshed/blob/master/stdlib/2and3/socket.pyi
from socket import errno as socket_errors # type: ignore from socket import errno as socket_errors # type: ignore
import unittest
import josepy as jose import josepy as jose
import mock import mock
+4 -5
View File
@@ -50,8 +50,7 @@ def path_surgery(cmd):
if util.exe_exists(cmd): if util.exe_exists(cmd):
return True return True
else: expanded = " expanded" if any(added) else ""
expanded = " expanded" if any(added) else "" logger.debug("Failed to find executable %s in%s PATH: %s", cmd,
logger.debug("Failed to find executable %s in%s PATH: %s", cmd, expanded, path)
expanded, path) return False
return False
+1 -1
View File
@@ -225,7 +225,7 @@ to serve all files under specified web root ({0})."""
self.performed[root_path].remove(achall) self.performed[root_path].remove(achall)
not_removed = [] # type: List[str] not_removed = [] # type: List[str]
while len(self._created_dirs) > 0: while self._created_dirs:
path = self._created_dirs.pop() path = self._created_dirs.pop()
try: try:
os.rmdir(path) os.rmdir(path)
+3 -3
View File
@@ -317,7 +317,7 @@ class Reverter(object):
# It is strongly advised to set newline = '' on Python 3 with CSV, # It is strongly advised to set newline = '' on Python 3 with CSV,
# and it fixes problems on Windows. # and it fixes problems on Windows.
kwargs = {'newline': ''} if sys.version_info[0] > 2 else {} kwargs = {'newline': ''} if sys.version_info[0] > 2 else {}
with open(filepath, 'r', **kwargs) as csvfile: # type: ignore # pylint: disable=star-args with open(filepath, 'r', **kwargs) as csvfile: # type: ignore
csvreader = csv.reader(csvfile) csvreader = csv.reader(csvfile)
for command in reversed(list(csvreader)): for command in reversed(list(csvreader)):
try: try:
@@ -418,9 +418,9 @@ class Reverter(object):
kwargs = {'newline': ''} if sys.version_info[0] > 2 else {} kwargs = {'newline': ''} if sys.version_info[0] > 2 else {}
try: try:
if os.path.isfile(commands_fp): if os.path.isfile(commands_fp):
command_file = open(commands_fp, "a", **kwargs) # type: ignore # pylint: disable=star-args command_file = open(commands_fp, "a", **kwargs) # type: ignore
else: else:
command_file = open(commands_fp, "w", **kwargs) # type: ignore # pylint: disable=star-args command_file = open(commands_fp, "w", **kwargs) # type: ignore
csvwriter = csv.writer(command_file) csvwriter = csv.writer(command_file)
csvwriter.writerow(command) csvwriter.writerow(command)
-1
View File
@@ -6,7 +6,6 @@ import os
import re import re
import shutil import shutil
import stat import stat
import shutil
import configobj import configobj
import parsedatetime import parsedatetime
+3 -3
View File
@@ -457,17 +457,17 @@ class ReportFailedAuthzrsTest(unittest.TestCase):
# Prevent future regressions if the error type changes # Prevent future regressions if the error type changes
self.assertTrue(kwargs["error"].description is not None) self.assertTrue(kwargs["error"].description is not None)
http_01 = messages.ChallengeBody(**kwargs) # pylint: disable=star-args http_01 = messages.ChallengeBody(**kwargs)
kwargs["chall"] = acme_util.TLSSNI01 kwargs["chall"] = acme_util.TLSSNI01
tls_sni_01 = messages.ChallengeBody(**kwargs) # pylint: disable=star-args tls_sni_01 = messages.ChallengeBody(**kwargs)
self.authzr1 = mock.MagicMock() self.authzr1 = mock.MagicMock()
self.authzr1.body.identifier.value = 'example.com' self.authzr1.body.identifier.value = 'example.com'
self.authzr1.body.challenges = [http_01, tls_sni_01] self.authzr1.body.challenges = [http_01, tls_sni_01]
kwargs["error"] = messages.Error(typ="dnssec", detail="detail") kwargs["error"] = messages.Error(typ="dnssec", detail="detail")
tls_sni_01_diff = messages.ChallengeBody(**kwargs) # pylint: disable=star-args tls_sni_01_diff = messages.ChallengeBody(**kwargs)
self.authzr2 = mock.MagicMock() self.authzr2 = mock.MagicMock()
self.authzr2.body.identifier.value = 'foo.bar' self.authzr2.body.identifier.value = 'foo.bar'
+1 -1
View File
@@ -229,7 +229,7 @@ class CertificatesTest(BaseCertManagerTest):
# pylint: disable=protected-access # pylint: disable=protected-access
out = get_report() out = get_report()
self.assertTrue('1 hour(s)' in out or '2 hour(s)' in out) self.assertTrue('1 hour(s)' in out or '2 hour(s)' in out)
self.assertTrue('VALID' in out and not 'INVALID' in out) self.assertTrue('VALID' in out and 'INVALID' not in out)
cert.target_expiry += datetime.timedelta(days=1) cert.target_expiry += datetime.timedelta(days=1)
# pylint: disable=protected-access # pylint: disable=protected-access
+3 -4
View File
@@ -7,14 +7,13 @@ import unittest
import mock import mock
from josepy import interfaces
import certbot.tests.util as test_util
from certbot import account from certbot import account
from certbot import errors from certbot import errors
from certbot import util from certbot import util
import certbot.tests.util as test_util
from josepy import interfaces
KEY = test_util.load_vector("rsa512_key.pem") KEY = test_util.load_vector("rsa512_key.pem")
CSR_SAN = test_util.load_vector("csr-san_512.pem") CSR_SAN = test_util.load_vector("csr-san_512.pem")
+5 -3
View File
@@ -1,9 +1,9 @@
"""Test certbot.display.completer.""" """Test certbot.display.completer."""
import os import os
try: try:
import readline # pylint: disable=import-error import readline # pylint: disable=import-error
except ImportError: except ImportError:
import certbot.display.dummy_readline as readline # type: ignore import certbot.display.dummy_readline as readline # type: ignore
import string import string
import sys import sys
import unittest import unittest
@@ -12,7 +12,8 @@ import mock
from six.moves import reload_module # pylint: disable=import-error from six.moves import reload_module # pylint: disable=import-error
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
import certbot.tests.util as test_util import certbot.tests.util as test_util # pylint: disable=ungrouped-imports
class CompleterTest(test_util.TempDirTestCase): class CompleterTest(test_util.TempDirTestCase):
"""Test certbot.display.completer.Completer.""" """Test certbot.display.completer.Completer."""
@@ -100,5 +101,6 @@ def enable_tab_completion(unused_command):
command = 'bind ^I rl_complete' if libedit else 'tab: complete' command = 'bind ^I rl_complete' if libedit else 'tab: complete'
readline.parse_and_bind(command) readline.parse_and_bind(command)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() # pragma: no cover unittest.main() # pragma: no cover
+1 -1
View File
@@ -312,7 +312,7 @@ class FileOutputDisplayTest(unittest.TestCase):
def test_methods_take_force_interactive(self): def test_methods_take_force_interactive(self):
# Every IDisplay method implemented by FileDisplay must take # Every IDisplay method implemented by FileDisplay must take
# force_interactive to prevent workflow regressions. # force_interactive to prevent workflow regressions.
for name in interfaces.IDisplay.names(): # pylint: disable=no-member for name in interfaces.IDisplay.names(): # pylint: disable=no-member,no-value-for-parameter
if six.PY2: if six.PY2:
getargspec = inspect.getargspec # pylint: disable=no-member getargspec = inspect.getargspec # pylint: disable=no-member
else: else:
-3
View File
@@ -148,9 +148,6 @@ class ExitHandlerTest(ErrorHandlerTest):
**self.init_kwargs) **self.init_kwargs)
func.assert_called_once_with() func.assert_called_once_with()
def test_bad_recovery_with_signal(self):
super(ExitHandlerTest, self).test_bad_recovery_with_signal()
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() # pragma: no cover unittest.main() # pragma: no cover
+5 -6
View File
@@ -153,8 +153,7 @@ class CertonlyTest(unittest.TestCase):
mock_find_cert.return_value = False, None mock_find_cert.return_value = False, None
self._call('certonly --webroot -d example.com'.split()) self._call('certonly --webroot -d example.com'.split())
def _assert_no_pause(self, message, pause=True): def _assert_no_pause(self, message, pause=True): # pylint: disable=unused-argument
# pylint: disable=unused-argument
self.assertFalse(pause) self.assertFalse(pause)
@mock.patch('certbot.cert_manager.lineage_for_certname') @mock.patch('certbot.cert_manager.lineage_for_certname')
@@ -533,14 +532,14 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
if mockisfile: if mockisfile:
orig_open = os.path.isfile orig_open = os.path.isfile
def mock_isfile(fn, *args, **kwargs): # pylint: disable=unused-argument def mock_isfile(fn, *args, **kwargs): # pylint: disable=unused-argument
"""Mock os.path.isfile()""" """Mock os.path.isfile()"""
if (fn.endswith("cert") or if (fn.endswith("cert") or
fn.endswith("chain") or fn.endswith("chain") or
fn.endswith("privkey")): fn.endswith("privkey")):
return True return True
else: return orig_open(fn)
return orig_open(fn)
with mock.patch("os.path.isfile") as mock_if: with mock.patch("os.path.isfile") as mock_if:
mock_if.side_effect = mock_isfile mock_if.side_effect = mock_isfile
+7 -7
View File
@@ -7,26 +7,26 @@ import logging
import os import os
import shutil import shutil
import stat import stat
import sys
import tempfile import tempfile
import unittest import unittest
import sys
from multiprocessing import Process, Event from multiprocessing import Process, Event
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
import mock
import OpenSSL import OpenSSL
import josepy as jose import josepy as jose
import mock
import pkg_resources
import six import six
from six.moves import reload_module # pylint: disable=import-error from six.moves import reload_module # pylint: disable=import-error
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from certbot import configuration
from certbot import constants from certbot import constants
from certbot import interfaces from certbot import interfaces
from certbot import storage
from certbot import configuration
from certbot import lock from certbot import lock
from certbot import storage
from certbot import util from certbot import util
from certbot.display import util as display_util from certbot.display import util as display_util
+4 -4
View File
@@ -312,10 +312,10 @@ class SafelyRemoveTest(test_util.TempDirTestCase):
# no error, yay! # no error, yay!
self.assertFalse(os.path.exists(self.path)) self.assertFalse(os.path.exists(self.path))
@mock.patch("certbot.util.os.remove") def test_other_error_passthrough(self):
def test_other_error_passthrough(self, mock_remove): with mock.patch("certbot.util.os.remove") as mock_remove:
mock_remove.side_effect = OSError mock_remove.side_effect = OSError
self.assertRaises(OSError, self._call) self.assertRaises(OSError, self._call)
class SafeEmailTest(unittest.TestCase): class SafeEmailTest(unittest.TestCase):
-2
View File
@@ -13,7 +13,6 @@ import platform
import re import re
import socket import socket
import subprocess import subprocess
from collections import OrderedDict
import configargparse import configargparse
import six import six
@@ -218,7 +217,6 @@ def safe_open(path, mode="w", chmod=None, buffering=None):
defaults if ``None``. defaults if ``None``.
""" """
# pylint: disable=star-args
open_args = () # type: Union[Tuple[()], Tuple[int]] open_args = () # type: Union[Tuple[()], Tuple[int]]
if chmod is not None: if chmod is not None:
open_args = (chmod,) open_args = (chmod,)
+2 -2
View File
@@ -52,13 +52,13 @@ install_requires = [
] ]
dev_extras = [ dev_extras = [
'astroid', 'astroid==1.6.5',
'coverage', 'coverage',
'ipdb', 'ipdb',
'pytest', 'pytest',
'pytest-cov', 'pytest-cov',
'pytest-xdist', 'pytest-xdist',
'pylint>=1.8.4', 'pylint==1.9.4',
'tox', 'tox',
'twine', 'twine',
'wheel', 'wheel',
@@ -1 +0,0 @@
../../archive/a.encryption-example.com/cert1.pem
@@ -0,0 +1 @@
../../archive/a.encryption-example.com/cert1.pem
@@ -1 +0,0 @@
../../archive/a.encryption-example.com/chain1.pem
@@ -0,0 +1 @@
../../archive/a.encryption-example.com/chain1.pem
@@ -1 +0,0 @@
../../archive/a.encryption-example.com/fullchain1.pem
@@ -0,0 +1 @@
../../archive/a.encryption-example.com/fullchain1.pem
@@ -1 +0,0 @@
../../archive/a.encryption-example.com/privkey1.pem
@@ -0,0 +1 @@
../../archive/a.encryption-example.com/privkey1.pem
@@ -1 +0,0 @@
../../archive/b.encryption-example.com/cert1.pem
@@ -0,0 +1 @@
../../archive/b.encryption-example.com/cert1.pem
@@ -1 +0,0 @@
../../archive/b.encryption-example.com/chain1.pem
@@ -0,0 +1 @@
../../archive/b.encryption-example.com/chain1.pem
@@ -1 +0,0 @@
../../archive/b.encryption-example.com/fullchain1.pem
@@ -0,0 +1 @@
../../archive/b.encryption-example.com/fullchain1.pem
@@ -1 +0,0 @@
../../archive/b.encryption-example.com/privkey1.pem
@@ -0,0 +1 @@
../../archive/b.encryption-example.com/privkey1.pem
+9 -2
View File
@@ -5,13 +5,15 @@
alabaster==0.7.10 alabaster==0.7.10
apipkg==1.4 apipkg==1.4
asn1crypto==0.22.0 asn1crypto==0.22.0
astroid==1.3.5 astroid==1.6.5
attrs==17.3.0 attrs==17.3.0
Babel==2.5.1 Babel==2.5.1
backports.functools-lru-cache==1.5
backports.shutil-get-terminal-size==1.0.0 backports.shutil-get-terminal-size==1.0.0
boto3==1.9.36 boto3==1.9.36
botocore==1.12.36 botocore==1.12.36
cloudflare==1.5.1 cloudflare==1.5.1
configparser==3.7.4
coverage==4.4.2 coverage==4.4.2
decorator==4.1.2 decorator==4.1.2
dns-lexicon==3.0.8 dns-lexicon==3.0.8
@@ -26,12 +28,15 @@ imagesize==0.7.1
ipdb==0.10.2 ipdb==0.10.2
ipython==5.5.0 ipython==5.5.0
ipython-genutils==0.2.0 ipython-genutils==0.2.0
isort==4.2.5
Jinja2==2.9.6 Jinja2==2.9.6
jmespath==0.9.3 jmespath==0.9.3
josepy==1.1.0 josepy==1.1.0
lazy-object-proxy==1.3.1
logger==1.4 logger==1.4
logilab-common==1.4.1 logilab-common==1.4.1
MarkupSafe==1.0 MarkupSafe==1.0
mccabe==0.6.1
mypy==0.600 mypy==0.600
ndg-httpsclient==0.3.2 ndg-httpsclient==0.3.2
oauth2client==2.0.0 oauth2client==2.0.0
@@ -46,7 +51,7 @@ py==1.4.34
pyasn1==0.1.9 pyasn1==0.1.9
pyasn1-modules==0.0.10 pyasn1-modules==0.0.10
Pygments==2.2.0 Pygments==2.2.0
pylint==1.4.2 pylint==1.9.4
pytest==3.2.5 pytest==3.2.5
pytest-cov==2.5.1 pytest-cov==2.5.1
pytest-forked==0.2 pytest-forked==0.2
@@ -62,6 +67,7 @@ rsa==3.4.2
s3transfer==0.1.11 s3transfer==0.1.11
scandir==1.6 scandir==1.6
simplegeneric==0.8.1 simplegeneric==0.8.1
singledispatch==3.4.0.3
snowballstemmer==1.2.1 snowballstemmer==1.2.1
Sphinx==1.7.5 Sphinx==1.7.5
sphinx-rtd-theme==0.2.4 sphinx-rtd-theme==0.2.4
@@ -76,3 +82,4 @@ typing==3.6.4
uritemplate==0.6 uritemplate==0.6
virtualenv==15.1.0 virtualenv==15.1.0
wcwidth==0.1.7 wcwidth==0.1.7
wrapt==1.11.1