You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2009/10/05 13:29:14 UTC
svn commit: r821761 [6/6] - in /qpid/branches/java-broker-0-10/qpid: ./ cpp/
cpp/bindings/qmf/ cpp/bindings/qmf/python/ cpp/bindings/qmf/python/qmf/
cpp/bindings/qmf/ruby/ cpp/bindings/qmf/tests/ cpp/boost-1.32-support/
cpp/build-aux/ cpp/examples/qmf-...
Modified: qpid/branches/java-broker-0-10/qpid/python/qpid/session.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/qpid/session.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/qpid/session.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/qpid/session.py Mon Oct 5 11:29:03 2009
@@ -22,9 +22,9 @@
from generator import command_invoker
from datatypes import RangedSet, Struct, Future
from codec010 import StringCodec
-from assembler import Segment
from queue import Queue
from datatypes import Message, serial
+from ops import Command, MessageTransfer
from util import wait, notify
from exceptions import *
from logging import getLogger
@@ -44,7 +44,7 @@
INCOMPLETE = object()
-class Session(command_invoker(SPEC)):
+class Session(command_invoker()):
def __init__(self, name, auto_sync=True, timeout=10, delegate=client):
self.name = name
@@ -67,8 +67,6 @@
self.results = {}
self.exceptions = []
- self.assembly = None
-
self.delegate = delegate(self)
def incoming(self, destination):
@@ -134,68 +132,51 @@
finally:
self.lock.release()
- def invoke(self, type, args, kwargs):
- # XXX
- if not hasattr(type, "track"):
- return type.new(args, kwargs)
-
- self.invoke_lock.acquire()
- try:
- return self.do_invoke(type, args, kwargs)
- finally:
- self.invoke_lock.release()
+ def invoke(self, op, args, kwargs):
+ if issubclass(op, Command):
+ self.invoke_lock.acquire()
+ try:
+ return self.do_invoke(op, args, kwargs)
+ finally:
+ self.invoke_lock.release()
+ else:
+ return op(*args, **kwargs)
- def do_invoke(self, type, args, kwargs):
+ def do_invoke(self, op, args, kwargs):
if self._closing:
raise SessionClosed()
if self.channel == None:
raise SessionDetached()
- if type.segments:
- if len(args) == len(type.fields) + 1:
+ if op == MessageTransfer:
+ if len(args) == len(op.FIELDS) + 1:
message = args[-1]
args = args[:-1]
else:
message = kwargs.pop("message", None)
- else:
- message = None
-
- hdr = Struct(self.spec["session.header"])
- hdr.sync = self.auto_sync or kwargs.pop("sync", False)
- self.need_sync = not hdr.sync
-
- cmd = type.new(args, kwargs)
- sc = StringCodec(self.spec)
- sc.write_command(hdr, cmd)
+ if message is not None:
+ kwargs["headers"] = message.headers
+ kwargs["payload"] = message.body
+
+ cmd = op(*args, **kwargs)
+ cmd.sync = self.auto_sync or cmd.sync
+ self.need_sync = not cmd.sync
+ cmd.channel = self.channel.id
- seg = Segment(True, (message == None or
- (message.headers == None and message.body == None)),
- type.segment_type, type.track, self.channel.id, sc.encoded)
-
- if type.result:
+ if op.RESULT:
result = Future(exception=SessionException)
self.results[self.sender.next_id] = result
- self.send(seg)
+ log.debug("SENDING %s", cmd)
- log.debug("SENT %s %s %s", seg.id, hdr, cmd)
+ self.send(cmd)
- if message != None:
- if message.headers != None:
- sc = StringCodec(self.spec)
- for st in message.headers:
- sc.write_struct32(st)
- seg = Segment(False, message.body == None, self.spec["segment_type.header"].value,
- type.track, self.channel.id, sc.encoded)
- self.send(seg)
- if message.body != None:
- seg = Segment(False, True, self.spec["segment_type.body"].value,
- type.track, self.channel.id, message.body)
- self.send(seg)
- msg.debug("SENT %s", message)
+ log.debug("SENT %s", cmd)
+ if op == MessageTransfer:
+ msg.debug("SENT %s", cmd)
- if type.result:
+ if op.RESULT:
if self.auto_sync:
return result.get(self.timeout)
else:
@@ -203,81 +184,47 @@
elif self.auto_sync:
self.sync(self.timeout)
- def received(self, seg):
- self.receiver.received(seg)
- if seg.first:
- assert self.assembly == None
- self.assembly = []
- self.assembly.append(seg)
- if seg.last:
- self.dispatch(self.assembly)
- self.assembly = None
-
- def dispatch(self, assembly):
- segments = assembly[:]
-
- hdr, cmd = assembly.pop(0).decode(self.spec)
- log.debug("RECV %s %s %s", cmd.id, hdr, cmd)
-
- args = []
-
- for st in cmd._type.segments:
- if assembly:
- seg = assembly[0]
- if seg.type == st.segment_type:
- args.append(seg.decode(self.spec))
- assembly.pop(0)
- continue
- args.append(None)
-
- assert len(assembly) == 0
-
- attr = cmd._type.qname.replace(".", "_")
- result = getattr(self.delegate, attr)(cmd, *args)
-
- if cmd._type.result:
+ def received(self, cmd):
+ self.receiver.received(cmd)
+ self.dispatch(cmd)
+
+ def dispatch(self, cmd):
+ log.debug("RECV %s", cmd)
+
+ result = getattr(self.delegate, cmd.NAME)(cmd)
+ if result is INCOMPLETE:
+ return
+ elif result is not None:
self.execution_result(cmd.id, result)
- if result is not INCOMPLETE:
- for seg in segments:
- self.receiver.completed(seg)
- # XXX: don't forget to obey sync for manual completion as well
- if hdr.sync:
- self.channel.session_completed(self.receiver._completed)
+ self.receiver.completed(cmd)
+ # XXX: don't forget to obey sync for manual completion as well
+ if cmd.sync:
+ self.channel.session_completed(self.receiver._completed)
- def send(self, seg):
- self.sender.send(seg)
-
- def __str__(self):
- return '<Session: %s, %s>' % (self.name, self.channel)
+ def send(self, cmd):
+ self.sender.send(cmd)
def __repr__(self):
- return str(self)
+ return '<Session: %s, %s>' % (self.name, self.channel)
class Receiver:
def __init__(self, session):
self.session = session
self.next_id = None
- self.next_offset = None
self._completed = RangedSet()
- def received(self, seg):
- if self.next_id == None or self.next_offset == None:
+ def received(self, cmd):
+ if self.next_id == None:
raise Exception("todo")
- seg.id = self.next_id
- seg.offset = self.next_offset
- if seg.last:
- self.next_id += 1
- self.next_offset = 0
- else:
- self.next_offset += len(seg.payload)
+ cmd.id = self.next_id
+ self.next_id += 1
- def completed(self, seg):
- if seg.id == None:
- raise ValueError("cannot complete unidentified segment")
- if seg.last:
- self._completed.add(seg.id)
+ def completed(self, cmd):
+ if cmd.id == None:
+ raise ValueError("cannot complete unidentified command")
+ self._completed.add(cmd.id)
def known_completed(self, commands):
completed = RangedSet()
@@ -294,30 +241,24 @@
def __init__(self, session):
self.session = session
self.next_id = serial(0)
- self.next_offset = 0
- self.segments = []
+ self.commands = []
self._completed = RangedSet()
- def send(self, seg):
- seg.id = self.next_id
- seg.offset = self.next_offset
- if seg.last:
- self.next_id += 1
- self.next_offset = 0
- else:
- self.next_offset += len(seg.payload)
- self.segments.append(seg)
+ def send(self, cmd):
+ cmd.id = self.next_id
+ self.next_id += 1
if self.session.send_id:
self.session.send_id = False
- self.session.channel.session_command_point(seg.id, seg.offset)
- self.session.channel.connection.write_segment(seg)
+ self.session.channel.session_command_point(cmd.id, 0)
+ self.commands.append(cmd)
+ self.session.channel.connection.write_op(cmd)
def completed(self, commands):
idx = 0
- while idx < len(self.segments):
- seg = self.segments[idx]
- if seg.id in commands:
- del self.segments[idx]
+ while idx < len(self.commands):
+ cmd = self.commands[idx]
+ if cmd.id in commands:
+ del self.commands[idx]
else:
idx += 1
for range in commands.ranges:
@@ -332,7 +273,7 @@
def start(self):
self.session.message_set_flow_mode(self.destination, self.session.flow_mode.credit)
- for unit in self.session.credit_unit.values():
+ for unit in self.session.credit_unit.VALUES:
self.session.message_flow(self.destination, unit, 0xFFFFFFFFL)
def stop(self):
@@ -356,9 +297,9 @@
class Client(Delegate):
- def message_transfer(self, cmd, headers, body):
- m = Message(body)
- m.headers = headers
+ def message_transfer(self, cmd):
+ m = Message(cmd.payload)
+ m.headers = cmd.headers
m.id = cmd.id
messages = self.session.incoming(cmd.destination)
messages.put(m)
Modified: qpid/branches/java-broker-0-10/qpid/python/qpid/spec.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/qpid/spec.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/qpid/spec.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/qpid/spec.py Mon Oct 5 11:29:03 2009
@@ -29,7 +29,7 @@
situations.
"""
-import os, mllib, spec08, spec010
+import os, mllib, spec08
def default():
try:
@@ -54,7 +54,7 @@
minor = doc["amqp/@minor"]
if major == "0" and minor == "10":
- return spec010.load(specfile, *errata)
+ return None
else:
return spec08.load(specfile, *errata)
Modified: qpid/branches/java-broker-0-10/qpid/python/qpid/testlib.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/qpid/testlib.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/qpid/testlib.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/qpid/testlib.py Mon Oct 5 11:29:03 2009
@@ -21,239 +21,13 @@
# Support library for qpid python tests.
#
-import sys, re, unittest, os, random, logging, traceback
-import qpid.client, qpid.spec, qmf.console
+import unittest, traceback, socket
+import qpid.client, qmf.console
import Queue
-from fnmatch import fnmatch
-from getopt import getopt, GetoptError
from qpid.content import Content
from qpid.message import Message
-
-#0-10 support
-from qpid.connection import Connection
-from qpid.spec010 import load
-from qpid.util import connect, ssl, URL
-
-def findmodules(root):
- """Find potential python modules under directory root"""
- found = []
- for dirpath, subdirs, files in os.walk(root):
- modpath = dirpath.replace(os.sep, '.')
- if not re.match(r'\.svn$', dirpath): # Avoid SVN directories
- for f in files:
- match = re.match(r'(.+)\.py$', f)
- if match and f != '__init__.py':
- found.append('.'.join([modpath, match.group(1)]))
- return found
-
-def default(value, default):
- if (value == None): return default
- else: return value
-
-class TestRunner:
-
- SPEC_FOLDER = "../specs"
- qpidd = os.getenv("QPIDD")
-
- """Runs unit tests.
-
- Parses command line arguments, provides utility functions for tests,
- runs the selected test suite.
- """
-
- def _die(self, message = None):
- if message: print message
- print """
-run-tests [options] [test*]
-The name of a test is package.module.ClassName.testMethod
-Options:
- -?/-h/--help : this message
- -s/--spec <spec.xml> : URL of AMQP XML specification or one of these abbreviations:
- 0-8 - use the default 0-8 specification.
- 0-9 - use the default 0-9 specification.
- 0-10-errata - use the 0-10 specification with qpid errata.
- -e/--errata <errata.xml> : file containing amqp XML errata
- -b/--broker [amqps://][<user>[/<password>]@]<host>[:<port>] : broker to connect to
- -B/--start-broker <broker-args> : start a local broker using broker-args; set QPIDD
- env to point to broker executable. broker-args will be
- prepended with "--daemon --port=0"
- -v/--verbose : verbose - lists tests as they are run.
- -d/--debug : enable debug logging.
- -i/--ignore <test> : ignore the named test.
- -I/--ignore-file : file containing patterns to ignore.
- -S/--skip-self-test : skips the client self tests in the 'tests folder'
- -F/--spec-folder : folder that contains the specs to be loaded
- """
- sys.exit(1)
-
- def startBroker(self, brokerArgs):
- """Start a single broker daemon"""
- if TestRunner.qpidd == None:
- self._die("QPIDD environment var must point to qpidd when using -B/--start-broker")
- cmd = "%s --daemon --port=0 %s" % (TestRunner.qpidd, brokerArgs)
- portStr = os.popen(cmd).read()
- if len(portStr) == 0:
- self._die("%s failed to start" % TestRunner.qpidd)
- port = int(portStr)
- pid = int(os.popen("%s -p %d -c" % (TestRunner.qpidd, port)).read())
- print "Started broker: pid=%d, port=%d" % (pid, port)
- self.brokerTuple = (pid, port)
- self.setBroker("localhost:%d" % port)
-
- def stopBroker(self):
- """Stop the broker using qpidd -q"""
- if self.brokerTuple:
- ret = os.spawnl(os.P_WAIT, TestRunner.qpidd, TestRunner.qpidd, "--port=%d" % self.brokerTuple[1], "-q")
- if ret != 0:
- self._die("stop_node(): pid=%d port=%d: qpidd -q returned %d" % (self.brokerTuple[0], self.brokerTuple[1], ret))
- print "Stopped broker: pid=%d, port=%d" % self.brokerTuple
-
- def killBroker(self):
- """Kill the broker using kill -9 (SIGTERM)"""
- if self.brokerTuple:
- os.kill(self.brokerTuple[0], signal.SIGTERM)
- print "Killed broker: pid=%d, port=%d" % self.brokerTuple
-
- def setBroker(self, broker):
- try:
- self.url = URL(broker)
- except ValueError:
- self._die("'%s' is not a valid broker" % (broker))
- self.user = default(self.url.user, "guest")
- self.password = default(self.url.password, "guest")
- self.host = self.url.host
- if self.url.scheme == URL.AMQPS:
- self.ssl = True
- default_port = 5671
- else:
- self.ssl = False
- default_port = 5672
- self.port = default(self.url.port, default_port)
-
- def ignoreFile(self, filename):
- f = file(filename)
- for line in f.readlines(): self.ignore.append(line.strip())
- f.close()
-
- def use08spec(self):
- "True if we are running with the old 0-8 spec."
- # NB: AMQP 0-8 identifies itself as 8-0 for historical reasons.
- return self.spec.major==8 and self.spec.minor==0
-
- def use09spec(self):
- "True if we are running with the 0-9 (non-wip) spec."
- return self.spec.major==0 and self.spec.minor==9
-
- def _parseargs(self, args):
- # Defaults
- self.setBroker("localhost")
- self.verbose = 1
- self.ignore = []
- self.specfile = "0-8"
- self.errata = []
- self.skip_self_test = False
-
- try:
- opts, self.tests = getopt(args, "s:e:b:B:h?dvSi:I:F:",
- ["help", "spec", "errata=", "broker=",
- "start-broker=", "verbose", "skip-self-test", "ignore",
- "ignore-file", "spec-folder"])
- except GetoptError, e:
- self._die(str(e))
- # check for mutually exclusive options
- if "-B" in opts or "--start-broker" in opts:
- if "-b" in opts or "--broker" in opts:
- self._die("Cannot use -B/--start-broker and -b/broker options together")
- for opt, value in opts:
- if opt in ("-?", "-h", "--help"): self._die()
- if opt in ("-s", "--spec"): self.specfile = value
- if opt in ("-e", "--errata"): self.errata.append(value)
- if opt in ("-b", "--broker"): self.setBroker(value)
- if opt in ("-B", "--start-broker"): self.startBroker(value)
- if opt in ("-v", "--verbose"): self.verbose = 2
- if opt in ("-d", "--debug"): logging.basicConfig(level=logging.DEBUG)
- if opt in ("-i", "--ignore"): self.ignore.append(value)
- if opt in ("-I", "--ignore-file"): self.ignoreFile(value)
- if opt in ("-S", "--skip-self-test"): self.skip_self_test = True
- if opt in ("-F", "--spec-folder"): TestRunner.SPEC_FOLDER = value
-
- # Abbreviations for default settings.
- if (self.specfile == "0-10"):
- self.spec = load(self.get_spec_file("amqp.0-10.xml"))
- elif (self.specfile == "0-10-errata"):
- self.spec = load(self.get_spec_file("amqp.0-10-qpid-errata.xml"))
- else:
- if (self.specfile == "0-8"):
- self.specfile = self.get_spec_file("amqp.0-8.xml")
- elif (self.specfile == "0-9"):
- self.specfile = self.get_spec_file("amqp.0-9.xml")
- self.errata.append(self.get_spec_file("amqp-errata.0-9.xml"))
-
- if (self.specfile == None):
- self._die("No XML specification provided")
- print "Using specification from:", self.specfile
-
- self.spec = qpid.spec.load(self.specfile, *self.errata)
-
- if len(self.tests) == 0:
- if not self.skip_self_test:
- self.tests=findmodules("tests")
- if self.use08spec() or self.use09spec():
- self.tests+=findmodules("tests_0-8")
- elif (self.spec.major == 99 and self.spec.minor == 0):
- self.tests+=findmodules("tests_0-10_preview")
- elif (self.spec.major == 0 and self.spec.minor == 10):
- self.tests+=findmodules("tests_0-10")
-
- def testSuite(self):
- class IgnoringTestSuite(unittest.TestSuite):
- def addTest(self, test):
- if isinstance(test, unittest.TestCase):
- for pattern in testrunner.ignore:
- if fnmatch(test.id(), pattern):
- return
- unittest.TestSuite.addTest(self, test)
-
- # Use our IgnoringTestSuite in the test loader.
- unittest.TestLoader.suiteClass = IgnoringTestSuite
- return unittest.defaultTestLoader.loadTestsFromNames(self.tests)
-
- def run(self, args=sys.argv[1:]):
- self.brokerTuple = None
- self._parseargs(args)
- runner = unittest.TextTestRunner(descriptions=False,
- verbosity=self.verbose)
- result = runner.run(self.testSuite())
-
- if (self.ignore):
- print "======================================="
- print "NOTE: the following tests were ignored:"
- for t in self.ignore: print t
- print "======================================="
-
- self.stopBroker()
- return result.wasSuccessful()
-
- def connect(self, host=None, port=None, spec=None, user=None, password=None, tune_params=None):
- """Connect to the broker, returns a qpid.client.Client"""
- host = host or self.host
- port = port or self.port
- spec = spec or self.spec
- user = user or self.user
- password = password or self.password
- client = qpid.client.Client(host, port, spec)
- if self.use08spec():
- client.start({"LOGIN": user, "PASSWORD": password}, tune_params=tune_params)
- else:
- client.start("\x00" + user + "\x00" + password, mechanism="PLAIN", tune_params=tune_params)
- return client
-
- def get_spec_file(self, fname):
- return TestRunner.SPEC_FOLDER + os.sep + fname
-
-# Global instance for tests to call connect.
-testrunner = TestRunner()
-
+from qpid.harness import Skipped
+from qpid.exceptions import VersionError
class TestBase(unittest.TestCase):
"""Base class for Qpid test cases.
@@ -267,6 +41,9 @@
resources to clean up later.
"""
+ def configure(self, config):
+ self.config = config
+
def setUp(self):
self.queues = []
self.exchanges = []
@@ -293,9 +70,26 @@
else:
self.client.close()
- def connect(self, *args, **keys):
+ def connect(self, host=None, port=None, user=None, password=None, tune_params=None):
"""Create a new connction, return the Client object"""
- return testrunner.connect(*args, **keys)
+ host = host or self.config.broker.host
+ port = port or self.config.broker.port or 5672
+ user = user or "guest"
+ password = password or "guest"
+ client = qpid.client.Client(host, port)
+ try:
+ if client.spec.major == 8 and client.spec.minor == 0:
+ client.start({"LOGIN": user, "PASSWORD": password}, tune_params=tune_params)
+ else:
+ client.start("\x00" + user + "\x00" + password, mechanism="PLAIN", tune_params=tune_params)
+ except qpid.client.Closed, e:
+ if isinstance(e.args[0], VersionError):
+ raise Skipped(e.args[0])
+ else:
+ raise e
+ except socket.error, e:
+ raise Skipped(e)
+ return client
def queue_declare(self, channel=None, *args, **keys):
channel = channel or self.channel
@@ -319,17 +113,8 @@
def consume(self, queueName):
"""Consume from named queue returns the Queue object."""
- if testrunner.use08spec() or testrunner.use09spec():
- reply = self.channel.basic_consume(queue=queueName, no_ack=True)
- return self.client.queue(reply.consumer_tag)
- else:
- if not "uniqueTag" in dir(self): self.uniqueTag = 1
- else: self.uniqueTag += 1
- consumer_tag = "tag" + str(self.uniqueTag)
- self.channel.message_subscribe(queue=queueName, destination=consumer_tag)
- self.channel.message_flow(destination=consumer_tag, unit=0, value=0xFFFFFFFFL)
- self.channel.message_flow(destination=consumer_tag, unit=1, value=0xFFFFFFFFL)
- return self.client.queue(consumer_tag)
+ reply = self.channel.basic_consume(queue=queueName, no_ack=True)
+ return self.client.queue(reply.consumer_tag)
def subscribe(self, channel=None, **keys):
channel = channel or self.channel
@@ -350,24 +135,14 @@
Publish to exchange and assert queue.get() returns the same message.
"""
body = self.uniqueString()
- if testrunner.use08spec() or testrunner.use09spec():
- self.channel.basic_publish(
- exchange=exchange,
- content=Content(body, properties=properties),
- routing_key=routing_key)
- else:
- self.channel.message_transfer(
- destination=exchange,
- content=Content(body, properties={'application_headers':properties,'routing_key':routing_key}))
+ self.channel.basic_publish(
+ exchange=exchange,
+ content=Content(body, properties=properties),
+ routing_key=routing_key)
msg = queue.get(timeout=1)
- if testrunner.use08spec() or testrunner.use09spec():
- self.assertEqual(body, msg.content.body)
- if (properties):
- self.assertEqual(properties, msg.content.properties)
- else:
- self.assertEqual(body, msg.content.body)
- if (properties):
- self.assertEqual(properties, msg.content['application_headers'])
+ self.assertEqual(body, msg.content.body)
+ if (properties):
+ self.assertEqual(properties, msg.content.properties)
def assertPublishConsume(self, queue="", exchange="", routing_key="", properties=None):
"""
@@ -394,11 +169,20 @@
self.assertEqual("close", message.method.name)
self.assertEqual(expectedCode, message.reply_code)
+#0-10 support
+from qpid.connection import Connection
+from qpid.util import connect, ssl, URL
+
class TestBase010(unittest.TestCase):
"""
Base class for Qpid test cases. using the final 0-10 spec
"""
+ def configure(self, config):
+ self.config = config
+ self.broker = config.broker
+ self.defines = self.config.defines
+
def setUp(self):
self.conn = self.connect()
self.session = self.conn.session("test-session", timeout=10)
@@ -406,15 +190,26 @@
def startQmf(self, handler=None):
self.qmf = qmf.console.Session(handler)
- self.qmf_broker = self.qmf.addBroker(str(testrunner.url))
+ self.qmf_broker = self.qmf.addBroker(str(self.broker))
def connect(self, host=None, port=None):
- sock = connect(host or testrunner.host, port or testrunner.port)
- if testrunner.url.scheme == URL.AMQPS:
+ url = self.broker
+ if url.scheme == URL.AMQPS:
+ default_port = 5671
+ else:
+ default_port = 5672
+ try:
+ sock = connect(host or url.host, port or url.port or default_port)
+ except socket.error, e:
+ raise Skipped(e)
+ if url.scheme == URL.AMQPS:
sock = ssl(sock)
- conn = Connection(sock, username=testrunner.user,
- password=testrunner.password)
- conn.start(timeout=10)
+ conn = Connection(sock, username=url.user or "guest",
+ password=url.password or "guest")
+ try:
+ conn.start(timeout=10)
+ except VersionError, e:
+ raise Skipped(e)
return conn
def tearDown(self):
Modified: qpid/branches/java-broker-0-10/qpid/python/qpid/tests/framing.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/qpid/tests/framing.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/qpid/tests/framing.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/qpid/tests/framing.py Mon Oct 5 11:29:03 2009
@@ -40,6 +40,50 @@
assert seg1.channel == seg2.channel, "expected: %r, got %r" % (seg1, seg2)
assert seg1.payload == seg2.payload, "expected: %r, got %r" % (seg1, seg2)
+ def cmp_list(self, l1, l2):
+ if l1 is None:
+ assert l2 is None
+ return
+
+ assert len(l1) == len(l2)
+ for v1, v2 in zip(l1, l2):
+ if isinstance(v1, Compound):
+ self.cmp_ops(v1, v2)
+ else:
+ assert v1 == v2
+
+ def cmp_ops(self, op1, op2):
+ if op1 is None:
+ assert op2 is None
+ return
+
+ assert op1.__class__ == op2.__class__
+ cls = op1.__class__
+ assert op1.NAME == op2.NAME
+ assert op1.CODE == op2.CODE
+ assert op1.FIELDS == op2.FIELDS
+ for f in cls.FIELDS:
+ v1 = getattr(op1, f.name)
+ v2 = getattr(op2, f.name)
+ if COMPOUND.has_key(f.type) or f.type == "struct32":
+ self.cmp_ops(v1, v2)
+ elif f.type in ("list", "array"):
+ self.cmp_list(v1, v2)
+ else:
+ assert v1 == v2, "expected: %r, got %r" % (v1, v2)
+
+ if issubclass(cls, Command) or issubclass(cls, Control):
+ assert op1.channel == op2.channel
+
+ if issubclass(cls, Command):
+ assert op1.sync == op2.sync, "expected: %r, got %r" % (op1.sync, op2.sync)
+ assert (op1.headers is None and op2.headers is None) or \
+ (op1.headers is not None and op2.headers is not None)
+ if op1.headers is not None:
+ assert len(op1.headers) == len(op2.headers)
+ for h1, h2 in zip(op1.headers, op2.headers):
+ self.cmp_ops(h1, h2)
+
class FrameTest(Base):
def enc_dec(self, frames, encoded=None):
@@ -171,3 +215,75 @@
for i in range(0, 8, 2)]
self.enc_dec([Segment(False, False, 0, 0, 0, "abcdefgh")], frames, ilvd, max_payload=2)
+
+from qpid.ops import *
+
+class OpTest(Base):
+
+ def enc_dec(self, ops):
+ enc = OpEncoder()
+ dec = OpDecoder()
+ enc.write(*ops)
+ segs = enc.read()
+ dec.write(*segs)
+ dops = dec.read()
+ assert len(ops) == len(dops)
+ for op1, op2 in zip(ops, dops):
+ self.cmp_ops(op1, op2)
+
+ def testEmtpyMT(self):
+ self.enc_dec([MessageTransfer()])
+
+ def testEmptyMTSync(self):
+ self.enc_dec([MessageTransfer(sync=True)])
+
+ def testMT(self):
+ self.enc_dec([MessageTransfer(destination="asdf")])
+
+ def testSyncMT(self):
+ self.enc_dec([MessageTransfer(destination="asdf", sync=True)])
+
+ def testEmptyPayloadMT(self):
+ self.enc_dec([MessageTransfer(payload="")])
+
+ def testPayloadMT(self):
+ self.enc_dec([MessageTransfer(payload="test payload")])
+
+ def testHeadersEmptyPayloadMT(self):
+ self.enc_dec([MessageTransfer(headers=[DeliveryProperties()])])
+
+ def testHeadersPayloadMT(self):
+ self.enc_dec([MessageTransfer(headers=[DeliveryProperties()], payload="test payload")])
+
+ def testMultiHeadersEmptyPayloadMT(self):
+ self.enc_dec([MessageTransfer(headers=[DeliveryProperties(), MessageProperties()])])
+
+ def testMultiHeadersPayloadMT(self):
+ self.enc_dec([MessageTransfer(headers=[MessageProperties(), DeliveryProperties()], payload="test payload")])
+
+ def testContentTypeHeadersPayloadMT(self):
+ self.enc_dec([MessageTransfer(headers=[MessageProperties(content_type="text/plain")], payload="test payload")])
+
+ def testMulti(self):
+ self.enc_dec([MessageTransfer(),
+ MessageTransfer(sync=True),
+ MessageTransfer(destination="one"),
+ MessageTransfer(destination="two", sync=True),
+ MessageTransfer(destination="three", payload="test payload")])
+
+ def testControl(self):
+ self.enc_dec([SessionAttach(name="asdf")])
+
+ def testMixed(self):
+ self.enc_dec([SessionAttach(name="fdsa"), MessageTransfer(destination="test")])
+
+ def testChannel(self):
+ self.enc_dec([SessionAttach(name="asdf", channel=3), MessageTransfer(destination="test", channel=1)])
+
+ def testCompound(self):
+ self.enc_dec([MessageTransfer(headers=[MessageProperties(reply_to=ReplyTo(exchange="exch", routing_key="rk"))])])
+
+ def testListCompound(self):
+ self.enc_dec([ExecutionResult(value=RecoverResult(in_doubt=[Xid(global_id="one"),
+ Xid(global_id="two"),
+ Xid(global_id="three")]))])
Modified: qpid/branches/java-broker-0-10/qpid/python/qpid/tests/messaging.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/qpid/tests/messaging.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/qpid/tests/messaging.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/qpid/tests/messaging.py Mon Oct 5 11:29:03 2009
@@ -22,6 +22,7 @@
import time
from qpid.tests import Test
+from qpid.harness import Skipped
from qpid.messaging import Connection, ConnectError, Disconnected, Empty, Message, UNLIMITED, uuid4
from Queue import Queue, Empty as QueueEmpty
@@ -42,7 +43,10 @@
def setup(self):
self.test_id = uuid4()
self.broker = self.config.broker
- self.conn = self.setup_connection()
+ try:
+ self.conn = self.setup_connection()
+ except ConnectError, e:
+ raise Skipped(e)
self.ssn = self.setup_session()
self.snd = self.setup_sender()
self.rcv = self.setup_receiver()
@@ -65,7 +69,7 @@
receiver = ssn.receiver("ping-queue")
msg = receiver.fetch(0)
ssn.acknowledge()
- assert msg.content == content
+ assert msg.content == content, "expected %r, got %r" % (content, msg.content)
def drain(self, rcv, limit=None):
contents = []
Modified: qpid/branches/java-broker-0-10/qpid/python/qpid_config.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/qpid_config.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/qpid_config.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/qpid_config.py Mon Oct 5 11:29:03 2009
@@ -21,3 +21,5 @@
qpid_home = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
amqp_spec = os.path.join(qpid_home, "specs", "amqp.0-10-qpid-errata.xml")
+amqp_spec_0_8 = os.path.join(qpid_home, "specs", "amqp.0-8.xml")
+amqp_spec_0_9 = os.path.join(qpid_home, "specs", "amqp.0-9.xml")
Modified: qpid/branches/java-broker-0-10/qpid/python/tests/__init__.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests/__init__.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests/__init__.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests/__init__.py Mon Oct 5 11:29:03 2009
@@ -19,12 +19,4 @@
# under the License.
#
-from codec import *
-from queue import *
-from spec import *
-from framer import *
-from assembler import *
-from datatypes import *
-from connection import *
-from spec010 import *
-from codec010 import *
+import codec, queue, datatypes, connection, spec010, codec010
Modified: qpid/branches/java-broker-0-10/qpid/python/tests/codec.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests/codec.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests/codec.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests/codec.py Mon Oct 5 11:29:03 2009
@@ -23,7 +23,6 @@
from qpid.spec import load
from cStringIO import StringIO
from qpid.reference import ReferenceId
-from qpid.testlib import testrunner
__doc__ = """
@@ -54,13 +53,8 @@
"""
-SPEC = None
-
-def spec():
- global SPEC
- if SPEC == None:
- SPEC = load(testrunner.get_spec_file("amqp.0-8.xml"))
- return SPEC
+from qpid_config import amqp_spec_0_8
+SPEC = load(amqp_spec_0_8)
# --------------------------------------
# --------------------------------------
@@ -76,7 +70,7 @@
"""
standard setUp for unitetest (refer unittest documentation for details)
"""
- self.codec = Codec(StringIO(), spec())
+ self.codec = Codec(StringIO(), SPEC)
# ------------------
def tearDown(self):
@@ -507,7 +501,7 @@
else:
values = [value]
stream = StringIO()
- codec = Codec(stream, spec())
+ codec = Codec(stream, SPEC)
for v in values:
codec.encode(type, v)
codec.flush()
Modified: qpid/branches/java-broker-0-10/qpid/python/tests/codec010.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests/codec010.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests/codec010.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests/codec010.py Mon Oct 5 11:29:03 2009
@@ -20,21 +20,17 @@
import time
from unittest import TestCase
-from qpid.spec010 import load
from qpid.codec010 import StringCodec
-from qpid.testlib import testrunner
from qpid.datatypes import timestamp, uuid4
+from qpid.ops import PRIMITIVE
class CodecTest(TestCase):
- def setUp(self):
- self.spec = load(testrunner.get_spec_file("amqp.0-10.xml"))
-
def check(self, type, value, compare=True):
- t = self.spec[type]
- sc = StringCodec(self.spec)
- t.encode(sc, value)
- decoded = t.decode(sc)
+ t = PRIMITIVE[type]
+ sc = StringCodec()
+ sc.write_primitive(t, value)
+ decoded = sc.read_primitive(t)
if compare:
assert decoded == value, "%s, %s" % (decoded, value)
return decoded
Modified: qpid/branches/java-broker-0-10/qpid/python/tests/connection.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests/connection.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests/connection.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests/connection.py Mon Oct 5 11:29:03 2009
@@ -22,10 +22,10 @@
from qpid.util import connect, listen
from qpid.connection import *
from qpid.datatypes import Message
-from qpid.testlib import testrunner
from qpid.delegates import Server
from qpid.queue import Queue
from qpid.session import Delegate
+from qpid.ops import QueueQueryResult
PORT = 1234
@@ -51,12 +51,12 @@
pass
def queue_query(self, qq):
- return qq._type.result.type.new((qq.queue,), {})
+ return QueueQueryResult(qq.queue)
- def message_transfer(self, cmd, headers, body):
+ def message_transfer(self, cmd):
if cmd.destination == "echo":
- m = Message(body)
- m.headers = headers
+ m = Message(cmd.payload)
+ m.headers = cmd.headers
self.session.message_transfer(cmd.destination, cmd.accept_mode,
cmd.acquire_mode, m)
elif cmd.destination == "abort":
@@ -64,7 +64,7 @@
elif cmd.destination == "heartbeat":
self.session.channel.connection_heartbeat()
else:
- self.queue.put((cmd, headers, body))
+ self.queue.put(cmd)
class ConnectionTest(TestCase):
@@ -134,17 +134,17 @@
ssn.message_transfer(d)
for d in destinations:
- cmd, header, body = self.queue.get(10)
+ cmd = self.queue.get(10)
assert cmd.destination == d
- assert header == None
- assert body == None
+ assert cmd.headers == None
+ assert cmd.payload == None
msg = Message("this is a test")
ssn.message_transfer("four", message=msg)
- cmd, header, body = self.queue.get(10)
+ cmd = self.queue.get(10)
assert cmd.destination == "four"
- assert header == None
- assert body == msg.body
+ assert cmd.headers == None
+ assert cmd.payload == msg.body
qq = ssn.queue_query("asdf")
assert qq.queue == "asdf"
Modified: qpid/branches/java-broker-0-10/qpid/python/tests/datatypes.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests/datatypes.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests/datatypes.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests/datatypes.py Mon Oct 5 11:29:03 2009
@@ -18,9 +18,8 @@
#
from unittest import TestCase
-from qpid.testlib import testrunner
-from qpid.spec010 import load
from qpid.datatypes import *
+from qpid.ops import DeliveryProperties, FragmentProperties, MessageProperties
class SerialTest(TestCase):
@@ -176,10 +175,9 @@
class MessageTest(TestCase):
def setUp(self):
- self.spec = load(testrunner.get_spec_file("amqp.0-10-qpid-errata.xml"))
- self.mp = Struct(self.spec["message.message_properties"])
- self.dp = Struct(self.spec["message.delivery_properties"])
- self.fp = Struct(self.spec["message.fragment_properties"])
+ self.mp = MessageProperties()
+ self.dp = DeliveryProperties()
+ self.fp = FragmentProperties()
def testHas(self):
m = Message(self.mp, self.dp, self.fp, "body")
@@ -207,7 +205,7 @@
def testSetReplace(self):
m = Message(self.mp, self.dp, self.fp, "body")
- dp = Struct(self.spec["message.delivery_properties"])
+ dp = DeliveryProperties()
assert m.get("delivery_properties") == self.dp
assert m.get("delivery_properties") != dp
m.set(dp)
Modified: qpid/branches/java-broker-0-10/qpid/python/tests/spec010.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests/spec010.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests/spec010.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests/spec010.py Mon Oct 5 11:29:03 2009
@@ -19,66 +19,56 @@
import os, tempfile, shutil, stat
from unittest import TestCase
-from qpid.spec010 import load
from qpid.codec010 import Codec, StringCodec
-from qpid.testlib import testrunner
-from qpid.datatypes import Struct
+from qpid.ops import *
class SpecTest(TestCase):
- def setUp(self):
- self.spec = load(testrunner.get_spec_file("amqp.0-10-qpid-errata.xml"))
-
def testSessionHeader(self):
- hdr = self.spec["session.header"]
- sc = StringCodec(self.spec)
- hdr.encode(sc, Struct(hdr, sync=True))
+ sc = StringCodec()
+ sc.write_compound(Header(sync=True))
assert sc.encoded == "\x01\x01"
- sc = StringCodec(self.spec)
- hdr.encode(sc, Struct(hdr, sync=False))
+ sc = StringCodec()
+ sc.write_compound(Header(sync=False))
assert sc.encoded == "\x01\x00"
- def encdec(self, type, value):
- sc = StringCodec(self.spec)
- type.encode(sc, value)
- decoded = type.decode(sc)
+ def encdec(self, value):
+ sc = StringCodec()
+ sc.write_compound(value)
+ decoded = sc.read_compound(value.__class__)
return decoded
def testMessageProperties(self):
- mp = self.spec["message.message_properties"]
- rt = self.spec["message.reply_to"]
-
- props = Struct(mp, content_length=3735928559L,
- reply_to=Struct(rt, exchange="the exchange name",
- routing_key="the routing key"))
- dec = self.encdec(mp, props)
+ props = MessageProperties(content_length=3735928559L,
+ reply_to=ReplyTo(exchange="the exchange name",
+ routing_key="the routing key"))
+ dec = self.encdec(props)
assert props.content_length == dec.content_length
assert props.reply_to.exchange == dec.reply_to.exchange
assert props.reply_to.routing_key == dec.reply_to.routing_key
def testMessageSubscribe(self):
- ms = self.spec["message.subscribe"]
- cmd = Struct(ms, exclusive=True, destination="this is a test")
- dec = self.encdec(self.spec["message.subscribe"], cmd)
+ cmd = MessageSubscribe(exclusive=True, destination="this is a test")
+ dec = self.encdec(cmd)
assert cmd.exclusive == dec.exclusive
assert cmd.destination == dec.destination
def testXid(self):
- xid = self.spec["dtx.xid"]
- sc = StringCodec(self.spec)
- st = Struct(xid, format=0, global_id="gid", branch_id="bid")
- xid.encode(sc, st)
+ sc = StringCodec()
+ xid = Xid(format=0, global_id="gid", branch_id="bid")
+ sc.write_compound(xid)
assert sc.encoded == '\x00\x00\x00\x10\x06\x04\x07\x00\x00\x00\x00\x00\x03gid\x03bid'
- assert xid.decode(sc).__dict__ == st.__dict__
+ dec = sc.read_compound(Xid)
+ assert xid.__dict__ == dec.__dict__
- def testLoadReadOnly(self):
- spec = "amqp.0-10-qpid-errata.xml"
- f = testrunner.get_spec_file(spec)
- dest = tempfile.mkdtemp()
- shutil.copy(f, dest)
- shutil.copy(os.path.join(os.path.dirname(f), "amqp.0-10.dtd"), dest)
- os.chmod(dest, stat.S_IRUSR | stat.S_IXUSR)
- fname = os.path.join(dest, spec)
- load(fname)
- assert not os.path.exists("%s.pcl" % fname)
+# def testLoadReadOnly(self):
+# spec = "amqp.0-10-qpid-errata.xml"
+# f = testrunner.get_spec_file(spec)
+# dest = tempfile.mkdtemp()
+# shutil.copy(f, dest)
+# shutil.copy(os.path.join(os.path.dirname(f), "amqp.0-10.dtd"), dest)
+# os.chmod(dest, stat.S_IRUSR | stat.S_IXUSR)
+# fname = os.path.join(dest, spec)
+# load(fname)
+# assert not os.path.exists("%s.pcl" % fname)
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-10/__init__.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-10/__init__.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-10/__init__.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-10/__init__.py Mon Oct 5 11:29:03 2009
@@ -24,6 +24,7 @@
from dtx import *
from example import *
from exchange import *
+from management import *
from message import *
from query import *
from queue import *
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-10/management.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-10/management.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-10/management.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-10/management.py Mon Oct 5 11:29:03 2009
@@ -29,7 +29,7 @@
Tests for the management hooks
"""
- def test_broker_connectivity_oldAPI (self):
+ def disabled_test_broker_connectivity_oldAPI (self):
"""
Call the "echo" method on the broker to verify it is alive and talking.
"""
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-10/message.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-10/message.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-10/message.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-10/message.py Mon Oct 5 11:29:03 2009
@@ -477,7 +477,7 @@
#send message A
ssn.message_transfer(message=Message(ssn.delivery_properties(routing_key="q"), "A"))
- for unit in ssn.credit_unit.values():
+ for unit in ssn.credit_unit.VALUES:
ssn.message_flow("c", unit, 0xFFFFFFFFL)
q = ssn.incoming("c")
@@ -490,7 +490,7 @@
ssn.channel.session_completed(ssn.receiver._completed)
ssn.message_accept(RangedSet(msgA.id))
- for unit in ssn.credit_unit.values():
+ for unit in ssn.credit_unit.VALUES:
ssn.message_flow("c", unit, 0xFFFFFFFFL)
#send message B
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-10/tx.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-10/tx.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-10/tx.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-10/tx.py Mon Oct 5 11:29:03 2009
@@ -19,7 +19,7 @@
from qpid.client import Client, Closed
from qpid.queue import Empty
from qpid.datatypes import Message, RangedSet
-from qpid.testlib import testrunner, TestBase010
+from qpid.testlib import TestBase010
class TxTests(TestBase010):
"""
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-8/__init__.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-8/__init__.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-8/__init__.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-8/__init__.py Mon Oct 5 11:29:03 2009
@@ -18,3 +18,5 @@
# specific language governing permissions and limitations
# under the License.
#
+
+import basic, broker, example, exchange, queue, testlib, tx
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-8/basic.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-8/basic.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-8/basic.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-8/basic.py Mon Oct 5 11:29:03 2009
@@ -19,7 +19,7 @@
from qpid.client import Client, Closed
from qpid.queue import Empty
from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.testlib import TestBase
class BasicTests(TestBase):
"""Tests for 'methods' on the amqp basic 'class'"""
@@ -219,10 +219,11 @@
channel.basic_ack(delivery_tag=msg4.delivery_tag, multiple=False) #Four
channel.basic_cancel(consumer_tag=subscription.consumer_tag)
- subscription2 = channel.basic_consume(queue="test-requeue")
- queue2 = self.client.queue(subscription2.consumer_tag)
channel.basic_recover(requeue=True)
+
+ subscription2 = channel.basic_consume(queue="test-requeue")
+ queue2 = self.client.queue(subscription2.consumer_tag)
msg3b = queue2.get(timeout=1)
msg5b = queue2.get(timeout=1)
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-8/broker.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-8/broker.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-8/broker.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-8/broker.py Mon Oct 5 11:29:03 2009
@@ -19,15 +19,15 @@
from qpid.client import Closed
from qpid.queue import Empty
from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.testlib import TestBase
class BrokerTests(TestBase):
"""Tests for basic Broker functionality"""
- def test_amqp_basic_13(self):
+ def test_ack_and_no_ack(self):
"""
First, this test tries to receive a message with a no-ack
- consumer. Second, this test tries to explicitely receive and
+ consumer. Second, this test tries to explicitly receive and
acknowledge a message with an acknowledging consumer.
"""
ch = self.channel
@@ -40,7 +40,7 @@
msg = self.client.queue(ctag).get(timeout = 5)
self.assert_(msg.content.body == body)
- # Acknowleding consumer
+ # Acknowledging consumer
self.queue_declare(ch, queue = "otherqueue")
ctag = ch.basic_consume(queue = "otherqueue", no_ack = False).consumer_tag
body = "test ack"
@@ -102,3 +102,19 @@
except Closed, e:
self.assertConnectionException(504, e.args[0])
+ def test_channel_flow(self):
+ channel = self.channel
+ channel.queue_declare(queue="flow_test_queue", exclusive=True)
+ ctag = channel.basic_consume(queue="flow_test_queue", no_ack=True).consumer_tag
+ incoming = self.client.queue(ctag)
+
+ channel.channel_flow(active=False)
+ channel.basic_publish(routing_key="flow_test_queue", content=Content("abcdefghijklmnopqrstuvwxyz"))
+ try:
+ incoming.get(timeout=1)
+ self.fail("Received message when flow turned off.")
+ except Empty: None
+
+ channel.channel_flow(active=True)
+ msg = incoming.get(timeout=1)
+ self.assertEqual("abcdefghijklmnopqrstuvwxyz", msg.content.body)
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-8/example.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-8/example.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-8/example.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-8/example.py Mon Oct 5 11:29:03 2009
@@ -18,7 +18,7 @@
#
from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.testlib import TestBase
class ExampleTest (TestBase):
"""
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-8/queue.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-8/queue.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-8/queue.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-8/queue.py Mon Oct 5 11:29:03 2009
@@ -19,7 +19,7 @@
from qpid.client import Client, Closed
from qpid.queue import Empty
from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.testlib import TestBase
class QueueTests(TestBase):
"""Tests for 'methods' on the amqp queue 'class'"""
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-8/testlib.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-8/testlib.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-8/testlib.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-8/testlib.py Mon Oct 5 11:29:03 2009
@@ -22,7 +22,7 @@
#
from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.testlib import TestBase
from Queue import Empty
import sys
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-8/tx.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-8/tx.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-8/tx.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-8/tx.py Mon Oct 5 11:29:03 2009
@@ -19,7 +19,7 @@
from qpid.client import Client, Closed
from qpid.queue import Empty
from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.testlib import TestBase
class TxTests(TestBase):
"""
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-9/__init__.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-9/__init__.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-9/__init__.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-9/__init__.py Mon Oct 5 11:29:03 2009
@@ -18,3 +18,5 @@
# specific language governing permissions and limitations
# under the License.
#
+
+import query, queue
Modified: qpid/branches/java-broker-0-10/qpid/python/tests_0-9/query.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/python/tests_0-9/query.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/python/tests_0-9/query.py (original)
+++ qpid/branches/java-broker-0-10/qpid/python/tests_0-9/query.py Mon Oct 5 11:29:03 2009
@@ -19,7 +19,7 @@
from qpid.client import Client, Closed
from qpid.queue import Empty
from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.testlib import TestBase
class QueryTests(TestBase):
"""Tests for various query methods introduced in 0-10 and available in 0-9 for preview"""
Modified: qpid/branches/java-broker-0-10/qpid/review/agenda.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/review/agenda.py?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/review/agenda.py (original)
+++ qpid/branches/java-broker-0-10/qpid/review/agenda.py Mon Oct 5 11:29:03 2009
@@ -21,17 +21,36 @@
import sys, re
from popen2 import popen2, popen3
from optparse import OptionParser
+from xml.dom.minidom import parse, parseString
prereqs = ["tr", "svn", "xsltproc", "sed", "grep", "wget"]
-svncmd = "svn log https://svn.apache.org/repos/asf/incubator/qpid --xml -r %s:HEAD | tr '\\n\\r|' ' -' | xsltproc svnlog2wiki.xsl - | grep r | sed -e 's/^ *//' | sed -e 's/\\(QPID-[0-9]*\\)/\\[\\1 | https:\\/\\/issues.apache.org\\/jira\\/browse\\/\\1 \]/g'"
+svncmd = "svn log https://svn.apache.org/repos/asf/qpid/trunk/qpid/java --xml -r %s:HEAD | tr '\\n\\r|' ' -' | xsltproc svnlog2wiki.xsl - | grep r | sed -e 's/^ *//' | sed -e 's/\\(QPID-[0-9]*\\)/\\[\\1 | https:\\/\\/issues.apache.org\\/jira\\/browse\\/\\1 \]/g'"
jiracmd = "wget -q -O - http://issues.apache.org/jira/sr/jira.issueviews:searchrequest-xml/12312564/SearchRequest-12312564.xml?tempMax=1000 | tr '[]|' '()-' | xsltproc jiraRSS2wiki.xsl - | grep '|' | sed -e 's/^ *//'"
def get_commits(revision):
(stdout, stdin) = popen2(svncmd % revision)
- return stdout.read()
+ return add_jira_status(stdout.read())
+
+def add_jira_status(commits):
+ commit_lines = commits.split("\n")
+ new_commits = []
+ for commit in commit_lines:
+ if re.match(".*https://issues.apache.org/.*", commit):
+ jira = re.findall("QPID-[0-9]*", commit)[0]
+ jira_xml_url = "http://issues.apache.org/jira/si/jira.issueviews:issue-xml/%s/%s.xml" % (jira, jira)
+ (stdout, stdin) = popen2("wget -q -O - %s" % jira_xml_url)
+
+ jira_dom = parse(stdout)
+ status = jira_dom.getElementsByTagName("status")[0]
+ new_commits.append("%s %s | " % (commit, status.lastChild.data))
+ else:
+ new_commits.append(commit)
+
+ return "\n".join(new_commits)
+
def get_jiras():
(stdout, stdin) = popen2(jiracmd)
Modified: qpid/branches/java-broker-0-10/qpid/review/svnlog2wiki.xsl
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/review/svnlog2wiki.xsl?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/review/svnlog2wiki.xsl (original)
+++ qpid/branches/java-broker-0-10/qpid/review/svnlog2wiki.xsl Mon Oct 5 11:29:03 2009
@@ -22,7 +22,7 @@
<xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform" version="1.0">
<xsl:output method="text"></xsl:output>
<xsl:template match="/">
-|| revision || committer || date || comment || review notes ||
+|| revision || committer || date || comment || review notes || jira status ||
<xsl:apply-templates select="log/logentry"></xsl:apply-templates>
</xsl:template>
<xsl:template match="logentry">
Modified: qpid/branches/java-broker-0-10/qpid/specs/amqp.0-9.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-0-10/qpid/specs/amqp.0-9.xml?rev=821761&r1=821760&r2=821761&view=diff
==============================================================================
--- qpid/branches/java-broker-0-10/qpid/specs/amqp.0-9.xml (original)
+++ qpid/branches/java-broker-0-10/qpid/specs/amqp.0-9.xml Mon Oct 5 11:29:03 2009
@@ -1659,29 +1659,28 @@
<doc>This method confirms the deletion of an exchange.</doc>
<chassis name = "client" implement = "MUST" />
</method>
-
- <!-- RG : Added Exchange.bound and Exchange.bound-ok -->
- <method name="bound" synchronous="1" index="22">
- <chassis name="server" implement="SHOULD"/>
- <field name="exchange" domain="exchange-name"/>
- <field name = "routing-key" type = "shortstr">
- Message routing key
- <doc>
- Specifies the routing key for the message. The routing key is
- used for routing messages depending on the exchange configuration.
- </doc>
- </field>
- <field name = "queue" domain = "queue name"/>
+
+ <!-- RG : Added Exchange.bound and Exchange.bound-ok -->
+ <method name="bound" synchronous="1" index="22">
+ <chassis name="server" implement="SHOULD"/>
+ <response name="bound-ok"/>
+ <field name="exchange" domain="exchange-name"/>
+ <field name = "routing-key" type = "shortstr">
+ Message routing key
+ <doc>
+ Specifies the routing key for the message. The routing key is
+ used for routing messages depending on the exchange configuration.
+ </doc>
+ </field>
+ <field name = "queue" domain = "queue name"/>
</method>
<method name="bound-ok" synchronous="1" index="23">
- <field name="reply-code" domain="reply-code"/>
- <field name="reply-text" domain="reply-text"/>
- <chassis name="client" implement="SHOULD"/>
+ <field name="reply-code" domain="reply-code"/>
+ <field name="reply-text" domain="reply-text"/>
+ <chassis name="client" implement="SHOULD"/>
</method>
-
-
</class>
<!-- == QUEUE ============================================================ -->
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org