Viewing file: test_unix.py (4.29 KB) -rw-r--r-- Select action/file-type: (+) | (+) | (+) | Code (+) | Session (+) | (+) | SDB (+) | (+) | (+) | (+) | (+) | (+) |
# Copyright (c) 2008-2010 Twisted Matrix Laboratories. # See LICENSE for details.
""" Tests for implementations of L{IReactorUNIX}. """
from stat import S_IMODE from os import stat from sys import platform
try: from socket import AF_UNIX except ImportError: AF_UNIX = None
from twisted.python.hashlib import md5 from twisted.internet.address import UNIXAddress from twisted.internet.protocol import ( ServerFactory, ClientFactory, DatagramProtocol) from twisted.internet.test.reactormixins import ReactorBuilder
class UNIXFamilyMixin: """ Test-helper defining mixin for things related to AF_UNIX sockets. """ if AF_UNIX is None: skip = "Platform does not support AF_UNIX sockets"
def _modeTest(self, methodName, path, factory): """ Assert that the mode of the created unix socket is set to the mode specified to the reactor method. """ mode = 0600 reactor = self.buildReactor() unixPort = getattr(reactor, methodName)(path, factory, mode=mode) unixPort.stopListening() self.assertEqual(S_IMODE(stat(path).st_mode), mode)
def _abstractPath(case): """ Return a new, unique abstract namespace path to be listened on. """ # Use the test cases's mktemp to get something unique, but also squash it # down to make sure it fits in the unix socket path limit (something around # 110 bytes). return md5(case.mktemp()).hexdigest()
class UNIXTestsBuilder(UNIXFamilyMixin, ReactorBuilder): """ Builder defining tests relating to L{IReactorUNIX}. """ def test_mode(self): """ The UNIX socket created by L{IReactorUNIX.listenUNIX} is created with the mode specified. """ self._modeTest('listenUNIX', self.mktemp(), ServerFactory())
def test_listenOnLinuxAbstractNamespace(self): """ On Linux, a UNIX socket path may begin with C{'\0'} to indicate a socket in the abstract namespace. L{IReactorUNIX.listenUNIX} accepts such a path. """ # Don't listen on a path longer than the maximum allowed. path = _abstractPath(self) reactor = self.buildReactor() port = reactor.listenUNIX('\0' + path, ServerFactory()) self.assertEquals(port.getHost(), UNIXAddress('\0' + path)) if platform != 'linux2': test_listenOnLinuxAbstractNamespace.skip = ( 'Abstract namespace UNIX sockets only supported on Linux.')
def test_connectToLinuxAbstractNamespace(self): """ L{IReactorUNIX.connectUNIX} also accepts a Linux abstract namespace path. """ path = _abstractPath(self) reactor = self.buildReactor() connector = reactor.connectUNIX('\0' + path, ClientFactory()) self.assertEquals( connector.getDestination(), UNIXAddress('\0' + path)) if platform != 'linux2': test_connectToLinuxAbstractNamespace.skip = ( 'Abstract namespace UNIX sockets only supported on Linux.')
class UNIXDatagramTestsBuilder(UNIXFamilyMixin, ReactorBuilder): """ Builder defining tests relating to L{IReactorUNIXDatagram}. """ # There's no corresponding test_connectMode because the mode parameter to # connectUNIXDatagram has been completely ignored since that API was first # introduced. def test_listenMode(self): """ The UNIX socket created by L{IReactorUNIXDatagram.listenUNIXDatagram} is created with the mode specified. """ self._modeTest('listenUNIXDatagram', self.mktemp(), DatagramProtocol())
def test_listenOnLinuxAbstractNamespace(self): """ On Linux, a UNIX socket path may begin with C{'\0'} to indicate a socket in the abstract namespace. L{IReactorUNIX.listenUNIXDatagram} accepts such a path. """ path = _abstractPath(self) reactor = self.buildReactor() port = reactor.listenUNIXDatagram('\0' + path, DatagramProtocol()) self.assertEquals(port.getHost(), UNIXAddress('\0' + path)) if platform != 'linux2': test_listenOnLinuxAbstractNamespace.skip = ( 'Abstract namespace UNIX sockets only supported on Linux.')
globals().update(UNIXTestsBuilder.makeTestCaseClasses()) globals().update(UNIXDatagramTestsBuilder.makeTestCaseClasses())
|