How to use the rally.consts function in rally

#
#    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.

from rally.common import db
from rally import consts
from rally import exceptions
from rally.verification import manager


class Verifier(object):
    """Represents a verifier object."""
    TIME_FORMAT = consts.TimeFormat.ISO8601

    def __init__(self, verifier):
        """Init a verifier object.

        :param verifier: Dict representation of a verifier in the database
        """
        self._db_entry = verifier

        self._deployment = None
        self._manager = None

    def __getattr__(self, attr):
        return self._db_entry[attr]

    def __getitem__(self, item):
        return self._db_entry[item]

#    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.

from rally import consts
from rally.plugins.openstack import scenario
from rally.plugins.openstack.scenarios.ceilometer import utils as ceiloutils
from rally.task import validation


"""Scenarios for Ceilometer Meters API."""


@validation.add("required_services",
                services=[consts.Service.CEILOMETER])
@validation.add("required_contexts", contexts=("ceilometer"))
@validation.add("required_platform", platform="openstack", users=True)
@scenario.configure(name="CeilometerMeters.list_meters", platform="openstack")
class ListMeters(ceiloutils.CeilometerScenario):

    def run(self, metadata_query=None, limit=None):
        """Check all available queries for list resource request.

        :param metadata_query: dict with metadata fields and values
        :param limit: limit of meters in response
        """

        scenario = ListMatchedMeters(self.context)
        scenario.run(filter_by_project_id=True)
        scenario.run(filter_by_user_id=True)
        scenario.run(filter_by_resource_id=True)

def upgrade():
    # Workaround for Alemic bug #89
    # https://bitbucket.org/zzzeek/alembic/issue/89

    with op.batch_alter_table("tasks") as batch_op:
        batch_op.add_column(sa.Column("new_status", sa.String(36),
                                      default=consts.TaskStatus.INIT))
    with op.batch_alter_table("subtasks") as batch_op:
        batch_op.add_column(sa.Column("new_status", sa.String(36),
                                      default=consts.SubtaskStatus.RUNNING))

    op.execute(
        task.update()
            .where(task.c.status.in_(WITHOUT_CHANGES))
            .values({"new_status": task.c.status}))

    for old, new in OLD_TO_NEW:
        op.execute(
            task.update()
                .where(task.c.status == op.inline_literal(old))
                .values({"new_status": new}))

    # NOTE(rvasilets): Assume that set_failed was used only in causes of

"additive": [$(additive_dd)],
  "complete": [$(complete_load)]
}
EOF
"""


@types.convert(image={"type": "glance_image"},
               flavor={"type": "nova_flavor"})
@validation.add("image_valid_on_flavor", flavor_param="flavor",
                image_param="image")
@validation.add("number", param_name="port", minval=1, maxval=65535,
                nullable=True, integer_only=True)
@validation.add("external_network_exists", param_name="floating_network")
@validation.add("required_services", services=[consts.Service.NOVA,
                                               consts.Service.CINDER])
@validation.add("required_platform", platform="openstack", users=True)
@scenario.configure(context={"cleanup@openstack": ["nova", "cinder"],
                             "keypair@openstack": {},
                             "allow_ssh@openstack": None},
                    name="VMTasks.dd_load_test",
                    platform="openstack")
class DDLoadTest(BootRuncommandDelete):
    @logging.log_deprecated_args(
        "Use 'interpreter' to specify the interpreter to execute script from.",
        "0.10.0", ["command"], once=True)
    def run(self, flavor, username, password=None,
            image=None, command=None, interpreter="/bin/sh",
            volume_args=None, floating_network=None, port=22,
            use_floating_ip=True, force_delete=False, wait_for_ping=True,
            max_log_length=None, **kwargs):
        """Boot a server from a custom image and performs dd load test.

