test_fixtures.py

# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import sqlalchemy
import sys

import fixtures as fx
import mock
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import uuidutils
import testtools

from nova.db.sqlalchemy import api as session
from nova import exception
from nova.objects import base as obj_base
from nova.tests import fixtures
from nova.tests.unit import conf_fixture
from nova import utils

CONF = cfg.CONF


class TestConfFixture(testtools.TestCase):
    """Test the Conf fixtures in Nova.

    This is a basic test that this fixture works like we expect.

    Expectations:

    1. before using the fixture, a default value (api_paste_config)
       comes through untouched.

    2. before using the fixture, a known default value that we
       override is correct.

    3. after using the fixture a known value that we override is the
       new value.

    4. after using the fixture we can set a default value to something
       random, and it will be reset once we are done.

    There are 2 copies of this test so that you can verify they do the
    right thing with:

       tox -e py27 test_fixtures -- --concurrency=1

    As regardless of run order, their initial asserts would be
    impacted if the reset behavior isn't working correctly.

    """
    def _test_override(self):
        self.assertEqual('api-paste.ini', CONF.api_paste_config)
        self.assertFalse(CONF.fake_network)
        self.useFixture(conf_fixture.ConfFixture())
        CONF.set_default('api_paste_config', 'foo')
        self.assertTrue(CONF.fake_network)

    def test_override1(self):
        self._test_override()

    def test_override2(self):
        self._test_override()


class TestOutputStream(testtools.TestCase):
    """Ensure Output Stream capture works as expected.

    This has the added benefit of providing a code example of how you
    can manipulate the output stream in your own tests.
    """
    def test_output(self):
        self.useFixture(fx.EnvironmentVariable('OS_STDOUT_CAPTURE', '1'))
        self.useFixture(fx.EnvironmentVariable('OS_STDERR_CAPTURE', '1'))

        out = self.useFixture(fixtures.OutputStreamCapture())
        sys.stdout.write("foo")
        sys.stderr.write("bar")
        self.assertEqual("foo", out.stdout)
        self.assertEqual("bar", out.stderr)
        # TODO(sdague): nuke the out and err buffers so it doesn't
        # make it to testr


class TestLogging(testtools.TestCase):
    def test_default_logging(self):
        stdlog = self.useFixture(fixtures.StandardLogging())
        root = logging.getLogger()
        # there should be a null handler as well at DEBUG
        self.assertEqual(2, len(root.handlers), root.handlers)
        log = logging.getLogger(__name__)
        log.info("at info")
        log.debug("at debug")
        self.assertIn("at info", stdlog.logger.output)
        self.assertNotIn("at debug", stdlog.logger.output)

        # broken debug messages should still explode, even though we
        # aren't logging them in the regular handler
        self.assertRaises(TypeError, log.debug, "this is broken %s %s", "foo")

        # and, ensure that one of the terrible log messages isn't
        # output at info
        warn_log = logging.getLogger('migrate.versioning.api')
        warn_log.info("warn_log at info, should be skipped")
        warn_log.error("warn_log at error")
        self.assertIn("warn_log at error", stdlog.logger.output)
        self.assertNotIn("warn_log at info", stdlog.logger.output)

    def test_debug_logging(self):
        self.useFixture(fx.EnvironmentVariable('OS_DEBUG', '1'))

        stdlog = self.useFixture(fixtures.StandardLogging())
        root = logging.getLogger()
        # there should no longer be a null handler
        self.assertEqual(1, len(root.handlers), root.handlers)
        log = logging.getLogger(__name__)
        log.info("at info")
        log.debug("at debug")
        self.assertIn("at info", stdlog.logger.output)
        self.assertIn("at debug", stdlog.logger.output)


class TestTimeout(testtools.TestCase):
    """Tests for our timeout fixture.

    Testing the actual timeout mechanism is beyond the scope of this
    test, because it's a pretty clear pass through to fixtures'
    timeout fixture, which tested in their tree.

    """
    def test_scaling(self):
        # a bad scaling factor
        self.assertRaises(ValueError, fixtures.Timeout, 1, 0.5)

        # various things that should work.
        timeout = fixtures.Timeout(10)
        self.assertEqual(10, timeout.test_timeout)
        timeout = fixtures.Timeout("10")
        self.assertEqual(10, timeout.test_timeout)
        timeout = fixtures.Timeout("10", 2)
        self.assertEqual(20, timeout.test_timeout)


