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