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