← Back to branch summary

~hudson-openstack/nova/trunk

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/internet/test/test_unix.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/internet/test/test_unix.py
 
1
# Copyright (c) 2008 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
"""
 
5
Tests for implementations of L{IReactorUNIX}.
 
6
"""
 
7
 
 
8
from stat import S_IMODE
 
9
from os import stat
 
10
from socket import socket, SOCK_DGRAM
 
11
try:
 
12
    from socket import AF_UNIX
 
13
except ImportError:
 
14
    AF_UNIX = None
 
15
 
 
16
from twisted.trial import util
 
17
from twisted.internet.protocol import ServerFactory, DatagramProtocol
 
18
from twisted.internet.protocol import ConnectedDatagramProtocol
 
19
from twisted.internet.test.reactormixins import ReactorBuilder
 
20
 
 
21
 
 
22
_deprecatedModeMessage = (
 
23
    'The mode parameter of %(interface)s.%(method)s will be removed.  Do '
 
24
    'not pass a value for it.  Set permissions on the containing directory '
 
25
    'before calling %(interface)s.%(method)s, instead.')
 
26
 
 
27
 
 
28
class UNIXFamilyMixin:
 
29
    """
 
30
    Test-helper defining mixin for things related to AF_UNIX sockets.
 
31
    """
 
32
    if AF_UNIX is None:
 
33
        skip = "Platform does not support AF_UNIX sockets"
 
34
 
 
35
    def _modeTest(self, methodName, path, factory):
 
36
        """
 
37
        Assert that the mode of the created unix socket is set to the mode
 
38
        specified to the reactor method.
 
39
        """
 
40
        mode = 0600
 
41
        reactor = self.buildReactor()
 
42
        unixPort = getattr(reactor, methodName)(path, factory, mode=mode)
 
43
        unixPort.stopListening()
 
44
        self.assertEqual(S_IMODE(stat(path).st_mode), mode)
 
45
 
 
46
 
 
47
    def _deprecatedModeTest(self, interfaceName, methodName, path, factory):
 
48
        """
 
49
        Assert that a deprecation warning is emitted when a value is specified
 
50
        for the mode parameter to the indicated reactor method.
 
51
        """
 
52
        reactor = self.buildReactor()
 
53
        method = getattr(reactor, methodName)
 
54
        port = self.assertWarns(
 
55
            DeprecationWarning,
 
56
            _deprecatedModeMessage % dict(
 
57
                interface=interfaceName, method=methodName),
 
58
            __file__,
 
59
            lambda: method(path, factory, mode=0246))
 
60
        port.stopListening()
 
61
 
 
62
 
 
63
 
 
64
class UNIXTestsBuilder(UNIXFamilyMixin, ReactorBuilder):
 
65
    """
 
66
    Builder defining tests relating to L{IReactorUNIX}.
 
67
    """
 
68
    def test_mode(self):
 
69
        """
 
70
        The UNIX socket created by L{IReactorUNIX.listenUNIX} is created with
 
71
        the mode specified.
 
72
        """
 
73
        self._modeTest('listenUNIX', self.mktemp(), ServerFactory())
 
74
    test_mode.suppress = [
 
75
        util.suppress(category=DeprecationWarning,
 
76
                      message=_deprecatedModeMessage % dict(
 
77
                interface='IReactorUNIX',
 
78
                method='listenUNIX'))]
 
79
 
 
80
 
 
81
    def test_deprecatedMode(self):
 
82
        """
 
83
        Passing any value for the C{mode} parameter of L{listenUNIX} causes a
 
84
        deprecation warning to be emitted.
 
85
        """
 
86
        self._deprecatedModeTest(
 
87
            'IReactorUNIX', 'listenUNIX', self.mktemp(), ServerFactory())
 
88
 
 
89
 
 
90
 
 
91
class UNIXDatagramTestsBuilder(UNIXFamilyMixin, ReactorBuilder):
 
92
    """
 
93
    Builder defining tests relating to L{IReactorUNIXDatagram}.
 
94
    """
 
95
    # There's no corresponding test_connectMode because the mode parameter to
 
96
    # connectUNIXDatagram has been completely ignored since that API was first
 
97
    # introduced.
 
98
    def test_listenMode(self):
 
99
        """
 
100
        The UNIX socket created by L{IReactorUNIXDatagram.listenUNIXDatagram}
 
101
        is created with the mode specified.
 
102
        """
 
103
        self._modeTest('listenUNIXDatagram', self.mktemp(), DatagramProtocol())
 
104
    test_listenMode.suppress = [
 
105
        util.suppress(category=DeprecationWarning,
 
106
                      message=_deprecatedModeMessage % dict(
 
107
                interface='IReactorUNIXDatagram',
 
108
                method='listenUNIXDatagram'))]
 
109
 
 
110
 
 
111
    def test_deprecatedListenMode(self):
 
112
        """
 
113
        Passing any value for the C{mode} parameter of L{listenUNIXDatagram}
 
114
        causes a deprecation warning to be emitted.
 
115
        """
 
116
        self._deprecatedModeTest(
 
117
            'IReactorUNIXDatagram', 'listenUNIXDatagram', self.mktemp(),
 
118
            DatagramProtocol())
 
119
 
 
120
 
 
121
    def test_deprecatedConnectMode(self):
 
122
        """
 
123
        Passing any value for the C{mode} parameter of L{connectUNIXDatagram}
 
124
        causes a deprecation warning to be emitted.
 
125
        """
 
126
        path = self.mktemp()
 
127
        server = socket(AF_UNIX, SOCK_DGRAM)
 
128
        server.bind(path)
 
129
        self.addCleanup(server.close)
 
130
 
 
131
        self._deprecatedModeTest(
 
132
            'IReactorUNIXDatagram', 'connectUNIXDatagram',
 
133
            path, ConnectedDatagramProtocol())
 
134
 
 
135
 
 
136
globals().update(UNIXTestsBuilder.makeTestCaseClasses())
 
137
globals().update(UNIXDatagramTestsBuilder.makeTestCaseClasses())