twisted/mail/test/test_options.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Tests for L{twisted.mail.tap}.
   6 """
   7
   8 from twisted.trial.unittest import TestCase
   9
  10 from twisted.python.usage import UsageError
  11 from twisted.mail import protocols
  12 from twisted.mail.tap import Options, makeService
  13 from twisted.python import deprecate, versions
  14 from twisted.python.filepath import FilePath
  15 from twisted.internet import endpoints, defer
  16
  17
  18 class OptionsTestCase(TestCase):
  19     """
  20     Tests for the command line option parser used for I{twistd mail}.
  21     """
  22     def setUp(self):
  23         self.aliasFilename = self.mktemp()
  24         aliasFile = file(self.aliasFilename, 'w')
  25         aliasFile.write('someuser:\tdifferentuser\n')
  26         aliasFile.close()
  27
  28
  29     def testAliasesWithoutDomain(self):
  30         """
  31         Test that adding an aliases(5) file before adding a domain raises a
  32         UsageError.
  33         """
  34         self.assertRaises(
  35             UsageError,
  36             Options().parseOptions,
  37             ['--aliases', self.aliasFilename])
  38
  39
  40     def testAliases(self):
  41         """
  42         Test that adding an aliases(5) file to an IAliasableDomain at least
  43         doesn't raise an unhandled exception.
  44         """
  45         Options().parseOptions([
  46             '--maildirdbmdomain', 'example.com=example.com',
  47             '--aliases', self.aliasFilename])
  48
  49
  50     def testPasswordfileDeprecation(self):
  51         """
  52         Test that the --passwordfile option will emit a correct warning.
  53         """
  54         passwd = FilePath(self.mktemp())
  55         passwd.setContent("")
  56         options = Options()
  57         options.opt_passwordfile(passwd.path)
  58         warnings = self.flushWarnings([self.testPasswordfileDeprecation])
  59         self.assertEqual(warnings[0]['category'], DeprecationWarning)
  60         self.assertEqual(len(warnings), 1)
  61         msg = deprecate.getDeprecationWarningString(options.opt_passwordfile,
  62                              versions.Version('twisted.mail', 11, 0, 0))
  63         self.assertEqual(warnings[0]['message'], msg)
  64
  65
  66     def test_barePort(self):
  67         """
  68         A bare port passed to I{--pop3} results in deprecation warning in
  69         addition to a TCP4ServerEndpoint.
  70         """
  71         options = Options()
  72         options.parseOptions(['--pop3', '8110'])
  73         self.assertEqual(len(options['pop3']), 1)
  74         self.assertIsInstance(
  75             options['pop3'][0], endpoints.TCP4ServerEndpoint)
  76         warnings = self.flushWarnings([options.opt_pop3])
  77         self.assertEqual(len(warnings), 1)
  78         self.assertEqual(warnings[0]['category'], DeprecationWarning)
  79         self.assertEqual(
  80             warnings[0]['message'],
  81             "Specifying plain ports and/or a certificate is deprecated since "
  82             "Twisted 11.0; use endpoint descriptions instead.")
  83
  84
  85     def _endpointTest(self, service):
  86         """
  87         Use L{Options} to parse a single service configuration parameter and
  88         verify that an endpoint of the correct type is added to the list for
  89         that service.
  90         """
  91         options = Options()
  92         options.parseOptions(['--' + service, 'tcp:1234'])
  93         self.assertEqual(len(options[service]), 1)
  94         self.assertIsInstance(
  95             options[service][0], endpoints.TCP4ServerEndpoint)
  96
  97
  98     def test_endpointSMTP(self):
  99         """
 100         When I{--smtp} is given a TCP endpoint description as an argument, a
 101         TCPServerEndpoint is added to the list of SMTP endpoints.
 102         """
 103         self._endpointTest('smtp')
 104
 105
 106     def test_endpointPOP3(self):
 107         """
 108         When I{--pop3} is given a TCP endpoint description as an argument, a
 109         TCPServerEndpoint is added to the list of POP3 endpoints.
 110         """
 111         self._endpointTest('pop3')
 112
 113
 114     def test_protoDefaults(self):
 115         """
 116         POP3 and SMTP each listen on a TCP4ServerEndpoint by default.
 117         """
 118         options = Options()
 119         options.parseOptions([])
 120
 121         self.assertEqual(len(options['pop3']), 1)
 122         self.assertIsInstance(
 123             options['pop3'][0], endpoints.TCP4ServerEndpoint)
 124
 125         self.assertEqual(len(options['smtp']), 1)
 126         self.assertIsInstance(
 127             options['smtp'][0], endpoints.TCP4ServerEndpoint)
 128
 129
 130     def test_protoDisable(self):
 131         """
 132         The I{--no-pop3} and I{--no-smtp} options disable POP3 and SMTP
 133         respectively.
 134         """
 135         options = Options()
 136         options.parseOptions(['--no-pop3'])
 137         self.assertEqual(options._getEndpoints(None, 'pop3'), [])
 138         self.assertNotEquals(options._getEndpoints(None, 'smtp'), [])
 139
 140         options = Options()
 141         options.parseOptions(['--no-smtp'])
 142         self.assertNotEquals(options._getEndpoints(None, 'pop3'), [])
 143         self.assertEqual(options._getEndpoints(None, 'smtp'), [])
 144
 145
 146     def test_allProtosDisabledError(self):
 147         """
 148         If all protocols are disabled, L{UsageError} is raised.
 149         """
 150         options = Options()
 151         self.assertRaises(
 152             UsageError, options.parseOptions, (['--no-pop3', '--no-smtp']))
 153
 154
 155     def test_pop3sBackwardCompatibility(self):
 156         """
 157         The deprecated I{--pop3s} and I{--certificate} options set up a POP3 SSL
 158         server.
 159         """
 160         cert = FilePath(self.mktemp())
 161         cert.setContent("")
 162         options = Options()
 163         options.parseOptions(['--pop3s', '8995',
 164                               '--certificate', cert.path])
 165         self.assertEqual(len(options['pop3']), 2)
 166         self.assertIsInstance(
 167             options['pop3'][0], endpoints.SSL4ServerEndpoint)
 168         self.assertIsInstance(
 169             options['pop3'][1], endpoints.TCP4ServerEndpoint)
 170
 171         warnings = self.flushWarnings([options.postOptions])
 172         self.assertEqual(len(warnings), 1)
 173         self.assertEqual(warnings[0]['category'], DeprecationWarning)
 174         self.assertEqual(
 175             warnings[0]['message'],
 176             "Specifying plain ports and/or a certificate is deprecated since "
 177             "Twisted 11.0; use endpoint descriptions instead.")
 178
 179
 180     def test_esmtpWithoutHostname(self):
 181         """
 182         If I{--esmtp} is given without I{--hostname}, L{Options.parseOptions}
 183         raises L{UsageError}.
 184         """
 185         options = Options()
 186         exc = self.assertRaises(UsageError, options.parseOptions, ['--esmtp'])
 187         self.assertEqual("--esmtp requires --hostname", str(exc))
 188
 189
 190     def test_auth(self):
 191         """
 192         Tests that the --auth option registers a checker.
 193         """
 194         options = Options()
 195         options.parseOptions(['--auth', 'memory:admin:admin:bob:password'])
 196         self.assertEqual(len(options['credCheckers']), 1)
 197         checker = options['credCheckers'][0]
 198         interfaces = checker.credentialInterfaces
 199         registered_checkers = options.service.smtpPortal.checkers
 200         for iface in interfaces:
 201             self.assertEqual(checker, registered_checkers[iface])
 202
 203
 204
 205 class SpyEndpoint(object):
 206     """
 207     SpyEndpoint remembers what factory it is told to listen with.
 208     """
 209     listeningWith = None
 210     def listen(self, factory):
 211         self.listeningWith = factory
 212         return defer.succeed(None)
 213
 214
 215
 216 class MakeServiceTests(TestCase):
 217     """
 218     Tests for L{twisted.mail.tap.makeService}
 219     """
 220     def _endpointServerTest(self, key, factoryClass):
 221         """
 222         Configure a service with two endpoints for the protocol associated with
 223         C{key} and verify that when the service is started a factory of type
 224         C{factoryClass} is used to listen on each of them.
 225         """
 226         cleartext = SpyEndpoint()
 227         secure = SpyEndpoint()
 228         config = Options()
 229         config[key] = [cleartext, secure]
 230         service = makeService(config)
 231         service.privilegedStartService()
 232         service.startService()
 233         self.addCleanup(service.stopService)
 234         self.assertIsInstance(cleartext.listeningWith, factoryClass)
 235         self.assertIsInstance(secure.listeningWith, factoryClass)
 236
 237
 238     def test_pop3(self):
 239         """
 240         If one or more endpoints is included in the configuration passed to
 241         L{makeService} for the C{"pop3"} key, a service for starting a POP3
 242         server is constructed for each of them and attached to the returned
 243         service.
 244         """
 245         self._endpointServerTest("pop3", protocols.POP3Factory)
 246
 247
 248     def test_smtp(self):
 249         """
 250         If one or more endpoints is included in the configuration passed to
 251         L{makeService} for the C{"smtp"} key, a service for starting an SMTP
 252         server is constructed for each of them and attached to the returned
 253         service.
 254         """
 255         self._endpointServerTest("smtp", protocols.SMTPFactory)