1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
|
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Tests for L{twisted.python.systemd}.
"""
import os
from twisted.trial.unittest import TestCase
from twisted.python.systemd import ListenFDs
class InheritedDescriptorsMixin(object):
"""
Mixin for a L{TestCase} subclass which defines test methods for some kind of
systemd sd-daemon class. In particular, it defines tests for a
C{inheritedDescriptors} method.
"""
def test_inheritedDescriptors(self):
"""
C{inheritedDescriptors} returns a list of integers giving the file
descriptors which were inherited from systemd.
"""
sddaemon = self.getDaemon(7, 3)
self.assertEqual([7, 8, 9], sddaemon.inheritedDescriptors())
def test_repeated(self):
"""
Any subsequent calls to C{inheritedDescriptors} return the same list.
"""
sddaemon = self.getDaemon(7, 3)
self.assertEqual(
sddaemon.inheritedDescriptors(),
sddaemon.inheritedDescriptors())
class MemoryOnlyMixin(object):
"""
Mixin for a L{TestCase} subclass which creates creating a fake, in-memory
implementation of C{inheritedDescriptors}. This provides verification that
the fake behaves in a compatible way with the real implementation.
"""
def getDaemon(self, start, count):
"""
Invent C{count} new I{file descriptors} (actually integers, attached to
no real file description), starting at C{start}. Construct and return a
new L{ListenFDs} which will claim those integers represent inherited
file descriptors.
"""
return ListenFDs(range(start, start + count))
class EnvironmentMixin(object):
"""
Mixin for a L{TestCase} subclass which creates a real implementation of
C{inheritedDescriptors} which is based on the environment variables set by
systemd. To facilitate testing, this mixin will also create a fake
environment dictionary and add keys to it to make it look as if some
descriptors have been inherited.
"""
def initializeEnvironment(self, count, pid):
"""
Create a copy of the process environment and add I{LISTEN_FDS} and
I{LISTEN_PID} (the environment variables set by systemd) to it.
"""
result = os.environ.copy()
result['LISTEN_FDS'] = str(count)
result['LISTEN_PID'] = str(pid)
return result
def getDaemon(self, start, count):
"""
Create a new L{ListenFDs} instance, initialized with a fake environment
dictionary which will be set up as systemd would have set it up if
C{count} descriptors were being inherited. The descriptors will also
start at C{start}.
"""
fakeEnvironment = self.initializeEnvironment(count, os.getpid())
return ListenFDs.fromEnvironment(environ=fakeEnvironment, start=start)
class MemoryOnlyTests(MemoryOnlyMixin, InheritedDescriptorsMixin, TestCase):
"""
Apply tests to L{ListenFDs}, explicitly constructed with some fake file
descriptors.
"""
class EnvironmentTests(EnvironmentMixin, InheritedDescriptorsMixin, TestCase):
"""
Apply tests to L{ListenFDs}, constructed based on an environment dictionary.
"""
def test_secondEnvironment(self):
"""
Only a single L{Environment} can extract inherited file descriptors.
"""
fakeEnvironment = self.initializeEnvironment(3, os.getpid())
first = ListenFDs.fromEnvironment(environ=fakeEnvironment)
second = ListenFDs.fromEnvironment(environ=fakeEnvironment)
self.assertEqual(range(3, 6), first.inheritedDescriptors())
self.assertEqual([], second.inheritedDescriptors())
def test_mismatchedPID(self):
"""
If the current process PID does not match the PID in the environment, no
inherited descriptors are reported.
"""
fakeEnvironment = self.initializeEnvironment(3, os.getpid() + 1)
sddaemon = ListenFDs.fromEnvironment(environ=fakeEnvironment)
self.assertEqual([], sddaemon.inheritedDescriptors())
def test_missingPIDVariable(self):
"""
If the I{LISTEN_PID} environment variable is not present, no inherited
descriptors are reported.
"""
fakeEnvironment = self.initializeEnvironment(3, os.getpid())
del fakeEnvironment['LISTEN_PID']
sddaemon = ListenFDs.fromEnvironment(environ=fakeEnvironment)
self.assertEqual([], sddaemon.inheritedDescriptors())
def test_nonIntegerPIDVariable(self):
"""
If the I{LISTEN_PID} environment variable is set to a string that cannot
be parsed as an integer, no inherited descriptors are reported.
"""
fakeEnvironment = self.initializeEnvironment(3, "hello, world")
sddaemon = ListenFDs.fromEnvironment(environ=fakeEnvironment)
self.assertEqual([], sddaemon.inheritedDescriptors())
def test_missingFDSVariable(self):
"""
If the I{LISTEN_FDS} environment variable is not present, no inherited
descriptors are reported.
"""
fakeEnvironment = self.initializeEnvironment(3, os.getpid())
del fakeEnvironment['LISTEN_FDS']
sddaemon = ListenFDs.fromEnvironment(environ=fakeEnvironment)
self.assertEqual([], sddaemon.inheritedDescriptors())
def test_nonIntegerFDSVariable(self):
"""
If the I{LISTEN_FDS} environment variable is set to a string that cannot
be parsed as an integer, no inherited descriptors are reported.
"""
fakeEnvironment = self.initializeEnvironment("hello, world", os.getpid())
sddaemon = ListenFDs.fromEnvironment(environ=fakeEnvironment)
self.assertEqual([], sddaemon.inheritedDescriptors())
def test_defaultEnviron(self):
"""
If the process environment is not explicitly passed to
L{Environment.__init__}, the real process environment dictionary is
used.
"""
self.patch(os, 'environ', {
'LISTEN_PID': str(os.getpid()),
'LISTEN_FDS': '5'})
sddaemon = ListenFDs.fromEnvironment()
self.assertEqual(range(3, 3 + 5), sddaemon.inheritedDescriptors())
|
