# Copyright (c) 2008 Divmod, Inc.
# See LICENSE for details.

"""
Tests for L{xmantissa.port}.
"""

import sys
import os
from StringIO import StringIO

from zope.interface import implements
from zope.interface.verify import verifyObject

from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
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.iaxiom import IAxiomaticCommand
from axiom.store import Store
from axiom.item import Item
from axiom.attributes import inmemory, integer
from axiom.dependency import installOn
from axiom.scripts.axiomatic import Options as AxiomaticOptions
from axiom.test.util import CommandStub

from xmantissa.ixmantissa import IProtocolFactoryFactory
from xmantissa.port import TCPPort, SSLPort, StringEndpointPort
from xmantissa.port import PortConfiguration


CERTIFICATE_DATA = """
-----BEGIN CERTIFICATE-----
<KEY>END CERTIFICATE-----
"""

PRIVATEKEY_DATA = """
-----<KEY>
"""



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, interface=''):
        self.portNumber = portNumber
        self.factory = factory
        self.contextFactory = contextFactory
        self.interface = interface


    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.
    """
    implements(IProtocolFactoryFactory)

    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
    someInterface = u'127.0.0.1'

    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.filesdir = self.mktemp()
        self.store = Store(filesdir=self.filesdir)
        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_interfaceAttribute(self):
        """
        Test that C{self.portType} remembers the interface it is told to listen
        on.
        """
        port = self.port(store=self.store, interface=self.someInterface)
        self.assertEqual(port.interface, self.someInterface)


    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 privilege.
        """
        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 high-numbered port with the reactor
        when it is started without privilege.
        """
        ports = self._startService(self.highPortNumber)
        self.assertEqual(len(ports), 1)
        self.checkPort(ports[0], self.highPortNumber)


    def test_startServiceNoInterface(self):
        """
        Test that C{self.portType} binds to all interfaces if no interface is
        explicitly specified.
        """
        port = self.port(store=self.store, portNumber=self.highPortNumber, factory=self.factory)
        port._listen = self.listen
        port.startService()
        self.assertEqual(self.ports[0].interface, '')


    def test_startServiceInterface(self):
        """
        Test that C{self.portType} binds to only the specified interface when
        instructed to.
        """
        port = self.port(store=self.store, portNumber=self.highPortNumber, factory=self.factory, interface=self.someInterface)
        port._listen = self.listen
        port.startService()
        self.assertEqual(self.ports[0].interface, self.someInterface)


    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, interface=''):
        self.ports.append(DummyPort(port, factory, interface=interface))
        return self.ports[-1]



class SSLPortTests(PortTestsMixin, TestCase):
    """
    Tests for L{xmantissa.port.SSLPort}.
    """
    portType = SSLPort

    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(CERTIFICATE_DATA + PRIVATEKEY_DATA)
        return self.portType(certificatePath=certificatePath, **kw)


    def listen(self, port, factory, contextFactory, interface=''):
        self.ports.append(DummyPort(port, factory, contextFactory, interface=interface))
        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)



class _FakeService(object):
    """
    Fake L{twisted.application.service.IService} implementation for testing
    L{xmantissa.port.StringEndpointPort}'s wrapping behaviour.
    """
    def __init__(self, description, factory):
        self.description = description
        self.factory = factory
        self.privilegedStarted = False
        self.started = False
        self.stopped = False


    def privilegedStartService(self):
        self.privilegedStarted = True


    def startService(self):
        self.started = True


    def stopService(self):
        self.stopped = True



class StringEndpointPortTests(TestCase):
    """
    Tests for L{xmantissa.port.StringEndpointPort}.
    """
    def _fakeService(self, description, factory):
        """
        A fake for L{twisted.application.strports.service} that just constructs
        our fake service object.
        """
        self._service = _FakeService(description, factory)
        return self._service


    def port(self, **kw):
        store = Store()
        factory = DummyFactory(store=store)
        factory.realFactory = ServerFactory()
        port = StringEndpointPort(store=store, factory=factory, **kw)
        port._endpointService = self._fakeService
        return port


    def test_startService(self):
        """
        The underlying endpoint service is started when
        L{xmantissa.port.StringEndpointPort} is started.
        """
        port = self.port(description=u'foo')
        port.privilegedStartService()
        self.assertTrue(self._service.privilegedStarted)
        port.startService()
        self.assertTrue(self._service.started)


    def test_description(self):
        """
        The underlying endpoint service is created with the description
        specified by the L{xmantissa.port.StringEndpointPort}.
        """
        port = self.port(description=u'foo')
        port.startService()
        self.assertEqual(u'foo', self._service.description)


    def test_factory(self):
        """
        The underlying endpoint service is created with the factory specified
        by invoking C{getFactory()} on the C{factory} attribute.
        """
        port = self.port(description=u'foo')
        port.startService()
        self.assertIdentical(self._service.factory, port.factory.realFactory)


    def test_stopService(self):
        """
        The underlying endpoint service is stopped when
        L{xmantissa.port.StringEndpointPort} is stopped.
        """
        port = self.port(description=u'foo')
        port.startService()
        port.stopService()
        self.assertTrue(self._service.stopped)



class PortConfigurationCommandTests(TestCase):
    """
    Tests for the I{axiomatic port} command.
    """
    def setUp(self):
        """
        Override C{sys.stdout} to capture anything written by the port
        subcommand.
        """
        self.oldColumns = os.environ.get('COLUMNS')
        os.environ['COLUMNS'] = '80'
        self.stdout = sys.stdout
        sys.stdout = StringIO()


    def tearDown(self):
        """
        Restore the original value of C{sys.stdout}.
        """
        sys.stdout = self.stdout
        if self.oldColumns is not None:
            os.environ['COLUMNS'] = self.oldColumns

    def _makeConfig(self, store):
        """
        Create a L{PortConfiguration} instance with a properly set C{parent}
        attribute.
        """
        config = PortConfiguration()
        config.parent = CommandStub(store, "port")
        return config


    def assertSuccessStatus(self, options, arguments):
        """
        Parse the given arguments with the given options object and assert that
        L{SystemExit} is raised with an exit code of C{0}.
        """
        self.assertFailStatus(0, options, arguments)


    def assertFailStatus(self, code, options, arguments):
        """
        Parse the given arguments with the given options object and assert that
        L{SystemExit} is raised with the specified exit code.
        """
        exc = self.assertRaises(SystemExit, options.parseOptions, arguments)
        self.assertEqual(exc.args, (code,))


    def assertSpacelessEqual(self, first, second):
        """
        Assert the equality of two strings without respect to their whitespace
        or order.
        """
        self.assertEqual(set(first.split()), set(second.split()))


    def test_providesCommandInterface(self):
        """
        L{PortConfiguration} provides L{IAxiomaticCommand}.
        """
        verifyObject(IAxiomaticCommand, PortConfiguration)


    def test_axiomaticSubcommand(self):
        """
        L{PortConfiguration} is available as a subcommand of I{axiomatic}.
        """
        subCommands = AxiomaticOptions().subCommands
        [options] = [cmd[2] for cmd in subCommands if cmd[0] == 'port']
        self.assertIdentical(options, PortConfiguration)


    _portHelpText = (
        # This is a bit unfortunate.  It depends on what Options in Twisted
        # decides to spit out.  Note particularly the seemingly random amount
        # of trailing whitespace included on some lines.  The intent of tests
        # using this isn't really to ensure byte-identical results, but simply
        # to verify that help text is going to be shown to a user.  -exarkun
        "Usage: axiomatic [options] port [options]\n"
        "Options:\n"
        "      --version  Display Twisted version and exit.\n"
        "      --help     Display this help and exit.\n"
        "\n"
        "This command allows for the inspection and modification of the "
        "configuration of\n"
        "network services in an Axiom store.\n"
        "Commands:\n"
        "    list        Show existing ports and factories.\n"
        "    delete      Delete existing ports.\n"
        "    create      Create new ports.\n"
        "\n")

    def test_implicitPortHelp(self):
        """
        When I{axiomatic port} is invoked with no arguments, usage information
        is written to standard out and the process exits successfully.
        """
        self.assertSuccessStatus(self._makeConfig(None), [])
        self.assertSpacelessEqual(self._portHelpText, sys.stdout.getvalue())


    def test_explicitPortHelp(self):
        """
        When I{axiomatic port} is invoked with I{--help}, usage information is
        written to standard out.
        """
        self.assertSuccessStatus(self._makeConfig(None), ["--help"])
        self.assertSpacelessEqual(self._portHelpText, sys.stdout.getvalue())


    _listHelpText = (
        "Usage: axiomatic [options] port [options] list [options]\n"
        "Options:\n"
        "      --version  Display Twisted version and exit.\n"
        "      --help     Display this help and exit.\n"
        "\n"
        "Show the port/factory bindings in an Axiom store.\n"
        "\n")

    def test_explicitListHelp(self):
        """
        When I{axiomatic port list} is invoked with I{--help}, usage
        information for the C{list} subcommand is written to standard out.
        """
        self.assertSuccessStatus(self._makeConfig(None), ["list", "--help"])
        self.assertSpacelessEqual(self._listHelpText, sys.stdout.getvalue())


    def test_listEmpty(self):
        """
        When I{axiomatic port list} is invoked, the ports which are currently
        configured in the system are displayed.
        """
        store = Store()
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertIn("There are no ports configured.", sys.stdout.getvalue())


    def test_listTCPPort(self):
        """
        When I{axiomatic port list} is invoked for a L{Store} which has a
        L{TCPPort} in it, the details of that port, including its factory, are
        written to stdout.
        """
        store = Store()
        factory = DummyFactory(store=store)
        port = TCPPort(
            store=store, factory=factory, portNumber=1234, interface=u"foo")
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) TCP, interface %s, port %d\n" % (
                port.storeID, port.interface, port.portNumber),
            sys.stdout.getvalue())


    def test_listSSLPort(self):
        """
        When I{axiomatic port list} is invoked for a L{Store} which has an
        L{SSLPort} in it, the details of that port, including its factory, are
        written to stdout.
        """
        store = Store(filesdir=self.mktemp())
        factory = DummyFactory(store=store)
        port = SSLPort(
            store=store, factory=factory, portNumber=1234, interface=u"foo",
            certificatePath=store.filesdir.child("bar"))
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) SSL, interface %s, port %d, certificate %s\n" % (
                port.storeID, port.interface, port.portNumber,
                port.certificatePath.path),
            sys.stdout.getvalue())


    def test_listStringEndpointPort(self):
        """
        When I{axiomatic port list} is invoked for a L{Store} which has an
        L{StringEndpointPort} in it, the endpoint description and factory are
        written to stdout.
        """
        store = Store()
        factory = DummyFactory(store=store)
        port = StringEndpointPort(
            store=store, factory=factory, description=u'tcp:1234')
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "{:d}) {!r} listening on:\n".format(factory.storeID, factory) +
            "  {:d}) Endpoint {!r}\n".format(port.storeID, port.description),
            sys.stdout.getvalue())


    def test_listAnyInterface(self):
        """
        I{axiomatic port list} displays a special string for a port bound to
        C{INADDR_ANY}.
        """
        store = Store()
        factory = DummyFactory(store=store)
        port = TCPPort(
            store=store, factory=factory, portNumber=1234, interface=u"")
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) TCP, any interface, port %d\n" % (port.storeID, port.portNumber),
            sys.stdout.getvalue())


    def test_listSSLPortWithoutAttributes(self):
        """
        If there is an L{SSLPort} with no certificate or no port number (a
        rather invalid configuration), I{axiomatic port list} should show this
        in its output without producing an error.
        """
        store = Store()
        factory = DummyFactory(store=store)
        port = SSLPort(store=store, factory=factory)
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) SSL, any interface, NO PORT, NO CERTIFICATE\n" % (
                port.storeID,),
            sys.stdout.getvalue())


    def test_listTwoPorts(self):
        """
        I{axiomatic port list} displays two different ports bound to the same
        factory together beneath that factory.
        """
        store = Store()
        factory = DummyFactory(store=store)
        portOne = TCPPort(
            store=store, factory=factory, portNumber=1234, interface=u"foo")
        portTwo = TCPPort(
            store=store, factory=factory, portNumber=2345, interface=u"bar")
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) TCP, interface %s, port %d\n" % (
                portOne.storeID, portOne.interface, portOne.portNumber) +
            "  %d) TCP, interface %s, port %d\n" % (
                portTwo.storeID, portTwo.interface, portTwo.portNumber),
            sys.stdout.getvalue())


    def test_listTwoFactories(self):
        """
        I{axiomatic port list} displays two different factories separately from
        each other.
        """
        store = Store()
        factoryOne = DummyFactory(store=store)
        factoryTwo = DummyFactory(store=store)
        portOne = TCPPort(
            store=store, factory=factoryOne, portNumber=10, interface=u"foo")
        portTwo = TCPPort(
            store=store, factory=factoryTwo, portNumber=20, interface=u"bar")
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factoryOne.storeID, factoryOne) +
            "  %d) TCP, interface %s, port %d\n" % (
                portOne.storeID, portOne.interface, portOne.portNumber) +
            "%d) %r listening on:\n" % (factoryTwo.storeID, factoryTwo) +
            "  %d) TCP, interface %s, port %d\n" % (
                portTwo.storeID, portTwo.interface, portTwo.portNumber),
            sys.stdout.getvalue())


    def test_listUnlisteningFactory(self):
        """
        I{axiomatic port list} displays factories even if they aren't associate
        with any port.
        """
        store = Store()
        factory = DummyFactory(store=store)
        store.powerUp(factory, IProtocolFactoryFactory)
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r is not listening.\n" % (factory.storeID, factory),
            sys.stdout.getvalue())


    _deleteHelpText = (
        "Usage: axiomatic [options] port [options] delete [options]\n"
        "Options:\n"
        "      --port-identifier=  Identify a port for deletion.\n"
        "      --version           Display Twisted version and exit.\n"
        "      --help              Display this help and exit.\n"
        "\n"
        "Delete an existing port binding from a factory. If a server is "
        "currently running using the database from which the port is deleted, "
        "the factory will *not* stop listening on that port until the server "
        "is restarted.")

    def test_explicitDeleteHelp(self):
        """
        If I{axiomatic port delete} is invoked with I{--help}, usage
        information for the C{delete} subcommand is written to standard out.
        """
        store = Store()
        self.assertSuccessStatus(self._makeConfig(store), ["delete", "--help"])
        self.assertSpacelessEqual(self._deleteHelpText, sys.stdout.getvalue())


    def test_implicitDeleteHelp(self):
        """
        If I{axiomatic port delete} is invoked with no arguments, usage
        information for the C{delete} subcommand is written to standard out.
        """
        store = Store()
        self.assertSuccessStatus(self._makeConfig(store), ["delete"])
        self.assertSpacelessEqual(self._deleteHelpText, sys.stdout.getvalue())


    def test_deletePorts(self):
        """
        I{axiomatic port delete} deletes each ports with a C{storeID} which is
        specified.
        """
        store = Store(filesdir=self.mktemp())
        factory = DummyFactory(store=store)
        deleteTCP = TCPPort(
            store=store, factory=factory, portNumber=10, interface=u"foo")
        keepTCP = TCPPort(
            store=store, factory=factory, portNumber=10, interface=u"bar")
        deleteSSL = SSLPort(
            store=store, factory=factory, portNumber=10, interface=u"baz",
            certificatePath=store.filesdir.child("baz"))
        keepSSL = SSLPort(
            store=store, factory=factory, portNumber=10, interface=u"quux",
            certificatePath=store.filesdir.child("quux"))
        deleteEndpoint = StringEndpointPort(
            store=store, factory=factory, description=u'tcp:1234')
        keepEndpoint = StringEndpointPort(
            store=store, factory=factory, description=u'tcp:1235')
        self.assertSuccessStatus(
            self._makeConfig(store),
            ["delete",
             "--port-identifier", str(deleteTCP.storeID),
             "--port-identifier", str(deleteSSL.storeID),
             "--port-identifier", str(deleteEndpoint.storeID)])
        self.assertEqual("Deleted.\n", sys.stdout.getvalue())
        self.assertEqual(list(store.query(TCPPort)), [keepTCP])
        self.assertEqual(list(store.query(SSLPort)), [keepSSL])
        self.assertEqual(list(store.query(StringEndpointPort)), [keepEndpoint])


    def test_cannotDeleteOtherStuff(self):
        """
        I{axiomatic port delete} will not delete something which is neither a
        L{TCPPort} nor an L{SSLPort} and does not delete anything if an invalid
        port identifier is present in the command.
        """
        store = Store()
        factory = DummyFactory(store=store)
        tcp = TCPPort(
            store=store, factory=factory, interface=u"foo", portNumber=1234)
        self.assertFailStatus(
            1,
            self._makeConfig(store),
            ["delete",
             "--port-identifier", str(tcp.storeID),
             "--port-identifier", str(factory.storeID)])
        self.assertEqual(
            "%d does not identify a port.\n" % (factory.storeID,),
            sys.stdout.getvalue())
        self.assertEqual(list(store.query(DummyFactory)), [factory])
        self.assertEqual(list(store.query(TCPPort)), [tcp])


    def test_cannotDeleteNonExistent(self):
        """
        I{axiomatic port delete} writes a short error to standard output when a
        port-identifier is specified for which there is no corresponding store
        ID.
        """
        store = Store()
        self.assertFailStatus(
            1, self._makeConfig(store),
            ["delete", "--port-identifier", "12345"])
        self.assertEqual(
            "12345 does not identify an item.\n",
            sys.stdout.getvalue())


    _createHelpText = (
        "Usage: axiomatic [options] port [options] create [options]\n"
        "Options:\n"
        "      --strport=             A Twisted strports description of a "
        "port to add.\n"
        "      --factory-identifier=  Identifier for a protocol factory to "
        "associate with\n"
        "                             the new port.\n"
        "      --version              Display Twisted version and exit.\n"
        "      --help                 Display this help and exit.\n"
        "\n"
        "Create a new port binding for an existing factory.  If a server is "
        "currently\n"
        "running using the database in which the port is created, the "
        "factory will *not*\n"
        "be started on that port until the server is restarted.\n"
        "\n")

    def test_createImplicitHelp(self):
        """
        If I{axiomatic port create} is invoked with no arguments, usage
        information for the C{create} subcommand is written to standard out.
        """
        self.assertSuccessStatus(self._makeConfig(None), ["create"])
        self.assertSpacelessEqual(self._createHelpText, sys.stdout.getvalue())


    def test_createExplicitHelp(self):
        """
        If I{axiomatic port create} is invoked with C{--help} as an argument,
        usage information for the C{add} subcommand is written to standard out.
        """
        self.assertSuccessStatus(self._makeConfig(None), ["create", "--help"])
        self.assertSpacelessEqual(self._createHelpText, sys.stdout.getvalue())


    def test_createInvalidPortDescription(self):
        """
        If an invalid string is given for the C{strport} option of I{axiomatic
        port create}, a short error is written to standard output.
        """
        store = Store()
        factory = DummyFactory(store=store)
        self.assertFailStatus(
            1, self._makeConfig(store),
            ["create", "--strport", "xyz",
             "--factory-identifier", str(factory.storeID)])
        self.assertEqual(
            "'xyz' is not a valid port description.\n", sys.stdout.getvalue())


    def test_createNonExistentFactoryIdentifier(self):
        """
        If a storeID which is not associated with any item is given for the
        C{factory-identifier} option of I{axiomatic port create}, a short error
        is written to standard output.
        """
        store = Store()
        self.assertFailStatus(
            1, self._makeConfig(store),
            ["create", "--strport", "tcp:8080",
             "--factory-identifier", "123"])
        self.assertEqual(
            "123 does not identify an item.\n", sys.stdout.getvalue())


    def test_createNonFactoryIdentifier(self):
        """
        If a storeID which is associated with an item which does not provide
        L{IProtocolFactoryFactory} is given for the C{factory-identifier}
        option of I{axiomatic port create}, a short error is written to
        standard output.
        """
        store = Store()
        storeID = TCPPort(store=store).storeID
        self.assertFailStatus(
            1, self._makeConfig(store),
            ["create", "--strport", "tcp:8080",
             "--factory-identifier", str(storeID)])
        self.assertEqual(
            "%d does not identify a factory.\n" % (storeID,),
            sys.stdout.getvalue())


    def test_createPort(self):
        """
        I{axiomatic port create} creates a new
        L{xmantissa.port.StringEndpointPort} with the specified description,
        referring to the specified factory. The port is also powered up on the
        store for L{IService}.
        """
        store = Store()
        factory = DummyFactory(store=store)
        self.assertSuccessStatus(
            self._makeConfig(store),
            ["create", "--strport", "tcp:8080",
             "--factory-identifier", str(factory.storeID)])
        self.assertEqual("Created.\n", sys.stdout.getvalue())
        [port] = list(store.query(StringEndpointPort))
        self.assertEqual(u'tcp:8080', port.description)
        self.assertEqual(list(store.interfacesFor(port)), [IService])

# Copyright (c) 2008 Divmod, Inc.
# See LICENSE for details.

"""
Tests for L{xmantissa.port}.
"""

import sys
import os
from StringIO import StringIO

from zope.interface import implements
from zope.interface.verify import verifyObject

from twisted.trial.unittest import TestCase
from twisted.python.filepath import FilePath
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.iaxiom import IAxiomaticCommand
from axiom.store import Store
from axiom.item import Item
from axiom.attributes import inmemory, integer
from axiom.dependency import installOn
from axiom.scripts.axiomatic import Options as AxiomaticOptions
from axiom.test.util import CommandStub

from xmantissa.ixmantissa import IProtocolFactoryFactory
from xmantissa.port import TCPPort, SSLPort, StringEndpointPort
from xmantissa.port import PortConfiguration


CERTIFICATE_DATA = """
-----BEGIN CERTIFICATE-----
<KEY>END CERTIFICATE-----
"""

PRIVATEKEY_DATA = """
-----<KEY>
"""



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, interface=''):
        self.portNumber = portNumber
        self.factory = factory
        self.contextFactory = contextFactory
        self.interface = interface


    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.
    """
    implements(IProtocolFactoryFactory)

    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
    someInterface = u'127.0.0.1'

    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.filesdir = self.mktemp()
        self.store = Store(filesdir=self.filesdir)
        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_interfaceAttribute(self):
        """
        Test that C{self.portType} remembers the interface it is told to listen
        on.
        """
        port = self.port(store=self.store, interface=self.someInterface)
        self.assertEqual(port.interface, self.someInterface)


    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 privilege.
        """
        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 high-numbered port with the reactor
        when it is started without privilege.
        """
        ports = self._startService(self.highPortNumber)
        self.assertEqual(len(ports), 1)
        self.checkPort(ports[0], self.highPortNumber)


    def test_startServiceNoInterface(self):
        """
        Test that C{self.portType} binds to all interfaces if no interface is
        explicitly specified.
        """
        port = self.port(store=self.store, portNumber=self.highPortNumber, factory=self.factory)
        port._listen = self.listen
        port.startService()
        self.assertEqual(self.ports[0].interface, '')


    def test_startServiceInterface(self):
        """
        Test that C{self.portType} binds to only the specified interface when
        instructed to.
        """
        port = self.port(store=self.store, portNumber=self.highPortNumber, factory=self.factory, interface=self.someInterface)
        port._listen = self.listen
        port.startService()
        self.assertEqual(self.ports[0].interface, self.someInterface)


    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, interface=''):
        self.ports.append(DummyPort(port, factory, interface=interface))
        return self.ports[-1]



