~launchpad-pqm/launchpad/devel

# Copyright 2009-2011 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

from __future__ import with_statement

""" Test layers

Note that many tests are performed at run time in the layers themselves
to confirm that the environment hasn't been corrupted by tests
"""

__metaclass__ = type

from contextlib import nested
from cStringIO import StringIO
import os
import signal
import smtplib
from urllib import urlopen

from amqplib import client_0_8 as amqp
from fixtures import (
    EnvironmentVariableFixture,
    Fixture,
    TestWithFixtures,
    )
from lazr.config import as_host_port
import testtools
from zope.component import (
    ComponentLookupError,
    getUtility,
    )

from canonical.lazr.pidfile import pidfile_path
from lp.services.config import config
from lp.services.librarian.client import (
    LibrarianClient,
    UploadFailed,
    )
from lp.services.librarian.interfaces.client import ILibrarianClient
from lp.services.memcache.client import memcache_client_factory
from lp.testing.layers import (
    AppServerLayer,
    BaseLayer,
    DatabaseLayer,
    FunctionalLayer,
    LaunchpadFunctionalLayer,
    LaunchpadLayer,
    LaunchpadScriptLayer,
    LaunchpadTestSetup,
    LaunchpadZopelessLayer,
    LayerInvariantError,
    LayerIsolationError,
    LayerProcessController,
    LibrarianLayer,
    MemcachedLayer,
    RabbitMQLayer,
    ZopelessLayer,
    )


class BaseLayerIsolator(Fixture):
    """A fixture for isolating BaseLayer.

    This is useful to test interactions with LP_PERSISTENT_TEST_SERVICES
    which makes tests within layers unable to test that easily.
    """

    def __init__(self, with_persistent=False):
        """Create a BaseLayerIsolator.

        :param with_persistent: If True LP_PERSISTENT_TEST_SERVICES will
            be enabled during setUp.
        """
        super(BaseLayerIsolator, self).__init__()
        self.with_persistent = with_persistent

    def setUp(self):
        super(BaseLayerIsolator, self).setUp()
        if self.with_persistent:
            env_value = ''
        else:
            env_value = None
        self.useFixture(EnvironmentVariableFixture(
            'LP_PERSISTENT_TEST_SERVICES', env_value))
        self.useFixture(EnvironmentVariableFixture('LP_TEST_INSTANCE'))


class LayerFixture(Fixture):
    """Adapt a layer to a fixture.

    Note that the layer setup/teardown are called, not the base class ones.

    :ivar layer: The adapted layer.
    """

    def __init__(self, layer):
        """Create a LayerFixture.

        :param layer: The layer to use.
        """
        super(LayerFixture, self).__init__()
        self.layer = layer

    def setUp(self):
        super(LayerFixture, self).setUp()
        self.layer.setUp()
        self.addCleanup(self.layer.tearDown)


class TestBaseLayer(testtools.TestCase, TestWithFixtures):

    def test_allocates_LP_TEST_INSTANCE(self):
        self.useFixture(BaseLayerIsolator())
        with LayerFixture(BaseLayer):
            self.assertEqual(
                str(os.getpid()),
                os.environ.get('LP_TEST_INSTANCE'))
        self.assertEqual(None, os.environ.get('LP_TEST_INSTANCE'))

    def test_persist_test_services_disables_LP_TEST_INSTANCE(self):
        self.useFixture(BaseLayerIsolator(with_persistent=True))
        with LayerFixture(BaseLayer):
            self.assertEqual(None, os.environ.get('LP_TEST_INSTANCE'))
        self.assertEqual(None, os.environ.get('LP_TEST_INSTANCE'))

    def test_generates_unique_config(self):
        config.setInstance('testrunner')
        orig_instance = config.instance_name
        self.useFixture(
            EnvironmentVariableFixture('LP_PERSISTENT_TEST_SERVICES'))
        self.useFixture(EnvironmentVariableFixture('LP_TEST_INSTANCE'))
        self.useFixture(EnvironmentVariableFixture('LPCONFIG'))
        with LayerFixture(BaseLayer):
            self.assertEqual(
                'testrunner_%s' % os.environ['LP_TEST_INSTANCE'],
                config.instance_name)
        self.assertEqual(orig_instance, config.instance_name)

    def test_generates_unique_config_dirs(self):
        self.useFixture(
            EnvironmentVariableFixture('LP_PERSISTENT_TEST_SERVICES'))
        self.useFixture(EnvironmentVariableFixture('LP_TEST_INSTANCE'))
        self.useFixture(EnvironmentVariableFixture('LPCONFIG'))
        with LayerFixture(BaseLayer):
            runner_root = 'configs/%s' % config.instance_name
            runner_appserver_root = 'configs/testrunner-appserver_%s' % \
                os.environ['LP_TEST_INSTANCE']
            self.assertTrue(os.path.isfile(
                runner_root + '/launchpad-lazr.conf'))
            self.assertTrue(os.path.isfile(
                runner_appserver_root + '/launchpad-lazr.conf'))
        self.assertFalse(os.path.exists(runner_root))
        self.assertFalse(os.path.exists(runner_appserver_root))


