QPID-1142: made session.sync() always set the sync flag on execution_sync

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@668344 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 23 insertions, 7 deletions

diff --git a/python/qpid/session.py b/python/qpid/session.py
index 3d8c0aa7fd..2f70461ab6 100644
--- a/python/qpid/session.py
+++ b/python/qpid/session.py
@@ -91,10 +91,11 @@ class Session(Invoker):
       return tuple(exc)
 
   def sync(self, timeout=None):
-    if currentThread() == self.channel.connection.thread:
+    ch = self.channel
+    if ch is not None and currentThread() == ch.connection.thread:
       raise SessionException("deadlock detected")
     if not self.auto_sync:
-      self.execution_sync()
+      self.execution_sync(sync=True)
     last = self.sender.next_id - 1
     if not wait(self.condition, lambda:
                   last in self.sender._completed or self.exceptions,
@@ -174,10 +175,11 @@ class Session(Invoker):
     else:
       message = None
 
+    hdr = Struct(self.spec["session.header"])
+    hdr.sync = self.auto_sync or kwargs.pop("sync", False)
+
     cmd = type.new(args, kwargs)
     sc = StringCodec(self.spec)
-    hdr = Struct(self.spec["session.header"])
-    hdr.sync = self.auto_sync
     sc.write_command(hdr, cmd)
 
     seg = Segment(True, (message == None or
diff --git a/python/tests/connection.py b/python/tests/connection.py
index 88620bc1c6..23e0c937fb 100644
--- a/python/tests/connection.py
+++ b/python/tests/connection.py
@@ -48,6 +48,9 @@ class TestSession(Delegate):
     self.session = session
     self.queue = queue
 
+  def execution_sync(self, es):
+    pass
+
   def queue_query(self, qq):
     return qq._type.result.type.new((qq.queue,), {})
 
@@ -90,8 +93,11 @@ class ConnectionTest(TestCase):
     connect("0.0.0.0", PORT).close()
     self.server.join(3)
 
+  def connect(self):
+    return Connection(connect("0.0.0.0", PORT), self.spec)
+
   def test(self):
-    c = Connection(connect("0.0.0.0", PORT), self.spec)
+    c = self.connect()
     c.start(10)
 
     ssn1 = c.session("test1", timeout=10)
@@ -144,7 +150,7 @@ class ConnectionTest(TestCase):
     c.close(5)
 
   def testCloseGet(self):
-    c = Connection(connect("0.0.0.0", PORT), self.spec)
+    c = self.connect()
     c.start(10)
     ssn = c.session("test", timeout=10)
     echos = ssn.incoming("echo")
@@ -166,7 +172,7 @@ class ConnectionTest(TestCase):
       pass
 
   def testCloseListen(self):
-    c = Connection(connect("0.0.0.0", PORT), self.spec)
+    c = self.connect()
     c.start(10)
     ssn = c.session("test", timeout=10)
     echos = ssn.incoming("echo")
@@ -198,3 +204,11 @@ class ConnectionTest(TestCase):
       assert m.body == "test%d" % i
 
     assert len(exceptions) == 1
+
+  def testSync(self):
+    c = self.connect()
+    c.start(10)
+    s = c.session("test")
+    s.auto_sync = False
+    s.message_transfer("echo", message=Message("test"))
+    s.sync(10)