class SSLPortTests(PortTestsMixin, TestCase):
    """
    Tests for L{xmantissa.port.SSLPort}.
    """
    portType = SSLPort

    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(CERTIFICATE_DATA + PRIVATEKEY_DATA)
        return self.portType(certificatePath=certificatePath, **kw)


    def listen(self, port, factory, contextFactory, interface=''):
        self.ports.append(DummyPort(port, factory, contextFactory, interface=interface))
        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)



class _FakeService(object):
    """
    Fake L{twisted.application.service.IService} implementation for testing
    L{xmantissa.port.StringEndpointPort}'s wrapping behaviour.
    """
    def __init__(self, description, factory):
        self.description = description
        self.factory = factory
        self.privilegedStarted = False
        self.started = False
        self.stopped = False


    def privilegedStartService(self):
        self.privilegedStarted = True


    def startService(self):
        self.started = True


    def stopService(self):
        self.stopped = True



class StringEndpointPortTests(TestCase):
    """
    Tests for L{xmantissa.port.StringEndpointPort}.
    """
    def _fakeService(self, description, factory):
        """
        A fake for L{twisted.application.strports.service} that just constructs
        our fake service object.
        """
        self._service = _FakeService(description, factory)
        return self._service


    def port(self, **kw):
        store = Store()
        factory = DummyFactory(store=store)
        factory.realFactory = ServerFactory()
        port = StringEndpointPort(store=store, factory=factory, **kw)
        port._endpointService = self._fakeService
        return port


    def test_startService(self):
        """
        The underlying endpoint service is started when
        L{xmantissa.port.StringEndpointPort} is started.
        """
        port = self.port(description=u'foo')
        port.privilegedStartService()
        self.assertTrue(self._service.privilegedStarted)
        port.startService()
        self.assertTrue(self._service.started)


    def test_description(self):
        """
        The underlying endpoint service is created with the description
        specified by the L{xmantissa.port.StringEndpointPort}.
        """
        port = self.port(description=u'foo')
        port.startService()
        self.assertEqual(u'foo', self._service.description)


    def test_factory(self):
        """
        The underlying endpoint service is created with the factory specified
        by invoking C{getFactory()} on the C{factory} attribute.
        """
        port = self.port(description=u'foo')
        port.startService()
        self.assertIdentical(self._service.factory, port.factory.realFactory)


    def test_stopService(self):
        """
        The underlying endpoint service is stopped when
        L{xmantissa.port.StringEndpointPort} is stopped.
        """
        port = self.port(description=u'foo')
        port.startService()
        port.stopService()
        self.assertTrue(self._service.stopped)



