mirror of
https://github.com/certbot/certbot.git
synced 2026-08-02 00:22:28 +02:00
[Windows] Create the CI logic (#6374)
So here we are: after #6361 has been merged, time is to provide an environment to execute the automated testing on Windows. Here are the assertions used to build the CI on Windows: every test running on Linux should ultimately be runnable on Windows, in a cross-platform compatible manner (there is one or two exception, when a test does not have any meaning for Windows), currently some tests are not runnable on Windows: theses tests are ignored by default when the environment is Windows using a custom decorator: @broken_on_windows, test environment should have functionalities similar to Travis, in particular an execution test matrix against various versions of Python and Windows, so test execution is done through AppVeyor, as it supports the requirements: it add a CI step along Travis and Codecov for each PR, all of this ensuring that Certbot is entirely functional on both Linux and Windows, code in tests can be changed, but code in Certbot should be changed as little as possible, to avoid regression risks. So far in this PR, I focused on the tests on Certbot core and ACME library. Concerning the plugins, it will be done later, for plugins which have an interest on Windows. Test are executed against Python 3.4, 3.5, 3.6 and 3.7, for Windows Server 2012 R2 and Windows Server 2016. I succeeded at making 258/259 of acme tests to work, and 828/868 of certbot core tests to work. Most of the errors where not because of Certbot itself, but because of how the tests are written. After redesigning some test utilitaries, and things like file path handling, or CRLF/LF, a lot of the errors vanished. I needed also to ignore a lot of IO errors typically occurring when a tearDown test process tries to delete a file before it has been closed: this kind of behavior is acceptable for Linux, but not for Windows. As a consequence, and until the tearDown process is improved, a lot of temporary files are not cleared on Windows after a test campaign. Remaining broken tests requires a more subtile approach to solve the errors, I will correct them progressively in future PR. Last words about tox. I did not used the existing tox.ini for now. It is just to far from what is supported on Windows: lot of bash scripts that should be rewritten completely, and that contain test logic not ready/relevant for Windows (plugin tests, Docker compilation/test, GNU distribution versatility handling and so on). So I use an independent file tox-win.ini for now, with the goal to merge it ultimately with the existing logic. * Define a tox configuration for windows, to execute tests against Python 3.4, 3.5, 3.6 and 3.7 + code coverage on Codecov.io * Correct windows compatibility on certbot codebase * Correct windows compatibility on certbot display functionalities * Correct windows compatibility on certbot plugins * Correct test utils to run tests on windows. Add decorator to skip (permanently) or mark broken (temporarily) tests on windows * Correct tests on certbot core to run them both on windows and linux. Mark some of them as broken on windows for now. * Lock tests are completely skipped on windows. Planned to be replace in next PR. * Correct tests on certbot display to run them both on windows and linux. Mark some of them as broken on windows for now. * Correct test utils for acme on windows. Add decorator to skip (permanently) or mark broken (temporarily) tests on windows. * Correct acme tests to run them both on windows and linux. Allow a reduction of code coverage of 1% on acme code base. * Create AppVeyor CI for Certbot on Windows, to run the test matrix (py34,35,36,37+coverage) on Windows Server 2012 R2 and Windows Server 2016. * Update changelog with Windows compatibility of Certbot. * Corrections about tox, pyreadline and CI logic * Correct english * Some corrections for acme * Newlines corrections * Remove changelog * Use os.devnull instead of /dev/null to be used on Windows * Uid is a always a number now. * Correct linting * PR https://github.com/python/typeshed/pull/2136 has been merge to third-party upstream 6 months ago, so code patch can be removed. * And so acme coverage should be 100% again. * More compatible tests Windows+Linux * Use stable line separator * Remove unused import * Do not rely on pytest in certbot tests * Use json.dumps to another json embedding weird characters * Change comment * Add import * Test rolling builds #1 * Test rolling builds #2 * Correction on json serialization * It seems that rolling builds are not canceling jobs on PR. Revert back to fail fast code in the pipeline.
This commit is contained in:
committed by
Brad Warren
parent
8dd68a6551
commit
1e8c13ebf9
@@ -1038,8 +1038,8 @@ class ClientNetworkTest(unittest.TestCase):
|
|||||||
|
|
||||||
# Requests Library Exceptions
|
# Requests Library Exceptions
|
||||||
except requests.exceptions.ConnectionError as z: #pragma: no cover
|
except requests.exceptions.ConnectionError as z: #pragma: no cover
|
||||||
self.assertEqual("('Connection aborted.', "
|
self.assertTrue("('Connection aborted.', error(111, 'Connection refused'))"
|
||||||
"error(111, 'Connection refused'))", str(z))
|
== str(z) or "[WinError 10061]" in str(z))
|
||||||
|
|
||||||
class ClientNetworkWithMockedResponseTest(unittest.TestCase):
|
class ClientNetworkWithMockedResponseTest(unittest.TestCase):
|
||||||
"""Tests for acme.client.ClientNetwork which mock out response."""
|
"""Tests for acme.client.ClientNetwork which mock out response."""
|
||||||
|
|||||||
@@ -136,22 +136,16 @@ def probe_sni(name, host, port=443, timeout=300,
|
|||||||
|
|
||||||
socket_kwargs = {'source_address': source_address}
|
socket_kwargs = {'source_address': source_address}
|
||||||
|
|
||||||
host_protocol_agnostic = host
|
|
||||||
if host == '::' or host == '0':
|
|
||||||
# https://github.com/python/typeshed/pull/2136
|
|
||||||
# while PR is not merged, we need to ignore
|
|
||||||
host_protocol_agnostic = None
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
# pylint: disable=star-args
|
# pylint: disable=star-args
|
||||||
logger.debug(
|
logger.debug(
|
||||||
"Attempting to connect to %s:%d%s.", host_protocol_agnostic, port,
|
"Attempting to connect to %s:%d%s.", host, port,
|
||||||
" from {0}:{1}".format(
|
" from {0}:{1}".format(
|
||||||
source_address[0],
|
source_address[0],
|
||||||
source_address[1]
|
source_address[1]
|
||||||
) if socket_kwargs else ""
|
) if socket_kwargs else ""
|
||||||
)
|
)
|
||||||
socket_tuple = (host_protocol_agnostic, port) # type: Tuple[Optional[str], int]
|
socket_tuple = (host, port) # type: Tuple[str, int]
|
||||||
sock = socket.create_connection(socket_tuple, **socket_kwargs) # type: ignore
|
sock = socket.create_connection(socket_tuple, **socket_kwargs) # type: ignore
|
||||||
except socket.error as error:
|
except socket.error as error:
|
||||||
raise errors.Error(error)
|
raise errors.Error(error)
|
||||||
|
|||||||
@@ -48,7 +48,7 @@ class TLSSNI01ServerTest(unittest.TestCase):
|
|||||||
test_util.load_cert('rsa2048_cert.pem'),
|
test_util.load_cert('rsa2048_cert.pem'),
|
||||||
)}
|
)}
|
||||||
from acme.standalone import TLSSNI01Server
|
from acme.standalone import TLSSNI01Server
|
||||||
self.server = TLSSNI01Server(("", 0), certs=self.certs)
|
self.server = TLSSNI01Server(('localhost', 0), certs=self.certs)
|
||||||
# pylint: disable=no-member
|
# 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()
|
||||||
@@ -133,8 +133,11 @@ class BaseDualNetworkedServersTest(unittest.TestCase):
|
|||||||
self.address_family = socket.AF_INET
|
self.address_family = socket.AF_INET
|
||||||
socketserver.TCPServer.__init__(self, *args, **kwargs)
|
socketserver.TCPServer.__init__(self, *args, **kwargs)
|
||||||
if ipv6:
|
if ipv6:
|
||||||
|
# NB: On Windows, socket.IPPROTO_IPV6 constant may be missing.
|
||||||
|
# We use the corresponding value (41) instead.
|
||||||
|
level = getattr(socket, "IPPROTO_IPV6", 41)
|
||||||
# pylint: disable=no-member
|
# pylint: disable=no-member
|
||||||
self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
|
self.socket.setsockopt(level, socket.IPV6_V6ONLY, 1)
|
||||||
try:
|
try:
|
||||||
self.server_bind()
|
self.server_bind()
|
||||||
self.server_activate()
|
self.server_activate()
|
||||||
@@ -147,15 +150,15 @@ class BaseDualNetworkedServersTest(unittest.TestCase):
|
|||||||
mock_bind.side_effect = socket.error
|
mock_bind.side_effect = socket.error
|
||||||
from acme.standalone import BaseDualNetworkedServers
|
from acme.standalone import BaseDualNetworkedServers
|
||||||
self.assertRaises(socket.error, BaseDualNetworkedServers,
|
self.assertRaises(socket.error, BaseDualNetworkedServers,
|
||||||
BaseDualNetworkedServersTest.SingleProtocolServer,
|
BaseDualNetworkedServersTest.SingleProtocolServer,
|
||||||
("", 0),
|
('', 0),
|
||||||
socketserver.BaseRequestHandler)
|
socketserver.BaseRequestHandler)
|
||||||
|
|
||||||
def test_ports_equal(self):
|
def test_ports_equal(self):
|
||||||
from acme.standalone import BaseDualNetworkedServers
|
from acme.standalone import BaseDualNetworkedServers
|
||||||
servers = BaseDualNetworkedServers(
|
servers = BaseDualNetworkedServers(
|
||||||
BaseDualNetworkedServersTest.SingleProtocolServer,
|
BaseDualNetworkedServersTest.SingleProtocolServer,
|
||||||
("", 0),
|
('', 0),
|
||||||
socketserver.BaseRequestHandler)
|
socketserver.BaseRequestHandler)
|
||||||
socknames = servers.getsocknames()
|
socknames = servers.getsocknames()
|
||||||
prev_port = None
|
prev_port = None
|
||||||
@@ -177,7 +180,7 @@ class TLSSNI01DualNetworkedServersTest(unittest.TestCase):
|
|||||||
test_util.load_cert('rsa2048_cert.pem'),
|
test_util.load_cert('rsa2048_cert.pem'),
|
||||||
)}
|
)}
|
||||||
from acme.standalone import TLSSNI01DualNetworkedServers
|
from acme.standalone import TLSSNI01DualNetworkedServers
|
||||||
self.servers = TLSSNI01DualNetworkedServers(("", 0), certs=self.certs)
|
self.servers = TLSSNI01DualNetworkedServers(('localhost', 0), certs=self.certs)
|
||||||
self.servers.serve_forever()
|
self.servers.serve_forever()
|
||||||
|
|
||||||
def tearDown(self):
|
def tearDown(self):
|
||||||
@@ -245,6 +248,7 @@ class HTTP01DualNetworkedServersTest(unittest.TestCase):
|
|||||||
self.assertFalse(self._test_http01(add=False))
|
self.assertFalse(self._test_http01(add=False))
|
||||||
|
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
class TestSimpleTLSSNI01Server(unittest.TestCase):
|
class TestSimpleTLSSNI01Server(unittest.TestCase):
|
||||||
"""Tests for acme.standalone.simple_tls_sni_01_server."""
|
"""Tests for acme.standalone.simple_tls_sni_01_server."""
|
||||||
|
|
||||||
|
|||||||
@@ -4,6 +4,7 @@
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
import os
|
import os
|
||||||
|
import sys
|
||||||
import pkg_resources
|
import pkg_resources
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
@@ -94,3 +95,11 @@ def skip_unless(condition, reason): # pragma: no cover
|
|||||||
return lambda cls: cls
|
return lambda cls: cls
|
||||||
else:
|
else:
|
||||||
return lambda cls: None
|
return lambda cls: None
|
||||||
|
|
||||||
|
def broken_on_windows(function):
|
||||||
|
"""Decorator to skip temporarily a broken test on Windows."""
|
||||||
|
reason = 'Test is broken and ignored on windows but should be fixed.'
|
||||||
|
return unittest.skipIf(
|
||||||
|
sys.platform == 'win32'
|
||||||
|
and os.environ.get('SKIP_BROKEN_TESTS_ON_WINDOWS', 'true') == 'true',
|
||||||
|
reason)(function)
|
||||||
|
|||||||
+26
-4
@@ -1,12 +1,34 @@
|
|||||||
# AppVeyor CI pipeline, executed on Windows Server 2016/2012 R2
|
image:
|
||||||
|
# => Windows Server 2012 R2
|
||||||
|
- Visual Studio 2015
|
||||||
|
# => Windows Server 2016
|
||||||
|
- Visual Studio 2017
|
||||||
|
|
||||||
branches:
|
branches:
|
||||||
only:
|
only:
|
||||||
- master
|
- master
|
||||||
- /^\d+\.\d+\.x$/ # version branches like X.X.X
|
- /^\d+\.\d+\.x$/ # Version branches like X.X.X
|
||||||
- /^test-.*$/
|
- /^test-.*$/
|
||||||
|
|
||||||
|
install:
|
||||||
|
# Fail fast if a newer build is queued for the same PR
|
||||||
|
- ps: if ($env:APPVEYOR_PULL_REQUEST_NUMBER -and $env:APPVEYOR_BUILD_NUMBER -ne ((Invoke-RestMethod `
|
||||||
|
https://ci.appveyor.com/api/projects/$env:APPVEYOR_ACCOUNT_NAME/$env:APPVEYOR_PROJECT_SLUG/history?recordsNumber=50).builds | `
|
||||||
|
Where-Object pullRequestId -eq $env:APPVEYOR_PULL_REQUEST_NUMBER)[0].buildNumber) { `
|
||||||
|
throw "There are newer queued builds for this pull request, failing early." }
|
||||||
|
# Use Python 3.7 by default
|
||||||
|
- "SET PATH=C:\\Python37;C:\\Python37\\Scripts;%PATH%"
|
||||||
|
# Check env
|
||||||
|
- "python --version"
|
||||||
|
# Upgrade pip to avoid warnings
|
||||||
|
- "python -m pip install --upgrade pip"
|
||||||
|
# Ready to install tox and coverage
|
||||||
|
- "pip install tox codecov"
|
||||||
|
|
||||||
build: off
|
build: off
|
||||||
|
|
||||||
test_script:
|
test_script:
|
||||||
- ps: Write-Host "Hello, world!"
|
- tox -c tox-win.ini -e py34,py35,py36,py37-cover
|
||||||
|
|
||||||
|
on_success:
|
||||||
|
- codecov
|
||||||
|
|||||||
+11
-1
@@ -2,7 +2,7 @@
|
|||||||
Compatibility layer to run certbot both on Linux and Windows.
|
Compatibility layer to run certbot both on Linux and Windows.
|
||||||
|
|
||||||
The approach used here is similar to Modernizr for Web browsers.
|
The approach used here is similar to Modernizr for Web browsers.
|
||||||
We do not check the plateform type to determine if a particular logic is supported.
|
We do not check the platform type to determine if a particular logic is supported.
|
||||||
Instead, we apply a logic, and then fallback to another logic if first logic
|
Instead, we apply a logic, and then fallback to another logic if first logic
|
||||||
is not supported at runtime.
|
is not supported at runtime.
|
||||||
|
|
||||||
@@ -13,6 +13,7 @@ import select
|
|||||||
import sys
|
import sys
|
||||||
import errno
|
import errno
|
||||||
import ctypes
|
import ctypes
|
||||||
|
import stat
|
||||||
|
|
||||||
from certbot import errors
|
from certbot import errors
|
||||||
|
|
||||||
@@ -138,3 +139,12 @@ def release_locked_file(fd, path):
|
|||||||
raise
|
raise
|
||||||
finally:
|
finally:
|
||||||
os.close(fd)
|
os.close(fd)
|
||||||
|
|
||||||
|
def compare_file_modes(mode1, mode2):
|
||||||
|
"""Return true if the two modes can be considered as equals for this platform"""
|
||||||
|
if 'fcntl' in sys.modules:
|
||||||
|
# Linux specific: standard compare
|
||||||
|
return oct(stat.S_IMODE(mode1)) == oct(stat.S_IMODE(mode2))
|
||||||
|
# Windows specific: most of mode bits are ignored on Windows. Only check user R/W rights.
|
||||||
|
return (stat.S_IMODE(mode1) & stat.S_IREAD == stat.S_IMODE(mode2) & stat.S_IREAD
|
||||||
|
and stat.S_IMODE(mode1) & stat.S_IWRITE == stat.S_IMODE(mode2) & stat.S_IWRITE)
|
||||||
|
|||||||
@@ -449,14 +449,17 @@ def _notAfterBefore(cert_path, method):
|
|||||||
def sha256sum(filename):
|
def sha256sum(filename):
|
||||||
"""Compute a sha256sum of a file.
|
"""Compute a sha256sum of a file.
|
||||||
|
|
||||||
|
NB: In given file, platform specific newlines characters will be converted
|
||||||
|
into their equivalent unicode counterparts before calculating the hash.
|
||||||
|
|
||||||
:param str filename: path to the file whose hash will be computed
|
:param str filename: path to the file whose hash will be computed
|
||||||
|
|
||||||
:returns: sha256 digest of the file in hexadecimal
|
:returns: sha256 digest of the file in hexadecimal
|
||||||
:rtype: str
|
:rtype: str
|
||||||
"""
|
"""
|
||||||
sha256 = hashlib.sha256()
|
sha256 = hashlib.sha256()
|
||||||
with open(filename, 'rb') as f:
|
with open(filename, 'rU') as file_d:
|
||||||
sha256.update(f.read())
|
sha256.update(file_d.read().encode('UTF-8'))
|
||||||
return sha256.hexdigest()
|
return sha256.hexdigest()
|
||||||
|
|
||||||
def cert_and_chain_from_fullchain(fullchain_pem):
|
def cert_and_chain_from_fullchain(fullchain_pem):
|
||||||
|
|||||||
@@ -49,9 +49,9 @@ class Completer(object):
|
|||||||
readline.set_completer(self.complete)
|
readline.set_completer(self.complete)
|
||||||
readline.set_completer_delims(' \t\n;')
|
readline.set_completer_delims(' \t\n;')
|
||||||
|
|
||||||
# readline can be implemented using GNU readline or libedit
|
# readline can be implemented using GNU readline, pyreadline or libedit
|
||||||
# which have different configuration syntax
|
# which have different configuration syntax
|
||||||
if 'libedit' in readline.__doc__:
|
if readline.__doc__ is not None and 'libedit' in readline.__doc__:
|
||||||
readline.parse_and_bind('bind ^I rl_complete')
|
readline.parse_and_bind('bind ^I rl_complete')
|
||||||
else:
|
else:
|
||||||
readline.parse_and_bind('tab: complete')
|
readline.parse_and_bind('tab: complete')
|
||||||
|
|||||||
@@ -53,7 +53,7 @@ def _wrap_lines(msg):
|
|||||||
break_long_words=False,
|
break_long_words=False,
|
||||||
break_on_hyphens=False))
|
break_on_hyphens=False))
|
||||||
|
|
||||||
return os.linesep.join(fixed_l)
|
return '\n'.join(fixed_l)
|
||||||
|
|
||||||
|
|
||||||
def input_with_timeout(prompt=None, timeout=36000.0):
|
def input_with_timeout(prompt=None, timeout=36000.0):
|
||||||
|
|||||||
+2
-1
@@ -1135,7 +1135,8 @@ def _csr_get_and_save_cert(config, le_client):
|
|||||||
"Dry run: skipping saving certificate to %s", config.cert_path)
|
"Dry run: skipping saving certificate to %s", config.cert_path)
|
||||||
return None, None
|
return None, None
|
||||||
cert_path, _, fullchain_path = le_client.save_certificate(
|
cert_path, _, fullchain_path = le_client.save_certificate(
|
||||||
cert, chain, config.cert_path, config.chain_path, config.fullchain_path)
|
cert, chain, os.path.normpath(config.cert_path),
|
||||||
|
os.path.normpath(config.chain_path), os.path.normpath(config.fullchain_path))
|
||||||
return cert_path, fullchain_path
|
return cert_path, fullchain_path
|
||||||
|
|
||||||
def renew_cert(config, plugins, lineage):
|
def renew_cert(config, plugins, lineage):
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import unittest
|
|||||||
|
|
||||||
import six
|
import six
|
||||||
import mock
|
import mock
|
||||||
|
import sys
|
||||||
|
|
||||||
from acme import challenges
|
from acme import challenges
|
||||||
|
|
||||||
@@ -75,12 +76,14 @@ class AuthenticatorTest(test_util.TempDirTestCase):
|
|||||||
def test_script_perform(self):
|
def test_script_perform(self):
|
||||||
self.config.manual_public_ip_logging_ok = True
|
self.config.manual_public_ip_logging_ok = True
|
||||||
self.config.manual_auth_hook = (
|
self.config.manual_auth_hook = (
|
||||||
'echo ${CERTBOT_DOMAIN}; '
|
'{0} -c "from __future__ import print_function;'
|
||||||
'echo ${CERTBOT_TOKEN:-notoken}; '
|
'import os; print(os.environ.get(\'CERTBOT_DOMAIN\'));'
|
||||||
'echo ${CERTBOT_CERT_PATH:-nocert}; '
|
'print(os.environ.get(\'CERTBOT_TOKEN\', \'notoken\'));'
|
||||||
'echo ${CERTBOT_KEY_PATH:-nokey}; '
|
'print(os.environ.get(\'CERTBOT_CERT_PATH\', \'nocert\'));'
|
||||||
'echo ${CERTBOT_SNI_DOMAIN:-nosnidomain}; '
|
'print(os.environ.get(\'CERTBOT_KEY_PATH\', \'nokey\'));'
|
||||||
'echo ${CERTBOT_VALIDATION:-novalidation};')
|
'print(os.environ.get(\'CERTBOT_SNI_DOMAIN\', \'nosnidomain\'));'
|
||||||
|
'print(os.environ.get(\'CERTBOT_VALIDATION\', \'novalidation\'));"'
|
||||||
|
.format(sys.executable))
|
||||||
dns_expected = '{0}\n{1}\n{2}\n{3}\n{4}\n{5}'.format(
|
dns_expected = '{0}\n{1}\n{2}\n{3}\n{4}\n{5}'.format(
|
||||||
self.dns_achall.domain, 'notoken',
|
self.dns_achall.domain, 'notoken',
|
||||||
'nocert', 'nokey', 'nosnidomain',
|
'nocert', 'nokey', 'nosnidomain',
|
||||||
@@ -128,6 +131,7 @@ class AuthenticatorTest(test_util.TempDirTestCase):
|
|||||||
achall.validation(achall.account_key) in args[0])
|
achall.validation(achall.account_key) in args[0])
|
||||||
self.assertFalse(kwargs['wrap'])
|
self.assertFalse(kwargs['wrap'])
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_cleanup(self):
|
def test_cleanup(self):
|
||||||
self.config.manual_public_ip_logging_ok = True
|
self.config.manual_public_ip_logging_ok = True
|
||||||
self.config.manual_auth_hook = 'echo foo;'
|
self.config.manual_auth_hook = 'echo foo;'
|
||||||
|
|||||||
@@ -4,13 +4,14 @@ import unittest
|
|||||||
|
|
||||||
import mock
|
import mock
|
||||||
|
|
||||||
|
|
||||||
class GetPrefixTest(unittest.TestCase):
|
class GetPrefixTest(unittest.TestCase):
|
||||||
"""Tests for certbot.plugins.get_prefixes."""
|
"""Tests for certbot.plugins.get_prefixes."""
|
||||||
def test_get_prefix(self):
|
def test_get_prefix(self):
|
||||||
from certbot.plugins.util import get_prefixes
|
from certbot.plugins.util import get_prefixes
|
||||||
self.assertEqual(get_prefixes('/a/b/c'), ['/a/b/c', '/a/b', '/a', '/'])
|
self.assertEqual(
|
||||||
self.assertEqual(get_prefixes('/'), ['/'])
|
get_prefixes('/a/b/c'),
|
||||||
|
[os.path.normpath(path) for path in ['/a/b/c', '/a/b', '/a', '/']])
|
||||||
|
self.assertEqual(get_prefixes('/'), [os.path.normpath('/')])
|
||||||
self.assertEqual(get_prefixes('a'), ['a'])
|
self.assertEqual(get_prefixes('a'), ['a'])
|
||||||
|
|
||||||
class PathSurgeryTest(unittest.TestCase):
|
class PathSurgeryTest(unittest.TestCase):
|
||||||
|
|||||||
@@ -4,9 +4,9 @@ from __future__ import print_function
|
|||||||
|
|
||||||
import argparse
|
import argparse
|
||||||
import errno
|
import errno
|
||||||
|
import json
|
||||||
import os
|
import os
|
||||||
import shutil
|
import shutil
|
||||||
import stat
|
|
||||||
import tempfile
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
@@ -17,6 +17,7 @@ import six
|
|||||||
from acme import challenges
|
from acme import challenges
|
||||||
|
|
||||||
from certbot import achallenges
|
from certbot import achallenges
|
||||||
|
from certbot import compat
|
||||||
from certbot import errors
|
from certbot import errors
|
||||||
from certbot.display import util as display_util
|
from certbot.display import util as display_util
|
||||||
|
|
||||||
@@ -142,6 +143,7 @@ class AuthenticatorTest(unittest.TestCase):
|
|||||||
self.assertRaises(errors.PluginError, self.auth.perform, [])
|
self.assertRaises(errors.PluginError, self.auth.perform, [])
|
||||||
os.chmod(self.path, 0o700)
|
os.chmod(self.path, 0o700)
|
||||||
|
|
||||||
|
@test_util.skip_on_windows('On Windows, there is no chown.')
|
||||||
@mock.patch("certbot.plugins.webroot.os.chown")
|
@mock.patch("certbot.plugins.webroot.os.chown")
|
||||||
def test_failed_chown(self, mock_chown):
|
def test_failed_chown(self, mock_chown):
|
||||||
mock_chown.side_effect = OSError(errno.EACCES, "msg")
|
mock_chown.side_effect = OSError(errno.EACCES, "msg")
|
||||||
@@ -169,16 +171,14 @@ class AuthenticatorTest(unittest.TestCase):
|
|||||||
# Remove exec bit from permission check, so that it
|
# Remove exec bit from permission check, so that it
|
||||||
# matches the file
|
# matches the file
|
||||||
self.auth.perform([self.achall])
|
self.auth.perform([self.achall])
|
||||||
path_permissions = stat.S_IMODE(os.stat(self.validation_path).st_mode)
|
self.assertTrue(compat.compare_file_modes(os.stat(self.validation_path).st_mode, 0o644))
|
||||||
self.assertEqual(path_permissions, 0o644)
|
|
||||||
|
|
||||||
# Check permissions of the directories
|
# Check permissions of the directories
|
||||||
|
|
||||||
for dirpath, dirnames, _ in os.walk(self.path):
|
for dirpath, dirnames, _ in os.walk(self.path):
|
||||||
for directory in dirnames:
|
for directory in dirnames:
|
||||||
full_path = os.path.join(dirpath, directory)
|
full_path = os.path.join(dirpath, directory)
|
||||||
dir_permissions = stat.S_IMODE(os.stat(full_path).st_mode)
|
self.assertTrue(compat.compare_file_modes(os.stat(full_path).st_mode, 0o755))
|
||||||
self.assertEqual(dir_permissions, 0o755)
|
|
||||||
|
|
||||||
parent_gid = os.stat(self.path).st_gid
|
parent_gid = os.stat(self.path).st_gid
|
||||||
parent_uid = os.stat(self.path).st_uid
|
parent_uid = os.stat(self.path).st_uid
|
||||||
@@ -274,7 +274,7 @@ class WebrootActionTest(unittest.TestCase):
|
|||||||
|
|
||||||
def test_webroot_map_action(self):
|
def test_webroot_map_action(self):
|
||||||
args = self.parser.parse_args(
|
args = self.parser.parse_args(
|
||||||
["--webroot-map", '{{"thing.com":"{0}"}}'.format(self.path)])
|
["--webroot-map", json.dumps({'thing.com': self.path})])
|
||||||
self.assertEqual(args.webroot_map["thing.com"], self.path)
|
self.assertEqual(args.webroot_map["thing.com"], self.path)
|
||||||
|
|
||||||
def test_domain_before_webroot(self):
|
def test_domain_before_webroot(self):
|
||||||
|
|||||||
+1
-1
@@ -301,7 +301,7 @@ def renew_cert(config, domains, le_client, lineage):
|
|||||||
domains = lineage.names()
|
domains = lineage.names()
|
||||||
# The private key is the existing lineage private key if reuse_key is set.
|
# The private key is the existing lineage private key if reuse_key is set.
|
||||||
# Otherwise, generate a fresh private key by passing None.
|
# Otherwise, generate a fresh private key by passing None.
|
||||||
new_key = lineage.privkey if config.reuse_key else None
|
new_key = os.path.normpath(lineage.privkey) if config.reuse_key else None
|
||||||
new_cert, new_chain, new_key, _ = le_client.obtain_certificate(domains, new_key)
|
new_cert, new_chain, new_key, _ = le_client.obtain_certificate(domains, new_key)
|
||||||
if config.dry_run:
|
if config.dry_run:
|
||||||
logger.debug("Dry run: skipping updating lineage at %s",
|
logger.debug("Dry run: skipping updating lineage at %s",
|
||||||
|
|||||||
@@ -116,6 +116,7 @@ class AccountFileStorageTest(test_util.ConfigTestCase):
|
|||||||
def test_init_creates_dir(self):
|
def test_init_creates_dir(self):
|
||||||
self.assertTrue(os.path.isdir(self.config.accounts_dir))
|
self.assertTrue(os.path.isdir(self.config.accounts_dir))
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_save_and_restore(self):
|
def test_save_and_restore(self):
|
||||||
self.storage.save(self.acc, self.mock_client)
|
self.storage.save(self.acc, self.mock_client)
|
||||||
account_path = os.path.join(self.config.accounts_dir, self.acc.id)
|
account_path = os.path.join(self.config.accounts_dir, self.acc.id)
|
||||||
@@ -218,12 +219,14 @@ class AccountFileStorageTest(test_util.ConfigTestCase):
|
|||||||
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
|
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
|
||||||
self.assertEqual([], self.storage.find_all())
|
self.assertEqual([], self.storage.find_all())
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_upgrade_version_staging(self):
|
def test_upgrade_version_staging(self):
|
||||||
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
|
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
|
||||||
self.storage.save(self.acc, self.mock_client)
|
self.storage.save(self.acc, self.mock_client)
|
||||||
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
|
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
|
||||||
self.assertEqual([self.acc], self.storage.find_all())
|
self.assertEqual([self.acc], self.storage.find_all())
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_upgrade_version_production(self):
|
def test_upgrade_version_production(self):
|
||||||
self._set_server('https://acme-v01.api.letsencrypt.org/directory')
|
self._set_server('https://acme-v01.api.letsencrypt.org/directory')
|
||||||
self.storage.save(self.acc, self.mock_client)
|
self.storage.save(self.acc, self.mock_client)
|
||||||
@@ -241,6 +244,7 @@ class AccountFileStorageTest(test_util.ConfigTestCase):
|
|||||||
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
|
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
|
||||||
self.assertEqual([], self.storage.find_all())
|
self.assertEqual([], self.storage.find_all())
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_upgrade_load(self):
|
def test_upgrade_load(self):
|
||||||
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
|
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
|
||||||
self.storage.save(self.acc, self.mock_client)
|
self.storage.save(self.acc, self.mock_client)
|
||||||
@@ -249,6 +253,7 @@ class AccountFileStorageTest(test_util.ConfigTestCase):
|
|||||||
account = self.storage.load(self.acc.id)
|
account = self.storage.load(self.acc.id)
|
||||||
self.assertEqual(prev_account, account)
|
self.assertEqual(prev_account, account)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_upgrade_load_single_account(self):
|
def test_upgrade_load_single_account(self):
|
||||||
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
|
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
|
||||||
self.storage.save(self.acc, self.mock_client)
|
self.storage.save(self.acc, self.mock_client)
|
||||||
@@ -273,6 +278,7 @@ class AccountFileStorageTest(test_util.ConfigTestCase):
|
|||||||
errors.AccountStorageError, self.storage.save,
|
errors.AccountStorageError, self.storage.save,
|
||||||
self.acc, self.mock_client)
|
self.acc, self.mock_client)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_delete(self):
|
def test_delete(self):
|
||||||
self.storage.save(self.acc, self.mock_client)
|
self.storage.save(self.acc, self.mock_client)
|
||||||
self.storage.delete(self.acc.id)
|
self.storage.delete(self.acc.id)
|
||||||
@@ -307,10 +313,12 @@ class AccountFileStorageTest(test_util.ConfigTestCase):
|
|||||||
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
|
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
|
||||||
self.assertRaises(errors.AccountNotFound, self.storage.load, self.acc.id)
|
self.assertRaises(errors.AccountNotFound, self.storage.load, self.acc.id)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_delete_folders_up(self):
|
def test_delete_folders_up(self):
|
||||||
self._test_delete_folders('https://acme-staging.api.letsencrypt.org/directory')
|
self._test_delete_folders('https://acme-staging.api.letsencrypt.org/directory')
|
||||||
self._assert_symlinked_account_removed()
|
self._assert_symlinked_account_removed()
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_delete_folders_down(self):
|
def test_delete_folders_down(self):
|
||||||
self._test_delete_folders('https://acme-staging-v02.api.letsencrypt.org/directory')
|
self._test_delete_folders('https://acme-staging-v02.api.letsencrypt.org/directory')
|
||||||
self._assert_symlinked_account_removed()
|
self._assert_symlinked_account_removed()
|
||||||
@@ -320,10 +328,12 @@ class AccountFileStorageTest(test_util.ConfigTestCase):
|
|||||||
with open(os.path.join(self.config.accounts_dir, 'foo'), 'w') as f:
|
with open(os.path.join(self.config.accounts_dir, 'foo'), 'w') as f:
|
||||||
f.write('bar')
|
f.write('bar')
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_delete_shared_account_up(self):
|
def test_delete_shared_account_up(self):
|
||||||
self._set_server_and_stop_symlink('https://acme-staging-v02.api.letsencrypt.org/directory')
|
self._set_server_and_stop_symlink('https://acme-staging-v02.api.letsencrypt.org/directory')
|
||||||
self._test_delete_folders('https://acme-staging.api.letsencrypt.org/directory')
|
self._test_delete_folders('https://acme-staging.api.letsencrypt.org/directory')
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_delete_shared_account_down(self):
|
def test_delete_shared_account_down(self):
|
||||||
self._set_server_and_stop_symlink('https://acme-staging-v02.api.letsencrypt.org/directory')
|
self._set_server_and_stop_symlink('https://acme-staging-v02.api.letsencrypt.org/directory')
|
||||||
self._test_delete_folders('https://acme-staging-v02.api.letsencrypt.org/directory')
|
self._test_delete_folders('https://acme-staging-v02.api.letsencrypt.org/directory')
|
||||||
|
|||||||
@@ -204,7 +204,7 @@ class CertificatesTest(BaseCertManagerTest):
|
|||||||
shutil.rmtree(empty_tempdir)
|
shutil.rmtree(empty_tempdir)
|
||||||
|
|
||||||
@mock.patch('certbot.cert_manager.ocsp.RevocationChecker.ocsp_revoked')
|
@mock.patch('certbot.cert_manager.ocsp.RevocationChecker.ocsp_revoked')
|
||||||
def test_report_human_readable(self, mock_revoked):
|
def test_report_human_readable(self, mock_revoked): #pylint: disable=too-many-statements
|
||||||
mock_revoked.return_value = None
|
mock_revoked.return_value = None
|
||||||
from certbot import cert_manager
|
from certbot import cert_manager
|
||||||
import datetime, pytz
|
import datetime, pytz
|
||||||
@@ -228,7 +228,7 @@ class CertificatesTest(BaseCertManagerTest):
|
|||||||
cert.target_expiry += datetime.timedelta(hours=2)
|
cert.target_expiry += datetime.timedelta(hours=2)
|
||||||
# pylint: disable=protected-access
|
# pylint: disable=protected-access
|
||||||
out = get_report()
|
out = get_report()
|
||||||
self.assertTrue('1 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 not 'INVALID' in out)
|
||||||
|
|
||||||
cert.target_expiry += datetime.timedelta(days=1)
|
cert.target_expiry += datetime.timedelta(days=1)
|
||||||
|
|||||||
@@ -76,6 +76,7 @@ class ParseTest(unittest.TestCase): # pylint: disable=too-many-public-methods
|
|||||||
|
|
||||||
return output.getvalue()
|
return output.getvalue()
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch("certbot.cli.flag_default")
|
@mock.patch("certbot.cli.flag_default")
|
||||||
def test_cli_ini_domains(self, mock_flag_default):
|
def test_cli_ini_domains(self, mock_flag_default):
|
||||||
tmp_config = tempfile.NamedTemporaryFile()
|
tmp_config = tempfile.NamedTemporaryFile()
|
||||||
|
|||||||
@@ -140,7 +140,7 @@ class ImportCSRFileTest(unittest.TestCase):
|
|||||||
util.CSR(file=csrfile,
|
util.CSR(file=csrfile,
|
||||||
data=data_pem,
|
data=data_pem,
|
||||||
form="pem"),
|
form="pem"),
|
||||||
["Example.com"],),
|
["Example.com"]),
|
||||||
self._call(csrfile, data))
|
self._call(csrfile, data))
|
||||||
|
|
||||||
def test_pem_csr(self):
|
def test_pem_csr(self):
|
||||||
@@ -376,7 +376,6 @@ class NotAfterTest(unittest.TestCase):
|
|||||||
|
|
||||||
class Sha256sumTest(unittest.TestCase):
|
class Sha256sumTest(unittest.TestCase):
|
||||||
"""Tests for certbot.crypto_util.notAfter"""
|
"""Tests for certbot.crypto_util.notAfter"""
|
||||||
|
|
||||||
def test_sha256sum(self):
|
def test_sha256sum(self):
|
||||||
from certbot.crypto_util import sha256sum
|
from certbot.crypto_util import sha256sum
|
||||||
self.assertEqual(sha256sum(CERT_PATH),
|
self.assertEqual(sha256sum(CERT_PATH),
|
||||||
|
|||||||
@@ -1,6 +1,9 @@
|
|||||||
"""Test certbot.display.completer."""
|
"""Test certbot.display.completer."""
|
||||||
import os
|
import os
|
||||||
import readline
|
try:
|
||||||
|
import readline # pylint: disable=import-error
|
||||||
|
except ImportError:
|
||||||
|
import certbot.display.dummy_readline as readline # type: ignore
|
||||||
import string
|
import string
|
||||||
import sys
|
import sys
|
||||||
import unittest
|
import unittest
|
||||||
@@ -9,9 +12,9 @@ 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
|
||||||
from certbot.tests.util import TempDirTestCase
|
import certbot.tests.util as test_util
|
||||||
|
|
||||||
class CompleterTest(TempDirTestCase):
|
class CompleterTest(test_util.TempDirTestCase):
|
||||||
"""Test certbot.display.completer.Completer."""
|
"""Test certbot.display.completer.Completer."""
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
@@ -47,6 +50,8 @@ class CompleterTest(TempDirTestCase):
|
|||||||
completion = my_completer.complete(self.tempdir, num_paths)
|
completion = my_completer.complete(self.tempdir, num_paths)
|
||||||
self.assertEqual(completion, None)
|
self.assertEqual(completion, None)
|
||||||
|
|
||||||
|
@unittest.skipIf('readline' not in sys.modules,
|
||||||
|
reason='Not relevant if readline is not available.')
|
||||||
def test_import_error(self):
|
def test_import_error(self):
|
||||||
original_readline = sys.modules['readline']
|
original_readline = sys.modules['readline']
|
||||||
sys.modules['readline'] = None
|
sys.modules['readline'] = None
|
||||||
@@ -91,7 +96,7 @@ class CompleterTest(TempDirTestCase):
|
|||||||
|
|
||||||
def enable_tab_completion(unused_command):
|
def enable_tab_completion(unused_command):
|
||||||
"""Enables readline tab completion using the system specific syntax."""
|
"""Enables readline tab completion using the system specific syntax."""
|
||||||
libedit = 'libedit' in readline.__doc__
|
libedit = readline.__doc__ is not None and 'libedit' in readline.__doc__
|
||||||
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)
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,5 @@
|
|||||||
"""Test :mod:`certbot.display.util`."""
|
"""Test :mod:`certbot.display.util`."""
|
||||||
import inspect
|
import inspect
|
||||||
import os
|
|
||||||
import socket
|
import socket
|
||||||
import tempfile
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
@@ -10,7 +9,6 @@ import mock
|
|||||||
|
|
||||||
from certbot import errors
|
from certbot import errors
|
||||||
from certbot import interfaces
|
from certbot import interfaces
|
||||||
|
|
||||||
from certbot.display import util as display_util
|
from certbot.display import util as display_util
|
||||||
|
|
||||||
|
|
||||||
@@ -281,10 +279,10 @@ class FileOutputDisplayTest(unittest.TestCase):
|
|||||||
msg = ("This is just a weak test{0}"
|
msg = ("This is just a weak test{0}"
|
||||||
"This function is only meant to be for easy viewing{0}"
|
"This function is only meant to be for easy viewing{0}"
|
||||||
"Test a really really really really really really really really "
|
"Test a really really really really really really really really "
|
||||||
"really really really really long line...".format(os.linesep))
|
"really really really really long line...".format('\n'))
|
||||||
text = display_util._wrap_lines(msg)
|
text = display_util._wrap_lines(msg)
|
||||||
|
|
||||||
self.assertEqual(text.count(os.linesep), 3)
|
self.assertEqual(text.count('\n'), 3)
|
||||||
|
|
||||||
def test_get_valid_int_ans_valid(self):
|
def test_get_valid_int_ans_valid(self):
|
||||||
# pylint: disable=protected-access
|
# pylint: disable=protected-access
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import mock
|
|||||||
from acme.magic_typing import Callable, Dict, Union
|
from acme.magic_typing import Callable, Dict, Union
|
||||||
# pylint: enable=unused-import, no-name-in-module
|
# pylint: enable=unused-import, no-name-in-module
|
||||||
|
|
||||||
|
import certbot.tests.util as test_util
|
||||||
|
|
||||||
def get_signals(signums):
|
def get_signals(signums):
|
||||||
"""Get the handlers for an iterable of signums."""
|
"""Get the handlers for an iterable of signums."""
|
||||||
@@ -65,6 +66,8 @@ class ErrorHandlerTest(unittest.TestCase):
|
|||||||
self.init_func.assert_called_once_with(*self.init_args,
|
self.init_func.assert_called_once_with(*self.init_args,
|
||||||
**self.init_kwargs)
|
**self.init_kwargs)
|
||||||
|
|
||||||
|
# On Windows, this test kills pytest itself !
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_context_manager_with_signal(self):
|
def test_context_manager_with_signal(self):
|
||||||
init_signals = get_signals(self.signals)
|
init_signals = get_signals(self.signals)
|
||||||
with signal_receiver(self.signals) as signals_received:
|
with signal_receiver(self.signals) as signals_received:
|
||||||
@@ -95,6 +98,8 @@ class ErrorHandlerTest(unittest.TestCase):
|
|||||||
**self.init_kwargs)
|
**self.init_kwargs)
|
||||||
bad_func.assert_called_once_with()
|
bad_func.assert_called_once_with()
|
||||||
|
|
||||||
|
# On Windows, this test kills pytest itself !
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_bad_recovery_with_signal(self):
|
def test_bad_recovery_with_signal(self):
|
||||||
sig1 = self.signals[0]
|
sig1 = self.signals[0]
|
||||||
sig2 = self.signals[-1]
|
sig2 = self.signals[-1]
|
||||||
@@ -144,5 +149,10 @@ class ExitHandlerTest(ErrorHandlerTest):
|
|||||||
**self.init_kwargs)
|
**self.init_kwargs)
|
||||||
func.assert_called_once_with()
|
func.assert_called_once_with()
|
||||||
|
|
||||||
|
# On Windows, this test kills pytest itself !
|
||||||
|
@test_util.broken_on_windows
|
||||||
|
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
|
||||||
|
|||||||
@@ -37,6 +37,7 @@ class ValidateHookTest(util.TempDirTestCase):
|
|||||||
from certbot.hooks import validate_hook
|
from certbot.hooks import validate_hook
|
||||||
return validate_hook(*args, **kwargs)
|
return validate_hook(*args, **kwargs)
|
||||||
|
|
||||||
|
@util.broken_on_windows
|
||||||
def test_not_executable(self):
|
def test_not_executable(self):
|
||||||
file_path = os.path.join(self.tempdir, "foo")
|
file_path = os.path.join(self.tempdir, "foo")
|
||||||
# create a non-executable file
|
# create a non-executable file
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ from certbot import errors
|
|||||||
from certbot.tests import util as test_util
|
from certbot.tests import util as test_util
|
||||||
|
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
class LockDirTest(test_util.TempDirTestCase):
|
class LockDirTest(test_util.TempDirTestCase):
|
||||||
"""Tests for certbot.lock.lock_dir."""
|
"""Tests for certbot.lock.lock_dir."""
|
||||||
@classmethod
|
@classmethod
|
||||||
@@ -24,6 +25,7 @@ class LockDirTest(test_util.TempDirTestCase):
|
|||||||
test_util.lock_and_call(assert_raises, lock_path)
|
test_util.lock_and_call(assert_raises, lock_path)
|
||||||
|
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
class LockFileTest(test_util.TempDirTestCase):
|
class LockFileTest(test_util.TempDirTestCase):
|
||||||
"""Tests for certbot.lock.LockFile."""
|
"""Tests for certbot.lock.LockFile."""
|
||||||
@classmethod
|
@classmethod
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ import six
|
|||||||
from acme import messages
|
from acme import messages
|
||||||
from acme.magic_typing import Optional # pylint: disable=unused-import, no-name-in-module
|
from acme.magic_typing import Optional # pylint: disable=unused-import, no-name-in-module
|
||||||
|
|
||||||
|
from certbot import compat
|
||||||
from certbot import constants
|
from certbot import constants
|
||||||
from certbot import errors
|
from certbot import errors
|
||||||
from certbot import util
|
from certbot import util
|
||||||
@@ -259,7 +260,7 @@ class TempHandlerTest(unittest.TestCase):
|
|||||||
|
|
||||||
def test_permissions(self):
|
def test_permissions(self):
|
||||||
self.assertTrue(
|
self.assertTrue(
|
||||||
util.check_permissions(self.handler.path, 0o600, os.getuid()))
|
util.check_permissions(self.handler.path, 0o600, compat.os_geteuid()))
|
||||||
|
|
||||||
def test_delete(self):
|
def test_delete(self):
|
||||||
self.handler.close()
|
self.handler.close()
|
||||||
|
|||||||
+53
-28
@@ -4,6 +4,7 @@
|
|||||||
from __future__ import print_function
|
from __future__ import print_function
|
||||||
|
|
||||||
import itertools
|
import itertools
|
||||||
|
import json
|
||||||
import mock
|
import mock
|
||||||
import os
|
import os
|
||||||
import shutil
|
import shutil
|
||||||
@@ -11,6 +12,8 @@ import traceback
|
|||||||
import unittest
|
import unittest
|
||||||
import datetime
|
import datetime
|
||||||
import pytz
|
import pytz
|
||||||
|
import tempfile
|
||||||
|
import sys
|
||||||
|
|
||||||
import josepy as jose
|
import josepy as jose
|
||||||
import six
|
import six
|
||||||
@@ -588,12 +591,14 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
self.assertTrue(message in str(exc))
|
self.assertTrue(message in str(exc))
|
||||||
self.assertTrue(exc is not None)
|
self.assertTrue(exc is not None)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_noninteractive(self):
|
def test_noninteractive(self):
|
||||||
args = ['-n', 'certonly']
|
args = ['-n', 'certonly']
|
||||||
self._cli_missing_flag(args, "specify a plugin")
|
self._cli_missing_flag(args, "specify a plugin")
|
||||||
args.extend(['--standalone', '-d', 'eg.is'])
|
args.extend(['--standalone', '-d', 'eg.is'])
|
||||||
self._cli_missing_flag(args, "register before running")
|
self._cli_missing_flag(args, "register before running")
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch('certbot.main._report_new_cert')
|
@mock.patch('certbot.main._report_new_cert')
|
||||||
@mock.patch('certbot.main.client.acme_client.Client')
|
@mock.patch('certbot.main.client.acme_client.Client')
|
||||||
@mock.patch('certbot.main._determine_account')
|
@mock.patch('certbot.main._determine_account')
|
||||||
@@ -635,43 +640,46 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
@mock.patch('certbot.main.plug_sel.record_chosen_plugins')
|
@mock.patch('certbot.main.plug_sel.record_chosen_plugins')
|
||||||
@mock.patch('certbot.main.plug_sel.pick_installer')
|
@mock.patch('certbot.main.plug_sel.pick_installer')
|
||||||
def test_installer_certname(self, _inst, _rec, mock_install):
|
def test_installer_certname(self, _inst, _rec, mock_install):
|
||||||
mock_lineage = mock.MagicMock(cert_path="/tmp/cert", chain_path="/tmp/chain",
|
mock_lineage = mock.MagicMock(cert_path=test_util.temp_join('cert'),
|
||||||
fullchain_path="/tmp/chain",
|
chain_path=test_util.temp_join('chain'),
|
||||||
key_path="/tmp/privkey")
|
fullchain_path=test_util.temp_join('chain'),
|
||||||
|
key_path=test_util.temp_join('privkey'))
|
||||||
|
|
||||||
with mock.patch("certbot.cert_manager.lineage_for_certname") as mock_getlin:
|
with mock.patch("certbot.cert_manager.lineage_for_certname") as mock_getlin:
|
||||||
mock_getlin.return_value = mock_lineage
|
mock_getlin.return_value = mock_lineage
|
||||||
self._call(['install', '--cert-name', 'whatever'], mockisfile=True)
|
self._call(['install', '--cert-name', 'whatever'], mockisfile=True)
|
||||||
call_config = mock_install.call_args[0][0]
|
call_config = mock_install.call_args[0][0]
|
||||||
self.assertEqual(call_config.cert_path, "/tmp/cert")
|
self.assertEqual(call_config.cert_path, test_util.temp_join('cert'))
|
||||||
self.assertEqual(call_config.fullchain_path, "/tmp/chain")
|
self.assertEqual(call_config.fullchain_path, test_util.temp_join('chain'))
|
||||||
self.assertEqual(call_config.key_path, "/tmp/privkey")
|
self.assertEqual(call_config.key_path, test_util.temp_join('privkey'))
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch('certbot.main._install_cert')
|
@mock.patch('certbot.main._install_cert')
|
||||||
@mock.patch('certbot.main.plug_sel.record_chosen_plugins')
|
@mock.patch('certbot.main.plug_sel.record_chosen_plugins')
|
||||||
@mock.patch('certbot.main.plug_sel.pick_installer')
|
@mock.patch('certbot.main.plug_sel.pick_installer')
|
||||||
def test_installer_param_override(self, _inst, _rec, mock_install):
|
def test_installer_param_override(self, _inst, _rec, mock_install):
|
||||||
mock_lineage = mock.MagicMock(cert_path="/tmp/cert", chain_path="/tmp/chain",
|
mock_lineage = mock.MagicMock(cert_path=test_util.temp_join('cert'),
|
||||||
fullchain_path="/tmp/chain",
|
chain_path=test_util.temp_join('chain'),
|
||||||
key_path="/tmp/privkey")
|
fullchain_path=test_util.temp_join('chain'),
|
||||||
|
key_path=test_util.temp_join('privkey'))
|
||||||
with mock.patch("certbot.cert_manager.lineage_for_certname") as mock_getlin:
|
with mock.patch("certbot.cert_manager.lineage_for_certname") as mock_getlin:
|
||||||
mock_getlin.return_value = mock_lineage
|
mock_getlin.return_value = mock_lineage
|
||||||
self._call(['install', '--cert-name', 'whatever',
|
self._call(['install', '--cert-name', 'whatever',
|
||||||
'--key-path', '/tmp/overriding_privkey'], mockisfile=True)
|
'--key-path', test_util.temp_join('overriding_privkey')], mockisfile=True)
|
||||||
call_config = mock_install.call_args[0][0]
|
call_config = mock_install.call_args[0][0]
|
||||||
self.assertEqual(call_config.cert_path, "/tmp/cert")
|
self.assertEqual(call_config.cert_path, test_util.temp_join('cert'))
|
||||||
self.assertEqual(call_config.fullchain_path, "/tmp/chain")
|
self.assertEqual(call_config.fullchain_path, test_util.temp_join('chain'))
|
||||||
self.assertEqual(call_config.chain_path, "/tmp/chain")
|
self.assertEqual(call_config.chain_path, test_util.temp_join('chain'))
|
||||||
self.assertEqual(call_config.key_path, "/tmp/overriding_privkey")
|
self.assertEqual(call_config.key_path, test_util.temp_join('overriding_privkey'))
|
||||||
|
|
||||||
mock_install.reset()
|
mock_install.reset()
|
||||||
|
|
||||||
self._call(['install', '--cert-name', 'whatever',
|
self._call(['install', '--cert-name', 'whatever',
|
||||||
'--cert-path', '/tmp/overriding_cert'], mockisfile=True)
|
'--cert-path', test_util.temp_join('overriding_cert')], mockisfile=True)
|
||||||
call_config = mock_install.call_args[0][0]
|
call_config = mock_install.call_args[0][0]
|
||||||
self.assertEqual(call_config.cert_path, "/tmp/overriding_cert")
|
self.assertEqual(call_config.cert_path, test_util.temp_join('overriding_cert'))
|
||||||
self.assertEqual(call_config.fullchain_path, "/tmp/chain")
|
self.assertEqual(call_config.fullchain_path, test_util.temp_join('chain'))
|
||||||
self.assertEqual(call_config.key_path, "/tmp/privkey")
|
self.assertEqual(call_config.key_path, test_util.temp_join('privkey'))
|
||||||
|
|
||||||
@mock.patch('certbot.main.plug_sel.record_chosen_plugins')
|
@mock.patch('certbot.main.plug_sel.record_chosen_plugins')
|
||||||
@mock.patch('certbot.main.plug_sel.pick_installer')
|
@mock.patch('certbot.main.plug_sel.pick_installer')
|
||||||
@@ -686,15 +694,17 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
@mock.patch('certbot.cert_manager.get_certnames')
|
@mock.patch('certbot.cert_manager.get_certnames')
|
||||||
@mock.patch('certbot.main._install_cert')
|
@mock.patch('certbot.main._install_cert')
|
||||||
def test_installer_select_cert(self, mock_inst, mock_getcert, _inst, _rec):
|
def test_installer_select_cert(self, mock_inst, mock_getcert, _inst, _rec):
|
||||||
mock_lineage = mock.MagicMock(cert_path="/tmp/cert", chain_path="/tmp/chain",
|
mock_lineage = mock.MagicMock(cert_path=test_util.temp_join('cert'),
|
||||||
fullchain_path="/tmp/chain",
|
chain_path=test_util.temp_join('chain'),
|
||||||
key_path="/tmp/privkey")
|
fullchain_path=test_util.temp_join('chain'),
|
||||||
|
key_path=test_util.temp_join('privkey'))
|
||||||
with mock.patch("certbot.cert_manager.lineage_for_certname") as mock_getlin:
|
with mock.patch("certbot.cert_manager.lineage_for_certname") as mock_getlin:
|
||||||
mock_getlin.return_value = mock_lineage
|
mock_getlin.return_value = mock_lineage
|
||||||
self._call(['install'], mockisfile=True)
|
self._call(['install'], mockisfile=True)
|
||||||
self.assertTrue(mock_getcert.called)
|
self.assertTrue(mock_getcert.called)
|
||||||
self.assertTrue(mock_inst.called)
|
self.assertTrue(mock_inst.called)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch('certbot.main._report_new_cert')
|
@mock.patch('certbot.main._report_new_cert')
|
||||||
@mock.patch('certbot.util.exe_exists')
|
@mock.patch('certbot.util.exe_exists')
|
||||||
def test_configurator_selection(self, mock_exe_exists, unused_report):
|
def test_configurator_selection(self, mock_exe_exists, unused_report):
|
||||||
@@ -710,7 +720,8 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
# ret, _, _, _ = self._call(args)
|
# ret, _, _, _ = self._call(args)
|
||||||
# self.assertTrue("Too many flags setting" in ret)
|
# self.assertTrue("Too many flags setting" in ret)
|
||||||
|
|
||||||
args = ["install", "--nginx", "--cert-path", "/tmp/blah", "--key-path", "/tmp/blah",
|
args = ["install", "--nginx", "--cert-path",
|
||||||
|
test_util.temp_join('blah'), "--key-path", test_util.temp_join('blah'),
|
||||||
"--nginx-server-root", "/nonexistent/thing", "-d",
|
"--nginx-server-root", "/nonexistent/thing", "-d",
|
||||||
"example.com", "--debug"]
|
"example.com", "--debug"]
|
||||||
if "nginx" in real_plugins:
|
if "nginx" in real_plugins:
|
||||||
@@ -733,6 +744,7 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
self._call(["auth", "--standalone"])
|
self._call(["auth", "--standalone"])
|
||||||
self.assertEqual(1, mock_certonly.call_count)
|
self.assertEqual(1, mock_certonly.call_count)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_rollback(self):
|
def test_rollback(self):
|
||||||
_, _, _, client = self._call(['rollback'])
|
_, _, _, client = self._call(['rollback'])
|
||||||
self.assertEqual(1, client.rollback.call_count)
|
self.assertEqual(1, client.rollback.call_count)
|
||||||
@@ -760,6 +772,7 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
self._call_no_clientmock(['delete'])
|
self._call_no_clientmock(['delete'])
|
||||||
self.assertEqual(1, mock_cert_manager.call_count)
|
self.assertEqual(1, mock_cert_manager.call_count)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_plugins(self):
|
def test_plugins(self):
|
||||||
flags = ['--init', '--prepare', '--authenticators', '--installers']
|
flags = ['--init', '--prepare', '--authenticators', '--installers']
|
||||||
for args in itertools.chain(
|
for args in itertools.chain(
|
||||||
@@ -1014,7 +1027,8 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
# The location of the previous live privkey.pem is passed
|
# The location of the previous live privkey.pem is passed
|
||||||
# to obtain_certificate
|
# to obtain_certificate
|
||||||
mock_client.obtain_certificate.assert_called_once_with(['isnot.org'],
|
mock_client.obtain_certificate.assert_called_once_with(['isnot.org'],
|
||||||
os.path.join(self.config.config_dir, "live/sample-renewal/privkey.pem"))
|
os.path.normpath(os.path.join(
|
||||||
|
self.config.config_dir, "live/sample-renewal/privkey.pem")))
|
||||||
else:
|
else:
|
||||||
mock_client.obtain_certificate.assert_called_once_with(['isnot.org'], None)
|
mock_client.obtain_certificate.assert_called_once_with(['isnot.org'], None)
|
||||||
else:
|
else:
|
||||||
@@ -1039,6 +1053,7 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
self.assertTrue('fullchain.pem' in cert_msg)
|
self.assertTrue('fullchain.pem' in cert_msg)
|
||||||
self.assertTrue('donate' in get_utility().add_message.call_args[0][0])
|
self.assertTrue('donate' in get_utility().add_message.call_args[0][0])
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch('certbot.crypto_util.notAfter')
|
@mock.patch('certbot.crypto_util.notAfter')
|
||||||
def test_certonly_renewal_triggers(self, unused_notafter):
|
def test_certonly_renewal_triggers(self, unused_notafter):
|
||||||
# --dry-run should force renewal
|
# --dry-run should force renewal
|
||||||
@@ -1087,6 +1102,7 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
self.assertTrue('No renewals were attempted.' in stdout.getvalue())
|
self.assertTrue('No renewals were attempted.' in stdout.getvalue())
|
||||||
self.assertTrue('The following certs are not due for renewal yet:' in stdout.getvalue())
|
self.assertTrue('The following certs are not due for renewal yet:' in stdout.getvalue())
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_quiet_renew(self):
|
def test_quiet_renew(self):
|
||||||
test_util.make_lineage(self.config.config_dir, 'sample-renewal.conf')
|
test_util.make_lineage(self.config.config_dir, 'sample-renewal.conf')
|
||||||
args = ["renew", "--dry-run"]
|
args = ["renew", "--dry-run"]
|
||||||
@@ -1204,7 +1220,7 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
renewalparams = {'authenticator': 'webroot'}
|
renewalparams = {'authenticator': 'webroot'}
|
||||||
self._test_renew_common(
|
self._test_renew_common(
|
||||||
renewalparams=renewalparams, assert_oc_called=True,
|
renewalparams=renewalparams, assert_oc_called=True,
|
||||||
args=['renew', '--webroot-map', '{"example.com": "/tmp"}'])
|
args=['renew', '--webroot-map', json.dumps({'example.com': tempfile.gettempdir()})])
|
||||||
|
|
||||||
def test_renew_reconstitute_error(self):
|
def test_renew_reconstitute_error(self):
|
||||||
# pylint: disable=protected-access
|
# pylint: disable=protected-access
|
||||||
@@ -1234,7 +1250,9 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
def test_no_renewal_with_hooks(self):
|
def test_no_renewal_with_hooks(self):
|
||||||
_, _, stdout = self._test_renewal_common(
|
_, _, stdout = self._test_renewal_common(
|
||||||
due_for_renewal=False, extra_args=None, should_renew=False,
|
due_for_renewal=False, extra_args=None, should_renew=False,
|
||||||
args=['renew', '--post-hook', 'echo hello world'])
|
args=['renew', '--post-hook',
|
||||||
|
'{0} -c "from __future__ import print_function; print(\'hello world\');"'
|
||||||
|
.format(sys.executable)])
|
||||||
self.assertTrue('No hooks were run.' in stdout.getvalue())
|
self.assertTrue('No hooks were run.' in stdout.getvalue())
|
||||||
|
|
||||||
@test_util.patch_get_utility()
|
@test_util.patch_get_utility()
|
||||||
@@ -1254,13 +1272,19 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
chain = 'chain'
|
chain = 'chain'
|
||||||
mock_client = mock.MagicMock()
|
mock_client = mock.MagicMock()
|
||||||
mock_client.obtain_certificate_from_csr.return_value = (certr, chain)
|
mock_client.obtain_certificate_from_csr.return_value = (certr, chain)
|
||||||
cert_path = '/etc/letsencrypt/live/example.com/cert_512.pem'
|
cert_path = os.path.normpath(os.path.join(
|
||||||
full_path = '/etc/letsencrypt/live/example.com/fullchain.pem'
|
self.config.config_dir,
|
||||||
|
'live/example.com/cert_512.pem'))
|
||||||
|
full_path = os.path.normpath(os.path.join(
|
||||||
|
self.config.config_dir,
|
||||||
|
'live/example.com/fullchain.pem'))
|
||||||
mock_client.save_certificate.return_value = cert_path, None, full_path
|
mock_client.save_certificate.return_value = cert_path, None, full_path
|
||||||
with mock.patch('certbot.main._init_le_client') as mock_init:
|
with mock.patch('certbot.main._init_le_client') as mock_init:
|
||||||
mock_init.return_value = mock_client
|
mock_init.return_value = mock_client
|
||||||
with test_util.patch_get_utility() as mock_get_utility:
|
with test_util.patch_get_utility() as mock_get_utility:
|
||||||
chain_path = '/etc/letsencrypt/live/example.com/chain.pem'
|
chain_path = os.path.normpath(os.path.join(
|
||||||
|
self.config.config_dir,
|
||||||
|
'live/example.com/chain.pem'))
|
||||||
args = ('-a standalone certonly --csr {0} --cert-path {1} '
|
args = ('-a standalone certonly --csr {0} --cert-path {1} '
|
||||||
'--chain-path {2} --fullchain-path {3}').format(
|
'--chain-path {2} --fullchain-path {3}').format(
|
||||||
CSR, cert_path, chain_path, full_path).split()
|
CSR, cert_path, chain_path, full_path).split()
|
||||||
@@ -1334,6 +1358,7 @@ class MainTest(test_util.ConfigTestCase): # pylint: disable=too-many-public-met
|
|||||||
self._call(['-c', test_util.vector_path('cli.ini')])
|
self._call(['-c', test_util.vector_path('cli.ini')])
|
||||||
self.assertTrue(mocked_run.called)
|
self.assertTrue(mocked_run.called)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_register(self):
|
def test_register(self):
|
||||||
with mock.patch('certbot.main.client') as mocked_client:
|
with mock.patch('certbot.main.client') as mocked_client:
|
||||||
acc = mock.MagicMock()
|
acc = mock.MagicMock()
|
||||||
|
|||||||
@@ -50,6 +50,7 @@ class ReverterCheckpointLocalTest(test_util.ConfigTestCase):
|
|||||||
x = f.read()
|
x = f.read()
|
||||||
self.assertTrue("No changes" in x)
|
self.assertTrue("No changes" in x)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_basic_add_to_temp_checkpoint(self):
|
def test_basic_add_to_temp_checkpoint(self):
|
||||||
# These shouldn't conflict even though they are both named config.txt
|
# These shouldn't conflict even though they are both named config.txt
|
||||||
self.reverter.add_to_temp_checkpoint(self.sets[0], "save1")
|
self.reverter.add_to_temp_checkpoint(self.sets[0], "save1")
|
||||||
@@ -91,6 +92,7 @@ class ReverterCheckpointLocalTest(test_util.ConfigTestCase):
|
|||||||
self.assertRaises(errors.ReverterError, self.reverter.add_to_checkpoint,
|
self.assertRaises(errors.ReverterError, self.reverter.add_to_checkpoint,
|
||||||
set([config3]), "invalid save")
|
set([config3]), "invalid save")
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_multiple_saves_and_temp_revert(self):
|
def test_multiple_saves_and_temp_revert(self):
|
||||||
self.reverter.add_to_temp_checkpoint(self.sets[0], "save1")
|
self.reverter.add_to_temp_checkpoint(self.sets[0], "save1")
|
||||||
update_file(self.config1, "updated-directive")
|
update_file(self.config1, "updated-directive")
|
||||||
@@ -120,6 +122,7 @@ class ReverterCheckpointLocalTest(test_util.ConfigTestCase):
|
|||||||
self.assertFalse(os.path.isfile(config3))
|
self.assertFalse(os.path.isfile(config3))
|
||||||
self.assertFalse(os.path.isfile(config4))
|
self.assertFalse(os.path.isfile(config4))
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_multiple_registration_same_file(self):
|
def test_multiple_registration_same_file(self):
|
||||||
self.reverter.register_file_creation(True, self.config1)
|
self.reverter.register_file_creation(True, self.config1)
|
||||||
self.reverter.register_file_creation(True, self.config1)
|
self.reverter.register_file_creation(True, self.config1)
|
||||||
@@ -144,6 +147,7 @@ class ReverterCheckpointLocalTest(test_util.ConfigTestCase):
|
|||||||
errors.ReverterError, self.reverter.register_file_creation,
|
errors.ReverterError, self.reverter.register_file_creation,
|
||||||
"filepath")
|
"filepath")
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_register_undo_command(self):
|
def test_register_undo_command(self):
|
||||||
coms = [
|
coms = [
|
||||||
["a2dismod", "ssl"],
|
["a2dismod", "ssl"],
|
||||||
@@ -166,6 +170,7 @@ class ReverterCheckpointLocalTest(test_util.ConfigTestCase):
|
|||||||
errors.ReverterError, self.reverter.register_undo_command,
|
errors.ReverterError, self.reverter.register_undo_command,
|
||||||
True, ["command"])
|
True, ["command"])
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch("certbot.util.run_script")
|
@mock.patch("certbot.util.run_script")
|
||||||
def test_run_undo_commands(self, mock_run):
|
def test_run_undo_commands(self, mock_run):
|
||||||
mock_run.side_effect = ["", errors.SubprocessError]
|
mock_run.side_effect = ["", errors.SubprocessError]
|
||||||
@@ -229,6 +234,7 @@ class ReverterCheckpointLocalTest(test_util.ConfigTestCase):
|
|||||||
self.assertRaises(
|
self.assertRaises(
|
||||||
errors.ReverterError, self.reverter.revert_temporary_config)
|
errors.ReverterError, self.reverter.revert_temporary_config)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch("certbot.reverter.logger.warning")
|
@mock.patch("certbot.reverter.logger.warning")
|
||||||
def test_recover_checkpoint_missing_new_files(self, mock_warn):
|
def test_recover_checkpoint_missing_new_files(self, mock_warn):
|
||||||
self.reverter.register_file_creation(
|
self.reverter.register_file_creation(
|
||||||
@@ -243,6 +249,7 @@ class ReverterCheckpointLocalTest(test_util.ConfigTestCase):
|
|||||||
self.assertRaises(
|
self.assertRaises(
|
||||||
errors.ReverterError, self.reverter.revert_temporary_config)
|
errors.ReverterError, self.reverter.revert_temporary_config)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_recovery_routine_temp_and_perm(self):
|
def test_recovery_routine_temp_and_perm(self):
|
||||||
# Register a new perm checkpoint file
|
# Register a new perm checkpoint file
|
||||||
config3 = os.path.join(self.dir1, "config3.txt")
|
config3 = os.path.join(self.dir1, "config3.txt")
|
||||||
@@ -306,6 +313,7 @@ class TestFullCheckpointsReverter(test_util.ConfigTestCase):
|
|||||||
self.assertRaises(
|
self.assertRaises(
|
||||||
errors.ReverterError, self.reverter.rollback_checkpoints, "one")
|
errors.ReverterError, self.reverter.rollback_checkpoints, "one")
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_rollback_finalize_checkpoint_valid_inputs(self):
|
def test_rollback_finalize_checkpoint_valid_inputs(self):
|
||||||
|
|
||||||
config3 = self._setup_three_checkpoints()
|
config3 = self._setup_three_checkpoints()
|
||||||
@@ -357,6 +365,7 @@ class TestFullCheckpointsReverter(test_util.ConfigTestCase):
|
|||||||
self.assertRaises(
|
self.assertRaises(
|
||||||
errors.ReverterError, self.reverter.finalize_checkpoint, "Title")
|
errors.ReverterError, self.reverter.finalize_checkpoint, "Title")
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch("certbot.reverter.logger")
|
@mock.patch("certbot.reverter.logger")
|
||||||
def test_rollback_too_many(self, mock_logger):
|
def test_rollback_too_many(self, mock_logger):
|
||||||
# Test no exist warning...
|
# Test no exist warning...
|
||||||
@@ -369,6 +378,7 @@ class TestFullCheckpointsReverter(test_util.ConfigTestCase):
|
|||||||
self.reverter.rollback_checkpoints(4)
|
self.reverter.rollback_checkpoints(4)
|
||||||
self.assertEqual(mock_logger.warning.call_count, 1)
|
self.assertEqual(mock_logger.warning.call_count, 1)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
def test_multi_rollback(self):
|
def test_multi_rollback(self):
|
||||||
config3 = self._setup_three_checkpoints()
|
config3 = self._setup_three_checkpoints()
|
||||||
self.reverter.rollback_checkpoints(3)
|
self.reverter.rollback_checkpoints(3)
|
||||||
|
|||||||
@@ -480,6 +480,7 @@ class RenewableCertTests(BaseRenewableCertTest):
|
|||||||
self.assertTrue(self.test_rc.should_autorenew())
|
self.assertTrue(self.test_rc.should_autorenew())
|
||||||
mock_ocsp.return_value = False
|
mock_ocsp.return_value = False
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch("certbot.storage.relevant_values")
|
@mock.patch("certbot.storage.relevant_values")
|
||||||
def test_save_successor(self, mock_rv):
|
def test_save_successor(self, mock_rv):
|
||||||
# Mock relevant_values() to claim that all values are relevant here
|
# Mock relevant_values() to claim that all values are relevant here
|
||||||
|
|||||||
+44
-2
@@ -9,6 +9,8 @@ import pkg_resources
|
|||||||
import shutil
|
import shutil
|
||||||
import tempfile
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
|
import sys
|
||||||
|
import warnings
|
||||||
|
|
||||||
from cryptography.hazmat.backends import default_backend
|
from cryptography.hazmat.backends import default_backend
|
||||||
from cryptography.hazmat.primitives import serialization
|
from cryptography.hazmat.primitives import serialization
|
||||||
@@ -36,8 +38,15 @@ def vector_path(*names):
|
|||||||
def load_vector(*names):
|
def load_vector(*names):
|
||||||
"""Load contents of a test vector."""
|
"""Load contents of a test vector."""
|
||||||
# luckily, resource_string opens file in binary mode
|
# luckily, resource_string opens file in binary mode
|
||||||
return pkg_resources.resource_string(
|
data = pkg_resources.resource_string(
|
||||||
__name__, os.path.join('testdata', *names))
|
__name__, os.path.join('testdata', *names))
|
||||||
|
# Try at most to convert CRLF to LF when data is text
|
||||||
|
try:
|
||||||
|
return data.decode().replace('\r\n', '\n').encode()
|
||||||
|
except ValueError:
|
||||||
|
# Failed to process the file with standard encoding.
|
||||||
|
# Most likely not a text file, return its bytes untouched.
|
||||||
|
return data
|
||||||
|
|
||||||
|
|
||||||
def _guess_loader(filename, loader_pem, loader_der):
|
def _guess_loader(filename, loader_pem, loader_der):
|
||||||
@@ -314,10 +323,21 @@ class TempDirTestCase(unittest.TestCase):
|
|||||||
"""Base test class which sets up and tears down a temporary directory"""
|
"""Base test class which sets up and tears down a temporary directory"""
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
|
"""Execute before test"""
|
||||||
self.tempdir = tempfile.mkdtemp()
|
self.tempdir = tempfile.mkdtemp()
|
||||||
|
|
||||||
def tearDown(self):
|
def tearDown(self):
|
||||||
shutil.rmtree(self.tempdir)
|
"""Execute after test"""
|
||||||
|
# Then we have various files which are not correctly closed at the time of tearDown.
|
||||||
|
# On Windows, it is visible for the same reasons as above.
|
||||||
|
# For know, we log them until a proper file close handling is written.
|
||||||
|
def onerror_handler(_, path, excinfo):
|
||||||
|
"""On error handler"""
|
||||||
|
message = ('Following error occurred when deleting the tempdir {0}'
|
||||||
|
' for path {1} during tearDown process: {2}'
|
||||||
|
.format(self.tempdir, path, str(excinfo)))
|
||||||
|
warnings.warn(message)
|
||||||
|
shutil.rmtree(self.tempdir, onerror=onerror_handler)
|
||||||
|
|
||||||
class ConfigTestCase(TempDirTestCase):
|
class ConfigTestCase(TempDirTestCase):
|
||||||
"""Test class which sets up a NamespaceConfig object.
|
"""Test class which sets up a NamespaceConfig object.
|
||||||
@@ -378,3 +398,25 @@ def hold_lock(cv, lock_path): # pragma: no cover
|
|||||||
cv.notify()
|
cv.notify()
|
||||||
cv.wait()
|
cv.wait()
|
||||||
my_lock.release()
|
my_lock.release()
|
||||||
|
|
||||||
|
def skip_on_windows(reason):
|
||||||
|
"""Decorator to skip permanently a test on Windows. A reason is required."""
|
||||||
|
def wrapper(function):
|
||||||
|
"""Wrapped version"""
|
||||||
|
return unittest.skipIf(sys.platform == 'win32', reason)(function)
|
||||||
|
return wrapper
|
||||||
|
|
||||||
|
def broken_on_windows(function):
|
||||||
|
"""Decorator to skip temporarily a broken test on Windows."""
|
||||||
|
reason = 'Test is broken and ignored on windows but should be fixed.'
|
||||||
|
return unittest.skipIf(
|
||||||
|
sys.platform == 'win32'
|
||||||
|
and os.environ.get('SKIP_BROKEN_TESTS_ON_WINDOWS', 'true') == 'true',
|
||||||
|
reason)(function)
|
||||||
|
|
||||||
|
def temp_join(path):
|
||||||
|
"""
|
||||||
|
Return the given path joined to the tempdir path for the current platform
|
||||||
|
Eg.: 'cert' => /tmp/cert (Linux) or 'C:\\Users\\currentuser\\AppData\\Temp\\cert' (Windows)
|
||||||
|
"""
|
||||||
|
return os.path.join(tempfile.gettempdir(), path)
|
||||||
|
|||||||
+14
-14
@@ -3,7 +3,6 @@ import argparse
|
|||||||
import errno
|
import errno
|
||||||
import os
|
import os
|
||||||
import shutil
|
import shutil
|
||||||
import stat
|
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
import mock
|
import mock
|
||||||
@@ -89,6 +88,7 @@ class LockDirUntilExit(test_util.TempDirTestCase):
|
|||||||
import certbot.util
|
import certbot.util
|
||||||
reload_module(certbot.util)
|
reload_module(certbot.util)
|
||||||
|
|
||||||
|
@test_util.broken_on_windows
|
||||||
@mock.patch('certbot.util.logger')
|
@mock.patch('certbot.util.logger')
|
||||||
@mock.patch('certbot.util.atexit_register')
|
@mock.patch('certbot.util.atexit_register')
|
||||||
def test_it(self, mock_register, mock_logger):
|
def test_it(self, mock_register, mock_logger):
|
||||||
@@ -140,9 +140,9 @@ class MakeOrVerifyDirTest(test_util.TempDirTestCase):
|
|||||||
super(MakeOrVerifyDirTest, self).setUp()
|
super(MakeOrVerifyDirTest, self).setUp()
|
||||||
|
|
||||||
self.path = os.path.join(self.tempdir, "foo")
|
self.path = os.path.join(self.tempdir, "foo")
|
||||||
os.mkdir(self.path, 0o400)
|
os.mkdir(self.path, 0o600)
|
||||||
|
|
||||||
self.uid = os.getuid()
|
self.uid = compat.os_geteuid()
|
||||||
|
|
||||||
def _call(self, directory, mode):
|
def _call(self, directory, mode):
|
||||||
from certbot.util import make_or_verify_dir
|
from certbot.util import make_or_verify_dir
|
||||||
@@ -152,14 +152,15 @@ class MakeOrVerifyDirTest(test_util.TempDirTestCase):
|
|||||||
path = os.path.join(self.tempdir, "bar")
|
path = os.path.join(self.tempdir, "bar")
|
||||||
self._call(path, 0o650)
|
self._call(path, 0o650)
|
||||||
self.assertTrue(os.path.isdir(path))
|
self.assertTrue(os.path.isdir(path))
|
||||||
self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o650)
|
self.assertTrue(compat.compare_file_modes(os.stat(path).st_mode, 0o650))
|
||||||
|
|
||||||
def test_existing_correct_mode_does_not_fail(self):
|
def test_existing_correct_mode_does_not_fail(self):
|
||||||
self._call(self.path, 0o400)
|
self._call(self.path, 0o600)
|
||||||
self.assertEqual(stat.S_IMODE(os.stat(self.path).st_mode), 0o400)
|
self.assertTrue(compat.compare_file_modes(os.stat(self.path).st_mode, 0o600))
|
||||||
|
|
||||||
|
@test_util.skip_on_windows('Umask modes are mostly ignored on Windows.')
|
||||||
def test_existing_wrong_mode_fails(self):
|
def test_existing_wrong_mode_fails(self):
|
||||||
self.assertRaises(errors.Error, self._call, self.path, 0o600)
|
self.assertRaises(errors.Error, self._call, self.path, 0o400)
|
||||||
|
|
||||||
def test_reraises_os_error(self):
|
def test_reraises_os_error(self):
|
||||||
with mock.patch.object(os, "makedirs") as makedirs:
|
with mock.patch.object(os, "makedirs") as makedirs:
|
||||||
@@ -178,7 +179,7 @@ class CheckPermissionsTest(test_util.TempDirTestCase):
|
|||||||
def setUp(self):
|
def setUp(self):
|
||||||
super(CheckPermissionsTest, self).setUp()
|
super(CheckPermissionsTest, self).setUp()
|
||||||
|
|
||||||
self.uid = os.getuid()
|
self.uid = compat.os_geteuid()
|
||||||
|
|
||||||
def _call(self, mode):
|
def _call(self, mode):
|
||||||
from certbot.util import check_permissions
|
from certbot.util import check_permissions
|
||||||
@@ -212,8 +213,8 @@ class UniqueFileTest(test_util.TempDirTestCase):
|
|||||||
self.assertEqual(open(name).read(), "bar")
|
self.assertEqual(open(name).read(), "bar")
|
||||||
|
|
||||||
def test_right_mode(self):
|
def test_right_mode(self):
|
||||||
self.assertEqual(0o700, os.stat(self._call(0o700)[1]).st_mode & 0o777)
|
self.assertTrue(compat.compare_file_modes(0o700, os.stat(self._call(0o700)[1]).st_mode))
|
||||||
self.assertEqual(0o100, os.stat(self._call(0o100)[1]).st_mode & 0o777)
|
self.assertTrue(compat.compare_file_modes(0o600, os.stat(self._call(0o600)[1]).st_mode))
|
||||||
|
|
||||||
def test_default_exists(self):
|
def test_default_exists(self):
|
||||||
name1 = self._call()[1] # create 0000_foo.txt
|
name1 = self._call()[1] # create 0000_foo.txt
|
||||||
@@ -513,17 +514,16 @@ class OsInfoTest(unittest.TestCase):
|
|||||||
|
|
||||||
def test_systemd_os_release(self):
|
def test_systemd_os_release(self):
|
||||||
from certbot.util import (get_os_info, get_systemd_os_info,
|
from certbot.util import (get_os_info, get_systemd_os_info,
|
||||||
get_os_info_ua)
|
get_os_info_ua)
|
||||||
|
|
||||||
with mock.patch('os.path.isfile', return_value=True):
|
with mock.patch('os.path.isfile', return_value=True):
|
||||||
self.assertEqual(get_os_info(
|
self.assertEqual(get_os_info(
|
||||||
test_util.vector_path("os-release"))[0], 'systemdos')
|
test_util.vector_path("os-release"))[0], 'systemdos')
|
||||||
self.assertEqual(get_os_info(
|
self.assertEqual(get_os_info(
|
||||||
test_util.vector_path("os-release"))[1], '42')
|
test_util.vector_path("os-release"))[1], '42')
|
||||||
self.assertEqual(get_systemd_os_info("/dev/null"), ("", ""))
|
self.assertEqual(get_systemd_os_info(os.devnull), ("", ""))
|
||||||
self.assertEqual(get_os_info_ua(
|
self.assertEqual(get_os_info_ua(
|
||||||
test_util.vector_path("os-release")),
|
test_util.vector_path("os-release")), "SystemdOS")
|
||||||
"SystemdOS")
|
|
||||||
with mock.patch('os.path.isfile', return_value=False):
|
with mock.patch('os.path.isfile', return_value=False):
|
||||||
self.assertEqual(get_systemd_os_info(), ("", ""))
|
self.assertEqual(get_systemd_os_info(), ("", ""))
|
||||||
|
|
||||||
|
|||||||
+2
-2
@@ -12,7 +12,6 @@ import platform
|
|||||||
import re
|
import re
|
||||||
import six
|
import six
|
||||||
import socket
|
import socket
|
||||||
import stat
|
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
@@ -21,6 +20,7 @@ from collections import OrderedDict
|
|||||||
import configargparse
|
import configargparse
|
||||||
|
|
||||||
from acme.magic_typing import Tuple, Union # pylint: disable=unused-import, no-name-in-module
|
from acme.magic_typing import Tuple, Union # pylint: disable=unused-import, no-name-in-module
|
||||||
|
from certbot import compat
|
||||||
from certbot import constants
|
from certbot import constants
|
||||||
from certbot import errors
|
from certbot import errors
|
||||||
from certbot import lock
|
from certbot import lock
|
||||||
@@ -204,7 +204,7 @@ def check_permissions(filepath, mode, uid=0):
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
file_stat = os.stat(filepath)
|
file_stat = os.stat(filepath)
|
||||||
return stat.S_IMODE(file_stat.st_mode) == mode and file_stat.st_uid == uid
|
return compat.compare_file_modes(file_stat.st_mode, mode) and file_stat.st_uid == uid
|
||||||
|
|
||||||
|
|
||||||
def safe_open(path, mode="w", chmod=None, buffering=None):
|
def safe_open(path, mode="w", chmod=None, buffering=None):
|
||||||
|
|||||||
+13
@@ -0,0 +1,13 @@
|
|||||||
|
[tox]
|
||||||
|
skipsdist = True
|
||||||
|
envlist = py{34,35,36,37}-cover
|
||||||
|
|
||||||
|
[testenv]
|
||||||
|
deps = -e acme[dev]
|
||||||
|
-e .[dev]
|
||||||
|
commands = pytest -n auto --pyargs acme
|
||||||
|
pytest -n auto --pyargs certbot
|
||||||
|
|
||||||
|
[testenv:cover]
|
||||||
|
commands = pytest -n auto --cov acme --pyargs acme
|
||||||
|
pytest -n auto --cov certbot --cov-append --pyargs certbot
|
||||||
Reference in New Issue
Block a user