Repository: qpid-dispatch Updated Branches: refs/heads/master ddbeb5f7a -> 2f9bbadde
NO-JIRA: Removed some redundant, slow and unreliable tests. Removed redundant SSL tests in system_tests_qdmanage and system_tests_two_routers. The tests take a long time to run and SSL is covered elsewhere. Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/2f9bbadd Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/2f9bbadd Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/2f9bbadd Branch: refs/heads/master Commit: 2f9bbadde809ec587e73097585ca70ceee81f63c Parents: ddbeb5f Author: Alan Conway <[email protected]> Authored: Mon Mar 27 12:35:25 2017 -0400 Committer: Alan Conway <[email protected]> Committed: Mon Mar 27 12:36:55 2017 -0400 ---------------------------------------------------------------------- tests/system_tests_qdmanage.py | 207 +++--------------------------- tests/system_tests_two_routers.py | 227 ++++++++++++--------------------- 2 files changed, 101 insertions(+), 333 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2f9bbadd/tests/system_tests_qdmanage.py ---------------------------------------------------------------------- diff --git a/tests/system_tests_qdmanage.py b/tests/system_tests_qdmanage.py index 120e959..585fc4b 100644 --- a/tests/system_tests_qdmanage.py +++ b/tests/system_tests_qdmanage.py @@ -17,13 +17,12 @@ # under the License # -import re, json, unittest, os -from time import sleep +import json, unittest, os + from system_test import TestCase, Process, Qdrouterd, main_module, TIMEOUT, DIR, wait_port from subprocess import PIPE, STDOUT from qpid_dispatch_internal.compat import OrderedDict, dictify from qpid_dispatch_internal.management.qdrouter import QdSchema -from proton import Url DUMMY = "org.apache.qpid.dispatch.dummy" @@ -211,18 +210,11 @@ class QdmanageTest(TestCase): log_levels_present += 1 else: log_levels_missing += 1 - if log_levels_present == n_log_levels: good_logs += 1 self.assertEquals ( good_logs, len(logs) ) - def test_ssl(self): - """Simple test for SSL connection. Note system_tests_qdstat has a more complete SSL test""" - url = Url(self.router_1.addresses[1], scheme="amqps") - schema = dictify(QdSchema().dump()) - actual = self.run_qdmanage("GET-JSON-SCHEMA") - self.assertEquals(schema, dictify(json.loads(actual))) def test_update(self): exception = False @@ -264,15 +256,8 @@ class QdmanageTest(TestCase): def test_specify_container_id_connection_link_route(self): long_type = 'org.apache.qpid.dispatch.router.config.linkRoute' create_command = 'CREATE --type=' + long_type + ' prefix=abc containerId=id1 connection=conn1 dir=out' - bad_create = False - try: - output = json.loads(self.run_qdmanage(create_command)) - print output - except Exception as e: - bad_create = True - self.assertTrue("Both connection and containerId cannot be specified. Specify only one" in e.message) - - self.assertTrue(bad_create) + output = self.run_qdmanage(create_command, expect=Process.EXIT_FAIL) + self.assertIn("Both connection and containerId cannot be specified", output) def test_check_auto_link_name(self): long_type = 'org.apache.qpid.dispatch.router.config.autoLink' @@ -285,15 +270,8 @@ class QdmanageTest(TestCase): def test_specify_container_id_connection_auto_link(self): long_type = 'org.apache.qpid.dispatch.router.config.autoLink' create_command = 'CREATE --type=' + long_type + ' addr=abc containerId=id1 connection=conn1 dir=out' - bad_create = False - try: - output = json.loads(self.run_qdmanage(create_command)) - print output - except Exception as e: - bad_create = True - self.assertTrue("Both connection and containerId cannot be specified. Specify only one" in e.message) - - self.assertTrue(bad_create) + output = self.run_qdmanage(create_command, expect=Process.EXIT_FAIL) + self.assertIn("Both connection and containerId cannot be specified", output) def test_create_delete_connector(self): long_type = 'org.apache.qpid.dispatch.connector' @@ -327,19 +305,9 @@ class QdmanageTest(TestCase): command = "CREATE --type=connector --name=eaconn1 port=" + str(port) + " host=0.0.0.0" output = json.loads(self.run_qdmanage(command)) self.assertEqual("normal", output['role']) - - exception = False - try: - port = self.get_port() - # provide the same connector name (eaconn1) and make sure there is a duplicate value failure - command = "CREATE --type=connector --name=eaconn1 port=" + str(port) + " host=0.0.0.0" - output = json.loads(self.run_qdmanage(command)) - except Exception as e: - self.assertTrue("Duplicate value 'eaconn1' for unique attribute 'name'" in e.message) - exception = True - - self.assertTrue(exception) - + # provide the same connector name (eaconn1), expect duplicate value failure + self.assertRaises(Exception, self.run_qdmanage, + "CREATE --type=connector --name=eaconn1 port=12345 host=0.0.0.0") port = self.get_port() # provide role as 'normal' and make sure that it is preserved command = "CREATE --type=connector --name=eaconn2 port=" + str(port) + " host=0.0.0.0 role=normal" @@ -358,14 +326,6 @@ class QdmanageTest(TestCase): exception_occurred = False - try: - # Try to connect to the port that was closed, it should not return an error - wait_port(listener_port, timeout=2) - except Exception as e: - exception_occurred = True - - self.assertFalse(exception_occurred) - delete_command = 'DELETE --type=' + long_type + ' --name=' + name self.run_qdmanage(delete_command) @@ -379,153 +339,18 @@ class QdmanageTest(TestCase): self.assertTrue(exception_occurred) - try: - # Try to connect to that port, it should not return an error - wait_port(listener_port, timeout=2) - except Exception as e: - exception_occurred = True - - self.assertTrue(exception_occurred) - - # Now try the same thing with a short_type - short_type = 'listener' - - listener_port = self.get_port() - - listener = self.create(long_type, name, str(listener_port)) - self.assertEquals(listener['type'], long_type) - self.assertEquals(listener['name'], name) - - delete_command = 'DELETE --type=' + short_type + ' --name=' + name - self.run_qdmanage(delete_command) - - exception_occurred = False - - try: - # Try to connect to that port, it should not return an error - wait_port(listener_port, timeout=2) - except Exception as e: - exception_occurred = True - - self.assertTrue(exception_occurred) - -class QdmanageTestSsl(QdmanageTest): - - @classmethod - def setUpClass(cls): - super(QdmanageTestSsl, cls).setUpClass() - - def address(self): - return self.router_1.addresses[1] - - 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', - '--ssl-trustfile=' + self.ssl_file('ca-certificate.pem'), - '--ssl-certificate=' + self.ssl_file('client-certificate.pem'), - '--ssl-key=' + self.ssl_file('client-private-key.pem'), - '--ssl-password=client-password', - '--timeout', str(TIMEOUT), - '--ssl-disable-peer-name-verify'], - stdin=PIPE, stdout=PIPE, stderr=STDOUT, expect=expect) - out = p.communicate(input)[0] - try: - p.teardown() - except Exception, e: - raise Exception("%s\n%s" % (e, out)) - return out - - def test_create_delete_connector(self): - long_type = 'org.apache.qpid.dispatch.connector' - query_command = 'QUERY --type=' + long_type - output = json.loads(self.run_qdmanage(query_command)) - name = output[0]['name'] - - # Delete an existing connector - delete_command = 'DELETE --type=' + long_type + ' --name=' + name - self.run_qdmanage(delete_command) - output = json.loads(self.run_qdmanage(query_command)) - self.assertEqual(output, []) - - # Re-create the connector and then try wait_connectors - self.create(long_type, name, str(QdmanageTestSsl.inter_router_port)) - - - outputs = json.loads(self.run_qdmanage(query_command)) - created = False - for output in outputs: - conn_name = 'connector/127.0.0.1:%s' % QdmanageTestSsl.inter_router_port - conn_name_1 = 'connector/0.0.0.0:%s' % QdmanageTestSsl.inter_router_port - if conn_name == output['name'] or conn_name_1 == output['name']: - created = True - break - self.assertTrue(created) - def test_create_delete_ssl_profile(self): - long_type = 'org.apache.qpid.dispatch.sslProfile' ssl_profile_name = 'ssl-profile-test' - ssl_create_command = 'CREATE --type=' + long_type + ' certFile=' + self.ssl_file('server-certificate.pem') + \ + ssl_create_command = 'CREATE --type=sslProfile certFile=' + self.ssl_file('server-certificate.pem') + \ ' keyFile=' + self.ssl_file('server-private-key.pem') + ' password=server-password' + \ ' name=' + ssl_profile_name + ' certDb=' + self.ssl_file('ca-certificate.pem') - output = json.loads(self.run_qdmanage(ssl_create_command)) - name = output['name'] - self.assertEqual(name, ssl_profile_name) - - long_type = 'org.apache.qpid.dispatch.listener' - listener_name = 'sslListener' - port = self.get_port() - listener_create_command = 'CREATE --type=' + long_type + ' --name=sslListener host=127.0.0.1 port=' + str(port) + \ - ' saslMechanisms=EXTERNAL sslProfile=' + ssl_profile_name + \ - ' requireSsl=yes authenticatePeer=yes' - output = json.loads(self.run_qdmanage(listener_create_command)) - name = output['name'] - self.assertEqual(name, listener_name) - - sleep(1) - query_command = 'QUERY --type=listener' - - # Query on the port that was created by the preceding listener create - output = json.loads(self.run_qdmanage(query_command, address="127.0.0.1:"+str(port))) - - ssl_listener_present = False - - for out in output: - if out['name'] == 'sslListener': - ssl_listener_present = True - self.assertEqual(out['sslProfile'], 'ssl-profile-test') - - self.assertEqual(len(output), 3) - self.assertTrue(ssl_listener_present) - - # Delete the SSL Profile. This will fail because there is a listener referencing the SSL profile. - delete_command = 'DELETE --type=sslProfile --name=' + ssl_profile_name - cannot_delete = False - try: - json.loads(self.run_qdmanage(delete_command)) - except Exception as e: - cannot_delete = True - self.assertTrue('ForbiddenStatus: SSL Profile is referenced by other listeners/connectors' in e.message) - - self.assertTrue(cannot_delete) - - # Deleting the listener first and then the SSL profile must work. - delete_command = 'DELETE --type=listener --name=' + listener_name - self.run_qdmanage(delete_command) - - delete_command = 'DELETE --type=sslProfile --name=' + ssl_profile_name - self.run_qdmanage(delete_command) - - def test_zzz_router_query(self): - long_type = 'org.apache.qpid.dispatch.router' - query_command = 'QUERY --type=' + long_type - output = json.loads(self.run_qdmanage(query_command)) - self.assertEqual(3, output[0]['connectionCount']) - self.assertEqual(1, output[0]['linkRouteCount']) - self.assertEqual(1, output[0]['autoLinkCount']) - self.assertEqual('interior', output[0]['mode']) - self.assertEqual('org.apache.qpid.dispatch.router', output[0]['type']) + self.assertEqual(output['name'], ssl_profile_name) + self.run_qdmanage('DELETE --type=sslProfile --name=' + ssl_profile_name) + # Try to delete the server-ssl profile which is in use. + output = self.run_qdmanage('DELETE --type=sslProfile --name=server-ssl', + expect=Process.EXIT_FAIL) + self.assertIn("ForbiddenStatus", output) if __name__ == '__main__': unittest.main(main_module()) http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2f9bbadd/tests/system_tests_two_routers.py ---------------------------------------------------------------------- diff --git a/tests/system_tests_two_routers.py b/tests/system_tests_two_routers.py index db82838..75bc545 100644 --- a/tests/system_tests_two_routers.py +++ b/tests/system_tests_two_routers.py @@ -17,9 +17,9 @@ # under the License. # -import unittest, os, json +import unittest, os, json, logging from subprocess import PIPE, STDOUT -from proton import Message, PENDING, ACCEPTED, REJECTED, RELEASED, SSLDomain, SSLUnavailable, Timeout +from proton import Message, PENDING, ACCEPTED, REJECTED, RELEASED, Timeout from system_test import TestCase, Qdrouterd, main_module, DIR, TIMEOUT, Process from proton.handlers import MessagingHandler from proton.reactor import Container, AtMostOnce, AtLeastOnce @@ -35,43 +35,39 @@ class RouterTest(TestCase): inter_router_port = None - @staticmethod - def ssl_config(client_server, connection): - return [] # Over-ridden by RouterTestSsl - @classmethod def setUpClass(cls): """Start a router and a messenger""" super(RouterTest, cls).setUpClass() def router(name, client_server, connection): - - config = cls.ssl_config(client_server, connection) + [ + + config = [ ('router', {'mode': 'interior', 'id': 'QDR.%s'%name}), - + ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'no'}), - + # The following listeners were exclusively added to test the stripAnnotations attribute in qdrouterd.conf file # Different listeners will be used to test all allowed values of stripAnnotations ('no', 'both', 'out', 'in') ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'no'}), ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'both'}), ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'out'}), ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'in'}), - + ('address', {'prefix': 'closest', 'distribution': 'closest'}), ('address', {'prefix': 'spread', 'distribution': 'balanced'}), ('address', {'prefix': 'multicast', 'distribution': 'multicast'}), connection ] - + config = Qdrouterd.Config(config) cls.routers.append(cls.tester.qdrouterd(name, config, wait=True)) cls.routers = [] - + inter_router_port = cls.tester.get_port() - + router('A', 'server', ('listener', {'role': 'inter-router', 'port': inter_router_port})) router('B', 'client', @@ -396,70 +392,70 @@ class RouterTest(TestCase): M1.stop() M2.stop() - - + + #The stripAnnotations property is set to 'no' def test_08a_test_strip_message_annotations_no(self): addr = "amqp:/message_annotations_strip_no/1" - + M1 = self.messenger() M2 = self.messenger() - + M1.route("amqp:/*", self.routers[0].addresses[1]+"/$1") M2.route("amqp:/*", self.routers[1].addresses[1]+"/$1") - + M1.start() M2.start() M2.subscribe(addr) self.routers[0].wait_address("message_annotations_strip_no/1", 0, 1) - + ingress_message = Message() ingress_message.address = addr ingress_message.body = {'message': 'Hello World!'} ingress_message_annotations = {'work': 'hard', 'stay': 'humble'} - + ingress_message.annotations = ingress_message_annotations - + M1.put(ingress_message) M1.send() - + # Receive the message M2.recv(1) egress_message = Message() M2.get(egress_message) - + #Make sure 'Hello World!' is in the message body dict self.assertEqual('Hello World!', egress_message.body['message']) - - + + egress_message_annotations = egress_message.annotations - + self.assertEqual(egress_message_annotations.__class__, dict) self.assertEqual(egress_message_annotations['x-opt-qd.ingress'], '0/QDR.A') self.assertEqual(egress_message_annotations['work'], 'hard') self.assertEqual(egress_message_annotations['stay'], 'humble') self.assertEqual(egress_message_annotations['x-opt-qd.trace'], ['0/QDR.A', '0/QDR.B']) - + M1.stop() M2.stop() - + # This unit test is currently skipped because dispatch router do not pass thru custom message annotations. # Once the feature is added the @unittest.skip decorator can be removed. # The stripAnnotations property is set to 'no' def test_08a_strip_message_annotations_custom(self): addr = "amqp:/message_annotations_strip_no_custom/1" - + M1 = self.messenger() M2 = self.messenger() - + M1.route("amqp:/*", self.routers[0].addresses[1]+"/$1") M2.route("amqp:/*", self.routers[1].addresses[1]+"/$1") - + M1.start() M2.start() M2.subscribe(addr) self.routers[0].wait_address("message_annotations_strip_no_custom/1", 0, 1) - + ingress_message = Message() ingress_message.address = addr ingress_message.body = {'message': 'Hello World!'} @@ -467,98 +463,98 @@ class RouterTest(TestCase): ingress_message_annotations['custom-annotation'] = '1/Custom_Annotation' ingress_message.annotations = ingress_message_annotations - + M1.put(ingress_message) M1.send() - + # Receive the message M2.recv(1) egress_message = Message() M2.get(egress_message) - + # Make sure 'Hello World!' is in the message body dict self.assertEqual('Hello World!', egress_message.body['message']) egress_message_annotations = egress_message.annotations - + self.assertEqual(egress_message_annotations.__class__, dict) self.assertEqual(egress_message_annotations['custom-annotation'], '1/Custom_Annotation') self.assertEqual(egress_message_annotations['x-opt-qd.ingress'], '0/QDR.A') self.assertEqual(egress_message_annotations['x-opt-qd.trace'], ['0/QDR.A', '0/QDR.B']) - + M1.stop() M2.stop() - + #The stripAnnotations property is set to 'no' def test_08a_test_strip_message_annotations_no_add_trace(self): addr = "amqp:/strip_message_annotations_no_add_trace/1" - + M1 = self.messenger() M2 = self.messenger() - + M1.route("amqp:/*", self.routers[0].addresses[1]+"/$1") M2.route("amqp:/*", self.routers[1].addresses[1]+"/$1") - + M1.start() M2.start() M2.subscribe(addr) self.routers[0].wait_address("strip_message_annotations_no_add_trace/1", 0, 1) - + ingress_message = Message() ingress_message.address = addr ingress_message.body = {'message': 'Hello World!'} - + ## ## Pre-existing ingress and trace ## #ingress_message_annotations = {'x-opt-qd.ingress': 'ingress-router', 'x-opt-qd.trace': ['0/QDR.1']} ingress_message_annotations = {'x-opt-qd.trace': ['0/QDR.1']} ingress_message.annotations = ingress_message_annotations - + ingress_message.annotations = ingress_message_annotations - + M1.put(ingress_message) M1.send() - + # Receive the message M2.recv(1) egress_message = Message() M2.get(egress_message) - + #Make sure 'Hello World!' is in the message body dict self.assertEqual('Hello World!', egress_message.body['message']) - - + + egress_message_annotations = egress_message.annotations - + self.assertEqual(egress_message_annotations.__class__, dict) self.assertEqual(egress_message_annotations['x-opt-qd.ingress'], '0/QDR.A') self.assertEqual(egress_message_annotations['x-opt-qd.trace'], ['0/QDR.1', '0/QDR.A', '0/QDR.B']) - + M1.stop() M2.stop() - + # Test to see if the dispatch router specific annotations were stripped. # The stripAnnotations property is set to 'both' # Send a message to the router with pre-existing ingress and trace annotations and make sure that nothing comes out. def test_08a_test_strip_message_annotations_both_add_ingress_trace(self): addr = "amqp:/strip_message_annotations_both_add_ingress_trace/1" - + M1 = self.messenger() M2 = self.messenger() - + M1.route("amqp:/*", self.routers[0].addresses[2]+"/$1") M2.route("amqp:/*", self.routers[1].addresses[2]+"/$1") - + M1.start() M2.start() M2.subscribe(addr) self.routers[0].wait_address("strip_message_annotations_both_add_ingress_trace/1", 0, 1) - + ingress_message = Message() ingress_message.address = addr ingress_message.body = {'message': 'Hello World!'} - + ## # Pre-existing ingress and trace. Intentionally populate the trace with the 0/QDR.A which is the trace # of the first router. If the inbound annotations were not stripped, the router would drop this message @@ -569,11 +565,11 @@ class RouterTest(TestCase): 'work': 'hard', 'x-opt-qd': 'humble'} ingress_message.annotations = ingress_message_annotations - + #Put and send the message M1.put(ingress_message) M1.send() - + # Receive the message M2.recv(1) egress_message = Message() @@ -581,7 +577,7 @@ class RouterTest(TestCase): # Router specific annotations (annotations with prefix "x-opt-qd.") will be stripped. User defined annotations will not be stripped. self.assertEqual(egress_message.annotations, {'work': 'hard', 'x-opt-qd': 'humble'}) - + M1.stop() M2.stop() @@ -589,38 +585,38 @@ class RouterTest(TestCase): # stripAnnotations property is set to "in" def test_08a_test_strip_message_annotations_out(self): addr = "amqp:/strip_message_annotations_out/1" - + M1 = self.messenger() M2 = self.messenger() - + M1.route("amqp:/*", self.routers[0].addresses[3]+"/$1") M2.route("amqp:/*", self.routers[1].addresses[3]+"/$1") - + M1.start() M2.start() M2.subscribe(addr) self.routers[0].wait_address("strip_message_annotations_out/1", 0, 1) - + ingress_message = Message() ingress_message.address = addr ingress_message.body = {'message': 'Hello World!'} - + #Put and send the message M1.put(ingress_message) M1.send() - + # Receive the message egress_message = Message() M2.recv(1) M2.get(egress_message) - + #Make sure 'Hello World!' is in the message body dict self.assertEqual('Hello World!', egress_message.body['message']) - + egress_message_annotations = egress_message.annotations - + self.assertEqual(egress_message.annotations, None) - + M1.stop() M2.stop() @@ -664,53 +660,53 @@ class RouterTest(TestCase): M1.stop() M2.stop() - + #Send in pre-existing trace and ingress and annotations and make sure that they are not in the outgoing annotations. #stripAnnotations property is set to "in" def test_08a_test_strip_message_annotations_in(self): addr = "amqp:/strip_message_annotations_in/1" - + M1 = self.messenger() M2 = self.messenger() M1.route("amqp:/*", self.routers[0].addresses[4]+"/$1") M2.route("amqp:/*", self.routers[1].addresses[4]+"/$1") - + M1.start() M2.start() M2.subscribe(addr) self.routers[0].wait_address("strip_message_annotations_in/1", 0, 1) - + ingress_message = Message() ingress_message.address = addr ingress_message.body = {'message': 'Hello World!'} - + ## ## Pre-existing ingress and trace ## ingress_message_annotations = {'x-opt-qd.ingress': 'ingress-router', 'x-opt-qd.trace': ['X/QDR']} ingress_message.annotations = ingress_message_annotations - + #Put and send the message M1.put(ingress_message) M1.send() - + # Receive the message egress_message = Message() M2.recv(1) M2.get(egress_message) - + #Make sure 'Hello World!' is in the message body dict self.assertEqual('Hello World!', egress_message.body['message']) - + egress_message_annotations = egress_message.annotations - + self.assertEqual(egress_message_annotations.__class__, dict) self.assertEqual(egress_message_annotations['x-opt-qd.ingress'], '0/QDR.A') self.assertEqual(egress_message_annotations['x-opt-qd.trace'], ['0/QDR.A', '0/QDR.B']) - + M1.stop() M2.stop() - + def test_09_management(self): M = self.messenger() @@ -1123,6 +1119,7 @@ class ExcessDeliveriesReleasedTest(MessagingHandler): class AttachOnInterRouterTest(MessagingHandler): + """Expect an error when attaching a link to an inter-router listener""" def __init__(self, address): super(AttachOnInterRouterTest, self).__init__(prefetch=0) self.address = address @@ -1144,65 +1141,11 @@ class AttachOnInterRouterTest(MessagingHandler): self.timer.cancel() def run(self): - Container(self).run() - -try: - SSLDomain(SSLDomain.MODE_CLIENT) - - class RouterTestSsl(RouterTest): - - @staticmethod - def ssl_file(name): - return os.path.join(DIR, 'ssl_certs', name) - - 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) - out = p.communicate(input)[0] - try: - p.teardown() - except Exception, e: - raise Exception("%s\n%s" % (e, out)) - return out - - @staticmethod - def ssl_config(client_server, connection): - connection[1]['sslProfile'] = 'test-ssl' - - return [ - ('sslProfile', { - 'name': 'test-ssl', - 'certDb': RouterTestSsl.ssl_file('ca-certificate.pem'), - 'certFile': RouterTestSsl.ssl_file(client_server+'-certificate.pem'), - 'keyFile': RouterTestSsl.ssl_file(client_server+'-private-key.pem'), - 'password': client_server+'-password'})] - - def test_zzz_delete_ssl_profile(self): - """ - Delete an ssl profile before deleting the connector and make sure it fails. - """ - delete_command = 'DELETE --type=sslProfile --name=test-ssl' - cannot_delete = False - try: - json.loads(self.run_qdmanage(delete_command, address=self.routers[1].addresses[0])) - except Exception as e: - cannot_delete = True - self.assertTrue('ForbiddenStatus: SSL Profile is referenced by other listeners/connectors' in e.message) - - self.assertTrue(cannot_delete) - - # Deleting the listener first and then the SSL profile must work. - delete_command = 'DELETE --type=connector --name=connectorToA' - self.run_qdmanage(delete_command, address=self.routers[1].addresses[0]) - - delete_command = 'DELETE --type=sslProfile --name=test-ssl' - self.run_qdmanage(delete_command, address=self.routers[1].addresses[0]) - -except SSLUnavailable: - class RouterTestSsl(TestCase): - def test_skip(self): - self.skipTest("Proton SSL support unavailable.") + logging.disable(logging.ERROR) # Hide expected log errors + try: + Container(self).run() + finally: + logging.disable(logging.NOTSET) # Restore to normal if __name__ == '__main__': unittest.main(main_module()) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