class TestOSAPIFixture(testtools.TestCase):
    @mock.patch('nova.objects.Service.get_by_host_and_binary')
    @mock.patch('nova.objects.Service.create')
    def test_responds_to_version(self, mock_service_create, mock_get):
        """Ensure the OSAPI server responds to calls sensibly."""
        self.useFixture(fixtures.OutputStreamCapture())
        self.useFixture(fixtures.StandardLogging())
        self.useFixture(conf_fixture.ConfFixture())
        self.useFixture(fixtures.RPCFixture('nova.test'))
        api = self.useFixture(fixtures.OSAPIFixture()).api

        # request the API root, which provides us the versions of the API
        resp = api.api_request('/', strip_version=True)
        self.assertEqual(200, resp.status_code, resp.content)

        # request a bad root url, should be a 404
        #
        # NOTE(sdague): this currently fails, as it falls into the 300
        # dispatcher instead. This is a bug. The test case is left in
        # here, commented out until we can address it.
        #
        # resp = api.api_request('/foo', strip_version=True)
        # self.assertEqual(resp.status_code, 400, resp.content)

        # request a known bad url, and we should get a 404
        resp = api.api_request('/foo')
        self.assertEqual(404, resp.status_code, resp.content)


class TestDatabaseFixture(testtools.TestCase):
    def test_fixture_reset(self):
        # because this sets up reasonable db connection strings
        self.useFixture(conf_fixture.ConfFixture())
        self.useFixture(fixtures.Database())
        engine = session.get_engine()
        conn = engine.connect()
        result = conn.execute("select * from instance_types")
        rows = result.fetchall()
        self.assertEqual(5, len(rows), "Rows %s" % rows)

        # insert a 6th instance type, column 5 below is an int id
        # which has a constraint on it, so if new standard instance
        # types are added you have to bump it.
        conn.execute("insert into instance_types VALUES "
                     "(NULL, NULL, NULL, 't1.test', 6, 4096, 2, 0, NULL, '87'"
                     ", 1.0, 40, 0, 0, 1, 0)")
        result = conn.execute("select * from instance_types")
        rows = result.fetchall()
        self.assertEqual(6, len(rows), "Rows %s" % rows)

        # reset by invoking the fixture again
        #
        # NOTE(sdague): it's important to reestablish the db
        # connection because otherwise we have a reference to the old
        # in mem db.
        self.useFixture(fixtures.Database())
        conn = engine.connect()
        result = conn.execute("select * from instance_types")
        rows = result.fetchall()
        self.assertEqual(5, len(rows), "Rows %s" % rows)

    def test_api_fixture_reset(self):
        # This sets up reasonable db connection strings
        self.useFixture(conf_fixture.ConfFixture())
        self.useFixture(fixtures.Database(database='api'))
        engine = session.get_api_engine()
        conn = engine.connect()
        result = conn.execute("select * from cell_mappings")
        rows = result.fetchall()
        self.assertEqual(0, len(rows), "Rows %s" % rows)

        uuid = uuidutils.generate_uuid()
        conn.execute("insert into cell_mappings (uuid, name) VALUES "
                     "('%s', 'fake-cell')" % (uuid,))
        result = conn.execute("select * from cell_mappings")
        rows = result.fetchall()
        self.assertEqual(1, len(rows), "Rows %s" % rows)

        # reset by invoking the fixture again
        #
        # NOTE(sdague): it's important to reestablish the db
        # connection because otherwise we have a reference to the old
        # in mem db.
        self.useFixture(fixtures.Database(database='api'))
        conn = engine.connect()
        result = conn.execute("select * from cell_mappings")
        rows = result.fetchall()
        self.assertEqual(0, len(rows), "Rows %s" % rows)

    def test_fixture_cleanup(self):
        # because this sets up reasonable db connection strings
        self.useFixture(conf_fixture.ConfFixture())
        fix = fixtures.Database()
        self.useFixture(fix)

        # manually do the cleanup that addCleanup will do
        fix.cleanup()

        # ensure the db contains nothing
        engine = session.get_engine()
        conn = engine.connect()
        schema = "".join(line for line in conn.connection.iterdump())
        self.assertEqual(schema, "BEGIN TRANSACTION;COMMIT;")

    def test_api_fixture_cleanup(self):
        # This sets up reasonable db connection strings
        self.useFixture(conf_fixture.ConfFixture())
        fix = fixtures.Database(database='api')
        self.useFixture(fix)

        # No data inserted by migrations so we need to add a row
        engine = session.get_api_engine()
        conn = engine.connect()
        uuid = uuidutils.generate_uuid()
        conn.execute("insert into cell_mappings (uuid, name) VALUES "
                     "('%s', 'fake-cell')" % (uuid,))
        result = conn.execute("select * from cell_mappings")
        rows = result.fetchall()
        self.assertEqual(1, len(rows), "Rows %s" % rows)

        # Manually do the cleanup that addCleanup will do
        fix.cleanup()

        # Ensure the db contains nothing
        engine = session.get_api_engine()
        conn = engine.connect()
        schema = "".join(line for line in conn.connection.iterdump())
        self.assertEqual("BEGIN TRANSACTION;COMMIT;", schema)


