tests/system_tests_http.py - qpid-dispatch - Git at Google

 #
 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
 # distributed with this work for additional information
 # regarding copyright ownership.  The ASF licenses this file
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License.  You may obtain a copy of the License at
 #
 #   http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
 #
 import errno
 import os
 import threading
 import ssl
 from subprocess import PIPE, STDOUT

 from urllib.request import urlopen, build_opener, HTTPSHandler
 from urllib.error import HTTPError, URLError

 import qpid_dispatch_site
 from system_test import TIMEOUT, Process, QdManager, retry
 from system_test import TestCase, Qdrouterd, main_module, DIR
 from system_test import unittest


 class RouterTestHttp(TestCase):

     @staticmethod
     def ssl_file(name):
         return os.path.join(DIR, 'ssl_certs', name)

     @classmethod
     def setUpClass(cls):
         super(RouterTestHttp, cls).setUpClass()
         # DISPATCH-1513, DISPATCH-2299: Http listener delete was broken in LWS v4 until v4.2.0
         cls.skip_delete_http_listener_test = qpid_dispatch_site.SKIP_DELETE_HTTP_LISTENER

     @classmethod
     def get(cls, url, use_ca=True):
         if use_ca:
             http_data = urlopen(url, cafile=cls.ssl_file('ca-certificate.pem'))
         else:
             http_data = urlopen(url)
         return http_data.read().decode('utf-8')

     @classmethod
     def get_cert(cls, url):
         context = ssl.create_default_context()
         context.load_cert_chain(cls.ssl_file('client-certificate.pem'),
                                 cls.ssl_file('client-private-key.pem'),
                                 'client-password')
         context.load_verify_locations(cls.ssl_file('ca-certificate.pem'))
         opener = build_opener(HTTPSHandler(context=context))
         return opener.open(url).read().decode('utf-8')

     def run_qdmanage(self, cmd, input=None, expect=Process.EXIT_OK, address=None):
         p = self.popen(
             ['qdmanage'] + cmd.split(' ') + ['--bus', address or self.address(), '--indent=-1', '--timeout', str(TIMEOUT)],
             stdin=PIPE, stdout=PIPE, stderr=STDOUT, expect=expect,
             universal_newlines=True)
         out = p.communicate(input)[0]
         try:
             p.teardown()
         except Exception as e:
             raise Exception(out if out else str(e))
         return out

     def assert_get(self, url):
         self.assertEqual('HTTP test\n', self.get("%s/system_tests_http.txt" % url))

     def assert_get_cert(self, url):
         self.assertEqual('HTTP test\n', self.get_cert("%s/system_tests_http.txt" % url))

     def test_listen_error(self):
         """Make sure a router exits if an initial HTTP listener fails, doesn't hang"""
         listen_port = self.get_port()
         config = Qdrouterd.Config([
             ('router', {'mode': 'standalone', 'id': 'bad'}),
             ('listener', {'port': listen_port, 'maxFrameSize': '2048', 'stripAnnotations': 'no'}),
             ('listener', {'port': listen_port, 'http': True})])
         r = Qdrouterd(name="expect_fail", config=config, wait=False)
         self.assertEqual(1, r.wait())

     def is_get_request_failing(self, url, use_ca=True, use_get_cert=False):
         try:
             if use_get_cert:
                 self.get_cert(url)
             else:
                 self.get(url, use_ca=use_ca)
             return False
         except OSError as e:
             # EADDRNOTAVAIL happens in Docker when connecting to localhost:port where nobody listens
             expected = (errno.ECONNREFUSED, errno.ECONNRESET, errno.EADDRNOTAVAIL)
             if e.errno in expected or e.reason.errno in expected:
                 return True
             raise e

     def test_http_listener_delete(self):
         name = 'delete_listener'
         name_1 = 'delete_listener_1'
         normal_listen_port = self.get_port()
         #
         # Open listeners on two HTTP enabled ports. Delete one of the
         # HTTP listeners and make sure that it is really gone and also
         # make sure that the other HTTP listener is still working.
         #
         http_delete_listen_port_1 = self.get_port()
         http_delete_listen_port_2 = self.get_port()
         config = Qdrouterd.Config([
             ('router', {'mode': 'standalone', 'id': 'A'}),
             ('listener', {'port': normal_listen_port}),
             ('listener', {'httpRootDir': os.path.dirname(__file__), 'name': name, 'port': http_delete_listen_port_1, 'http': True}),
             ('listener', {'httpRootDir': os.path.dirname(__file__), 'name': name_1, 'port': http_delete_listen_port_2, 'http': True})])
         router = self.qdrouterd(name="expect_fail_1", config=config, wait=True)

         def address():
             return router.addresses[0]

         # Perform a GET request on the http_delete_listen_port_1 just to make
         # sure that it is up and running.
         url_1 = "%s/system_tests_http.txt" % "http://localhost:%d" % http_delete_listen_port_1
         out = self.get(url_1, use_ca=False)

         # Perform a GET request on the http_delete_listen_port_2 just to make
         # sure that it is up and running.
         url_2 = "%s/system_tests_http.txt" % "http://localhost:%d" % http_delete_listen_port_2
         out = self.get(url_2, use_ca=False)

         # Now both http_delete_listen_port_1 and http_delete_listen_port_2
         # are working.

         # Delete the listener on port http_delete_listen_port_1
         long_type = 'org.apache.qpid.dispatch.listener'
         mgmt = QdManager(address=address())

         if self.skip_delete_http_listener_test:
             # You are not allowed to delete a http:yes listener
             # Try deleting it and make sure you get an exception.
             try:
                 mgmt.delete(long_type, name=name)
             except Exception as e:
                 if "BadRequestStatus: HTTP listeners cannot be deleted" in str(e):
                     exception_raised = True
             self.assertTrue(exception_raised)
         else:
             mgmt.delete(long_type, name=name)

             # Once again try to perform a GET request. Now since the listener
             # is gone, the GET will fail.
             ret_val = retry(lambda: self.is_get_request_failing(url_1, use_ca=False), timeout=10, delay=2)
             self.assertTrue(ret_val)

             # HTTP listener on port http_delete_listen_port_1 has been
             # deleted successfully. Make sure that the listener on port
             # http_delete_listen_port_2 is still working.
             out = self.get(url_2, use_ca=False)

     def test_http_get(self):
         config = Qdrouterd.Config([
             ('router', {'id': 'QDR.HTTP'}),
             # httpRoot has been deprecated. We are using it here to test backward compatibility.
             ('listener', {'port': self.get_port(), 'httpRoot': os.path.dirname(__file__)}),
             ('listener', {'port': self.get_port(), 'httpRootDir': os.path.dirname(__file__)}),
         ])
         r = self.qdrouterd('http-test-router', config)

         def test(port):
             self.assert_get("http://localhost:%d" % port)
             self.assertRaises(HTTPError, urlopen, "http://localhost:%d/nosuch" % port)

         # Sequential calls on multiple ports
         for port in r.ports:
             test(port)

         # Concurrent calls on multiple ports
         class TestThread(threading.Thread):
             def __init__(self, port):
                 threading.Thread.__init__(self)
                 self.port, self.ex = port, None
                 self.start()

             def run(self):
                 try:
                     test(self.port)
                 except Exception as e:
                     self.ex = e
         threads = [TestThread(p) for p in r.ports + r.ports]
         for t in threads:
             t.join()
         for t in threads:
             if t.ex:
                 raise t.ex

         # https not configured
         self.assertRaises(URLError, urlopen, "https://localhost:%d/nosuch" % r.ports[0])

     def test_http_metrics(self):
         config = Qdrouterd.Config([
             ('router', {'id': 'QDR.METRICS'}),
             ('listener', {'port': self.get_port(), 'http': 'yes'}),
             ('listener', {'port': self.get_port(), 'httpRootDir': os.path.dirname(__file__)}),
         ])
         r = self.qdrouterd('metrics-test-router', config)

         def test(port):
             result = urlopen("http://localhost:%d/metrics" % port, cafile=self.ssl_file('ca-certificate.pem'))
             self.assertEqual(200, result.getcode())
             data = result.read().decode('utf-8')
             assert 'connections' in data
             assert 'deliveries_ingress' in data
             assert 'deliveries_delayed_1sec' in data
             assert 'deliveries_delayed_10sec' in data
             assert 'deliveries_redirected_to_fallback' in data

         # Sequential calls on multiple ports
         for port in r.ports:
             test(port)

         # Concurrent calls on multiple ports
         class TestThread(threading.Thread):
             def __init__(self, port):
                 threading.Thread.__init__(self)
                 self.port, self.ex = port, None
                 self.start()

             def run(self):
                 try:
                     test(self.port)
                 except Exception as e:
                     self.ex = e
         threads = [TestThread(p) for p in r.ports + r.ports]
         for t in threads:
             t.join()
         for t in threads:
             if t.ex:
                 raise t.ex

     def test_http_healthz(self):
         config = Qdrouterd.Config([
             ('router', {'id': 'QDR.HEALTHZ'}),
             ('listener', {'port': self.get_port(), 'http': 'yes'}),
             ('listener', {'port': self.get_port(), 'httpRootDir': os.path.dirname(__file__)}),
         ])
         r = self.qdrouterd('metrics-test-router', config)

         def test(port):
             result = urlopen("http://localhost:%d/healthz" % port, cafile=self.ssl_file('ca-certificate.pem'))
             self.assertEqual(200, result.getcode())

         # Sequential calls on multiple ports
         for port in r.ports:
             test(port)

         # Concurrent calls on multiple ports
         class TestThread(threading.Thread):
             def __init__(self, port):
                 threading.Thread.__init__(self)
                 self.port, self.ex = port, None
                 self.start()

             def run(self):
                 try:
                     test(self.port)
                 except Exception as e:
                     self.ex = e
         threads = [TestThread(p) for p in r.ports + r.ports]
         for t in threads:
             t.join()
         for t in threads:
             if t.ex:
                 raise t.ex

     def test_https_get(self):
         def http_listener(**kwargs):
             args = dict(kwargs)
             args.update({'port': self.get_port(), 'http': 'yes', 'httpRootDir': os.path.dirname(__file__)})
             return ('listener', args)

         def listener(**kwargs):
             args = dict(kwargs)
             args.update({'port': self.get_port()})
             return ('listener', args)

         name = 'delete-me'
         config = Qdrouterd.Config([
             ('router', {'id': 'QDR.HTTPS'}),
             ('sslProfile', {'name': 'simple-ssl',
                             'caCertFile': self.ssl_file('ca-certificate.pem'),
                             'certFile': self.ssl_file('server-certificate.pem'),
                             'privateKeyFile': self.ssl_file('server-private-key.pem'),
                             'ciphers': 'ECDH+AESGCM:DH+AESGCM:ECDH+AES256:DH+AES256:ECDH+AES128:DH+AES:RSA+AESGCM:RSA+AES:!aNULL:!MD5:!DSS',
                             'password': 'server-password'
                             }),
             http_listener(sslProfile='simple-ssl', requireSsl=False, authenticatePeer=False),
             http_listener(sslProfile='simple-ssl', requireSsl=True, authenticatePeer=False),
             http_listener(sslProfile='simple-ssl', requireSsl=True, authenticatePeer=True),
             http_listener(name=name, sslProfile='simple-ssl', requireSsl=True, authenticatePeer=True),
             listener(name='mgmt_listener', authenticatePeer=False)])
         # saslMechanisms='EXTERNAL'

         r = self.qdrouterd('https-test-router', config)
         r.wait_ready()

         def address():
             return r.addresses[4]

         self.assert_get("https://localhost:%s" % r.ports[0])
         # requireSsl=false Allows simple-ssl HTTP

         # DISPATCH-1513: libwebsockets versions 3.2.0 introduces a new flag called
         # LWS_SERVER_OPTION_ALLOW_HTTP_ON_HTTPS_LISTENER
         # The new flag allows (as the flag says) HTTP over HTTPS listeners.
         # Since this flag is not available before lws 3.2.0 we need
         # to selectively disable this check
         if qpid_dispatch_site.LIBWEBSOCKETS_VERSION >= (3, 2, 0):
             self.assert_get("http://localhost:%s" % r.ports[0])

         self.assert_get("https://localhost:%s" % r.ports[1])
         # requireSsl=True does not allow simple-ssl HTTP
         self.assertRaises(Exception, self.assert_get, "http://localhost:%s" % r.ports[1])

         # authenticatePeer=True requires a client cert
         self.assertRaises((URLError, ssl.SSLError), self.assert_get, "https://localhost:%s" % r.ports[2])

         # Provide client cert
         self.assert_get_cert("https://localhost:%d" % r.ports[2])

         # Try a get on the HTTP listener we are going to delete
         self.assert_get_cert("https://localhost:%d" % r.ports[3])

         if not self.skip_delete_http_listener_test:
             # Delete the listener with name 'delete-me'
             long_type = 'org.apache.qpid.dispatch.listener'
             mgmt = QdManager(address=address())
             mgmt.delete(long_type, name=name)

             # Make sure that the listener got deleted.
             ret_val = retry(lambda: self.is_get_request_failing("https://localhost:%s/system_tests_http.txt" % r.ports[3], use_get_cert=True), timeout=10, delay=2)
             self.assertTrue(ret_val)

             # Make sure other ports are working normally after the above delete.
             self.assert_get_cert("https://localhost:%d" % r.ports[2])


 if __name__ == '__main__':
     unittest.main(main_module())
	#
	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.
	#
	import errno
	import os
	import threading
	import ssl
	from subprocess import PIPE, STDOUT

	from urllib.request import urlopen, build_opener, HTTPSHandler
	from urllib.error import HTTPError, URLError

	import qpid_dispatch_site
	from system_test import TIMEOUT, Process, QdManager, retry
	from system_test import TestCase, Qdrouterd, main_module, DIR
	from system_test import unittest


	class RouterTestHttp(TestCase):

	@staticmethod
	def ssl_file(name):
	return os.path.join(DIR, 'ssl_certs', name)

	@classmethod
	def setUpClass(cls):
	super(RouterTestHttp, cls).setUpClass()
	# DISPATCH-1513, DISPATCH-2299: Http listener delete was broken in LWS v4 until v4.2.0
	cls.skip_delete_http_listener_test = qpid_dispatch_site.SKIP_DELETE_HTTP_LISTENER

	@classmethod
	def get(cls, url, use_ca=True):
	if use_ca:
	http_data = urlopen(url, cafile=cls.ssl_file('ca-certificate.pem'))
	else:
	http_data = urlopen(url)
	return http_data.read().decode('utf-8')

	@classmethod
	def get_cert(cls, url):
	context = ssl.create_default_context()
	context.load_cert_chain(cls.ssl_file('client-certificate.pem'),
	cls.ssl_file('client-private-key.pem'),
	'client-password')
	context.load_verify_locations(cls.ssl_file('ca-certificate.pem'))
	opener = build_opener(HTTPSHandler(context=context))
	return opener.open(url).read().decode('utf-8')

	def run_qdmanage(self, cmd, input=None, expect=Process.EXIT_OK, address=None):
	p = self.popen(
	['qdmanage'] + cmd.split(' ') + ['--bus', address or self.address(), '--indent=-1', '--timeout', str(TIMEOUT)],
	stdin=PIPE, stdout=PIPE, stderr=STDOUT, expect=expect,
	universal_newlines=True)
	out = p.communicate(input)[0]
	try:
	p.teardown()
	except Exception as e:
	raise Exception(out if out else str(e))
	return out

	def assert_get(self, url):
	self.assertEqual('HTTP test\n', self.get("%s/system_tests_http.txt" % url))

	def assert_get_cert(self, url):
	self.assertEqual('HTTP test\n', self.get_cert("%s/system_tests_http.txt" % url))

	def test_listen_error(self):
	"""Make sure a router exits if an initial HTTP listener fails, doesn't hang"""
	listen_port = self.get_port()
	config = Qdrouterd.Config([
	('router', {'mode': 'standalone', 'id': 'bad'}),
	('listener', {'port': listen_port, 'maxFrameSize': '2048', 'stripAnnotations': 'no'}),
	('listener', {'port': listen_port, 'http': True})])
	r = Qdrouterd(name="expect_fail", config=config, wait=False)
	self.assertEqual(1, r.wait())

	def is_get_request_failing(self, url, use_ca=True, use_get_cert=False):
	try:
	if use_get_cert:
	self.get_cert(url)
	else:
	self.get(url, use_ca=use_ca)
	return False
	except OSError as e:
	# EADDRNOTAVAIL happens in Docker when connecting to localhost:port where nobody listens
	expected = (errno.ECONNREFUSED, errno.ECONNRESET, errno.EADDRNOTAVAIL)
	if e.errno in expected or e.reason.errno in expected:
	return True
	raise e

	def test_http_listener_delete(self):
	name = 'delete_listener'
	name_1 = 'delete_listener_1'
	normal_listen_port = self.get_port()
	#
	# Open listeners on two HTTP enabled ports. Delete one of the
	# HTTP listeners and make sure that it is really gone and also
	# make sure that the other HTTP listener is still working.
	#
	http_delete_listen_port_1 = self.get_port()
	http_delete_listen_port_2 = self.get_port()
	config = Qdrouterd.Config([
	('router', {'mode': 'standalone', 'id': 'A'}),
	('listener', {'port': normal_listen_port}),
	('listener', {'httpRootDir': os.path.dirname(__file__), 'name': name, 'port': http_delete_listen_port_1, 'http': True}),
	('listener', {'httpRootDir': os.path.dirname(__file__), 'name': name_1, 'port': http_delete_listen_port_2, 'http': True})])
	router = self.qdrouterd(name="expect_fail_1", config=config, wait=True)

	def address():
	return router.addresses[0]

	# Perform a GET request on the http_delete_listen_port_1 just to make
	# sure that it is up and running.
	url_1 = "%s/system_tests_http.txt" % "http://localhost:%d" % http_delete_listen_port_1
	out = self.get(url_1, use_ca=False)

	# Perform a GET request on the http_delete_listen_port_2 just to make
	# sure that it is up and running.
	url_2 = "%s/system_tests_http.txt" % "http://localhost:%d" % http_delete_listen_port_2
	out = self.get(url_2, use_ca=False)

	# Now both http_delete_listen_port_1 and http_delete_listen_port_2
	# are working.

	# Delete the listener on port http_delete_listen_port_1
	long_type = 'org.apache.qpid.dispatch.listener'
	mgmt = QdManager(address=address())

	if self.skip_delete_http_listener_test:
	# You are not allowed to delete a http:yes listener
	# Try deleting it and make sure you get an exception.
	try:
	mgmt.delete(long_type, name=name)
	except Exception as e:
	if "BadRequestStatus: HTTP listeners cannot be deleted" in str(e):
	exception_raised = True
	self.assertTrue(exception_raised)
	else:
	mgmt.delete(long_type, name=name)

	# Once again try to perform a GET request. Now since the listener
	# is gone, the GET will fail.
	ret_val = retry(lambda: self.is_get_request_failing(url_1, use_ca=False), timeout=10, delay=2)
	self.assertTrue(ret_val)

	# HTTP listener on port http_delete_listen_port_1 has been
	# deleted successfully. Make sure that the listener on port
	# http_delete_listen_port_2 is still working.
	out = self.get(url_2, use_ca=False)

	def test_http_get(self):
	config = Qdrouterd.Config([
	('router', {'id': 'QDR.HTTP'}),
	# httpRoot has been deprecated. We are using it here to test backward compatibility.
	('listener', {'port': self.get_port(), 'httpRoot': os.path.dirname(__file__)}),
	('listener', {'port': self.get_port(), 'httpRootDir': os.path.dirname(__file__)}),
	])
	r = self.qdrouterd('http-test-router', config)

	def test(port):
	self.assert_get("http://localhost:%d" % port)
	self.assertRaises(HTTPError, urlopen, "http://localhost:%d/nosuch" % port)

	# Sequential calls on multiple ports
	for port in r.ports:
	test(port)

	# Concurrent calls on multiple ports
	class TestThread(threading.Thread):
	def __init__(self, port):
	threading.Thread.__init__(self)
	self.port, self.ex = port, None
	self.start()

	def run(self):
	try:
	test(self.port)
	except Exception as e:
	self.ex = e
	threads = [TestThread(p) for p in r.ports + r.ports]
	for t in threads:
	t.join()
	for t in threads:
	if t.ex:
	raise t.ex

	# https not configured
	self.assertRaises(URLError, urlopen, "https://localhost:%d/nosuch" % r.ports[0])

	def test_http_metrics(self):
	config = Qdrouterd.Config([
	('router', {'id': 'QDR.METRICS'}),
	('listener', {'port': self.get_port(), 'http': 'yes'}),
	('listener', {'port': self.get_port(), 'httpRootDir': os.path.dirname(__file__)}),
	])
	r = self.qdrouterd('metrics-test-router', config)

	def test(port):
	result = urlopen("http://localhost:%d/metrics" % port, cafile=self.ssl_file('ca-certificate.pem'))
	self.assertEqual(200, result.getcode())
	data = result.read().decode('utf-8')
	assert 'connections' in data
	assert 'deliveries_ingress' in data
	assert 'deliveries_delayed_1sec' in data
	assert 'deliveries_delayed_10sec' in data
	assert 'deliveries_redirected_to_fallback' in data

	# Sequential calls on multiple ports
	for port in r.ports:
	test(port)

	# Concurrent calls on multiple ports
	class TestThread(threading.Thread):
	def __init__(self, port):
	threading.Thread.__init__(self)
	self.port, self.ex = port, None
	self.start()

	def run(self):
	try:
	test(self.port)
	except Exception as e:
	self.ex = e
	threads = [TestThread(p) for p in r.ports + r.ports]
	for t in threads:
	t.join()
	for t in threads:
	if t.ex:
	raise t.ex

	def test_http_healthz(self):
	config = Qdrouterd.Config([
	('router', {'id': 'QDR.HEALTHZ'}),
	('listener', {'port': self.get_port(), 'http': 'yes'}),
	('listener', {'port': self.get_port(), 'httpRootDir': os.path.dirname(__file__)}),
	])
	r = self.qdrouterd('metrics-test-router', config)

	def test(port):
	result = urlopen("http://localhost:%d/healthz" % port, cafile=self.ssl_file('ca-certificate.pem'))
	self.assertEqual(200, result.getcode())

	# Sequential calls on multiple ports
	for port in r.ports:
	test(port)

	# Concurrent calls on multiple ports
	class TestThread(threading.Thread):
	def __init__(self, port):
	threading.Thread.__init__(self)
	self.port, self.ex = port, None
	self.start()

	def run(self):
	try:
	test(self.port)
	except Exception as e:
	self.ex = e
	threads = [TestThread(p) for p in r.ports + r.ports]
	for t in threads:
	t.join()
	for t in threads:
	if t.ex:
	raise t.ex

	def test_https_get(self):
	def http_listener(**kwargs):
	args = dict(kwargs)
	args.update({'port': self.get_port(), 'http': 'yes', 'httpRootDir': os.path.dirname(__file__)})
	return ('listener', args)

	def listener(**kwargs):
	args = dict(kwargs)
	args.update({'port': self.get_port()})
	return ('listener', args)

	name = 'delete-me'
	config = Qdrouterd.Config([
	('router', {'id': 'QDR.HTTPS'}),
	('sslProfile', {'name': 'simple-ssl',
	'caCertFile': self.ssl_file('ca-certificate.pem'),
	'certFile': self.ssl_file('server-certificate.pem'),
	'privateKeyFile': self.ssl_file('server-private-key.pem'),
	'ciphers': 'ECDH+AESGCM:DH+AESGCM:ECDH+AES256:DH+AES256:ECDH+AES128:DH+AES:RSA+AESGCM:RSA+AES:!aNULL:!MD5:!DSS',
	'password': 'server-password'
	}),
	http_listener(sslProfile='simple-ssl', requireSsl=False, authenticatePeer=False),
	http_listener(sslProfile='simple-ssl', requireSsl=True, authenticatePeer=False),
	http_listener(sslProfile='simple-ssl', requireSsl=True, authenticatePeer=True),
	http_listener(name=name, sslProfile='simple-ssl', requireSsl=True, authenticatePeer=True),
	listener(name='mgmt_listener', authenticatePeer=False)])
	# saslMechanisms='EXTERNAL'

	r = self.qdrouterd('https-test-router', config)
	r.wait_ready()

	def address():
	return r.addresses[4]

	self.assert_get("https://localhost:%s" % r.ports[0])
	# requireSsl=false Allows simple-ssl HTTP

	# DISPATCH-1513: libwebsockets versions 3.2.0 introduces a new flag called
	# LWS_SERVER_OPTION_ALLOW_HTTP_ON_HTTPS_LISTENER
	# The new flag allows (as the flag says) HTTP over HTTPS listeners.
	# Since this flag is not available before lws 3.2.0 we need
	# to selectively disable this check
	if qpid_dispatch_site.LIBWEBSOCKETS_VERSION >= (3, 2, 0):
	self.assert_get("http://localhost:%s" % r.ports[0])

	self.assert_get("https://localhost:%s" % r.ports[1])
	# requireSsl=True does not allow simple-ssl HTTP
	self.assertRaises(Exception, self.assert_get, "http://localhost:%s" % r.ports[1])

	# authenticatePeer=True requires a client cert
	self.assertRaises((URLError, ssl.SSLError), self.assert_get, "https://localhost:%s" % r.ports[2])

	# Provide client cert
	self.assert_get_cert("https://localhost:%d" % r.ports[2])

	# Try a get on the HTTP listener we are going to delete
	self.assert_get_cert("https://localhost:%d" % r.ports[3])

	if not self.skip_delete_http_listener_test:
	# Delete the listener with name 'delete-me'
	long_type = 'org.apache.qpid.dispatch.listener'
	mgmt = QdManager(address=address())
	mgmt.delete(long_type, name=name)

	# Make sure that the listener got deleted.
	ret_val = retry(lambda: self.is_get_request_failing("https://localhost:%s/system_tests_http.txt" % r.ports[3], use_get_cert=True), timeout=10, delay=2)
	self.assertTrue(ret_val)

	# Make sure other ports are working normally after the above delete.
	self.assert_get_cert("https://localhost:%d" % r.ports[2])


	if __name__ == '__main__':
	unittest.main(main_module())