How to use rally - 10 common examples

import sys
import netaddr
import six
from collections import defaultdict

from rally import exceptions
from rally_ovs.plugins.ovs import scenario
from rally.task import atomic
from rally.common import logging
from rally.common import objects

from netaddr.ip import IPRange
from rally_ovs.plugins.ovs.consts import ResourceType

LOG = logging.getLogger(__name__)


class SandboxScenario(scenario.OvsScenario):


    def _add_controller_resource(self, deployment, host_container,
                                 controller_cidr):
        dep = objects.Deployment.get(deployment)
        resources = dep.get_resources(type=ResourceType.CONTROLLER)
        if resources == None:
            dep.add_resource(provider_name=deployment,
                            type=ResourceType.CONTROLLER,
                            info={"ip":controller_cidr.split('/')[0],
                                  "deployment_name":deployment,
                                  "host_container":host_container})
            return

def test_stdout_capture(self):
        stdout = sys.stdout
        messages = ["abcdef", "defgaga"]
        with utils.StdOutCapture() as out:
            for msg in messages:
                print(msg)

        self.assertEqual(messages, out.getvalue().rstrip("\n").split("\n"))
        self.assertEqual(sys.stdout, stdout)

def test_verified_keystone_user_not_admin(self):
        self.auth_ref.role_names = ["notadmin"]
        self.assertRaises(exceptions.InvalidAdminException,
                          self.clients.verified_keystone)

def test_validate_too_small_network(self):
        config = self.config.copy()
        config["containers_per_host"] = 42
        self.assertRaises(exceptions.InvalidConfigException,
                          lxc.LxcProvider, self.deployment, config)

self.clients("nova").servers.get(return_value=fake_server)
        nova_scenario = utils.NovaScenario(context=self.context)
        nova_scenario._migrate(fake_server, skip_compute_nodes_check=True,
                               skip_host_check=True)

        self.mock_wait_for_status.mock.assert_called_once_with(
            fake_server,
            ready_statuses=["VERIFY_RESIZE"],
            update_resource=self.mock_get_from_manager.mock.return_value,
            check_interval=CONF.openstack.nova_server_migrate_poll_interval,
            timeout=CONF.openstack.nova_server_migrate_timeout)
        self.mock_get_from_manager.mock.assert_called_once_with()
        self._test_atomic_action_timer(nova_scenario.atomic_actions(),
                                       "nova.migrate")

        self.assertRaises(rally_exceptions.RallyException,
                          nova_scenario._migrate,
                          fake_server, skip_compute_nodes_check=True,
                          skip_host_check=False)
        self.assertRaises(rally_exceptions.RallyException,
                          nova_scenario._migrate,
                          fake_server, skip_compute_nodes_check=False,
                          skip_host_check=True)

def test_create_and_list_volume_types_with_fails(self):
        # Negative case: type isn't listed
        mock_service = self.mock_cinder.return_value
        fake_type = mock.Mock()
        pool_list = [mock.Mock(), mock.Mock(), mock.Mock()]
        description = "rally tests creating types"
        is_public = False

        scenario = volume_types.CreateAndListVolumeTypes(self._get_context())
        mock_service.create_volume_type.return_value = fake_type
        mock_service.list_types.return_value = pool_list
        self.assertRaises(rally_exceptions.RallyAssertionError,
                          scenario.run,
                          description=description, is_public=is_public)

        mock_service.create_volume_type.assert_called_once_with(
            description=description, is_public=is_public)
        mock_service.list_types.assert_called_once_with()

clients = self.context[
            "users"][0]["credential"].clients.return_value
        clients.glance().images.get = mock.Mock()

        result = self.validator.validate(self.context, config, None, None)
        self.assertIsNone(result)

        mock_glance_image.assert_called_once_with(
            context={"admin": {
                "credential": self.context["users"][0]["credential"]}})
        mock_glance_image.return_value.pre_process.assert_called_once_with(
            config["args"]["image"], config={})
        clients.glance().images.get.assert_called_with("image_id")

        exs = [exceptions.InvalidScenarioArgument(),
               glance_exc.HTTPNotFound()]
        for ex in exs:
            clients.glance().images.get.side_effect = ex

            e = self.assertRaises(
                validators.validation.ValidationError,
                self.validator.validate, self.context, config, None, None)

            self.assertEqual("Image 'fake_image' not found", e.message)

def test_logcatcher(self):
        LOG = log.getLogger("testlogger")
        LOG.logger.setLevel(log.INFO)

        with log.LogCatcher(LOG) as catcher:
            LOG.warning("Warning")
            LOG.info("Info")
            LOG.debug("Debug")

        catcher.assertInLogs("Warning")
        self.assertRaises(AssertionError, catcher.assertInLogs, "Error")

        self.assertEqual(["Warning", "Info"], catcher.fetchLogs())
        self.assertEqual(2, len(catcher.fetchLogRecords()))

def test_setup(self, mock_oslogging, mock_handlers, mock_conf):

        proj = "fakep"
        version = "fakev"
        mock_handlers.ColorHandler.LEVEL_COLORS = {
            logging.DEBUG: "debug_color"}
        mock_conf.rally_debug = True

        rally_logging.setup(proj, version)

        self.assertIn(logging.RDEBUG, mock_handlers.ColorHandler.LEVEL_COLORS)
        self.assertEqual(
            mock_handlers.ColorHandler.LEVEL_COLORS[logging.DEBUG],
            mock_handlers.ColorHandler.LEVEL_COLORS[logging.RDEBUG])

        mock_oslogging.setup.assert_called_once_with(mock_conf, proj, version)
        mock_oslogging.getLogger(None).logger.setLevel.assert_called_once_with(
            logging.RDEBUG)

def test__request_fail(self, mock_find_exception, mock_request):
        api_inst = api.API(skip_db_check=True)
        method = "test"
        path = "path"
        response = mock_request.return_value
        mock_find_exception.return_value = exceptions.RallyException()
        response.status_code = 201
        response.json.return_value = {"result": "test"}
        self.assertRaises(exceptions.RallyException,
                          api_inst._request, path=path, method=method)