diff options
Diffstat (limited to 'python/qpid')
| -rw-r--r-- | python/qpid/connection.py | 11 | ||||
| -rw-r--r-- | python/qpid/delegates.py | 12 | ||||
| -rw-r--r-- | python/qpid/managementdata.py | 22 | ||||
| -rw-r--r-- | python/qpid/testlib.py | 3 |
4 files changed, 35 insertions, 13 deletions
diff --git a/python/qpid/connection.py b/python/qpid/connection.py index 39f882e9c3..4ed430249b 100644 --- a/python/qpid/connection.py +++ b/python/qpid/connection.py @@ -37,6 +37,8 @@ class ChannelsBusy(Exception): pass class SessionBusy(Exception): pass +class ConnectionFailed(Exception): pass + def client(*args): return delegates.Client(*args) @@ -45,7 +47,7 @@ def server(*args): class Connection(Assembler): - def __init__(self, sock, spec=None, delegate=client): + def __init__(self, sock, spec=None, delegate=client, **args): Assembler.__init__(self, sock) if spec == None: spec = load(default()) @@ -58,13 +60,14 @@ class Connection(Assembler): self.condition = Condition() self.opened = False + self.failed = False self.thread = Thread(target=self.run) self.thread.setDaemon(True) self.channel_max = 65535 - self.delegate = delegate(self) + self.delegate = delegate(self, args) def attach(self, name, ch, delegate, force=False): self.lock.acquire() @@ -127,8 +130,10 @@ class Connection(Assembler): def start(self, timeout=None): self.delegate.start() self.thread.start() - if not wait(self.condition, lambda: self.opened, timeout): + if not wait(self.condition, lambda: self.opened or self.failed, timeout): raise Timeout() + if (self.failed): + raise ConnectionFailed() def run(self): # XXX: we don't really have a good way to exit this loop without diff --git a/python/qpid/delegates.py b/python/qpid/delegates.py index f31d9a0f09..cdff132219 100644 --- a/python/qpid/delegates.py +++ b/python/qpid/delegates.py @@ -52,6 +52,9 @@ class Delegate: def connection_close(self, ch, close): ch.connection_close_ok() self.connection.sock.close() + if not self.connection.opened: + self.connection.failed = True + notify(self.connection.condition) def connection_close_ok(self, ch, close_ok): self.connection.opened = False @@ -124,12 +127,19 @@ class Client(Delegate): "version": "development", "platform": os.name} + def __init__(self, connection, args={}): + Delegate.__init__(self, connection) + self.username = args.get('username', 'guest') + self.password = args.get('password', 'guest') + self.mechanism = args.get('mechanism', 'PLAIN') + def start(self): self.connection.write_header(self.spec.major, self.spec.minor) self.connection.read_header() def connection_start(self, ch, start): - ch.connection_start_ok(client_properties=Client.PROPERTIES, mechanism="ANONYMOUS") + r = "\0%s\0%s" % (self.username, self.password) + ch.connection_start_ok(client_properties=Client.PROPERTIES, mechanism=self.mechanism, response=r) def connection_tune(self, ch, tune): ch.connection_tune_ok() diff --git a/python/qpid/managementdata.py b/python/qpid/managementdata.py index c908483354..a0197ba7d2 100644 --- a/python/qpid/managementdata.py +++ b/python/qpid/managementdata.py @@ -20,6 +20,7 @@ # import qpid +import re import socket import struct import os @@ -32,14 +33,18 @@ from qpid.util import connect class Broker: def __init__ (self, text): - colon = text.find (":") - if colon == -1: - host = text - self.port = 5672 - else: - host = text[:colon] - self.port = int (text[colon+1:]) + rex = re.compile(r""" + # [ <user> [ / <password> ] @] <host> [ :<port> ] + ^ (?: ([^/]*) (?: / ([^@]*) )? @)? ([^:]+) (?: :([0-9]+))?$""", re.X) + match = rex.match(text) + if not match: raise ValueError("'%s' is not a valid broker url" % (text)) + user, password, host, port = match.groups() + self.host = socket.gethostbyname (host) + if port: self.port = int(port) + else: self.port = 5672 + self.username = user or "guest" + self.password = password or "guest" def name (self): return self.host + ":" + str (self.port) @@ -174,7 +179,8 @@ class ManagementData: self.sessionId = "%s.%d" % (os.uname()[1], os.getpid()) self.broker = Broker (host) - self.conn = Connection (connect (self.broker.host, self.broker.port), self.spec) + self.conn = Connection (connect (self.broker.host, self.broker.port), self.spec, + username=self.broker.username, password=self.broker.password) self.conn.start () self.mclient = managementClient (self.spec, self.ctrlHandler, self.configHandler, diff --git a/python/qpid/testlib.py b/python/qpid/testlib.py index adda1a650f..b5aa59f586 100644 --- a/python/qpid/testlib.py +++ b/python/qpid/testlib.py @@ -353,7 +353,8 @@ class TestBase010(unittest.TestCase): def setUp(self): spec = testrunner.spec - self.conn = Connection(connect(testrunner.host, testrunner.port), spec) + self.conn = Connection(connect(testrunner.host, testrunner.port), spec, + username=testrunner.user, password=testrunner.password) self.conn.start(timeout=10) self.session = self.conn.session("test-session", timeout=10) |