class BaseTestCase(testtools.TestCase):
    """Both the Base layer tests, as well as the base Test Case
    for all the other Layer tests.
    """
    layer = BaseLayer

    # These flags will be overridden in subclasses to describe the
    # environment they expect to have available.
    want_component_architecture = False
    want_librarian_running = False
    want_launchpad_database = False
    want_functional_flag = False
    want_zopeless_flag = False
    want_memcached = False
    want_rabbitmq = False

    def testBaseIsSetUpFlag(self):
        self.failUnlessEqual(BaseLayer.isSetUp, True)

    def testFunctionalIsSetUp(self):
        self.failUnlessEqual(
                FunctionalLayer.isSetUp, self.want_functional_flag
                )

    def testZopelessIsSetUp(self):
        self.failUnlessEqual(
                ZopelessLayer.isSetUp, self.want_zopeless_flag
                )

    def testComponentArchitecture(self):
        try:
            getUtility(ILibrarianClient)
        except ComponentLookupError:
            self.failIf(
                    self.want_component_architecture,
                    'Component Architecture should be available.'
                    )
        else:
            self.failUnless(
                    self.want_component_architecture,
                    'Component Architecture should not be available.'
                    )

    def testLibrarianRunning(self):
        # Check that the librarian is running. Note that even if the
        # librarian is running, it may not be able to actually store
        # or retrieve files if, for example, the Launchpad database is
        # not currently available.
        try:
            urlopen(config.librarian.download_url).read()
            self.failUnless(
                    self.want_librarian_running,
                    'Librarian should not be running.'
                    )
        except IOError:
            self.failIf(
                    self.want_librarian_running,
                    'Librarian should be running.'
                    )

    def testLibrarianWorking(self):
        # Check that the librian is actually working. This means at
        # a minimum the Librarian service is running and is connected
        # to the Launchpad database.
        want_librarian_working = (
                self.want_librarian_running and self.want_launchpad_database
                and self.want_component_architecture
                )
        client = LibrarianClient()
        data = 'Whatever'
        try:
            client.addFile(
                    'foo.txt', len(data), StringIO(data), 'text/plain'
                    )
        except UploadFailed:
            self.failIf(
                    want_librarian_working,
                    'Librarian should be fully operational'
                    )
        except (AttributeError, ComponentLookupError):
            self.failIf(
                    want_librarian_working,
                    'Librarian not operational as component architecture '
                    'not loaded'
                    )
        else:
            self.failUnless(
                    want_librarian_working,
                    'Librarian should not be operational'
                    )

    def testLaunchpadDbAvailable(self):
        if not self.want_launchpad_database:
            self.assertEqual(None, DatabaseLayer._db_fixture)
            return
        con = DatabaseLayer.connect()
        cur = con.cursor()
        cur.execute("SELECT id FROM Person LIMIT 1")
        self.assertNotEqual(None, cur.fetchone())

    def testMemcachedWorking(self):
        client = MemcachedLayer.client or memcache_client_factory()
        key = "BaseTestCase.testMemcachedWorking"
        client.forget_dead_hosts()
        is_live = client.set(key, "live")
        if self.want_memcached:
            self.assertEqual(
                is_live, True, "memcached not live when it should be.")
        else:
            self.assertEqual(
                is_live, False, "memcached is live but should not be.")

    def testRabbitWorking(self):
        rabbitmq = config.rabbitmq
        if not self.want_rabbitmq:
            self.assertEqual(None, rabbitmq.host)
        else:
            self.assertNotEqual(None, rabbitmq.host)
            conn = amqp.Connection(
                host=rabbitmq.host,
                userid=rabbitmq.userid,
                password=rabbitmq.password,
                virtual_host=rabbitmq.virtual_host,
                insist=False)
            conn.close()


class MemcachedTestCase(BaseTestCase):
    layer = MemcachedLayer
    want_memcached = True


