twisted/test/test_strports.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Tests for L{twisted.application.strports}.
   6 """
   7
   8 from twisted.trial.unittest import TestCase
   9 from twisted.application import strports
  10 from twisted.application import internet
  11 from twisted.internet.test.test_endpoints import ParserTestCase
  12 from twisted.internet.protocol import Factory
  13 from twisted.internet.endpoints import TCP4ServerEndpoint, UNIXServerEndpoint
  14
  15
  16
  17 class DeprecatedParseTestCase(ParserTestCase):
  18     """
  19     L{strports.parse} is deprecated.  It's an alias for a method that is now
  20     private in L{twisted.internet.endpoints}.
  21     """
  22
  23     def parse(self, *a, **kw):
  24         result = strports.parse(*a, **kw)
  25         warnings = self.flushWarnings([self.parse])
  26         self.assertEqual(len(warnings), 1)
  27         self.assertEqual(
  28             warnings[0]['message'],
  29             "twisted.application.strports.parse was deprecated "
  30             "in Twisted 10.2.0: in favor of twisted.internet.endpoints.serverFromString")
  31         return result
  32
  33
  34     def test_simpleNumeric(self):
  35         """
  36         Base numeric ports should be parsed as TCP.
  37         """
  38         self.assertEqual(self.parse('80', self.f),
  39                          ('TCP', (80, self.f), {'interface':'', 'backlog':50}))
  40
  41
  42     def test_allKeywords(self):
  43         """
  44         A collection of keyword arguments with no prefixed type, like 'port=80',
  45         will be parsed as keyword arguments to 'tcp'.
  46         """
  47         self.assertEqual(self.parse('port=80', self.f),
  48                          ('TCP', (80, self.f), {'interface':'', 'backlog':50}))
  49
  50
  51
  52 class ServiceTestCase(TestCase):
  53     """
  54     Tests for L{strports.service}.
  55     """
  56
  57     def test_service(self):
  58         """
  59         L{strports.service} returns a L{StreamServerEndpointService}
  60         constructed with an endpoint produced from
  61         L{endpoint.serverFromString}, using the same syntax.
  62         """
  63         reactor = object() # the cake is a lie
  64         aFactory = Factory()
  65         aGoodPort = 1337
  66         svc = strports.service(
  67             'tcp:'+str(aGoodPort), aFactory, reactor=reactor)
  68         self.assertIsInstance(svc, internet.StreamServerEndpointService)
  69
  70         # See twisted.application.test.test_internet.TestEndpointService.
  71         # test_synchronousRaiseRaisesSynchronously
  72         self.assertEqual(svc._raiseSynchronously, True)
  73         self.assertIsInstance(svc.endpoint, TCP4ServerEndpoint)
  74         # Maybe we should implement equality for endpoints.
  75         self.assertEqual(svc.endpoint._port, aGoodPort)
  76         self.assertIdentical(svc.factory, aFactory)
  77         self.assertIdentical(svc.endpoint._reactor, reactor)
  78
  79
  80     def test_serviceDefaultReactor(self):
  81         """
  82         L{strports.service} will use the default reactor when none is provided
  83         as an argument.
  84         """
  85         from twisted.internet import reactor as globalReactor
  86         aService = strports.service("tcp:80", None)
  87         self.assertIdentical(aService.endpoint._reactor, globalReactor)
  88
  89
  90     def test_serviceDeprecatedDefault(self):
  91         """
  92         L{strports.service} still accepts a 'default' argument, which will
  93         affect the parsing of 'default' (i.e. 'not containing a colon')
  94         endpoint descriptions, but this behavior is deprecated.
  95         """
  96         svc = strports.service("8080", None, "unix")
  97         self.assertIsInstance(svc.endpoint, UNIXServerEndpoint)
  98         warnings = self.flushWarnings([self.test_serviceDeprecatedDefault])
  99         self.assertEqual(warnings[0]['category'], DeprecationWarning)
 100         self.assertEqual(
 101             warnings[0]['message'],
 102             "The 'default' parameter was deprecated in Twisted 10.2.0.  "
 103             "Use qualified endpoint descriptions; for example, 'tcp:8080'.")
 104         self.assertEqual(len(warnings), 1)
 105
 106         # Almost the same case, but slightly tricky - explicitly passing the old
 107         # default value, None, also must trigger a deprecation warning.
 108         svc = strports.service("tcp:8080", None, None)
 109         self.assertIsInstance(svc.endpoint, TCP4ServerEndpoint)
 110         warnings = self.flushWarnings([self.test_serviceDeprecatedDefault])
 111         self.assertEqual(warnings[0]['category'], DeprecationWarning)
 112         self.assertEqual(
 113             warnings[0]['message'],
 114             "The 'default' parameter was deprecated in Twisted 10.2.0.")
 115         self.assertEqual(len(warnings), 1)
 116
 117
 118     def test_serviceDeprecatedUnqualified(self):
 119         """
 120         Unqualified strport descriptions, i.e. "8080", are deprecated.
 121         """
 122         svc = strports.service("8080", None)
 123         self.assertIsInstance(svc.endpoint, TCP4ServerEndpoint)
 124         warnings = self.flushWarnings(
 125             [self.test_serviceDeprecatedUnqualified])
 126         self.assertEqual(warnings[0]['category'], DeprecationWarning)
 127         self.assertEqual(
 128             warnings[0]['message'],
 129             "Unqualified strport description passed to 'service'."
 130             "Use qualified endpoint descriptions; for example, 'tcp:8080'.")
 131         self.assertEqual(len(warnings), 1)
 132
 133