class ConstantForDurationScenarioRunner(runner.ScenarioRunner):
    """Creates constant load executing a scenario for an interval of time.

    This runner will place a constant load on the cloud under test by
    executing each scenario iteration without pausing between iterations
    until a specified interval of time has elapsed.

    The concurrency parameter of the scenario config controls the
    number of concurrent iterations which execute during a single
    sceanario in order to simulate the activities of multiple users
    placing load on the cloud under test.
    """

    CONFIG_SCHEMA = {
        "type": "object",
        "$schema": consts.JSON_SCHEMA,
        "properties": {
            "concurrency": {
                "type": "integer",
                "minimum": 1,
                "description": "The number of parallel iteration executions."
            },
            "duration": {
                "type": "number",
                "minimum": 0.0,
                "description": "The number of seconds during which to generate"
                               " a load."
            },
            "timeout": {
                "type": "number",
                "minimum": 1,
                "description": "Operation's timeout."

from rally.plugins.openstack.cleanup import manager as resource_manager
from rally.plugins.openstack.scenarios.ec2 import utils as ec2_utils
from rally.plugins.openstack import types
from rally.task import context


LOG = logging.getLogger(__name__)


@context.configure(name="ec2_servers", platform="openstack", order=460)
class EC2ServerGenerator(context.Context):
    """Creates specified amount of nova servers in each tenant uses ec2 API."""

    CONFIG_SCHEMA = {
        "type": "object",
        "$schema": consts.JSON_SCHEMA,
        "properties": {
            "image": {
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string"
                    }
                },
                "additionalProperties": False
            },
            "flavor": {
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string"
                    }

#
#         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.

from rally.common import db
from rally import consts


class Verification(object):
    """Represents a verification object."""
    TIME_FORMAT = consts.TimeFormat.ISO8601

    def __init__(self, verification):
        """Init a verification object.

        :param verification: Dict representation of a verification
                             in the database
        """
        self._db_entry = verification
        self._db_entry["deployment_uuid"] = self._db_entry["env_uuid"]

    def __getattr__(self, attr):
        return self._db_entry[attr]

    def __getitem__(self, item):
        return self._db_entry[item]

#
# 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.

from rally import consts
from rally.plugins.openstack import scenario
from rally.task import atomic
from rally.task import validation


@validation.add("required_services", services=[consts.Service.NOVA])
@validation.add("required_platform", platform="openstack", users=True)
@scenario.configure(name="ScenarioPlugin.list_flavors")
class ListFlavors(scenario.OpenStackScenario):

    @atomic.action_timer("list_flavors")
    def _list_flavors(self):
        """Sample of usage clients - list flavors

        You can use self.context, self.admin_clients and self.clients which are
        initialized on scenario instance creation.
        """
        self.clients("nova").flavors.list()

    @atomic.action_timer("list_flavors_as_admin")
    def _list_flavors_as_admin(self):
        """The same with admin clients."""

def discover_impl(self):
        impl_cls, impls = super(UnifiedOpenStackService, self).discover_impl()
        if not impl_cls:
            # Nova-network is not listed in keystone catalog and we can not
            # assume that it is enabled if neutron is missed. Since such
            # discovery needs an external call, it is done only if needed.
            for impl in impls:
                o = impl._meta_get("impl")
                if (o._meta_get("name") == consts.Service.NOVA_NET and
                        impl.is_applicable(self._clients)):
                    return impl, impls
        return impl_cls, impls

#    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.

from rally import consts
from rally.plugins.openstack import scenario
from rally.plugins.openstack.scenarios.nova import utils
from rally.task import validation


"""Scenarios for create nova floating IP by range."""


@validation.restricted_parameters("pool")
@validation.required_services(consts.Service.NOVA, consts.Service.NOVA_NET)
@validation.add("required_platform", platform="openstack", admin=True)
@scenario.configure(context={"admin_cleanup": ["nova"]},
                    name=("NovaFloatingIpsBulk"
                          ".create_and_list_floating_ips_bulk"))
class CreateAndListFloatingIpsBulk(utils.NovaScenario):

    def run(self, start_cidr, **kwargs):
        """Create nova floating IP by range and list it.

        This scenario creates a floating IP by range and then lists all.

        :param start_cidr: Floating IP range
        :param kwargs: Optional additional arguments for range IP creation
        """

        ips_bulk = self._create_floating_ips_bulk(start_cidr, **kwargs)