class LibrarianTestCase(BaseTestCase):
    layer = LibrarianLayer

    want_launchpad_database = True
    want_librarian_running = True

    def testUploadsSucceed(self):
        # This layer is able to be used on its own as it depends on
        # DatabaseLayer.
        # We can test this using remoteAddFile (it does not need the CA
        # loaded)
        client = LibrarianClient()
        data = 'This is a test'
        client.remoteAddFile(
            'foo.txt', len(data), StringIO(data), 'text/plain')


class LibrarianLayerTest(testtools.TestCase, TestWithFixtures):

    def test_makes_unique_instance(self):
        # Capture the original settings
        default_root = config.librarian_server.root
        download_port = config.librarian.download_port
        restricted_download_port = config.librarian.restricted_download_port
        self.useFixture(BaseLayerIsolator())
        with nested(
            LayerFixture(BaseLayer),
            LayerFixture(DatabaseLayer),
            ):
            with LayerFixture(LibrarianLayer):
                active_root = config.librarian_server.root
                # The config settings have changed:
                self.assertNotEqual(default_root, active_root)
                self.assertNotEqual(
                    download_port, config.librarian.download_port)
                self.assertNotEqual(
                    restricted_download_port,
                    config.librarian.restricted_download_port)
                self.assertTrue(os.path.exists(active_root))
            # This needs more sophistication in the config system (tearDown on
            # the layer needs to pop the config fragment off of disk - and
            # perhaps notify other processes that its done this. So for now we
            # leave the new config in place).
            # self.assertEqual(default_root, config.librarian_server.root)
            # The working dir has to have been deleted.
            self.assertFalse(os.path.exists(active_root))


class LibrarianResetTestCase(testtools.TestCase):
    """Our page tests need to run multple tests without destroying
    the librarian database in between.
    """
    layer = LibrarianLayer

    sample_data = 'This is a test'

    def test_librarian_is_reset(self):
        # Add a file. We use remoteAddFile because it does not need the CA
        # loaded to work.
        client = LibrarianClient()
        LibrarianTestCase.url = client.remoteAddFile(
                self.sample_data, len(self.sample_data),
                StringIO(self.sample_data), 'text/plain'
                )
        self.failUnlessEqual(
                urlopen(LibrarianTestCase.url).read(), self.sample_data
                )
        # Perform the librarian specific between-test code:
        LibrarianLayer.testTearDown()
        LibrarianLayer.testSetUp()
        # Which should have nuked the old file.
        # XXX: StuartBishop 2006-06-30 Bug=51370:
        # We should get a DownloadFailed exception here.
        data = urlopen(LibrarianTestCase.url).read()
        self.failIfEqual(data, self.sample_data)


class LibrarianHideTestCase(testtools.TestCase):
    layer = LaunchpadLayer

    def testHideLibrarian(self):
        # First perform a successful upload:
        client = LibrarianClient()
        data = 'foo'
        client.remoteAddFile(
            'foo', len(data), StringIO(data), 'text/plain')
        # The database was committed to, but not by this process, so we need
        # to ensure that it is fully torn down and recreated.
        DatabaseLayer.force_dirty_database()

        # Hide the librarian, and show that the upload fails:
        LibrarianLayer.hide()
        self.assertRaises(UploadFailed, client.remoteAddFile,
                          'foo', len(data), StringIO(data), 'text/plain')

        # Reveal the librarian again, allowing uploads:
        LibrarianLayer.reveal()
        client.remoteAddFile(
            'foo', len(data), StringIO(data), 'text/plain')


class RabbitMQTestCase(BaseTestCase):
    layer = RabbitMQLayer
    want_rabbitmq = True


class DatabaseTestCase(BaseTestCase):
    layer = DatabaseLayer

    want_launchpad_database = True

    def testConnect(self):
        DatabaseLayer.connect()

    def getWikinameCount(self, con):
        cur = con.cursor()
        cur.execute("SELECT COUNT(*) FROM Wikiname")
        num = cur.fetchone()[0]
        return num

    def test_db_is_reset(self):
        con = DatabaseLayer.connect()
        cur = con.cursor()
        cur.execute("DELETE FROM Wikiname")
        self.failUnlessEqual(self.getWikinameCount(con), 0)
        con.commit()
        # Run the per-test code for the Database layer.
        DatabaseLayer.testTearDown()
        DatabaseLayer.testSetUp()
        # Wikiname table should have been restored.
        con = DatabaseLayer.connect()
        self.assertNotEqual(0, self.getWikinameCount(con))


class LaunchpadTestCase(BaseTestCase):
    layer = LaunchpadLayer

    want_launchpad_database = True
    want_librarian_running = True
    want_memcached = True
    want_rabbitmq = True


class FunctionalTestCase(BaseTestCase):
    layer = FunctionalLayer

    want_component_architecture = True
    want_functional_flag = True


