twisted/python/test/test_systemd.py

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