1from PyObjCTools.TestSupport import * 2from CoreFoundation import * 3 4 5class TestMessagePort (TestCase): 6 def testTypes(self): 7 self.failUnlessIsCFType(CFMessagePortRef) 8 9 def testConstants(self): 10 self.failUnless(kCFMessagePortSuccess == 0) 11 self.failUnless(kCFMessagePortSendTimeout == -1) 12 self.failUnless(kCFMessagePortReceiveTimeout == -2) 13 self.failUnless(kCFMessagePortIsInvalid == -3) 14 self.failUnless(kCFMessagePortTransportError == -4) 15 16 def testTypeID(self): 17 self.failUnless(isinstance(CFMessagePortGetTypeID(), (int, long))) 18 19 def testInteraction(self): 20 class Context: pass 21 context = Context() 22 23 def callout(port, messageid, data, info): 24 pass 25 26 port, shouldFree = CFMessagePortCreateLocal(None, u"name", callout, context, None) 27 self.failUnless(isinstance(port, CFMessagePortRef)) 28 self.failUnless(shouldFree is True or shouldFree is False) 29 30 self.failIf(CFMessagePortIsRemote(port)) 31 ctx = CFMessagePortGetContext(port) 32 self.failUnless(ctx is context) 33 34 35 port = CFMessagePortCreateRemote(None, u"name") 36 self.failUnless(isinstance(port, CFMessagePortRef)) 37 38 self.failUnlessResultIsBOOL(CFMessagePortIsRemote) 39 self.failUnless(CFMessagePortIsRemote(port)) 40 self.failUnless(CFMessagePortGetName(port), u"name") 41 42 CFMessagePortSetName(port, "newname") 43 self.failUnless(CFMessagePortGetName(port), u"newname") 44 45 cb = CFMessagePortGetInvalidationCallBack(port) 46 self.failUnless(cb is None) 47 48 global didInvalidate 49 didInvalidate = False 50 51 @objc.callbackFor(CFMessagePortSetInvalidationCallBack) 52 def invalidate(port, info): 53 global didInvalidate 54 didInvalidate = True 55 56 CFMessagePortSetInvalidationCallBack(port, invalidate) 57 cb = CFMessagePortGetInvalidationCallBack(port) 58 59 # XXX: Without writing a custom wrapper we cannot guarantee this 60 #self.failUnless(cb is invalidate) 61 62 cb(None, None) 63 self.failUnless(didInvalidate is True) 64 didInvalidate = False 65 66 rls = CFMessagePortCreateRunLoopSource(None, port, 0) 67 self.failUnless(isinstance(rls, CFRunLoopSourceRef)) 68 69 self.failUnlessResultIsBOOL(CFMessagePortIsValid) 70 self.failUnless(CFMessagePortIsValid(port)) 71 CFMessagePortInvalidate(port) 72 self.failIf(CFMessagePortIsValid(port)) 73 self.failUnless(didInvalidate) 74 75 @min_os_level('10.5') 76 def testSending(self): 77 # FIXME: Crash on Tiger 78 context = [] 79 def callout(port, messageid, data, info): 80 print "callout" 81 info.append((port, messageid, data)) 82 return buffer("hello world") 83 84 port, shouldFree = CFMessagePortCreateLocal(None, u"pyobjc.test", callout, context, None) 85 self.failUnlessIsInstance(port, CFMessagePortRef) 86 87 self.failUnlessArgIsOut(CFMessagePortSendRequest, 6) 88 rls = CFMessagePortCreateRunLoopSource(None, port, 0) 89 err, data = CFMessagePortSendRequest(port, 99, None, 1.0, 1.0, None, None) 90 self.failUnlessEqual(err, 0) 91 self.failUnlessEqual(data, None) 92 93 94if __name__ == "__main__": 95 main() 96