Start adding tests for nginx configurator

This commit is contained in:
yan
2015-04-17 22:24:19 -07:00
parent d2588de4fd
commit 154db5a757
7 changed files with 201 additions and 206 deletions
@@ -91,7 +91,7 @@ class NginxConfigurator(object):
# Set Version
if self.version is None:
self.version = self._get_version()
self.version = self.get_version()
# Get all of the available vhosts
self.vhosts = self.parser.get_vhosts()
@@ -214,7 +214,7 @@ class NginxConfigurator(object):
matches.append({'vhost': vhost,
'name': name,
'rank': 6 if vhost.ssl else 7})
return sorted(matches, key=lambda x: x['rank'], reverse=True)
return sorted(matches, key=lambda x: x['rank'])
def get_all_names(self):
"""Returns all names found in the Nginx Configuration.
@@ -303,7 +303,7 @@ class NginxConfigurator(object):
try:
return self._enhance_func[enhancement](
self.choose_vhost(domain), options)
except ValueError:
except (KeyError, ValueError):
raise errors.LetsEncryptConfiguratorError(
"Unsupported enhancement: {}".format(enhancement))
except errors.LetsEncryptConfiguratorError:
@@ -360,7 +360,7 @@ class NginxConfigurator(object):
le_util.make_or_verify_dir(self.config.work_dir, 0o755, uid)
le_util.make_or_verify_dir(self.config.backup_dir, 0o755, uid)
def _get_version(self):
def get_version(self):
"""Return version of Nginx Server.
Version is returned as tuple. (ie. 2.4.7 = (2, 4, 7))
@@ -440,11 +440,11 @@ class NginxConfigurator(object):
self.reverter.add_to_checkpoint(save_files,
self.save_notes)
# Don't override original files for now.
self.parser.filedump('le')
self.parser.filedump(ext='')
if title and not temporary:
self.reverter.finalize_checkpoint(title)
# Refresh the vhosts
self.vhosts = self.parser.get_vhosts()
return True
+3 -2
View File
@@ -169,10 +169,11 @@ class NginxParser(object):
names = re.sub(whitespace_re, ' ', names)
return names.split(' ')
def _parse_files(self, filepath):
def _parse_files(self, filepath, override=False):
"""Parse files from a glob
:param str filepath: Nginx config file path
:param bool override: Whether to parse a file that has been parsed
:returns: list of parsed tree structures
:rtype: list
@@ -180,7 +181,7 @@ class NginxParser(object):
files = glob.glob(filepath)
trees = []
for f in files:
if f in self.parsed:
if f in self.parsed and not override:
continue
try:
with open(f) as fo:
@@ -1,6 +1,4 @@
"""Test for letsencrypt.client.plugins.nginx.configurator."""
import os
import re
import shutil
import unittest
@@ -8,145 +6,151 @@ import mock
from letsencrypt.acme import challenges
from letsencrypt.client import achallenges
from letsencrypt.client import errors
from letsencrypt.client import le_util
from letsencrypt.client.plugins.nginx import configurator
from letsencrypt.client.plugins.nginx import obj
from letsencrypt.client.plugins.nginx import parser
from letsencrypt.client.plugins.nginx.tests import util
class TwoVhost80Test(util.NginxTest):
"""Test two standard well configured HTTP vhosts."""
class NginxConfiguratorTest(util.NginxTest):
"""Test a semi complex vhost configuration."""
def setUp(self):
super(TwoVhost80Test, self).setUp()
super(NginxConfiguratorTest, self).setUp()
with mock.patch("letsencrypt.client.plugins.nginx.configurator."
"mod_loaded") as mock_load:
mock_load.return_value = True
self.config = util.get_nginx_configurator(
self.config_path, self.config_dir, self.work_dir,
self.ssl_options)
self.vh_truth = util.get_vh_truth(
self.temp_dir, "debian_nginx_2_4/two_vhost_80")
self.config = util.get_nginx_configurator(
self.config_path, self.config_dir, self.work_dir,
self.ssl_options)
def tearDown(self):
shutil.rmtree(self.temp_dir)
shutil.rmtree(self.config_dir)
shutil.rmtree(self.work_dir)
def test_prepare(self):
self.assertEquals((1, 6, 2), self.config.version)
self.assertEquals(5, len(self.config.vhosts))
def test_get_all_names(self):
names = self.config.get_all_names()
self.assertEqual(names, set(
["letsencrypt.demo", "encryption-example.demo", "ip-172-30-0-17"]))
["*.www.foo.com", "somename", "another.alias",
"alias", "localhost", ".example.com",
"155.225.50.69.nephoscale.net", "*.www.example.com",
"example.*", "www.example.org", "myhost"]))
def test_get_virtual_hosts(self):
"""Make sure all vhosts are being properly found.
def test_supported_enhancements(self):
self.assertEqual([], self.config.supported_enhancements())
.. note:: If test fails, only finding 1 Vhost... it is likely that
it is a problem with is_enabled.
def test_enhance(self):
self.assertRaises(errors.LetsEncryptConfiguratorError,
self.config.enhance,
'myhost',
'redirect')
"""
vhs = self.config.get_virtual_hosts()
self.assertEqual(len(vhs), 4)
found = 0
def test_get_chall_pref(self):
self.assertEqual([challenges.DVSNI],
self.config.get_chall_pref('myhost'))
for vhost in vhs:
for truth in self.vh_truth:
if vhost == truth:
found += 1
break
self.assertEqual(found, 4)
def test_is_site_enabled(self):
"""Test if site is enabled.
.. note:: This test currently fails for hard links
(which may happen if you move dirs incorrectly)
.. warning:: This test does not work when running using the
unittest.main() function. It incorrectly copies symlinks.
"""
self.assertTrue(self.config.is_site_enabled(self.vh_truth[0].filep))
self.assertFalse(self.config.is_site_enabled(self.vh_truth[1].filep))
self.assertTrue(self.config.is_site_enabled(self.vh_truth[2].filep))
self.assertTrue(self.config.is_site_enabled(self.vh_truth[3].filep))
def test_deploy_cert(self):
# Get the default 443 vhost
self.config.assoc["random.demo"] = self.vh_truth[1]
self.config.deploy_cert(
"random.demo",
"example/cert.pem", "example/key.pem", "example/cert_chain.pem")
def test_save(self):
filep = self.config.parser.abs_path('sites-enabled/example.com')
self.config.parser.add_server_directives(
filep, set(['.example.com', 'example.*']),
[['listen', '443 ssl']])
self.config.save()
loc_cert = self.config.parser.find_dir(
parser.case_i("sslcertificatefile"),
re.escape("example/cert.pem"), self.vh_truth[1].path)
loc_key = self.config.parser.find_dir(
parser.case_i("sslcertificateKeyfile"),
re.escape("example/key.pem"), self.vh_truth[1].path)
loc_chain = self.config.parser.find_dir(
parser.case_i("SSLCertificateChainFile"),
re.escape("example/cert_chain.pem"), self.vh_truth[1].path)
# pylint: disable=protected-access
parsed = self.config.parser._parse_files(filep, override=True)
self.assertEqual([[['server'], [['listen', '69.50.225.155:9000'],
['listen', '127.0.0.1'],
['server_name', '.example.com'],
['server_name', 'example.*'],
['listen', '443 ssl']]]],
parsed[0])
# Verify one directive was found in the correct file
self.assertEqual(len(loc_cert), 1)
self.assertEqual(configurator.get_file_path(loc_cert[0]),
self.vh_truth[1].filep)
def test_choose_vhost(self):
localhost_conf = set(['localhost'])
server_conf = set(['somename', 'another.alias', 'alias'])
example_conf = set(['.example.com', 'example.*'])
foo_conf = set(['*.www.foo.com', '*.www.example.com'])
self.assertEqual(len(loc_key), 1)
self.assertEqual(configurator.get_file_path(loc_key[0]),
self.vh_truth[1].filep)
results = {'localhost': localhost_conf,
'alias': server_conf,
'example.com': example_conf,
'example.com.uk.test': example_conf,
'www.example.com': example_conf,
'test.www.example.com': foo_conf,
'abc.www.foo.com': foo_conf}
bad_results = ['www.foo.com', 'example', '69.255.225.155']
self.assertEqual(len(loc_chain), 1)
self.assertEqual(configurator.get_file_path(loc_chain[0]),
self.vh_truth[1].filep)
for name in results:
self.assertEqual(results[name],
self.config.choose_vhost(name).names)
for name in bad_results:
self.assertEqual(None, self.config.choose_vhost(name))
def test_is_name_vhost(self):
addr = obj.Addr.fromstring("*:80")
self.assertTrue(self.config.is_name_vhost(addr))
self.config.version = (2, 2)
self.assertFalse(self.config.is_name_vhost(addr))
def test_more_info(self):
self.assertTrue('nginx.conf' in self.config.more_info())
def test_add_name_vhost(self):
self.config.add_name_vhost("*:443")
self.assertTrue(self.config.parser.find_dir(
"NameVirtualHost", re.escape("*:443")))
def test_deploy_cert(self):
pass
# Get the default 443 vhost
# self.config.assoc["random.demo"] = self.vh_truth[1]
# self.config.deploy_cert(
# "random.demo",
# "example/cert.pem", "example/key.pem", "example/cert_chain.pem")
# self.config.save()
#
# loc_cert = self.config.parser.find_dir(
# parser.case_i("sslcertificatefile"),
# re.escape("example/cert.pem"), self.vh_truth[1].path)
# loc_key = self.config.parser.find_dir(
# parser.case_i("sslcertificateKeyfile"),
# re.escape("example/key.pem"), self.vh_truth[1].path)
# loc_chain = self.config.parser.find_dir(
# parser.case_i("SSLCertificateChainFile"),
# re.escape("example/cert_chain.pem"), self.vh_truth[1].path)
#
# # Verify one directive was found in the correct file
# self.assertEqual(len(loc_cert), 1)
# self.assertEqual(configurator.get_file_path(loc_cert[0]),
# self.vh_truth[1].filep)
#
# self.assertEqual(len(loc_key), 1)
# self.assertEqual(configurator.get_file_path(loc_key[0]),
# self.vh_truth[1].filep)
#
# self.assertEqual(len(loc_chain), 1)
# self.assertEqual(configurator.get_file_path(loc_chain[0]),
# self.vh_truth[1].filep)
def test_make_vhost_ssl(self):
ssl_vhost = self.config.make_vhost_ssl(self.vh_truth[0])
self.assertEqual(
ssl_vhost.filep,
os.path.join(self.config_path, "sites-available",
"encryption-example-le-ssl.conf"))
self.assertEqual(ssl_vhost.path,
"/files" + ssl_vhost.filep + "/IfModule/VirtualHost")
self.assertEqual(len(ssl_vhost.addrs), 1)
self.assertEqual(set([obj.Addr.fromstring("*:443")]), ssl_vhost.addrs)
self.assertEqual(ssl_vhost.names, set(["encryption-example.demo"]))
self.assertTrue(ssl_vhost.ssl)
self.assertFalse(ssl_vhost.enabled)
self.assertTrue(self.config.parser.find_dir(
"SSLCertificateFile", None, ssl_vhost.path))
self.assertTrue(self.config.parser.find_dir(
"SSLCertificateKeyFile", None, ssl_vhost.path))
self.assertTrue(self.config.parser.find_dir(
"Include", self.ssl_options, ssl_vhost.path))
self.assertEqual(self.config.is_name_vhost(self.vh_truth[0]),
self.config.is_name_vhost(ssl_vhost))
self.assertEqual(len(self.config.vhosts), 5)
pass
# ssl_vhost = self.config.make_vhost_ssl(self.vh_truth[0])
#
# self.assertEqual(
# ssl_vhost.filep,
# os.path.join(self.config_path, "sites-available",
# "encryption-example-le-ssl.conf"))
#
# self.assertEqual(ssl_vhost.path,
# "/files" + ssl_vhost.filep + "/IfModule/VirtualHost")
# self.assertEqual(len(ssl_vhost.addrs), 1)
# self.assertEqual(set([obj.Addr.fromstring("*:443")]), ssl_vhost.addrs)
# self.assertEqual(ssl_vhost.names, set(["encryption-example.demo"]))
# self.assertTrue(ssl_vhost.ssl)
# self.assertFalse(ssl_vhost.enabled)
#
# self.assertTrue(self.config.parser.find_dir(
# "SSLCertificateFile", None, ssl_vhost.path))
# self.assertTrue(self.config.parser.find_dir(
# "SSLCertificateKeyFile", None, ssl_vhost.path))
# self.assertTrue(self.config.parser.find_dir(
# "Include", self.ssl_options, ssl_vhost.path))
#
# self.assertEqual(self.config.is_name_vhost(self.vh_truth[0]),
# self.config.is_name_vhost(ssl_vhost))
#
# self.assertEqual(len(self.config.vhosts), 5)
@mock.patch("letsencrypt.client.plugins.nginx.configurator."
"dvsni.NginxDvsni.perform")
@@ -155,56 +159,81 @@ class TwoVhost80Test(util.NginxTest):
def test_perform(self, mock_restart, mock_dvsni_perform):
# Only tests functionality specific to configurator.perform
# Note: As more challenges are offered this will have to be expanded
auth_key = le_util.Key(self.rsa256_file, self.rsa256_pem)
achall1 = achallenges.DVSNI(
chall=challenges.DVSNI(
r="jIq_Xy1mXGN37tb4L6Xj_es58fW571ZNyXekdZzhh7Q",
nonce="37bc5eb75d3e00a19b4f6355845e5a18"),
domain="encryption-example.demo", key=auth_key)
achall2 = achallenges.DVSNI(
chall=challenges.DVSNI(
r="uqnaPzxtrndteOqtrXb0Asl5gOJfWAnnx6QJyvcmlDU",
nonce="59ed014cac95f77057b1d7a1b2c596ba"),
domain="letsencrypt.demo", key=auth_key)
dvsni_ret_val = [
challenges.DVSNIResponse(s="randomS1"),
challenges.DVSNIResponse(s="randomS2"),
]
mock_dvsni_perform.return_value = dvsni_ret_val
responses = self.config.perform([achall1, achall2])
self.assertEqual(mock_dvsni_perform.call_count, 1)
self.assertEqual(responses, dvsni_ret_val)
self.assertEqual(mock_restart.call_count, 1)
pass
# auth_key = le_util.Key(self.rsa256_file, self.rsa256_pem)
# achall1 = achallenges.DVSNI(
# chall=challenges.DVSNI(
# r="jIq_Xy1mXGN37tb4L6Xj_es58fW571ZNyXekdZzhh7Q",
# nonce="37bc5eb75d3e00a19b4f6355845e5a18"),
# domain="encryption-example.demo", key=auth_key)
# achall2 = achallenges.DVSNI(
# chall=challenges.DVSNI(
# r="uqnaPzxtrndteOqtrXb0Asl5gOJfWAnnx6QJyvcmlDU",
# nonce="59ed014cac95f77057b1d7a1b2c596ba"),
# domain="letsencrypt.demo", key=auth_key)
#
# dvsni_ret_val = [
# challenges.DVSNIResponse(s="randomS1"),
# challenges.DVSNIResponse(s="randomS2"),
# ]
#
# mock_dvsni_perform.return_value = dvsni_ret_val
# responses = self.config.perform([achall1, achall2])
#
# self.assertEqual(mock_dvsni_perform.call_count, 1)
# self.assertEqual(responses, dvsni_ret_val)
#
# self.assertEqual(mock_restart.call_count, 1)
@mock.patch("letsencrypt.client.plugins.nginx.configurator."
"subprocess.Popen")
def test_get_version(self, mock_popen):
mock_popen().communicate.return_value = (
"Server Version: Nginx/2.4.2 (Debian)", "")
self.assertEqual(self.config.get_version(), (2, 4, 2))
"", "\n".join(["nginx version: nginx/1.4.2",
"built by clang 6.0 (clang-600.0.56)"
" (based on LLVM 3.5svn)",
"TLS SNI support enabled",
"configure arguments: --prefix=/usr/local/Cellar/"
"nginx/1.6.2 --with-http_ssl_module"]))
self.assertEqual(self.config.get_version(), (1, 4, 2))
mock_popen().communicate.return_value = (
"Server Version: Nginx/2 (Linux)", "")
self.assertEqual(self.config.get_version(), (2,))
"", "\n".join(["blah 0.0.1",
"TLS SNI support enabled"]))
self.assertRaises(errors.LetsEncryptConfiguratorError,
self.config.get_version)
mock_popen().communicate.return_value = (
"Server Version: Nginx (Debian)", "")
self.assertRaises(
errors.LetsEncryptConfiguratorError, self.config.get_version)
"", "\n".join(["nginx version: nginx/1.4.2",
""]))
self.assertRaises(errors.LetsEncryptConfiguratorError,
self.config.get_version)
mock_popen().communicate.return_value = (
"Server Version: Nginx/2.3{0} Nginx/2.4.7".format(os.linesep), "")
self.assertRaises(
errors.LetsEncryptConfiguratorError, self.config.get_version)
"", "\n".join(["nginx version: nginx/0.8.1",
""]))
self.assertRaises(errors.LetsEncryptConfiguratorError,
self.config.get_version)
mock_popen.side_effect = OSError("Can't find program")
self.assertRaises(
errors.LetsEncryptConfiguratorError, self.config.get_version)
@mock.patch("letsencrypt.client.plugins.nginx.configurator."
"subprocess.Popen")
def test_nginx_restart(self, mock_popen):
m = mock_popen()
m.communicate.return_value = ('', '')
m.returncode = 0
self.assertTrue(self.config.restart())
@mock.patch("letsencrypt.client.plugins.nginx.configurator."
"subprocess.Popen")
def test_config_test(self, mock_popen):
m = mock_popen()
m.communicate.return_value = ('', '')
m.returncode = 0
self.assertTrue(self.config.config_test())
if __name__ == "__main__":
unittest.main()
@@ -64,8 +64,8 @@ class TestRawNginxParser(unittest.TestCase):
parsed,
[['user', 'www-data'],
[['server'], [
['listen', '80'],
['server_name', 'foo.com'],
['listen', '*:80 default_server ssl'],
['server_name', '*.www.foo.com *.www.example.com'],
['root', '/home/ubuntu/sites/foo/'],
[['location', '/status'], [
['check_status'],
@@ -99,7 +99,8 @@ class NginxParserTest(util.NginxTest):
False, True, set(['www.example.org']), [])
vhost5 = VirtualHost(parser.abs_path('foo.conf'),
[Addr('*', '80', True, True)],
True, True, set(['*.www.foo.com']), [])
True, True, set(['*.www.foo.com',
'*.www.example.com']), [])
self.assertEqual(5, len(vhosts))
example_com = filter(lambda x: 'example.com' in x.filep, vhosts)[0]
+1 -1
View File
@@ -3,7 +3,7 @@ user www-data;
server {
listen *:80 default_server ssl;
server_name *.www.foo.com;
server_name *.www.foo.com *.www.example.com;
root /home/ubuntu/sites/foo/;
location /status {
+13 -49
View File
@@ -9,13 +9,13 @@ import mock
from letsencrypt.client import constants
from letsencrypt.client.plugins.nginx import configurator
from letsencrypt.client.plugins.nginx import obj
class NginxTest(unittest.TestCase): # pylint: disable=too-few-public-methods
def setUp(self):
super(NginxTest, self).setUp()
self.maxDiff = None
self.temp_dir, self.config_dir, self.work_dir = dir_setup(
"testdata")
@@ -59,59 +59,23 @@ def setup_nginx_ssl_options(config_dir):
def get_nginx_configurator(
config_path, config_dir, work_dir, ssl_options, version=(2, 4, 7)):
config_path, config_dir, work_dir, ssl_options, version=(1, 6, 2)):
"""Create an Nginx Configurator with the specified options."""
backups = os.path.join(work_dir, "backups")
with mock.patch("letsencrypt.client.plugins.nginx.configurator."
"subprocess.Popen") as mock_popen:
# This just states that the ssl module is already loaded
mock_popen().communicate.return_value = ("ssl_module", "")
config = configurator.NginxConfigurator(
mock.MagicMock(
nginx_server_root=config_path,
nginx_mod_ssl_conf=ssl_options,
le_vhost_ext="-le-ssl.conf",
backup_dir=backups,
config_dir=config_dir,
temp_checkpoint_dir=os.path.join(work_dir, "temp_checkpoints"),
in_progress_dir=os.path.join(backups, "IN_PROGRESS"),
work_dir=work_dir),
version)
config = configurator.NginxConfigurator(
mock.MagicMock(
nginx_server_root=config_path,
nginx_mod_ssl_conf=ssl_options,
le_vhost_ext="-le-ssl.conf",
backup_dir=backups,
config_dir=config_dir,
temp_checkpoint_dir=os.path.join(work_dir, "temp_checkpoints"),
in_progress_dir=os.path.join(backups, "IN_PROGRESS"),
work_dir=work_dir),
version)
config.prepare()
return config
def get_vh_truth(temp_dir, config_name):
"""Return the ground truth for the specified directory."""
if config_name == "debian_nginx_2_4/two_vhost_80":
prefix = os.path.join(
temp_dir, config_name, "nginx2/sites-available")
aug_pre = "/files" + prefix
vh_truth = [
obj.VirtualHost(
os.path.join(prefix, "encryption-example.conf"),
os.path.join(aug_pre, "encryption-example.conf/VirtualHost"),
set([obj.Addr.fromstring("*:80")]),
False, True, set(["encryption-example.demo"])),
obj.VirtualHost(
os.path.join(prefix, "default-ssl.conf"),
os.path.join(aug_pre, "default-ssl.conf/IfModule/VirtualHost"),
set([obj.Addr.fromstring("_default_:443")]), True, False),
obj.VirtualHost(
os.path.join(prefix, "000-default.conf"),
os.path.join(aug_pre, "000-default.conf/VirtualHost"),
set([obj.Addr.fromstring("*:80")]), False, True,
set(["ip-172-30-0-17"])),
obj.VirtualHost(
os.path.join(prefix, "letsencrypt.conf"),
os.path.join(aug_pre, "letsencrypt.conf/VirtualHost"),
set([obj.Addr.fromstring("*:80")]), False, True,
set(["letsencrypt.demo"])),
]
return vh_truth
return None