class PortConfigurationCommandTests(TestCase):
    """
    Tests for the I{axiomatic port} command.
    """
    def setUp(self):
        """
        Override C{sys.stdout} to capture anything written by the port
        subcommand.
        """
        self.oldColumns = os.environ.get('COLUMNS')
        os.environ['COLUMNS'] = '80'
        self.stdout = sys.stdout
        sys.stdout = StringIO()


    def tearDown(self):
        """
        Restore the original value of C{sys.stdout}.
        """
        sys.stdout = self.stdout
        if self.oldColumns is not None:
            os.environ['COLUMNS'] = self.oldColumns

    def _makeConfig(self, store):
        """
        Create a L{PortConfiguration} instance with a properly set C{parent}
        attribute.
        """
        config = PortConfiguration()
        config.parent = CommandStub(store, "port")
        return config


    def assertSuccessStatus(self, options, arguments):
        """
        Parse the given arguments with the given options object and assert that
        L{SystemExit} is raised with an exit code of C{0}.
        """
        self.assertFailStatus(0, options, arguments)


    def assertFailStatus(self, code, options, arguments):
        """
        Parse the given arguments with the given options object and assert that
        L{SystemExit} is raised with the specified exit code.
        """
        exc = self.assertRaises(SystemExit, options.parseOptions, arguments)
        self.assertEqual(exc.args, (code,))


    def assertSpacelessEqual(self, first, second):
        """
        Assert the equality of two strings without respect to their whitespace
        or order.
        """
        self.assertEqual(set(first.split()), set(second.split()))


    def test_providesCommandInterface(self):
        """
        L{PortConfiguration} provides L{IAxiomaticCommand}.
        """
        verifyObject(IAxiomaticCommand, PortConfiguration)


    def test_axiomaticSubcommand(self):
        """
        L{PortConfiguration} is available as a subcommand of I{axiomatic}.
        """
        subCommands = AxiomaticOptions().subCommands
        [options] = [cmd[2] for cmd in subCommands if cmd[0] == 'port']
        self.assertIdentical(options, PortConfiguration)


    _portHelpText = (
        # This is a bit unfortunate.  It depends on what Options in Twisted
        # decides to spit out.  Note particularly the seemingly random amount
        # of trailing whitespace included on some lines.  The intent of tests
        # using this isn't really to ensure byte-identical results, but simply
        # to verify that help text is going to be shown to a user.  -exarkun
        "Usage: axiomatic [options] port [options]\n"
        "Options:\n"
        "      --version  Display Twisted version and exit.\n"
        "      --help     Display this help and exit.\n"
        "\n"
        "This command allows for the inspection and modification of the "
        "configuration of\n"
        "network services in an Axiom store.\n"
        "Commands:\n"
        "    list        Show existing ports and factories.\n"
        "    delete      Delete existing ports.\n"
        "    create      Create new ports.\n"
        "\n")

    def test_implicitPortHelp(self):
        """
        When I{axiomatic port} is invoked with no arguments, usage information
        is written to standard out and the process exits successfully.
        """
        self.assertSuccessStatus(self._makeConfig(None), [])
        self.assertSpacelessEqual(self._portHelpText, sys.stdout.getvalue())


    def test_explicitPortHelp(self):
        """
        When I{axiomatic port} is invoked with I{--help}, usage information is
        written to standard out.
        """
        self.assertSuccessStatus(self._makeConfig(None), ["--help"])
        self.assertSpacelessEqual(self._portHelpText, sys.stdout.getvalue())


    _listHelpText = (
        "Usage: axiomatic [options] port [options] list [options]\n"
        "Options:\n"
        "      --version  Display Twisted version and exit.\n"
        "      --help     Display this help and exit.\n"
        "\n"
        "Show the port/factory bindings in an Axiom store.\n"
        "\n")

    def test_explicitListHelp(self):
        """
        When I{axiomatic port list} is invoked with I{--help}, usage
        information for the C{list} subcommand is written to standard out.
        """
        self.assertSuccessStatus(self._makeConfig(None), ["list", "--help"])
        self.assertSpacelessEqual(self._listHelpText, sys.stdout.getvalue())


    def test_listEmpty(self):
        """
        When I{axiomatic port list} is invoked, the ports which are currently
        configured in the system are displayed.
        """
        store = Store()
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertIn("There are no ports configured.", sys.stdout.getvalue())


    def test_listTCPPort(self):
        """
        When I{axiomatic port list} is invoked for a L{Store} which has a
        L{TCPPort} in it, the details of that port, including its factory, are
        written to stdout.
        """
        store = Store()
        factory = DummyFactory(store=store)
        port = TCPPort(
            store=store, factory=factory, portNumber=1234, interface=u"foo")
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) TCP, interface %s, port %d\n" % (
                port.storeID, port.interface, port.portNumber),
            sys.stdout.getvalue())


    def test_listSSLPort(self):
        """
        When I{axiomatic port list} is invoked for a L{Store} which has an
        L{SSLPort} in it, the details of that port, including its factory, are
        written to stdout.
        """
        store = Store(filesdir=self.mktemp())
        factory = DummyFactory(store=store)
        port = SSLPort(
            store=store, factory=factory, portNumber=1234, interface=u"foo",
            certificatePath=store.filesdir.child("bar"))
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) SSL, interface %s, port %d, certificate %s\n" % (
                port.storeID, port.interface, port.portNumber,
                port.certificatePath.path),
            sys.stdout.getvalue())


    def test_listStringEndpointPort(self):
        """
        When I{axiomatic port list} is invoked for a L{Store} which has an
        L{StringEndpointPort} in it, the endpoint description and factory are
        written to stdout.
        """
        store = Store()
        factory = DummyFactory(store=store)
        port = StringEndpointPort(
            store=store, factory=factory, description=u'tcp:1234')
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "{:d}) {!r} listening on:\n".format(factory.storeID, factory) +
            "  {:d}) Endpoint {!r}\n".format(port.storeID, port.description),
            sys.stdout.getvalue())


    def test_listAnyInterface(self):
        """
        I{axiomatic port list} displays a special string for a port bound to
        C{INADDR_ANY}.
        """
        store = Store()
        factory = DummyFactory(store=store)
        port = TCPPort(
            store=store, factory=factory, portNumber=1234, interface=u"")
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) TCP, any interface, port %d\n" % (port.storeID, port.portNumber),
            sys.stdout.getvalue())


    def test_listSSLPortWithoutAttributes(self):
        """
        If there is an L{SSLPort} with no certificate or no port number (a
        rather invalid configuration), I{axiomatic port list} should show this
        in its output without producing an error.
        """
        store = Store()
        factory = DummyFactory(store=store)
        port = SSLPort(store=store, factory=factory)
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) SSL, any interface, NO PORT, NO CERTIFICATE\n" % (
                port.storeID,),
            sys.stdout.getvalue())


    def test_listTwoPorts(self):
        """
        I{axiomatic port list} displays two different ports bound to the same
        factory together beneath that factory.
        """
        store = Store()
        factory = DummyFactory(store=store)
        portOne = TCPPort(
            store=store, factory=factory, portNumber=1234, interface=u"foo")
        portTwo = TCPPort(
            store=store, factory=factory, portNumber=2345, interface=u"bar")
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factory.storeID, factory) +
            "  %d) TCP, interface %s, port %d\n" % (
                portOne.storeID, portOne.interface, portOne.portNumber) +
            "  %d) TCP, interface %s, port %d\n" % (
                portTwo.storeID, portTwo.interface, portTwo.portNumber),
            sys.stdout.getvalue())


    def test_listTwoFactories(self):
        """
        I{axiomatic port list} displays two different factories separately from
        each other.
        """
        store = Store()
        factoryOne = DummyFactory(store=store)
        factoryTwo = DummyFactory(store=store)
        portOne = TCPPort(
            store=store, factory=factoryOne, portNumber=10, interface=u"foo")
        portTwo = TCPPort(
            store=store, factory=factoryTwo, portNumber=20, interface=u"bar")
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r listening on:\n" % (factoryOne.storeID, factoryOne) +
            "  %d) TCP, interface %s, port %d\n" % (
                portOne.storeID, portOne.interface, portOne.portNumber) +
            "%d) %r listening on:\n" % (factoryTwo.storeID, factoryTwo) +
            "  %d) TCP, interface %s, port %d\n" % (
                portTwo.storeID, portTwo.interface, portTwo.portNumber),
            sys.stdout.getvalue())


    def test_listUnlisteningFactory(self):
        """
        I{axiomatic port list} displays factories even if they aren't associate
        with any port.
        """
        store = Store()
        factory = DummyFactory(store=store)
        store.powerUp(factory, IProtocolFactoryFactory)
        self.assertSuccessStatus(self._makeConfig(store), ["list"])
        self.assertEqual(
            "%d) %r is not listening.\n" % (factory.storeID, factory),
            sys.stdout.getvalue())


    _deleteHelpText = (
        "Usage: axiomatic [options] port [options] delete [options]\n"
        "Options:\n"
        "      --port-identifier=  Identify a port for deletion.\n"
        "      --version           Display Twisted version and exit.\n"
        "      --help              Display this help and exit.\n"
        "\n"
        "Delete an existing port binding from a factory. If a server is "
        "currently running using the database from which the port is deleted, "
        "the factory will *not* stop listening on that port until the server "
        "is restarted.")

    def test_explicitDeleteHelp(self):
        """
        If I{axiomatic port delete} is invoked with I{--help}, usage
        information for the C{delete} subcommand is written to standard out.
        """
        store = Store()
        self.assertSuccessStatus(self._makeConfig(store), ["delete", "--help"])
        self.assertSpacelessEqual(self._deleteHelpText, sys.stdout.getvalue())


    def test_implicitDeleteHelp(self):
        """
        If I{axiomatic port delete} is invoked with no arguments, usage
        information for the C{delete} subcommand is written to standard out.
        """
        store = Store()
        self.assertSuccessStatus(self._makeConfig(store), ["delete"])
        self.assertSpacelessEqual(self._deleteHelpText, sys.stdout.getvalue())


    def test_deletePorts(self):
        """
        I{axiomatic port delete} deletes each ports with a C{storeID} which is
        specified.
        """
        store = Store(filesdir=self.mktemp())
        factory = DummyFactory(store=store)
        deleteTCP = TCPPort(
            store=store, factory=factory, portNumber=10, interface=u"foo")
        keepTCP = TCPPort(
            store=store, factory=factory, portNumber=10, interface=u"bar")
        deleteSSL = SSLPort(
            store=store, factory=factory, portNumber=10, interface=u"baz",
            certificatePath=store.filesdir.child("baz"))
        keepSSL = SSLPort(
            store=store, factory=factory, portNumber=10, interface=u"quux",
            certificatePath=store.filesdir.child("quux"))
        deleteEndpoint = StringEndpointPort(
            store=store, factory=factory, description=u'tcp:1234')
        keepEndpoint = StringEndpointPort(
            store=store, factory=factory, description=u'tcp:1235')
        self.assertSuccessStatus(
            self._makeConfig(store),
            ["delete",
             "--port-identifier", str(deleteTCP.storeID),
             "--port-identifier", str(deleteSSL.storeID),
             "--port-identifier", str(deleteEndpoint.storeID)])
        self.assertEqual("Deleted.\n", sys.stdout.getvalue())
        self.assertEqual(list(store.query(TCPPort)), [keepTCP])
        self.assertEqual(list(store.query(SSLPort)), [keepSSL])
        self.assertEqual(list(store.query(StringEndpointPort)), [keepEndpoint])


    def test_cannotDeleteOtherStuff(self):
        """
        I{axiomatic port delete} will not delete something which is neither a
        L{TCPPort} nor an L{SSLPort} and does not delete anything if an invalid
        port identifier is present in the command.
        """
        store = Store()
        factory = DummyFactory(store=store)
        tcp = TCPPort(
            store=store, factory=factory, interface=u"foo", portNumber=1234)
        self.assertFailStatus(
            1,
            self._makeConfig(store),
            ["delete",
             "--port-identifier", str(tcp.storeID),
             "--port-identifier", str(factory.storeID)])
        self.assertEqual(
            "%d does not identify a port.\n" % (factory.storeID,),
            sys.stdout.getvalue())
        self.assertEqual(list(store.query(DummyFactory)), [factory])
        self.assertEqual(list(store.query(TCPPort)), [tcp])


    def test_cannotDeleteNonExistent(self):
        """
        I{axiomatic port delete} writes a short error to standard output when a
        port-identifier is specified for which there is no corresponding store
        ID.
        """
        store = Store()
        self.assertFailStatus(
            1, self._makeConfig(store),
            ["delete", "--port-identifier", "12345"])
        self.assertEqual(
            "12345 does not identify an item.\n",
            sys.stdout.getvalue())


    _createHelpText = (
        "Usage: axiomatic [options] port [options] create [options]\n"
        "Options:\n"
        "      --strport=             A Twisted strports description of a "
        "port to add.\n"
        "      --factory-identifier=  Identifier for a protocol factory to "
        "associate with\n"
        "                             the new port.\n"
        "      --version              Display Twisted version and exit.\n"
        "      --help                 Display this help and exit.\n"
        "\n"
        "Create a new port binding for an existing factory.  If a server is "
        "currently\n"
        "running using the database in which the port is created, the "
        "factory will *not*\n"
        "be started on that port until the server is restarted.\n"
        "\n")

    def test_createImplicitHelp(self):
        """
        If I{axiomatic port create} is invoked with no arguments, usage
        information for the C{create} subcommand is written to standard out.
        """
        self.assertSuccessStatus(self._makeConfig(None), ["create"])
        self.assertSpacelessEqual(self._createHelpText, sys.stdout.getvalue())


    def test_createExplicitHelp(self):
        """
        If I{axiomatic port create} is invoked with C{--help} as an argument,
        usage information for the C{add} subcommand is written to standard out.
        """
        self.assertSuccessStatus(self._makeConfig(None), ["create", "--help"])
        self.assertSpacelessEqual(self._createHelpText, sys.stdout.getvalue())


    def test_createInvalidPortDescription(self):
        """
        If an invalid string is given for the C{strport} option of I{axiomatic
        port create}, a short error is written to standard output.
        """
        store = Store()
        factory = DummyFactory(store=store)
        self.assertFailStatus(
            1, self._makeConfig(store),
            ["create", "--strport", "xyz",
             "--factory-identifier", str(factory.storeID)])
        self.assertEqual(
            "'xyz' is not a valid port description.\n", sys.stdout.getvalue())


    def test_createNonExistentFactoryIdentifier(self):
        """
        If a storeID which is not associated with any item is given for the
        C{factory-identifier} option of I{axiomatic port create}, a short error
        is written to standard output.
        """
        store = Store()
        self.assertFailStatus(
            1, self._makeConfig(store),
            ["create", "--strport", "tcp:8080",
             "--factory-identifier", "123"])
        self.assertEqual(
            "123 does not identify an item.\n", sys.stdout.getvalue())


    def test_createNonFactoryIdentifier(self):
        """
        If a storeID which is associated with an item which does not provide
        L{IProtocolFactoryFactory} is given for the C{factory-identifier}
        option of I{axiomatic port create}, a short error is written to
        standard output.
        """
        store = Store()
        storeID = TCPPort(store=store).storeID
        self.assertFailStatus(
            1, self._makeConfig(store),
            ["create", "--strport", "tcp:8080",
             "--factory-identifier", str(storeID)])
        self.assertEqual(
            "%d does not identify a factory.\n" % (storeID,),
            sys.stdout.getvalue())


    def test_createPort(self):
        """
        I{axiomatic port create} creates a new
        L{xmantissa.port.StringEndpointPort} with the specified description,
        referring to the specified factory. The port is also powered up on the
        store for L{IService}.
        """
        store = Store()
        factory = DummyFactory(store=store)
        self.assertSuccessStatus(
            self._makeConfig(store),
            ["create", "--strport", "tcp:8080",
             "--factory-identifier", str(factory.storeID)])
        self.assertEqual("Created.\n", sys.stdout.getvalue())
        [port] = list(store.query(StringEndpointPort))
        self.assertEqual(u'tcp:8080', port.description)
        self.assertEqual(list(store.interfacesFor(port)), [IService])

