pyobjc / pyobjc-framework-CFNetwork / PyObjCTest / test_cfsocketstream.py

from PyObjCTools.TestSupport import *
from CFNetwork import *

try:
    long
except NameError:
    long = int

try:
    unicode
except NameError:
    unicode = str


class TestCFSocketStream (TestCase):
    @min_os_level('10.7')
    def testConstants10_7(self):
        self.assertIsInstance(kCFStreamNetworkServiceType, unicode)
        self.assertIsInstance(kCFStreamNetworkServiceTypeVoIP, unicode)
        self.assertIsInstance(kCFStreamNetworkServiceTypeVideo, unicode)
        self.assertIsInstance(kCFStreamNetworkServiceTypeBackground, unicode)
        self.assertIsInstance(kCFStreamNetworkServiceTypeVoice, unicode)

    @min_os_level('10.5')
    def testConstants10_5(self):
        self.assertIsInstance(kCFStreamPropertySSLPeerTrust, unicode)
        self.assertIsInstance(kCFStreamErrorDomainWinSock, (int, long))

    def testConstants(self):
        self.assertIsInstance(kCFStreamPropertySSLPeerCertificates, unicode)

        self.assertIsInstance(kCFStreamPropertySSLPeerCertificates, unicode)
        self.assertIsInstance(kCFStreamPropertySSLSettings, unicode)
        self.assertIsInstance(kCFStreamSSLLevel, unicode)
        self.assertIsInstance(kCFStreamSSLAllowsExpiredCertificates, unicode)
        self.assertIsInstance(kCFStreamSSLAllowsExpiredRoots, unicode)
        self.assertIsInstance(kCFStreamSSLAllowsAnyRoot, unicode)
        self.assertIsInstance(kCFStreamSSLValidatesCertificateChain, unicode)
        self.assertIsInstance(kCFStreamSSLPeerName, unicode)
        self.assertIsInstance(kCFStreamSSLCertificates, unicode)
        self.assertIsInstance(kCFStreamSSLIsServer, unicode)
        self.assertIsInstance(kCFStreamErrorDomainSOCKS, (int, long))

        self.assertEqual(kCFStreamErrorSOCKSSubDomainNone, 0)
        self.assertEqual(kCFStreamErrorSOCKSSubDomainVersionCode, 1)
        self.assertEqual(kCFStreamErrorSOCKS4SubDomainResponse, 2)
        self.assertEqual(kCFStreamErrorSOCKS5SubDomainUserPass, 3)
        self.assertEqual(kCFStreamErrorSOCKS5SubDomainMethod, 4)
        self.assertEqual(kCFStreamErrorSOCKS5SubDomainResponse, 5)
        self.assertEqual(kCFStreamErrorSOCKS5BadResponseAddr, 1)
        self.assertEqual(kCFStreamErrorSOCKS5BadState, 2)
        self.assertEqual(kCFStreamErrorSOCKSUnknownClientVersion, 3)
        self.assertEqual(kCFStreamErrorSOCKS4RequestFailed, 91)
        self.assertEqual(kCFStreamErrorSOCKS4IdentdFailed, 92)
        self.assertEqual(kCFStreamErrorSOCKS4IdConflict, 93)
        self.assertEqual(kSOCKS5NoAcceptableMethod, 0xFF)

        self.assertIsInstance(kCFStreamPropertySOCKSProxy, unicode)
        self.assertIsInstance(kCFStreamPropertySOCKSProxyHost, unicode)
        self.assertIsInstance(kCFStreamPropertySOCKSProxyPort, unicode)
        self.assertIsInstance(kCFStreamPropertySOCKSVersion, unicode)
        self.assertIsInstance(kCFStreamSocketSOCKSVersion4, unicode)
        self.assertIsInstance(kCFStreamSocketSOCKSVersion5, unicode)
        self.assertIsInstance(kCFStreamPropertySOCKSUser, unicode)
        self.assertIsInstance(kCFStreamPropertySOCKSPassword, unicode)
        self.assertIsInstance(kCFStreamErrorDomainSSL, (int, long))
        self.assertIsInstance(kCFStreamPropertySocketSecurityLevel, unicode)
        self.assertIsInstance(kCFStreamSocketSecurityLevelNone, unicode)
        self.assertIsInstance(kCFStreamSocketSecurityLevelSSLv2, unicode)
        self.assertIsInstance(kCFStreamSocketSecurityLevelSSLv3, unicode)
        self.assertIsInstance(kCFStreamSocketSecurityLevelTLSv1, unicode)
        self.assertIsInstance(kCFStreamSocketSecurityLevelNegotiatedSSL, unicode)
        self.assertIsInstance(kCFStreamPropertyShouldCloseNativeSocket, unicode)
        self.assertIsInstance(kCFStreamPropertySocketRemoteHost, unicode)
        self.assertIsInstance(kCFStreamPropertySocketRemoteNetService, unicode)
        self.assertEqual(kCFStreamSocketSecurityNone, 0)
        self.assertEqual(kCFStreamSocketSecuritySSLv2, 1)
        self.assertEqual(kCFStreamSocketSecuritySSLv3, 2)
        self.assertEqual(kCFStreamSocketSecuritySSLv23, 3)
        self.assertEqual(kCFStreamSocketSecurityTLSv1, 4)

        self.assertIsInstance(kCFStreamPropertyProxyLocalBypass, unicode)

    def testFunctions(self):
        err = CFStreamError()
        err.error = (1<<16)+2

        v = CFSocketStreamSOCKSGetErrorSubdomain(err)
        self.assertEqual(v, 1)

        v = CFSocketStreamSOCKSGetError(err)
        self.assertEqual(v, 2)

        self.assertArgIsCFRetained(CFStreamCreatePairWithSocketToCFHost, 3)
        self.assertArgIsCFRetained(CFStreamCreatePairWithSocketToCFHost, 4)
        self.assertArgIsOut(CFStreamCreatePairWithSocketToCFHost, 3)
        self.assertArgIsOut(CFStreamCreatePairWithSocketToCFHost, 4)

        host = CFHostCreateWithName(None, "connect.apple.com")
        rd, wr = CFStreamCreatePairWithSocketToCFHost(None, host, 443, None, None)
        self.assertIsInstance(rd, CFReadStreamRef)
        self.assertIsInstance(wr, CFWriteStreamRef)

        self.assertResultIsBOOL(CFSocketStreamPairSetSecurityProtocol)
        v = CFSocketStreamPairSetSecurityProtocol(rd, wr, kCFStreamSocketSecuritySSLv23)
        self.assertIsInstance(v, bool)

        self.assertArgIsCFRetained(CFStreamCreatePairWithSocketToNetService, 2)
        self.assertArgIsCFRetained(CFStreamCreatePairWithSocketToNetService, 3)
        self.assertArgIsOut(CFStreamCreatePairWithSocketToNetService, 2)
        self.assertArgIsOut(CFStreamCreatePairWithSocketToNetService, 3)
        service = CFNetServiceCreate(None, "pyobjc.local", "ssh", "pyobjc.test.local", 9999)
        rd, wr = CFStreamCreatePairWithSocketToNetService(None, service, None, None)
        self.assertIsInstance(rd, CFReadStreamRef)
        self.assertIsInstance(wr, CFWriteStreamRef)

    @min_os_level('10.8')
    def testConstants10_8(self):
        # XXX: iOS only:
        #self.assertIsInstance(kCFStreamPropertySSLContext, unicode)
        self.assertIsInstance(kCFStreamPropertyNoCellular, unicode)
        self.assertIsInstance(kCFStreamPropertyConnectionIsCellular, unicode)

if __name__ == "__main__":
    main()
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.