""" Tests for L{xmantissa.port}. """ from twisted.trial.unittest import TestCase from twisted.application.service import IService, IServiceCollection from twisted.internet.protocol import ServerFactory from twisted.internet.defer import Deferred from twisted.internet.ssl import CertificateOptions from axiom.store import Store from axiom.item import Item from axiom.attributes import inmemory, integer from axiom.dependency import installOn from xmantissa.port import TCPPort, SSLPort class DummyPort(object): """ Stub class used to track what reactor listen calls have been made and what created ports have been stopped. """ stopping = None def __init__(self, portNumber, factory, contextFactory=None): self.portNumber = portNumber self.factory = factory self.contextFactory = contextFactory def stopListening(self): assert self.stopping is None self.stopping = Deferred() return self.stopping class DummyFactory(Item): """ Helper class used as a stand-in for a real protocol factory by the unit tests. """ dummyAttribute = integer(doc=""" Meaningless attribute which serves only to make this a valid Item subclass. """) realFactory = inmemory(doc=""" A reference to the protocol factory which L{getFactory} will return. """) def getFactory(self): return self.realFactory class PortTestsMixin: """ Test method-defining mixin class for port types with C{portNumber} and C{factory} attributes. Included are tests for various persistence-related behaviors as well as the L{IService} implementation which all ports should have. @ivar portType: The L{Item} subclass which will be tested. @ivar lowPortNumber: A port number which requires privileges to bind on POSIX. Used to test L{privilegedStartService}. @ivar highPortNumber: A port number which does not require privileges to bind on POSIX. Used to test the interaction between L{privilegedStartService} and L{startService}. @ivar dbdir: The path at which to create the test L{Store}. This must be bound before L{setUp} is run, since that is the only method which examines its value. @ivar ports: A list of ports which have been bound using L{listen}. created in L{setUp}. """ portType = None lowPortNumber = 123 highPortNumber = 1234 dbdir = None def port(self, **kw): """ Create and return a new port instance with the given attribute values. """ return self.portType(**kw) def listen(self, *a, **kw): """ Pretend to bind a port. Used as a stub implementation of a reactor listen method. Subclasses should override and implement to append useful information to C{self.ports}. """ raise NotImplementedError def checkPort(self, port, alternatePort=None): """ Assert that the given port has been properly created. @type port: L{DummyPort} @param port: A port which has been created by the code being tested. @type alternatePort: C{int} @param alternatePort: If not C{None}, the port number on which C{port} should be listening. """ raise NotImplementedError def setUp(self): self.store = Store(self.dbdir) self.realFactory = ServerFactory() self.factory = DummyFactory(store=self.store, realFactory=self.realFactory) self.ports = [] def test_portNumberAttribute(self): """ Test that C{self.portType} remembers the port number it is told to listen on. """ port = self.port(store=self.store, portNumber=self.lowPortNumber) self.assertEqual(port.portNumber, self.lowPortNumber) def test_factoryAttribute(self): """ Test that C{self.portType} remembers the factory it is given to associate with its port. """ port = self.port(store=self.store, factory=self.factory) self.assertIdentical(port.factory, self.factory) def test_service(self): """ Test that C{self.portType} becomes a service on the store it is installed on. """ port = self.port(store=self.store) installOn(port, self.store) self.assertEqual( list(self.store.powerupsFor(IService)), [port]) def test_setServiceParent(self): """ Test that the C{self.portType.setServiceParent} method adds the C{self.portType} to the Axiom Store Service as a child. """ port = self.port(store=self.store) port.setServiceParent(self.store) self.failUnlessIn(port, list(IService(self.store))) def test_disownServiceParent(self): """ Test that the C{self.portType.disownServiceParent} method removes the C{self.portType} from the Axiom Store Service. """ port = self.port(store=self.store) port.setServiceParent(self.store) port.disownServiceParent() self.failIfIn(port, list(IService(self.store))) def test_serviceParent(self): """ Test that C{self.portType} is a child of the store service after it is installed. """ port = self.port(store=self.store) installOn(port, self.store) service = IServiceCollection(self.store) self.failUnlessIn(port, list(service)) def _start(self, portNumber, methodName): port = self.port(store=self.store, portNumber=portNumber, factory=self.factory) port._listen = self.listen getattr(port, methodName)() return self.ports def _privilegedStartService(self, portNumber): return self._start(portNumber, 'privilegedStartService') def _startService(self, portNumber): return self._start(portNumber, 'startService') def test_startPrivilegedService(self): """ Test that C{self.portType} binds a low-numbered port with the reactor when it is started with privilege. """ ports = self._privilegedStartService(self.lowPortNumber) self.assertEqual(len(ports), 1) self.checkPort(ports[0]) def test_dontStartPrivilegedService(self): """ Test that C{self.portType} doesn't bind a high-numbered port with the reactor when it is started with privilege. """ ports = self._privilegedStartService(self.highPortNumber) self.assertEqual(ports, []) def test_startServiceLow(self): """ Test that C{self.portType} binds a low-numbered port with the reactor when it is started without privileged. """ ports = self._startService(self.lowPortNumber) self.assertEqual(len(ports), 1) self.checkPort(ports[0]) def test_startServiceHigh(self): """ Test that C{self.portType} binds a low-numbered port with the reactor when it is started without privileged. """ ports = self._startService(self.highPortNumber) self.assertEqual(len(ports), 1) self.checkPort(ports[0], self.highPortNumber) def test_startedOnce(self): """ Test that C{self.portType} only binds one network port when C{privilegedStartService} and C{startService} are both called. """ port = self.port(store=self.store, portNumber=self.lowPortNumber, factory=self.factory) port._listen = self.listen port.privilegedStartService() self.assertEqual(len(self.ports), 1) self.checkPort(self.ports[0]) port.startService() self.assertEqual(len(self.ports), 1) def test_stopService(self): """ Test that C{self.portType} cleans up its listening port when it is stopped. """ port = self.port(store=self.store, portNumber=self.lowPortNumber, factory=self.factory) port._listen = self.listen port.startService() stopped = port.stopService() stopping = self.ports[0].stopping self.failIfIdentical(stopping, None) self.assertIdentical(stopped, stopping) def test_deletedFactory(self): """ Test that the deletion of a C{self.portType}'s factory item results in the C{self.portType} being deleted. """ port = self.port(store=self.store, portNumber=self.lowPortNumber, factory=self.factory) self.factory.deleteFromStore() self.assertEqual(list(self.store.query(self.portType)), []) def test_deletionDisownsParent(self): """ Test that a deleted C{self.portType} no longer shows up in the children list of the service which used to be its parent. """ port = self.port(store=self.store, portNumber=self.lowPortNumber, factory=self.factory) port.setServiceParent(self.store) port.deleteFromStore() service = IServiceCollection(self.store) self.failIfIn(port, list(service)) class TCPPortTests(PortTestsMixin, TestCase): """ Tests for L{xmantissa.port.TCPPort}. """ portType = TCPPort def checkPort(self, port, alternatePort=None): if alternatePort is None: alternatePort = self.lowPortNumber self.assertEqual(port.portNumber, alternatePort) self.assertEqual(port.factory, self.realFactory) def listen(self, port, factory): self.ports.append(DummyPort(port, factory)) return self.ports[-1] class SSLPortTests(PortTestsMixin, TestCase): """ Tests for L{xmantissa.port.SSLPort}. """ portType = SSLPort dummyCertificateData = """ -----BEGIN CERTIFICATE----- MIICmTCCAgICAQEwDQYJKoZIhvcNAQEEBQAwgZQxCzAJBgNVBAYTAlVTMRQwEgYD VQQDEwtleGFtcGxlLmNvbTERMA8GA1UEBxMITmV3IFlvcmsxEzARBgNVBAoTCkRp dm1vZCBMTEMxETAPBgNVBAgTCE5ldyBZb3JrMSIwIAYJKoZIhvcNAQkBFhNzdXBw b3J0QGV4YW1wbGUuY29tMRAwDgYDVQQLEwdUZXN0aW5nMB4XDTA2MTIzMDE5MDEx NloXDTA3MTIzMDE5MDExNlowgZQxCzAJBgNVBAYTAlVTMRQwEgYDVQQDEwtleGFt cGxlLmNvbTERMA8GA1UEBxMITmV3IFlvcmsxEzARBgNVBAoTCkRpdm1vZCBMTEMx ETAPBgNVBAgTCE5ldyBZb3JrMSIwIAYJKoZIhvcNAQkBFhNzdXBwb3J0QGV4YW1w bGUuY29tMRAwDgYDVQQLEwdUZXN0aW5nMIGfMA0GCSqGSIb3DQEBAQUAA4GNADCB iQKBgQCrmNNyXLHAETcDH8Uxhmbo8IhFFMx1C4i7oTHTKsmD84E3YFj/RdByrWrG TL4XskALpfmw1+LxQmMO8n4sIsN3QmjkAWhFhMEquKv6NNN+sRo6vF+ytEasuYn/ 7gY/iT7LYqUmKWckBsPYzT9elyOXi6miI0tFdeyfXRSxOslKewIDAQABMA0GCSqG SIb3DQEBBAUAA4GBABotNizqPoGWIG5BMsl8lxseqiw/8AwvoiQNpYTrC8W+Umsg oZEaMuVkf/NDJEa3TXdYcAzkFwGN9Cn/WCgHEkLxIZ66aHV0bfcE7YJjHRDrrLiY chPndOGGrD3iTuWaGnauUcsjJ+RsxqHMBu6NRQYgkoYNsOr0UA1ek7O1vjMy -----END CERTIFICATE----- -----BEGIN RSA PRIVATE KEY----- MIICXAIBAAKBgQCrmNNyXLHAETcDH8Uxhmbo8IhFFMx1C4i7oTHTKsmD84E3YFj/ RdByrWrGTL4XskALpfmw1+LxQmMO8n4sIsN3QmjkAWhFhMEquKv6NNN+sRo6vF+y tEasuYn/7gY/iT7LYqUmKWckBsPYzT9elyOXi6miI0tFdeyfXRSxOslKewIDAQAB AoGAHd9YCBOs+gPFMO0J9iowpiKhhm0tfr7ISemw89MCC8+LUimatK3hsOURrn3T peppDd4SDsA2iMuG1SZP4r0Wi9ZncZ+uj6KfVHg6rJZRDW2cPsGNyBw2HO8pFxnh NsfxioutzCqJ9A0KwqSNQsBpOAlRWzP13+/W5wYAGK+yrLECQQDYgOhVR+1KOhty CI0NVITNFL5IOZ254Eu46qbEGwPNJvkzdp+Wx5gsfCip9aiZgw3LMEeGXu9P1C4N AqDM4uozAkEAyua0F0nCRLzjLAAw4odC+vA6jnq6K4M7QT6cQVwmrxgOj6jGEOGu eaoWbXi2bKcxOGBNDZW0PVKmpq4hZblmmQJBALwFP0AIxg+HZRxkRrMD6oz77cBl oQ+ytbAywH9ggq2gohzKcRAN6J8BeIMZn8EpqkoCdKtCOQyX1SJhXOpySjcCQDds mZka7tQz/KISU0gtxqAhav1sjNpB+Lez0J8R+wctPR0E70XBQBW/3mx84uf/K7TI qYOidx+hKiCxxDGzWVECQHNVutQ1ABjmv6EDJTo28QQsm5hNbfS+tVY3bSihNjLM Y+O7ib90LsqfQ8r0GUphQVi4EA4QMJqaF7ZxKms79qA= -----END RSA PRIVATE KEY----- """ def setUp(self): """ Override the inherited setup to create an on-disk Store so that certificates can be created within it. """ self.dbdir = self.mktemp() return super(SSLPortTests, self).setUp() def checkPort(self, port, alternatePort=None): if alternatePort is None: alternatePort = self.lowPortNumber self.assertEqual(port.portNumber, alternatePort) self.assertEqual(port.factory, self.realFactory) self.failUnless(isinstance(port.contextFactory, CertificateOptions)) def port(self, certificatePath=None, **kw): if certificatePath is None: certificatePath = self.store.newFilePath('certificate.pem') assert not certificatePath.exists() certificatePath.setContent(self.dummyCertificateData) return self.portType(certificatePath=certificatePath, **kw) def listen(self, port, factory, contextFactory): self.ports.append(DummyPort(port, factory, contextFactory)) return self.ports[-1] def test_certificatePathAttribute(self): """ Test that L{SSLPort} remembers the certificate filename it is given. """ certificatePath = self.store.newFilePath('foo', 'bar') port = self.port(store=self.store, certificatePath=certificatePath) self.assertEqual(port.certificatePath, certificatePath)