py/test/Ice/operations/OnewaysNewAMI.py

   1 # **********************************************************************
   2 #
   3 # Copyright (c) 2003-2011 ZeroC, Inc. All rights reserved.
   4 #
   5 # This copy of Ice is licensed to you under the terms described in the
   6 # ICE_LICENSE file included in this distribution.
   7 #
   8 # **********************************************************************
   9
  10 import Ice, Test, threading
  11
  12 def test(b):
  13     if not b:
  14         raise RuntimeError('test assertion failed')
  15
  16 class CallbackBase:
  17     def __init__(self):
  18         self._called = False
  19         self._cond = threading.Condition()
  20
  21     def check(self):
  22         self._cond.acquire()
  23         try:
  24             while not self._called:
  25                 self._cond.wait()
  26             self._called = False
  27         finally:
  28             self._cond.release()
  29
  30     def called(self):
  31         self._cond.acquire()
  32         self._called = True
  33         self._cond.notify()
  34         self._cond.release()
  35
  36 class Callback(CallbackBase):
  37     def sent(self, sentSynchronously):
  38         self.called()
  39
  40     def noException(self, ex):
  41         test(False)
  42
  43 def onewaysNewAMI(communicator, proxy):
  44
  45     p = Test.MyClassPrx.uncheckedCast(proxy.ice_oneway())
  46
  47     cb = Callback()
  48     p.begin_ice_ping(None, cb.noException, cb.sent)
  49     cb.check()
  50
  51     try:
  52         p.begin_ice_isA(Test.MyClass.ice_staticId())
  53         test(False)
  54     except RuntimeError:
  55         pass
  56
  57     try:
  58         p.begin_ice_id()
  59         test(False)
  60     except RuntimeError:
  61         pass
  62
  63     try:
  64         p.begin_ice_ids()
  65         test(False)
  66     except RuntimeError:
  67         pass
  68
  69     cb = Callback()
  70     p.begin_opVoid(None, cb.noException, cb.sent)
  71     cb.check()
  72
  73     cb = Callback()
  74     p.begin_opIdempotent(None, cb.noException, cb.sent)
  75     cb.check()
  76
  77     cb = Callback()
  78     p.begin_opNonmutating(None, cb.noException, cb.sent)
  79     cb.check()
  80
  81     try:
  82         p.begin_opByte(0xff, 0x0f)
  83         test(False)
  84     except RuntimeError:
  85         pass