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