mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 00:35:50 +02:00
Add comments
This commit is contained in:
@@ -5,4 +5,4 @@ disable_warnings = module-not-imported,no-data-collected
|
||||
|
||||
[report]
|
||||
# Exclude unit tests in coverage during integration tests.
|
||||
omit = **/*_test.py,**/tests/*,**/certbot_nginx/parser_obj.py
|
||||
omit = **/*_test.py,**/tests/*,**/dns_common*,**/certbot_nginx/parser_obj.py
|
||||
|
||||
@@ -1,12 +1,21 @@
|
||||
"""This module contains advanced assertions for the certbot integration tests."""
|
||||
|
||||
import os
|
||||
import grp
|
||||
|
||||
|
||||
# Defining __all__ allows to make a static import "from assertions import *" without importing
|
||||
# also the modules that are part of this module.
|
||||
__all__ = ['assert_hook_execution', 'assert_save_renew_hook', 'assert_certs_count_for_lineage',
|
||||
'assert_equals_permissions', 'assert_equals_group_owner', 'assert_world_permissions']
|
||||
|
||||
|
||||
def assert_hook_execution(probe_path, probe_content):
|
||||
"""
|
||||
Assert that a certbot hook has been executed
|
||||
:param probe_path: path to the file that received the hook output
|
||||
:param probe_content: content expected when the hook is executed
|
||||
"""
|
||||
with open(probe_path, 'r') as file:
|
||||
lines = file.readlines()
|
||||
|
||||
@@ -15,10 +24,21 @@ def assert_hook_execution(probe_path, probe_content):
|
||||
|
||||
|
||||
def assert_save_renew_hook(config_dir, lineage):
|
||||
"""
|
||||
Assert that the renew hook configuration of a lineage has been saved.
|
||||
:param config_dir: location of the certbot configuration
|
||||
:param lineage: lineage domain name
|
||||
"""
|
||||
assert os.path.isfile(os.path.join(config_dir, 'renewal/{0}.conf'.format(lineage)))
|
||||
|
||||
|
||||
def assert_certs_count_for_lineage(config_dir, lineage, count):
|
||||
"""
|
||||
Assert the number of certificates generated for a lineage.
|
||||
:param config_dir: location of the certbot configuration
|
||||
:param lineage: lineage domain name
|
||||
:param count: number of expected certificates
|
||||
"""
|
||||
archive_dir = os.path.join(config_dir, 'archive')
|
||||
lineage_dir = os.path.join(archive_dir, lineage)
|
||||
certs = [file for file in os.listdir(lineage_dir) if file.startswith('cert')]
|
||||
@@ -26,6 +46,12 @@ def assert_certs_count_for_lineage(config_dir, lineage, count):
|
||||
|
||||
|
||||
def assert_equals_permissions(file1, file2, mask):
|
||||
"""
|
||||
Assert that permissions on two files are identical in respect to a given umask.
|
||||
:param file1: first file path to compare
|
||||
:param file2: second file path to compare
|
||||
:param mask: umask to apply before comparing file modes
|
||||
"""
|
||||
mode_file1 = os.stat(file1).st_mode & mask
|
||||
mode_file2 = os.stat(file2).st_mode & mask
|
||||
|
||||
@@ -33,6 +59,12 @@ def assert_equals_permissions(file1, file2, mask):
|
||||
|
||||
|
||||
def assert_equals_group_owner(file1, file2):
|
||||
"""
|
||||
Assert that two files have the same group owner.
|
||||
:param file1: first file path to compare
|
||||
:param file2: second file path to compare
|
||||
:return:
|
||||
"""
|
||||
group_owner_file1 = grp.getgrgid(os.stat(file1).st_gid)[0]
|
||||
group_owner_file2 = grp.getgrgid(os.stat(file2).st_gid)[0]
|
||||
|
||||
@@ -40,6 +72,11 @@ def assert_equals_group_owner(file1, file2):
|
||||
|
||||
|
||||
def assert_world_permissions(file, mode):
|
||||
"""
|
||||
Assert that a file has the expected world permission.
|
||||
:param file: file path to check
|
||||
:param mode: world permissions mode expected
|
||||
"""
|
||||
mode_file_all = os.stat(file).st_mode & 0o007
|
||||
|
||||
assert mode_file_all == mode
|
||||
|
||||
@@ -101,5 +101,14 @@ class IntegrationTestsContext(object):
|
||||
command.extend(args)
|
||||
return self.common_no_force_renew(command)
|
||||
|
||||
def wtf(self, prefix='le'):
|
||||
return '{0}.{1}.wtf'.format(prefix, self.worker_id)
|
||||
def wtf(self, subdomain='le'):
|
||||
"""
|
||||
Generate a certificate name suitable for distributed certbot integration tests.
|
||||
This is a requirement to let the distribution knows how to redirect the challenge check
|
||||
from the ACME server to the relevant pytest-xdist worker. This resolution is done by
|
||||
appending the pytest worker id to the domain, using this pattern:
|
||||
{subdomain}.{worker_id}.wtf
|
||||
:param subdomain: the subdomain to use in the generated domain (default 'le')
|
||||
:return: the well-formed domain suitable for redirection on
|
||||
"""
|
||||
return '{0}.{1}.wtf'.format(subdomain, self.worker_id)
|
||||
|
||||
@@ -23,6 +23,7 @@ def context(request):
|
||||
|
||||
|
||||
def test_basic_commands(context):
|
||||
"""Test simple commands on Certbot CLI."""
|
||||
# TMPDIR env variable is set to workspace for the certbot subprocess.
|
||||
# So tempdir module will create any temporary files/dirs in workspace,
|
||||
# and its content can be tested to check correct certbot cleanup.
|
||||
@@ -40,6 +41,7 @@ def test_basic_commands(context):
|
||||
|
||||
|
||||
def test_hook_dirs_creation(context):
|
||||
"""Test all hooks directory are created during Certbot startup."""
|
||||
context.common(['register'])
|
||||
|
||||
for hook_dir in misc.list_renewal_hooks_dirs(context.config_dir):
|
||||
@@ -47,6 +49,7 @@ def test_hook_dirs_creation(context):
|
||||
|
||||
|
||||
def test_registration_override(context):
|
||||
"""Test correct register/unregister, and registration override."""
|
||||
context.common(['register'])
|
||||
context.common(['unregister'])
|
||||
context.common(['register', '--email', 'ex1@domain.org,ex2@domain.org'])
|
||||
@@ -61,12 +64,16 @@ def test_registration_override(context):
|
||||
|
||||
|
||||
def test_prepare_plugins(context):
|
||||
"""Test that plugins are correctly instantiated and displayed."""
|
||||
output = context.common(['plugins', '--init', '--prepare'])
|
||||
|
||||
assert 'webroot' in output
|
||||
|
||||
|
||||
def test_http_01(context):
|
||||
"""Test the HTTP-01 challenge using standalone plugin."""
|
||||
# We start a server listening on the port for the
|
||||
# TLS-SNI challenge to prevent regressions in #3601.
|
||||
with misc.create_tcp_server(context.tls_alpn_01_port):
|
||||
certname = context.wtf('le2')
|
||||
context.common([
|
||||
@@ -82,6 +89,7 @@ def test_http_01(context):
|
||||
|
||||
|
||||
def test_manual_http_auth(context):
|
||||
"""Test the HTTP-01 challenge using manual plugin."""
|
||||
with misc.create_tcp_server(context.http_01_port) as webroot:
|
||||
manual_http_hooks = misc.manual_http_hooks(webroot)
|
||||
|
||||
@@ -102,6 +110,7 @@ def test_manual_http_auth(context):
|
||||
|
||||
|
||||
def test_manual_dns_auth(context):
|
||||
"""Test the DNS-01 challenge using manual plugin."""
|
||||
certname = context.wtf('dns')
|
||||
context.common([
|
||||
'-a', 'manual', '-d', certname, '--preferred-challenges', 'dns',
|
||||
@@ -119,10 +128,12 @@ def test_manual_dns_auth(context):
|
||||
|
||||
|
||||
def test_certonly(context):
|
||||
"""Test the certonly verb on certbot."""
|
||||
context.common(['certonly', '--cert-name', 'newname', '-d', context.wtf('newname')])
|
||||
|
||||
|
||||
def test_auth_and_install_with_csr(context):
|
||||
"""Test certificate issuance and install using an existing CSR."""
|
||||
certname = context.wtf('le3')
|
||||
key_path = join(context.workspace, 'key.pem')
|
||||
csr_path = join(context.workspace, 'csr.der')
|
||||
@@ -149,6 +160,7 @@ def test_auth_and_install_with_csr(context):
|
||||
|
||||
|
||||
def test_renew(context):
|
||||
"""Test various certificate renew scenarios."""
|
||||
# First, we create a target certificate, with all hook dirs instantiated.
|
||||
# We should have a new certificate, with hooks executed.
|
||||
# Check also file permissions.
|
||||
@@ -233,6 +245,7 @@ def test_renew(context):
|
||||
|
||||
|
||||
def test_hook_override(context):
|
||||
"""Test correct hook override on renew."""
|
||||
certname = context.wtf('override')
|
||||
context.common([
|
||||
'certonly', '-d', certname,
|
||||
@@ -246,6 +259,7 @@ def test_hook_override(context):
|
||||
assert_hook_execution(context.hook_probe, 'post')
|
||||
assert_hook_execution(context.hook_probe, 'deploy')
|
||||
|
||||
# Now we override all previous hooks during next renew.
|
||||
open(context.hook_probe, 'w').close()
|
||||
context.common([
|
||||
'renew', '--cert-name', certname,
|
||||
@@ -264,6 +278,7 @@ def test_hook_override(context):
|
||||
with pytest.raises(AssertionError):
|
||||
assert_hook_execution(context.hook_probe, 'deploy')
|
||||
|
||||
# Expect that this renew will reuse new hooks registered in the previous renew.
|
||||
open(context.hook_probe, 'w').close()
|
||||
context.common(['renew', '--cert-name', certname])
|
||||
|
||||
@@ -273,6 +288,8 @@ def test_hook_override(context):
|
||||
|
||||
|
||||
def test_invalid_domain_with_dns_challenge(context):
|
||||
"""Test certificate issuance failure with DNS-01 challenge."""
|
||||
# Manual dns auth hooks from misc are designed to fail if the domain contains 'fail-*'.
|
||||
certs = ','.join([context.wtf('dns1'), context.wtf('fail-dns1')])
|
||||
context.common([
|
||||
'-a', 'manual', '-d', certs,
|
||||
@@ -288,6 +305,7 @@ def test_invalid_domain_with_dns_challenge(context):
|
||||
|
||||
|
||||
def test_reuse_key(context):
|
||||
"""Test various scenarios where a key is reused."""
|
||||
certname = context.wtf('reusekey')
|
||||
context.common(['--domains', certname, '--reuse-key'])
|
||||
context.common(['renew', '--cert-name', certname])
|
||||
@@ -315,6 +333,7 @@ def test_reuse_key(context):
|
||||
|
||||
|
||||
def test_ecdsa(context):
|
||||
"""Test certificate issuance with ECDSA key."""
|
||||
key_path = join(context.workspace, 'privkey-p384.pem')
|
||||
csr_path = join(context.workspace, 'csr-p384.der')
|
||||
cert_path = join(context.workspace, 'cert-p384.pem')
|
||||
@@ -328,6 +347,7 @@ def test_ecdsa(context):
|
||||
|
||||
|
||||
def test_ocsp_must_staple(context):
|
||||
"""Test that OCSP Must-Staple is correctly set in the generated certificate."""
|
||||
certname = context.wtf('must-staple')
|
||||
context.common(['auth', '--must-staple', '--domains', certname])
|
||||
|
||||
@@ -337,6 +357,8 @@ def test_ocsp_must_staple(context):
|
||||
|
||||
|
||||
def test_revoke_simple(context):
|
||||
"""Test various scenarios that revokes a certificate."""
|
||||
# Default action after revoke is to delete the certificate.
|
||||
certname = context.wtf()
|
||||
cert_path = join(context.config_dir, 'live/{0}/cert.pem'.format(certname))
|
||||
context.common(['-d', certname])
|
||||
@@ -344,12 +366,14 @@ def test_revoke_simple(context):
|
||||
|
||||
assert not exists(cert_path)
|
||||
|
||||
# Check default deletion is overridden.
|
||||
certname = context.wtf('le1')
|
||||
cert_path = join(context.config_dir, 'live/{0}/cert.pem'.format(certname))
|
||||
context.common(['-d', certname])
|
||||
context.common(['revoke', '--cert-path', cert_path, '--no-delete-after-revoke'])
|
||||
|
||||
assert exists(cert_path)
|
||||
|
||||
context.common(['delete', '--cert-name', certname])
|
||||
|
||||
assert not exists(join(context.config_dir, 'archive/{0}'.format(certname)))
|
||||
@@ -364,6 +388,7 @@ def test_revoke_simple(context):
|
||||
|
||||
|
||||
def test_revoke_and_unregister(context):
|
||||
"""Test revoke with a reason then unregister."""
|
||||
cert1 = context.wtf('le1')
|
||||
cert2 = context.wtf('le2')
|
||||
cert3 = context.wtf('le3')
|
||||
@@ -391,6 +416,8 @@ def test_revoke_and_unregister(context):
|
||||
|
||||
|
||||
def test_revoke_corner_cases(context):
|
||||
"""Test specific revoke corner case."""
|
||||
# Cannot use --cert-path and --cert-name during a revoke.
|
||||
cert1 = context.wtf('le1')
|
||||
context.common(['-d', cert1])
|
||||
with pytest.raises(subprocess.CalledProcessError) as error:
|
||||
@@ -402,6 +429,7 @@ def test_revoke_corner_cases(context):
|
||||
|
||||
assert os.path.isfile(join(context.config_dir, 'renewal/{0}.conf'.format(cert1)))
|
||||
|
||||
# Revocation should not delete if multiple lineages share an archive dir
|
||||
cert2 = context.wtf('le2')
|
||||
context.common(['-d', cert2])
|
||||
with open(join(context.config_dir, 'renewal/{0}.conf'.format(cert2)), 'r') as file:
|
||||
@@ -423,6 +451,7 @@ def test_revoke_corner_cases(context):
|
||||
|
||||
|
||||
def test_wildcard_certificates(context):
|
||||
"""Test wildcard certificate issuance."""
|
||||
if context.acme_server == 'boulder-v1':
|
||||
pytest.skip('Wildcard certificates are not supported on ACME v1')
|
||||
|
||||
@@ -439,6 +468,7 @@ def test_wildcard_certificates(context):
|
||||
|
||||
|
||||
def test_ocsp_status(context):
|
||||
"""Test retrieval of OCSP statuses."""
|
||||
if context.acme_server == 'pebble':
|
||||
pytest.skip('Pebble does not support OCSP status requests.')
|
||||
|
||||
|
||||
Reference in New Issue
Block a user