class ZopelessTestCase(BaseTestCase):
    layer = ZopelessLayer

    want_component_architecture = True
    want_launchpad_database = False
    want_librarian_running = False
    want_zopeless_flag = True


class LaunchpadFunctionalTestCase(BaseTestCase):
    layer = LaunchpadFunctionalLayer

    want_component_architecture = True
    want_launchpad_database = True
    want_librarian_running = True
    want_functional_flag = True
    want_memcached = True
    want_rabbitmq = True


class LaunchpadZopelessTestCase(BaseTestCase):
    layer = LaunchpadZopelessLayer

    want_component_architecture = True
    want_launchpad_database = True
    want_librarian_running = True
    want_zopeless_flag = True
    want_memcached = True
    want_rabbitmq = True


class LaunchpadScriptTestCase(BaseTestCase):
    layer = LaunchpadScriptLayer

    want_component_architecture = True
    want_launchpad_database = True
    want_librarian_running = True
    want_zopeless_flag = True
    want_memcached = True
    want_rabbitmq = True


class LayerProcessControllerInvariantsTestCase(BaseTestCase):
    layer = AppServerLayer

    want_component_architecture = True
    want_launchpad_database = True
    want_librarian_running = True
    want_functional_flag = True
    want_zopeless_flag = False
    want_memcached = True
    want_rabbitmq = True

    def testAppServerIsAvailable(self):
        # Test that the app server is up and running.
        mainsite = LayerProcessController.appserver_config.vhost.mainsite
        home_page = urlopen(mainsite.rooturl).read()
        self.failUnless(
            'Is your project registered yet?' in home_page,
            "Home page couldn't be retrieved:\n%s" % home_page)

    def testSMTPServerIsAvailable(self):
        # Test that the SMTP server is up and running.
        smtpd = smtplib.SMTP()
        host, port = as_host_port(config.mailman.smtp)
        code, message = smtpd.connect(host, port)
        self.assertEqual(code, 220)

    def testStartingAppServerTwiceRaisesInvariantError(self):
        # Starting the appserver twice should raise an exception.
        self.assertRaises(LayerInvariantError,
                          LayerProcessController.startAppServer)

    def testStartingSMTPServerTwiceRaisesInvariantError(self):
        # Starting the SMTP server twice should raise an exception.
        self.assertRaises(LayerInvariantError,
                          LayerProcessController.startSMTPServer)


class LayerProcessControllerTestCase(testtools.TestCase):
    """Tests for the `LayerProcessController`."""
    # We need the database to be set up, no more.
    layer = DatabaseLayer

    def tearDown(self):
        super(LayerProcessControllerTestCase, self).tearDown()
        # Stop both servers.  It's okay if they aren't running.
        LayerProcessController.stopSMTPServer()
        LayerProcessController.stopAppServer()

    def test_stopAppServer(self):
        # Test that stopping the app server kills the process and remove the
        # PID file.
        LayerProcessController.setConfig()
        LayerProcessController.startAppServer()
        pid = LayerProcessController.appserver.pid
        pid_file = pidfile_path('launchpad',
                                LayerProcessController.appserver_config)
        LayerProcessController.stopAppServer()
        self.assertRaises(OSError, os.kill, pid, 0)
        self.failIf(os.path.exists(pid_file), "PID file wasn't removed")
        self.failUnless(LayerProcessController.appserver is None,
                        "appserver class attribute wasn't reset")

    def test_postTestInvariants(self):
        # A LayerIsolationError should be raised if the app server dies in the
        # middle of a test.
        LayerProcessController.setConfig()
        LayerProcessController.startAppServer()
        pid = LayerProcessController.appserver.pid
        os.kill(pid, signal.SIGTERM)
        LayerProcessController.appserver.wait()
        self.assertRaises(LayerIsolationError,
                          LayerProcessController.postTestInvariants)

    def test_postTestInvariants_dbIsReset(self):
        # The database should be reset by the test invariants.
        LayerProcessController.setConfig()
        LayerProcessController.startAppServer()
        LayerProcessController.postTestInvariants()
        # XXX: Robert Collins 2010-10-17 bug=661967 - this isn't a reset, its
        # a flag that it *needs* a reset, which is actually quite different;
        # the lack of a teardown will leak databases.
        self.assertEquals(True, LaunchpadTestSetup()._reset_db)


class TestNameTestCase(testtools.TestCase):
    layer = BaseLayer

    def testTestName(self):
        self.failUnlessEqual(
                BaseLayer.test_name,
                "testTestName "
                "(lp.testing.tests.test_layers_functional.TestNameTestCase)")