class TestDatabaseAtVersionFixture(testtools.TestCase):
    def test_fixture_schema_version(self):
        self.useFixture(conf_fixture.ConfFixture())

        # In/after 317 aggregates did have uuid
        self.useFixture(fixtures.DatabaseAtVersion(318))
        engine = session.get_engine()
        engine.connect()
        meta = sqlalchemy.MetaData(engine)
        aggregate = sqlalchemy.Table('aggregates', meta, autoload=True)
        self.assertTrue(hasattr(aggregate.c, 'uuid'))

        # Before 317, aggregates had no uuid
        self.useFixture(fixtures.DatabaseAtVersion(316))
        engine = session.get_engine()
        engine.connect()
        meta = sqlalchemy.MetaData(engine)
        aggregate = sqlalchemy.Table('aggregates', meta, autoload=True)
        self.assertFalse(hasattr(aggregate.c, 'uuid'))
        engine.dispose()

    def test_fixture_after_database_fixture(self):
        self.useFixture(conf_fixture.ConfFixture())
        self.useFixture(fixtures.Database())
        self.useFixture(fixtures.DatabaseAtVersion(318))


class TestIndirectionAPIFixture(testtools.TestCase):
    def test_indirection_api(self):
        # Should initially be None
        self.assertIsNone(obj_base.NovaObject.indirection_api)

        # make sure the fixture correctly sets the value
        fix = fixtures.IndirectionAPIFixture('foo')
        self.useFixture(fix)
        self.assertEqual('foo', obj_base.NovaObject.indirection_api)

        # manually do the cleanup that addCleanup will do
        fix.cleanup()

        # ensure the initial value is restored
        self.assertIsNone(obj_base.NovaObject.indirection_api)


class TestSpawnIsSynchronousFixture(testtools.TestCase):
    def test_spawn_patch(self):
        orig_spawn = utils.spawn_n

        fix = fixtures.SpawnIsSynchronousFixture()
        self.useFixture(fix)
        self.assertNotEqual(orig_spawn, utils.spawn_n)

    def test_spawn_passes_through(self):
        self.useFixture(fixtures.SpawnIsSynchronousFixture())
        tester = mock.MagicMock()
        utils.spawn_n(tester.function, 'foo', bar='bar')
        tester.function.assert_called_once_with('foo', bar='bar')

    def test_spawn_return_has_wait(self):
        self.useFixture(fixtures.SpawnIsSynchronousFixture())
        gt = utils.spawn(lambda x: '%s' % x, 'foo')
        foo = gt.wait()
        self.assertEqual('foo', foo)

    def test_spawn_n_return_has_wait(self):
        self.useFixture(fixtures.SpawnIsSynchronousFixture())
        gt = utils.spawn_n(lambda x: '%s' % x, 'foo')
        foo = gt.wait()
        self.assertEqual('foo', foo)

    def test_spawn_has_link(self):
        self.useFixture(fixtures.SpawnIsSynchronousFixture())
        gt = utils.spawn(mock.MagicMock)
        passed_arg = 'test'
        call_count = []

        def fake(thread, param):
            self.assertEqual(gt, thread)
            self.assertEqual(passed_arg, param)
            call_count.append(1)

        gt.link(fake, passed_arg)
        self.assertEqual(1, len(call_count))

    def test_spawn_n_has_link(self):
        self.useFixture(fixtures.SpawnIsSynchronousFixture())
        gt = utils.spawn_n(mock.MagicMock)
        passed_arg = 'test'
        call_count = []

        def fake(thread, param):
            self.assertEqual(gt, thread)
            self.assertEqual(passed_arg, param)
            call_count.append(1)

        gt.link(fake, passed_arg)
        self.assertEqual(1, len(call_count))


class TestBannedDBSchemaOperations(testtools.TestCase):
    def test_column(self):
        column = sqlalchemy.Column()
        with fixtures.BannedDBSchemaOperations(['Column']):
            self.assertRaises(exception.DBNotAllowed,
                              column.drop)
            self.assertRaises(exception.DBNotAllowed,
                              column.alter)

    def test_table(self):
        table = sqlalchemy.Table()
        with fixtures.BannedDBSchemaOperations(['Table']):
            self.assertRaises(exception.DBNotAllowed,
                              table.drop)
            self.assertRaises(exception.DBNotAllowed,
                              table.alter)


class TestStableObjectJsonFixture(testtools.TestCase):
    def test_changes_sort(self):
        class TestObject(obj_base.NovaObject):
            def obj_what_changed(self):
                return ['z', 'a']

        obj = TestObject()
        self.assertEqual(['z', 'a'],
                         obj.obj_to_primitive()['nova_object.changes'])
        with fixtures.StableObjectJsonFixture():
            self.assertEqual(['a', 'z'],
                             obj.obj_to_primitive()['nova_object.changes'])