x/deb-oslo.messaging
Code Issues Proposed changes

AMQP 1.0 - create only one Cyrus SASL configuration for the tests

Browse Source 
The Cyrus SASL library can only be initialized once per process. And
the configuration must remain available until all the tests have run.

Change-Id: I2810ecf53956fc5d4fcdc944f8e06237ca985e6a
Closes-Bug: #1591211
This commit is contained in:
Kenneth Giusti 2016-06-12 19:36:06 -04:00
parent 5afb60512a
commit 3316edb2ea
1 changed files with 35 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
oslo_messaging/tests/drivers/test_amqp_driver.py
View File

@ -18,7 +18,6 @@ import select
import shutil
import socket
import subprocess
import tempfile
import threading
import time
import uuid
@ -378,58 +377,71 @@ class TestAuthentication(test_utils.BaseTestCase):
class TestCyrusAuthentication(test_utils.BaseTestCase):
"""Test the driver's Cyrus SASL integration"""
def setUp(self):
"""Create a simple SASL configuration. This assumes saslpasswd2 is in
the OS path, otherwise the test will be skipped.
"""
super(TestCyrusAuthentication, self).setUp()
_conf_dir = None
# Note: don't add ANONYMOUS or EXTERNAL mechs without updating the
# test_authentication_bad_mechs test below
_mechs = "DIGEST-MD5 SCRAM-SHA-1 CRAM-MD5 PLAIN"
@classmethod
def setUpClass(cls):
# The Cyrus library can only be initialized once per _process_
# Create a SASL configuration and user database,
# add a user 'joe' with password 'secret':
self._conf_dir = tempfile.mkdtemp()
db = os.path.join(self._conf_dir, 'openstack.sasldb')
cls._conf_dir = "/tmp/amqp1_tests_%s" % os.getpid()
# no, we cannot use tempfile.mkdtemp() as it will 'helpfully' remove
# the temp dir after the first test is run (?why?)
os.makedirs(cls._conf_dir)
db = os.path.join(cls._conf_dir, 'openstack.sasldb')
_t = "echo secret | saslpasswd2 -c -p -f ${db} joe"
cmd = Template(_t).substitute(db=db)
try:
subprocess.check_call(args=cmd, shell=True)
except Exception:
shutil.rmtree(self._conf_dir, ignore_errors=True)
self._conf_dir = None
raise self.skip("Cyrus tool saslpasswd2 not installed")
shutil.rmtree(cls._conf_dir, ignore_errors=True)
cls._conf_dir = None
return
# configure the SASL broker:
conf = os.path.join(self._conf_dir, 'openstack.conf')
# Note: don't add ANONYMOUS or EXTERNAL without updating the
# test_authentication_bad_mechs test below
mechs = "DIGEST-MD5 SCRAM-SHA-1 CRAM-MD5 PLAIN"
# configure the SASL server:
conf = os.path.join(cls._conf_dir, 'openstack.conf')
t = Template("""sasldb_path: ${db}
pwcheck_method: auxprop
auxprop_plugin: sasldb
mech_list: ${mechs}
""")
with open(conf, 'w') as f:
f.write(t.substitute(db=db, mechs=mechs))
f.write(t.substitute(db=db, mechs=cls._mechs))
self._broker = FakeBroker(sasl_mechanisms=mechs,
@classmethod
def tearDownClass(cls):
if cls._conf_dir:
shutil.rmtree(cls._conf_dir, ignore_errors=True)
def setUp(self):
# fire up a test broker with the SASL config:
super(TestCyrusAuthentication, self).setUp()
if TestCyrusAuthentication._conf_dir is None:
self.skipTest("Cyrus SASL tools not installed")
_mechs = TestCyrusAuthentication._mechs
_dir = TestCyrusAuthentication._conf_dir
self._broker = FakeBroker(sasl_mechanisms=_mechs,
user_credentials=["\0joe\0secret"],
sasl_config_dir=self._conf_dir,
sasl_config_dir=_dir,
sasl_config_name="openstack")
self._broker.start()
self.messaging_conf.transport_driver = 'amqp'
self.conf = self.messaging_conf.conf
def tearDown(self):
super(TestCyrusAuthentication, self).tearDown()
if self._broker:
self._broker.stop()
self._broker = None
if self._conf_dir:
shutil.rmtree(self._conf_dir, ignore_errors=True)
super(TestCyrusAuthentication, self).tearDown()
def test_authentication_ok(self):
"""Verify that username and password given in TransportHost are
accepted by the broker.
"""
addr = "amqp://joe:secret@%s:%d" % (self._broker.host,
self._broker.port)
url = oslo_messaging.TransportURL.parse(self.conf, addr)