# Copyright (c) 2008 Divmod, Inc. # See LICENSE for details. Tests for L{xmantissa.port}. -----BEGIN CERTIFICATE----- <KEY>END CERTIFICATE----- -----<KEY> Stub class used to track what reactor listen calls have been made and what     created ports have been stopped. Helper class used as a stand-in for a real protocol factory by the unit     tests. Meaningless attribute which serves only to make this a valid Item subclass. A reference to the protocol factory which L{getFactory} will return. 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}. Create and return a new port instance with the given attribute values. 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}. 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. Test that C{self.portType} remembers the port number it is told to         listen on. Test that C{self.portType} remembers the interface it is told to listen         on. Test that C{self.portType} remembers the factory it is given to associate         with its port. Test that C{self.portType} becomes a service on the store it is installed on. Test that the C{self.portType.setServiceParent} method adds the C{self.portType} to         the Axiom Store Service as a child. Test that the C{self.portType.disownServiceParent} method removes the         C{self.portType} from the Axiom Store Service. Test that C{self.portType} is a child of the store service after it is         installed. Test that C{self.portType} binds a low-numbered port with the reactor when it         is started with privilege. Test that C{self.portType} doesn't bind a high-numbered port with the         reactor when it is started with privilege. Test that C{self.portType} binds a low-numbered port with the reactor         when it is started without privilege. Test that C{self.portType} binds a high-numbered port with the reactor         when it is started without privilege. Test that C{self.portType} binds to all interfaces if no interface is         explicitly specified. Test that C{self.portType} binds to only the specified interface when         instructed to. Test that C{self.portType} only binds one network port when         C{privilegedStartService} and C{startService} are both called. Test that C{self.portType} cleans up its listening port when it is stopped. Test that the deletion of a C{self.portType}'s factory item results in the         C{self.portType} being deleted. Test that a deleted C{self.portType} no longer shows up in the children list         of the service which used to be its parent. Tests for L{xmantissa.port.TCPPort}. Tests for L{xmantissa.port.SSLPort}. Test that L{SSLPort} remembers the certificate filename it is given. Fake L{twisted.application.service.IService} implementation for testing     L{xmantissa.port.StringEndpointPort}'s wrapping behaviour. Tests for L{xmantissa.port.StringEndpointPort}. A fake for L{twisted.application.strports.service} that just constructs         our fake service object. The underlying endpoint service is started when         L{xmantissa.port.StringEndpointPort} is started. The underlying endpoint service is created with the description         specified by the L{xmantissa.port.StringEndpointPort}. The underlying endpoint service is created with the factory specified         by invoking C{getFactory()} on the C{factory} attribute. The underlying endpoint service is stopped when         L{xmantissa.port.StringEndpointPort} is stopped. Tests for the I{axiomatic port} command. Override C{sys.stdout} to capture anything written by the port         subcommand. Restore the original value of C{sys.stdout}. Create a L{PortConfiguration} instance with a properly set C{parent}         attribute. Parse the given arguments with the given options object and assert that         L{SystemExit} is raised with an exit code of C{0}. Parse the given arguments with the given options object and assert that         L{SystemExit} is raised with the specified exit code. Assert the equality of two strings without respect to their whitespace         or order. L{PortConfiguration} provides L{IAxiomaticCommand}. L{PortConfiguration} is available as a subcommand of I{axiomatic}. # This is a bit unfortunate.  It depends on what Options in Twisted # decides to spit out.  Note particularly the seemingly random amount # of trailing whitespace included on some lines.  The intent of tests # using this isn't really to ensure byte-identical results, but simply # to verify that help text is going to be shown to a user.  -exarkun When I{axiomatic port} is invoked with no arguments, usage information         is written to standard out and the process exits successfully. When I{axiomatic port} is invoked with I{--help}, usage information is         written to standard out. When I{axiomatic port list} is invoked with I{--help}, usage         information for the C{list} subcommand is written to standard out. When I{axiomatic port list} is invoked, the ports which are currently         configured in the system are displayed. When I{axiomatic port list} is invoked for a L{Store} which has a         L{TCPPort} in it, the details of that port, including its factory, are         written to stdout. When I{axiomatic port list} is invoked for a L{Store} which has an         L{SSLPort} in it, the details of that port, including its factory, are         written to stdout. When I{axiomatic port list} is invoked for a L{Store} which has an         L{StringEndpointPort} in it, the endpoint description and factory are         written to stdout. I{axiomatic port list} displays a special string for a port bound to         C{INADDR_ANY}. If there is an L{SSLPort} with no certificate or no port number (a         rather invalid configuration), I{axiomatic port list} should show this         in its output without producing an error. I{axiomatic port list} displays two different ports bound to the same         factory together beneath that factory. I{axiomatic port list} displays two different factories separately from         each other. I{axiomatic port list} displays factories even if they aren't associate         with any port. If I{axiomatic port delete} is invoked with I{--help}, usage         information for the C{delete} subcommand is written to standard out. If I{axiomatic port delete} is invoked with no arguments, usage         information for the C{delete} subcommand is written to standard out. I{axiomatic port delete} deletes each ports with a C{storeID} which is         specified. I{axiomatic port delete} will not delete something which is neither a         L{TCPPort} nor an L{SSLPort} and does not delete anything if an invalid         port identifier is present in the command. I{axiomatic port delete} writes a short error to standard output when a         port-identifier is specified for which there is no corresponding store         ID. If I{axiomatic port create} is invoked with no arguments, usage         information for the C{create} subcommand is written to standard out. If I{axiomatic port create} is invoked with C{--help} as an argument,         usage information for the C{add} subcommand is written to standard out. If an invalid string is given for the C{strport} option of I{axiomatic         port create}, a short error is written to standard output. If a storeID which is not associated with any item is given for the         C{factory-identifier} option of I{axiomatic port create}, a short error         is written to standard output. If a storeID which is associated with an item which does not provide         L{IProtocolFactoryFactory} is given for the C{factory-identifier}         option of I{axiomatic port create}, a short error is written to         standard output. I{axiomatic port create} creates a new         L{xmantissa.port.StringEndpointPort} with the specified description,         referring to the specified factory. The port is also powered up on the         store for L{IService}.