## subunit2sql.db.api

this page last updated: 2019-07-16 22:32:18

# Source code for subunit2sql.db.api

    # Copyright 2014 Hewlett-Packard Development Company, L.P.
    #
    # 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 collections
    import datetime

    from oslo_config import cfg
    from oslo_db.sqlalchemy import session as db_session
    from oslo_db.sqlalchemy import utils as db_utils
    import six
    import sqlalchemy
    from sqlalchemy.engine.url import make_url

    import logging

    from subunit2sql.db import models
    from subunit2sql import exceptions
    from subunit2sql import read_subunit

    CONF = cfg.CONF
    CONF.register_cli_opt(cfg.BoolOpt('verbose', short='v', default=False,
                                      help='Verbose output including logging of '
                                           'SQL statements'))

    DAY_SECONDS = 60 * 60 * 24

    _facades = {}

    def _create_facade_lazily():
        global _facades
        db_url = make_url(CONF.database.connection)
        db_backend = db_url.get_backend_name()
        facade = _facades.get(db_backend)
        if facade is None:
            facade = db_session.EngineFacade(
                CONF.database.connection,
                **dict(six.iteritems(CONF.database)))
            _facades[db_backend] = facade
        return facade

    def get_session(autocommit=True, expire_on_commit=False):
        """Get a new sqlalchemy Session instance

        :param bool autocommit: Enable autocommit mode for the session.
        :param bool expire_on_commit: Expire the session on commit defaults False.
        """
        facade = _create_facade_lazily()
        session = facade.get_session(autocommit=autocommit,
                                     expire_on_commit=expire_on_commit)

        # if --verbose was specified, turn on SQL logging
        # note that this is done after the session has been initialized so that
        # we can override the default sqlalchemy logging
        if CONF.get('verbose', False):
            logging.basicConfig()
            logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

        return session

    def _filter_runs_by_date(query, start_date=None, stop_date=None):
        # Helper to apply a data range filter to a query on Run table
        if isinstance(start_date, str):
            start_date = datetime.datetime.strptime(start_date, '%Y-%m-%d')
        if isinstance(stop_date, str):
            stop_date = datetime.datetime.strptime(stop_date, '%Y-%m-%d')
        if start_date:
            query = query.filter(models.Run.run_at >= start_date)
        if stop_date:
            query = query.filter(models.Run.run_at <= stop_date)
        return query

    def _filter_test_runs_by_date(query, start_date=None, stop_date=None):
        # Helper to apply a data range filter to a query on Run table
        if isinstance(start_date, str):
            start_date = datetime.datetime.strptime(start_date, '%Y-%m-%d')
        if isinstance(stop_date, str):
            stop_date = datetime.datetime.strptime(stop_date, '%Y-%m-%d')
        if start_date:
            query = query.filter(models.TestRun.start_time >= start_date)
        if stop_date:
            query = query.filter(models.TestRun.start_time <= stop_date)
        return query

    def get_engine(use_slave=False):
        """Get a new sqlalchemy engine instance

        :param bool use_slave: If possible, use 'slave' database for this engine

        :return: The engine object for the database connection
        :rtype: sqlalchemy.engine.Engine
        """
        facade = _create_facade_lazily()
        return facade.get_engine(use_slave=use_slave)

    def create_test(test_id, run_count=0, success=0, failure=0, run_time=0.0,
                    session=None):
        """Create a new test record in the database.

        This method is used to add a new test in the database. Tests are used to
        track the run history of a unique test over all runs.

        :param str test_id: test_id identifying the test
        :param int run_count: Total number or runs defaults to 0
        :param int success: Number of successful runs defaults 0
        :param int failure: Number of failed runs defaults to 0
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :return: The test object stored in the DB
        :rtype: subunit2sql.models.Test

        :raises InvalidRunCount: If the run_count doesn't equal the sum of the
                                 successes and failures.
        """
        if run_count != success + failure:
            raise exceptions.InvalidRunCount()
        test = models.Test()
        test.test_id = test_id
        test.run_count = run_count
        test.success = success
        test.failure = failure
        test.run_time = run_time
        session = session or get_session()
        with session.begin():
            session.add(test)
        return test

    def update_test(values, test_id, session=None):
        """Update an individual test with new data.

        This method will take a dictionary of fields to update for a specific test.
        If a field is omitted it will not be changed in the DB.

        :param dict values: Dict of values to update the test with. The key is the
                            column name and the value is the new value to be stored
                            in the DB
        :param str test_id: The uuid of the test to update. (value of the id column
                            for the row to be updated)
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The updated test object stored in the DB
        :rtype: subunit2sql.models.Test
        """
        session = session or get_session()
        with session.begin():
            test = get_test_by_id(test_id, session)
            test.update(values)
        return test

    def create_run(skips=0, fails=0, passes=0, run_time=0, artifacts=None,
                   id=None, session=None, run_at=None):
        """Create a new run record in the database

        :param int skips: Total number of skipped tests defaults to 0
        :param int fails: Total number of failed tests defaults to 0
        :param int passes: Total number of passed tests defaults to 0
        :param float run_time: Total run timed defaults to 0
        :param str artifacts: A link to any artifacts from the test run defaults to
                              None
        :param str id: The run id for the new run, needs to be a unique value
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :param run_at: Optional time at which the run was started. If not specified
                       the time that data is added to the DB will be used instead

        :return: The run object stored in the DB
        :rtype: subunit2sql.models.Run
        """
        run = models.Run()
        if id:
            run.uuid = id
        if run_at:
            run.run_at = run_at
        run.skips = skips
        run.fails = fails
        run.passes = passes
        run.run_time = run_time
        run.artifacts = artifacts
        session = session or get_session()
        with session.begin():
            session.add(run)
        return run

    def update_run(values, run_id, session=None):
        """Update an individual run with new data.

        This method will take a dictionary of fields to update for a specific run.
        If a field is omitted it will not be changed in the DB.

        :param dict values: Dict of values to update the test with. The key is the
                            column name and the value is the new value to be stored
                            in the DB
        :param str run_id: The uuid of the run to update. (value of the id column
                            for the row to be updated)
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The updated run object stored in the DB
        :rtype: subunit2sql.models.Run
        """
        session = session or get_session()
        with session.begin():
            run = get_run_by_id(run_id, session)
            run.update(values)
        return run

    def update_test_run(values, test_run_id, session=None):
        """Update an individual test_run with new data.

        This method will take a dictionary of fields to update for a specific
        test_run. If a field is omitted it will not be changed in the DB.

        :param dict values: Dict of values to update the test with. The key is the
                            column name and the value is the new value to be stored
                            in the DB
        :param str test_run_id: The uuid of the test_run to update. (value of the
                            id column for the row to be updated)
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The updated test_run object stored in the DB
        :rtype: subunit2sql.models.TestRun
        """
        session = session or get_session()
        with session.begin():
            test_run = get_test_run_by_id(test_run_id, session)
            test_run.update(values)
        return test_run

    def add_run_metadata(meta_dict, run_id, session=None):
        """Add a metadata key value pairs for a specific run.

        This method will take a dictionary and store key value pair metadata in the
        DB associated with the specified run.

        :param dict meta_dict: A dictionary which will generate a separate key
                               value pair row associated with the run_id
        :param str run_id: The uuid of the run to update. (value of the id column
                           for the row to be updated)
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of created metadata objects
        :rtype: subunit2sql.models.RunMeta
        """

        session = session or get_session()
        metadata = []
        for key, value in meta_dict.items():
            meta = models.RunMetadata()
            meta.key = key
            meta.value = value
            meta.run_id = run_id
            with session.begin():
                session.add(meta)
            metadata.append(meta)
        return metadata

    def get_run_metadata(run_id, session=None):
        """Return all run metadata objects associated with a given run.

        :param str run_id: The uuid of the run to get all the metadata
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of metadata objects
        :rtype: subunit2sql.models.RunMetadata
        """
        session = session or get_session()
        query = db_utils.model_query(models.RunMetadata, session).join(
            models.Run,
            models.RunMetadata.run_id == models.Run.id).filter(
                models.Run.uuid == run_id)
        return query.all()

    def get_runs_by_key_value(key, value, session=None):
        """Return all run objects associated with a certain key/value metadata pair

        :param key: The key to be matched
        :param value: The value to be matched
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :return list: The list of runs
        :rtype: subunit2sql.models.Run
        """
        session = session or get_session()
        query = db_utils.model_query(models.Run, session=session).join(
            models.RunMetadata,
            models.Run.id == models.RunMetadata.run_id).filter_by(
                key=key, value=value)

        return query.all()

    def create_test_run(test_id, run_id, status, start_time=None,
                        end_time=None, session=None):
        """Create a new test run record in the database

        This method creates a new record in the database

        :param str test_id: UUID for test that was run
        :param str run_id: UUID for run that this was a member of
        :param str status: Status of the test run, normally success, fail, or skip
        :param datetime.Datetime start_time: When the test was started defaults to
                                             None
        :param datetime.Datetime end_time: When the test was finished defaults to
                                           None
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The test_run object stored in the DB
        :rtype: subunit2sql.models.TestRun
        """
        test_run = models.TestRun()
        test_run.test_id = test_id
        test_run.run_id = run_id
        test_run.status = status
        if start_time:
            start_time = start_time.replace(tzinfo=None)
            start_time_microsecond = start_time.microsecond
        else:
            start_time_microsecond = None
        if end_time:
            stop_time = end_time.replace(tzinfo=None)
            stop_time_microsecond = stop_time.microsecond
        else:
            stop_time = None
            stop_time_microsecond = None
        test_run.stop_time = stop_time
        test_run.stop_time_microsecond = stop_time_microsecond
        test_run.start_time = start_time
        test_run.start_time_microsecond = start_time_microsecond
        session = session or get_session()
        with session.begin():
            session.add(test_run)
        return test_run

    def add_test_run_metadata(meta_dict, test_run_id, session=None):
        """Add a metadata key value pairs for a specific run.

        This method will take a dictionary and store key value pair metadata in the
        DB associated with the specified run.

        :param dict meta_dict: A dictionary which will generate a separate key
                               value pair row associated with the test_run_id
        :param str test_run_id: The uuid of the test_run to update. (value of the
                                id column for the row to be updated)
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of created metadata objects
        :rtype: subunit2sql.models.TestRunMeta
        """
        metadata = []
        for key, value in meta_dict.items():
            meta = models.TestRunMetadata()
            meta.key = key
            meta.value = value
            meta.test_run_id = test_run_id
            session = session or get_session()
            with session.begin():
                session.add(meta)
            metadata.append(meta)
        return metadata

    def get_test_run_metadata(test_run_id, session=None):
        """Return all run metadata objects for associated with a given run.

        :param str test_run_id: The uuid of the test_run to get all the metadata
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of created metadata objects
        :rtype: subunit2sql.models.RunMeta
        """
        session = session or get_session()
        query = db_utils.model_query(models.TestRunMetadata, session).filter_by(
            test_run_id=test_run_id)
        return query.all()

    def add_test_metadata(meta_dict, test_id, session=None):
        """Add a metadata key value pairs for a specific test.

        This method will take a dictionary and store key value pair metadata in the
        DB associated with the specified run.

        :param dict meta_dict: A dictionary which will generate a separate key
                               value pair row associated with the test_run_id
        :param str test_id: The uuid of the test to update. (value of the
                            id column for the row to be updated)
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of created metadata objects
        :rtype: subunit2sql.models.TestMeta
        """
        metadata = []
        for key, value in meta_dict.items():
            meta = models.TestMetadata()
            meta.key = key
            meta.value = value
            meta.test_id = test_id
            session = session or get_session()
            with session.begin():
                session.add(meta)
            metadata.append(meta)
        return metadata

    def get_test_metadata(test_id, session=None):
        """Return all test metadata objects for associated with a given test.

        :param str test_id: The uuid of the test to get all the metadata
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of created metadata objects
        :rtype: subunit2sql.models.TestMetadata
        """
        session = session or get_session()
        query = db_utils.model_query(models.TestMetadata, session).filter_by(
            test_id=test_id)
        return query.all()

    def get_all_tests(session=None):
        """Return all tests from the DB.

        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of test objects
        :rtype: subunit2sql.models.Test
        """
        session = session or get_session()
        query = db_utils.model_query(models.Test, session)
        return query.all()

    def _get_test_prefixes_mysql(session):
        query = session.query(
            sqlalchemy.func.substring_index(models.Test.test_id, '.', 1))

        prefixes = set()
        for prefix in query.distinct().all():
            prefix = prefix[0]

            # strip out any wrapped function names, e.g. 'setUpClass (
            if '(' in prefix:
                prefix = prefix.split('(', 1)[1]

            prefixes.add(prefix)

        return list(prefixes)

    def _get_test_prefixes_other(session):
        query = session.query(models.Test.test_id)

        unique = set()
        for test_id in query:
            # get the first '.'-separated token (possibly including 'setUpClass (')
            prefix = test_id[0].split('.', 1)[0]
            if '(' in prefix:
                # strip out the function name and paren, e.g. 'setUpClass(a' -> 'a'
                prefix = prefix.split('(', 1)[1]

            unique.add(prefix)

        return list(unique)

    def get_test_prefixes(session=None):
        """Returns all test prefixes from the DB.

        This returns a list of unique test_id prefixes from the database, defined
        as the first dot-separated token in the test id. Prefixes wrapped in
        function syntax, such as 'setUpClass (a', will have this extra syntax
        stripped out of the returned value, up to and including the '(' character.

        As an example, given an input test with an ID 'prefix.test.Clazz.a_method',
        the derived prefix would be 'prefix'. Given a second test with an ID
        'setUpClass (prefix.test.Clazz)', the derived prefix would also be
        'prefix'. If this function were called on a database containing only these
        tests, a list with only one entry, 'prefix', would be returned.

        Note that this implementation assumes that tests ids are semantically
        separated by a period. If this is not the case (and no period characters
        occur at any position within test ids), the full test id will be considered
        the prefix, and the result of this function will be all unique test ids in
        the database.

        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :return list: A list of all unique prefix strings, with any extraneous
                      details removed, e.g. 'setUpClass ('.
        :rtype: str
        """
        session = session or get_session()

        backend = session.bind.dialect.name
        if backend == 'mysql':
            return _get_test_prefixes_mysql(session)
        else:
            return _get_test_prefixes_other(session)

    def _get_tests_by_prefix_mysql(prefix, session, limit, offset):
        # use mysql's substring_index to pull the prefix out of the full test_id
        func_filter = sqlalchemy.func.substring_index(models.Test.test_id, '.', 1)

        # query for tests against the prefix token, but use an ends-with compare
        # this way, if a test_id has a function call, e.g. 'setUpClass (a.b..c)' we
        # can still match it here
        # (we use an ugly 'like' query here, but this won't be operating on an
        # index regardless)
        query = db_utils.model_query(models.Test, session).filter(
            func_filter.like('%' + prefix)).order_by(models.Test.test_id.asc())

        return query.limit(limit).offset(offset).all()

    def _get_tests_by_prefix_other(prefix, session, limit, offset):
        query = db_utils.model_query(models.Test, session).order_by(
            models.Test.test_id.asc())

        # counter to track progress toward offset
        skipped = 0

        ret = []
        for test in query:
            test_prefix = test.test_id.split('.', 1)[0]
            # compare via endswith to match wrapped test_ids: given
            # 'setUpClass (a.b.c)',  the first token will be 'setUpClass (a',
            # which endswith() will catch
            if test_prefix.endswith(prefix):
                # manually track offset progress since we aren't checking for
                # matches on the database-side
                if offset > 0 and skipped < offset:
                    skipped += 1
                    continue

                ret.append(test)

                if len(ret) >= limit:
                    break

        return ret

    def get_tests_by_prefix(prefix, session=None, limit=100, offset=0):
        """Returns all tests with the given prefix in the DB.

        A test prefix is the first segment of a test_id when split using a period
        ('.'). This function will return a list of tests whose first
        period-separated token ends with the specified prefix. As a side-effect,
        given an input 'a', this will return tests with prefixes 'a', but also
        prefixes wrapped in function syntax, such as 'setUpClass (a'.

        Note that this implementation assumes that tests ids are semantically
        separated by a period. If no period character exists in a test id, its
        prefix will be considered the full test id, and this method may return
        unexpected results.

        :param str prefix: The test prefix to search for
        :param session: Optional session object: if one isn't provided, a new
                        session will be acquired for the duration of this operation
        :param int limit: The maximum number of results to return
        :param int offset: The starting index, for pagination purposes
        :return list: The list of matching test objects, ordered by their test id
        :rtype: subunit2sql.models.Test
        """
        session = session or get_session()

        backend = session.bind.dialect.name
        if backend == 'mysql':
            return _get_tests_by_prefix_mysql(prefix, session, limit, offset)
        else:
            return _get_tests_by_prefix_other(prefix, session, limit, offset)

    def get_all_runs_by_date(start_date=None, stop_date=None, session=None):
        """Return all runs from the DB.

        :param str: Optional start_date, if provided only runs started at or after
                    the start_date will be included in the response
        :param str: Optional end_date, if provided only runs started at or before
                    the end_date will be included in the response
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of run objects
        :rtype: subunit2sql.models.Run
        """

        session = session or get_session()
        query = db_utils.model_query(models.Run, session=session)

        # Process date bounds
        query = _filter_runs_by_date(query, start_date, stop_date)

        return query.all()

    def get_all_runs(session=None):
        """Return all runs from the DB.

        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of run objects
        :rtype: subunit2sql.models.Run
        """

        return get_all_runs_by_date(session=session)

    def get_all_test_runs(session=None):
        """Return all test runs from the DB.

        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of test run objects
        :rtype: subunit2sql.models.TestRun
        """
        session = session or get_session()
        query = db_utils.model_query(models.TestRun, session)
        return query.all()

    def get_latest_run(session=None):
        """Return the most recently created run from the DB.

        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The latest run object
        :rtype: subunit2sql.models.Run
        """
        session = session or get_session()
        query = db_utils.model_query(models.Run, session).order_by(
            models.Run.run_at.desc())
        return query.first()

    def get_failing_from_run(run_id, session=None):
        """Return the set of failing test runs for a give run.

        This method will return all the test run objects that failed during the
        specified run.

        :param str run_id: UUID for the run to find all the failing runs
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of failing test runs for the given run
        :rtype: subunit2sql.models.TestRun
        """
        session = session or get_session()
        query = db_utils.model_query(models.TestRun, session).filter_by(
            run_id=run_id, status='fail')
        return query.all()

    def get_test_by_id(id, session=None):
        """Get an individual test by it's uuid.

        :param str id: The uuid for the test (the id field in the DB)
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The specified test object
        :rtype: subunit2sql.models.Test
        """
        session = session or get_session()
        test = db_utils.model_query(models.Test, session).filter_by(
            id=id).first()
        return test

    def get_test_by_test_id(test_id, session=None):
        """Get an individual test by it's test_id.

        :param str test_id: The id (aka the test name) for the test (the test_id
                            field in the DB)
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The specified test object
        :rtype: subunit2sql.models.Test
        """
        session = session or get_session()
        test = db_utils.model_query(models.Test, session).filter_by(
            test_id=test_id).first()
        return test

    def get_tests_by_test_ids(test_ids, session=None):
        """Get tests that match input test_ids

        :param list test_ids: A list of test_ids (aka the test name) for the test
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :return: A list of the specified test objects
        :rtype: list
        """
        session = session or get_session()
        return db_utils.model_query(models.Test, session).filter(
            models.Test.test_id.in_(test_ids)).all()

    def get_run_id_from_uuid(uuid, session=None):
        """Get the id for a run by it's uuid

        :param str uuid: The uuid for the run
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :return: The id for the run with the provided uuid
        :rtype: int
        """
        session = session or get_session()
        run_id = session.query(models.Run.id).filter(
            models.Run.uuid == uuid).first()[0]
        return run_id

    def get_run_by_id(id, session=None):
        """Get an individual run by it's id.

        :param str id: The id for the run
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The specified run object
        :rtype: subunit2sql.models.Run
        """
        session = session or get_session()
        run = db_utils.model_query(models.Run, session).filter_by(id=id).first()
        return run

    def get_runs_by_ids(ids, session=None):
        """Get a list of runs by their ids.

        :param list id: A list of run ids for the run
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The list of the specified run objects
        :rtype: list
        """
        session = session or get_session()
        run = db_utils.model_query(models.Run, session).filter(
            models.Run.id.in_(ids)).all()
        return run

    def get_test_run_by_id(test_run_id, session=None):
        """Get an individual test run by it's id.

        :param str test_run_id: The id for the test run
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The specified test run object
        :rtype: subunit2sql.models.TestRun
        """
        session = session or get_session()
        test_run = db_utils.model_query(models.TestRun, session=session).filter_by(
            id=test_run_id).first()
        return test_run

    def get_test_runs_by_test_id(test_id, session=None):
        """Get all test runs for a specific test.

        :param str test_id: The test's uuid (the id column in the test table) which
                            to get all test runs for
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of test run objects for the specified test
        :rtype: subunit2sql.models.TestRun
        """
        session = session or get_session()
        test_runs = db_utils.model_query(models.TestRun,
                                         session=session).filter_by(
            test_id=test_id).all()
        return test_runs

    def get_test_runs_by_test_test_id(test_id, start_date=None, stop_date=None,
                                      session=None, key=None, value=None,
                                      most_recent_first=False):
        """Get all test runs for a specific test by the test'stest_id column

        :param str test_id: The test's test_id (the test_id column in the test
                            table) which to get all test runs for
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :param datetime.datetime start_date: The date to use as the start date for
                                             results
        :param datetime.datetime stop_date: The date to use as the cutoff date for
                                            results
        :param str key: An optional key for run metadata to filter the test runs
                        on. Must be specified with a value otherwise it does
                        nothing.
        :param str value: An optional value for run metadata to filter the test
                          runs on. Must be specified with a key otherwise it does
                          nothing.
        :param bool most_recent_first: If true order the results list by date of
                                       test_run start_time in descending order.

        :return list: The list of test run objects for the specified test
        :rtype: subunit2sql.models.TestRun
        """
        session = session or get_session()

        test_runs_query = db_utils.model_query(models.TestRun,
                                               session=session).join(
            models.Test, models.TestRun.test_id == models.Test.id).filter(
                models.Test.test_id == test_id)
        if start_date:
            test_runs_query = test_runs_query.filter(
                models.TestRun.start_time >= start_date)
        if stop_date:
            test_runs_query = test_runs_query.filter(
                models.TestRun.start_time <= stop_date)
        if key and value:
            test_runs_query = test_runs_query.join(
                models.RunMetadata,
                models.TestRun.run_id == models.RunMetadata.run_id).filter(
                    models.RunMetadata.key == key,
                    models.RunMetadata.value == value)
        if most_recent_first:
            test_runs_query = test_runs_query.order_by(
                models.TestRun.start_time.desc())
        test_runs = test_runs_query.all()
        return test_runs

    def get_test_runs_by_run_id(run_id, session=None):
        """Get all test runs for a specific run.

        :param str run_id: The run's uuid (the uuid column in the run table) which
                           to get all test runs for
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of test run objects for the specified test
        :rtype: subunit2sql.models.TestRun
        """
        session = session or get_session()
        test_runs = db_utils.model_query(
            models.TestRun, session=session).join(
                models.Run, models.TestRun.run_id == models.Run.id).filter(
                    models.Run.uuid == run_id).all()
        return test_runs

    def get_test_run_duration(test_run_id, session=None):
        """Get the run duration for a specific test_run.

        :param str test_run_id: The test_run's uuid (the id column in the test_run
                                table) to get the duration of
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: The duration of the test run in secs
        :rtype: float
        """
        session = session or get_session()
        test_run = get_test_run_by_id(test_run_id, session)
        start_time = test_run.start_time
        start_time = start_time.replace(
            microsecond=test_run.start_time_microsecond)
        stop_time = test_run.stop_time
        stop_time = stop_time.replace(microsecond=test_run.stop_time_microsecond)
        return read_subunit.get_duration(test_run.start_time, test_run.stop_time)

    def get_tests_from_run_id(run_id, session=None):
        """Return the all tests for a specific run.

        This method returns a list of all the Test objects that were executed as
        part of a specified run.

        :param str run_id: The run's uuid (the id column in the run table) which to
                           get all tests for
        :param session: Optional session object if one isn't provided a new session

        :return list: The list of test objects for the specified test
        :rtype: subunit2sql.models.Test
        """
        session = session or get_session()
        query = db_utils.model_query(models.Test, session=session).join(
            models.TestRun, models.Test.id == models.TestRun.test_id).filter_by(
                run_id=run_id)
        return query.all()

    def get_tests_run_dicts_from_run_id(run_id, session=None):
        """Returns all the stored data about test runs for a specific run.

        This method returns a dictionary containing all the information stored in
        the database regarding the test_runs. This includes the test_id from the
        tests table, all the stored key value pair metadata from the
        test_run_metadata table, and from the test_runs table the status,
        start_time, and stop_time.

        :param str run_id: The run's uuid (the id column in the run table) which to
                           use to select it's run ids to collect information for.
        :param session: Optional session object if one isn't provided a new session

        :return dict: A dictionary with the test_id from the tests for keys that
                      contains all the stored information about the test_runs.
        """
        session = session or get_session()
        query = db_utils.model_query(models.Test, session=session).join(
            models.TestRun, models.Test.id == models.TestRun.test_id).join(
                models.Run, models.TestRun.run_id == models.Run.id).filter(
                    models.Run.uuid == run_id).outerjoin(
                        models.TestRunMetadata,
                        models.TestRun.id == models.TestRunMetadata.
                        test_run_id).order_by(
                            models.TestRun.start_time,
                            models.TestRun.start_time_microsecond).values(
                                models.Test.test_id,
                                models.TestRun.status,
                                models.TestRun.start_time,
                                models.TestRun.start_time_microsecond,
                                models.TestRun.stop_time,
                                models.TestRun.stop_time_microsecond,
                                models.TestRunMetadata.key,
                                models.TestRunMetadata.value)
        test_runs = collections.OrderedDict()
        for test_run in query:
            if test_run[0] not in test_runs:
                # If there is no start_time set to None
                if test_run[2]:
                    start_time = test_run[2]
                    start_time = start_time.replace(microsecond=test_run[3])
                else:
                    start_time = None
                # If there is no stop_time set to None
                if test_run[4]:
                    stop_time = test_run[4]
                    stop_time = stop_time.replace(microsecond=test_run[5])
                else:
                    stop_time = None
                test_runs[test_run[0]] = {
                    'status': test_run[1],
                    'start_time': start_time,
                    'stop_time': stop_time,
                }
                if test_run[6]:
                    test_runs[test_run[0]]['metadata'] = {test_run[6]: test_run[7]}
            else:
                if test_run[6]:
                    test_runs[test_run[0]]['metadata'][test_run[6]] = test_run[7]
        return test_runs

    def get_test_run_time_series(test_id, session=None):
        """Returns a time series dict of run_times for successes of a single test

        :param str test_id: The test's uuid (the id column in the test table) which
                            will be used to get all the test run times for.
        :param session: Optional session object if one isn't provided a new session

        :return dict: A dictionary with the start times as the keys and the values
                      being the duration of the test that started at that time in
                      sec.
        """
        session = session or get_session()
        query = db_utils.model_query(models.TestRun, session=session).filter_by(
            test_id=test_id).filter_by(status='success').values(
                models.TestRun.start_time, models.TestRun.start_time_microsecond,
                models.TestRun.stop_time, models.TestRun.stop_time_microsecond)
        time_series = {}
        for test_run in query:
            start_time = test_run[0]
            start_time = start_time.replace(microsecond=test_run[1])
            stop_time = test_run[2]
            stop_time = stop_time.replace(microsecond=test_run[3])
            time_series[test_run[0]] = (stop_time - start_time).total_seconds()
        return time_series

    def get_test_run_series(start_date=None, stop_date=None, session=None,
                            key='build_queue', value='gate'):
        """Returns a time series dict of total daily run counts

        :param str start_date: Optional start date to filter results on
        :param str stop_date: Optional stop date to filter results on
        :param session: Optional session object if one isn't provided a new session
        :param str key: Optional run_metadata key to filter the runs used on. Key
                        must be specified with value for filtering to occur.
                        This defaults to 'build_queue' for backwards compatibility
                        with earlier versions. Note, this default will be removed
                        in the future.
        :param str value: Optional run_metadata value to filter the runs used on.
                          Value must be specified with key for filtering to occur.
                          This defaults to 'gate' for backwards
                          compatibility with earlier versions. Note, this default
                          will be removed in the future.
        :return dict: A dictionary with the dates as the keys and the values
                      being the total run count for that day. (The sum of success
                      and failures from all runs that started that day)
        """
        session = session or get_session()
        full_query = db_utils.model_query(models.Run, session=session)
        if key and value:
            full_query = full_query.join(
                models.RunMetadata,
                models.Run.id == models.RunMetadata.run_id).filter_by(
                    key=key, value=value)

        # Process date bounds
        full_query = _filter_runs_by_date(full_query, start_date, stop_date)

        query = full_query.values(models.Run.run_at, models.Run.passes,
                                  models.Run.fails)
        time_series = {}
        for test_run in query:
            start_time = test_run[0]
            # Sum of starts and failures is the count for the run
            local_run_count = test_run[1] + test_run[2]
            if start_time in time_series:
                time_series[start_time] = time_series[start_time] + local_run_count
            else:
                time_series[start_time] = local_run_count
        return time_series

    def get_test_status_time_series(test_id, session=None):
        """Returns a time series dict of test_run statuses of a single test

        :param str test_id: The test's uuid (the id column in the test table) which
                            will be used to get all the test run times for.
        :param session: Optional session object if one isn't provided a new session

        :return dict: A dictionary with the start times as the keys and the values
                      being the status of that test run.
        """
        session = session or get_session()
        query = db_utils.model_query(models.TestRun, session=session).filter_by(
            test_id=test_id).values(
                models.TestRun.start_time, models.TestRun.start_time_microsecond,
                models.TestRun.status)
        status_series = {}
        for test_run in query:
            start_time = test_run[0]
            start_time = start_time.replace(microsecond=test_run[1])
            status = test_run[2]
            status_series[start_time] = status
        return status_series

    def get_recent_successful_runs(num_runs=10, session=None, start_date=None):
        """Return a list of run uuid strings for the most recent successful runs

        :param int num_runs: The number of runs to return in the list
        :param session: Optional session object if one isn't provided a new session
        :param datetime start_date: An optional date to use as the starting point
                                    for getting recent runs. Only runs after this
                                    date will be returned.

        :return list: A list of run uuid strings (the id column in the runs table)
                      for the most recent runs.
        """
        session = session or get_session()
        results = db_utils.model_query(models.Run, session)
        results = _filter_runs_by_date(results, start_date)
        results = results.order_by(
            models.Run.run_at.desc()).filter_by(fails=0).limit(num_runs).all()
        return list(map(lambda x: x.uuid, results))

    def get_recent_failed_runs(num_runs=10, session=None, start_date=None):
        """Return a list of run uuid strings for the most recent failed runs

        :param int num_runs: The number of runs to return in the list
        :param session: Optional session object if one isn't provided a new session
        :param datetime start_date: An optional date to use as the starting point
                                    for getting recent runs. Only runs after this
                                    date will be returned.

        :return list: A list of run uuid strings (the id column in the runs table)
                      for the most recent runs.
        """
        session = session or get_session()
        results = db_utils.model_query(models.Run, session)
        results = _filter_runs_by_date(results, start_date)
        results = results.order_by(
            models.Run.run_at.desc()).filter(
            models.Run.fails > 0).limit(num_runs).all()
        return list(map(lambda x: x.uuid, results))

    def get_recent_runs_by_key_value_metadata(key, value, num_runs=10,
                                              session=None, start_date=None):
        """Get a list of runs for recent runs with a key value metadata pair

        :param int num_runs: The number of runs to return in the list
        :param session: Optional session object if one isn't provided a new session
        :param datetime start_date: An optional date to use as the starting point
                                    for getting recent runs. Only runs after this
                                    date will be returned.

        :return list: A list of run objects for the most recent runs.
        :rtype: subunit2sql.db.models.Run
        """
        session = session or get_session()
        results = db_utils.model_query(models.Run, session).join(
            models.RunMetadata,
            models.Run.id == models.RunMetadata.run_id)
        results = _filter_runs_by_date(results, start_date)
        results = results.filter(
            models.RunMetadata.key == key,
            models.RunMetadata.value == value).order_by(
                models.Run.run_at.desc()).limit(num_runs).all()
        return results

    def delete_old_runs(expire_age=186, session=None):
        """Delete all runs and associated metadata older than the provided age

        :param int expire_age: The number of days into the past to use as the
                               expiration date for deleting the runs
        :param session: Optional session object if one isn't provided a new session
        """
        session = session or get_session()
        expire_date = datetime.date.today() - datetime.timedelta(days=expire_age)

        # Delete the run_metadata
        sub_query = session.query(models.Run.id).filter(
            models.Run.run_at < expire_date).subquery()
        db_utils.model_query(models.RunMetadata, session).filter(
            models.RunMetadata.run_id.in_(sub_query)).delete(
                synchronize_session=False)
        # Delete the runs
        db_utils.model_query(models.Run, session).filter(
            models.Run.run_at < expire_date).delete(synchronize_session=False)

    def delete_old_test_runs(expire_age=186, session=None):
        """Delete all test runs and associated metadata older than the provided age

        :param int expire_age: The number of days into the past to use as the
                               expiration date for deleting the test runs
        :param session: Optional session object if one isn't provided a new session
        """
        session = session or get_session()
        expire_date = datetime.date.today() - datetime.timedelta(days=expire_age)

        # Delete the test run metadata
        sub_query = session.query(models.TestRun.id).filter(
            models.TestRun.start_time < expire_date).subquery()
        db_utils.model_query(models.TestRunMetadata, session).filter(
            models.TestRunMetadata.test_run_id.in_(sub_query)).delete(
                synchronize_session=False)
        # Delete the test runs
        db_utils.model_query(models.TestRun, session).filter(
            models.TestRun.start_time < expire_date).delete(
                synchronize_session=False)

    def delete_run_by_uuid(run_uuid, session=None):
        """Delete a single test run based on its uuid.

        :param str uuid: The uuid of the run that will be deleted.
        :param session: Optional session object. if one isn't provided, a
                        new session will be created.
        """
        session = session or get_session()
        run_id = get_run_id_from_uuid(run_uuid, session)

        # find and delete any runs
        db_utils.model_query(models.Run, session).filter(
            models.Run.uuid == run_uuid).delete(synchronize_session=False)
        # search for any metadata linked to the run uuid
        db_utils.model_query(models.RunMetadata, session).filter(
            models.RunMetadata.run_id == run_id).delete(synchronize_session=False)

    def delete_test_runs_by_run_uuid(run_uuid, session=None):
        """Delete all test runs included in a given run

        :param str uuid: The uuid of the run that will be deleted.
        :param session: Optional session object. if one isn't provided, a
                        new session will be created.
        """
        session = session or get_session()

        # find and delete all runs associated the given run
        test_runs = get_test_runs_by_run_id(run_uuid, session)
        test_run_ids = [x.id for x in test_runs]
        for test_run_id in test_run_ids:
            db_utils.model_query(models.TestRun, session). \
                filter(models.TestRun.id == test_run_id). \
                delete(synchronize_session=False)
            # Find and delete metadata for this test run
            db_utils.model_query(models.TestRunMetadata, session). \
                filter(models.TestRunMetadata.test_run_id == test_run_id). \
                delete(synchronize_session=False)
            # find and delete attachements for this test run
            db_utils.model_query(models.Attachments, session). \
                filter(models.Attachments.test_run_id == test_run_id). \
                delete(synchronize_session=False)

    def get_id_from_test_id(test_id, session=None):
        """Return the id (uuid primary key) for a test given it's test_id value

        :param str test_id: The test_id's string (not UUID) to identify the test
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :return: The id for the specified test
        :rtype: str
        """
        session = session or get_session()
        return db_utils.model_query(models.Test, session).filter_by(
            test_id=test_id).value('id')

    def get_ids_for_all_tests(session=None):
        """Return an iterator of ids (uuid primary key) for all tests in the database

        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :return: The iterator of all ids for tests in the tests table
        :rtype: iterator
        """
        session = session or get_session()
        return db_utils.model_query(models.Test, session).values(models.Test.id)

    def get_run_times_all_test_runs(tests=None, start_date=None, stop_date=None,
                                    session=None):
        """Return the all the individual duration times for each test_run

        This function will return a dictionary where each key is a test_id and
        the value is a list of all the durations for each run of that test

        :param list tests: the list of test_ids to get results for, if none is
                           specified all tests
        :param str start_date: The date to use as the start date for results
        :param str stop_date: The date to use as the cutoff date for results
        :param session: optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return run_times: all the durations for test_runs grouped by test_id
        :rtype: dict
        """
        session = session or get_session()
        run_times_query = db_utils.model_query(models.TestRun, session).filter(
            models.TestRun.status == 'success').join(
                models.Test, models.TestRun.test_id == models.Test.id)
        if tests:
            run_times_query = run_times_query.filter(
                models.Test.test_id.in_(tests))
        if start_date:
            run_times_query = run_times_query.filter(
                models.TestRun.start_time >= start_date)
        if stop_date:
            run_times_query = run_times_query.filter(
                models.TestRun.start_time <= stop_date)
        run_times = run_times_query.values(models.Test.test_id,
                                           models.TestRun.start_time,
                                           models.TestRun.start_time_microsecond,
                                           models.TestRun.stop_time,
                                           models.TestRun.stop_time_microsecond)
        run_times_dict = {}
        for run_time in run_times:
            test_id = run_time[0]
            if run_time[1]:
                start_time = run_time[1]
                start_time = start_time.replace(microsecond=run_time[2])
            else:
                continue
            if run_time[3]:
                stop_time = run_time[3]
                stop_time = stop_time.replace(microsecond=run_time[4])
            duration = read_subunit.get_duration(start_time, stop_time)
            if test_id in run_times_dict:
                run_times_dict[test_id].append(duration)
            else:
                run_times_dict[test_id] = [duration]
        return run_times_dict

    def get_run_times_grouped_by_run_metadata_key(key, start_date=None,
                                                  stop_date=None, session=None,
                                                  match_key=None,
                                                  match_value=None):
        """Return the aggregate run times for all runs grouped by a metadata key

        The results of the output can be limited to runs with a different matching
        key value run_metadata pair using the match_key and match_value parameters.

        :param key: The run_metadata key to use for grouping runs
        :param session: Optional session object if one isn't provided a new session
                            will be acquired for the duration of this operation
        :param str match_key: An optional key as part of a key value run_metadata
                              pair to filter the runs used to. This can not be the
                              same as the key parameter. If match_value is not also
                              specified this does nothing
        :param str match_value: An optional value as part of a key value
                                run_metadata pair to filter the runs used to. If
                                match_key is not also specified this does nothing.

        :return: A dictionary where keys are the value of the provided metadata key
                 and the values are a list of run_times for successful runs with
                 that metadata value
        :rtype: dict
        """
        if key == match_key:
            raise ValueError('match_key cannot have the same value as key')
        session = session or get_session()
        run_times_query = db_utils.model_query(models.Run, session).filter(
            models.Run.fails == 0, models.Run.passes > 0).join(
                models.RunMetadata,
                models.Run.id == models.RunMetadata.run_id).filter(
                    models.RunMetadata.key == key)
        if match_key and match_value:
            subquery = session.query(models.RunMetadata.run_id).filter(
                models.RunMetadata.key == match_key,
                models.RunMetadata.value == match_value).subquery()
            run_times_query = run_times_query.filter(models.Run.id.in_(subquery))

        run_times_query = _filter_runs_by_date(run_times_query, start_date,
                                               stop_date)
        run_times = run_times_query.values(models.Run.run_at, models.Run.run_time,
                                           models.RunMetadata.value)
        result = {}
        for run in run_times:
            if result.get(run.value):
                result[run.value].append(run.run_time)
            else:
                result[run.value] = [run.run_time]
        return result

    def get_run_times_time_series_grouped_by_run_metadata_key(key, start_date=None,
                                                              stop_date=None,
                                                              session=None,
                                                              match_key=None,
                                                              match_value=None):
        """Get a times series dict of aggregate run times grouped by a metadata key

        The results of the output can be limited to runs with a different matching
        key value run_metadata pair using the match_key and match_value parameters.

        :param key: The run_metadata key to use for grouping runs
        :param session: Optional session object if one isn't provided a new session
                            will be acquired for the duration of this operation
        :param str match_key: An optional key as part of a key value run_metadata
                              pair to filter the runs used to. This can not be the
                              same as the key parameter. If match_value is not also
                              specified this does nothing
        :param str match_value: An optional value as part of a key value
                                run_metadata pair to filter the runs used to. If
                                match_key is not also specified this does nothing.

        :return: A dictionary where keys are the time stamp and the value is a dict
                 with the key being the value of the provided metadata key
                 and the values are run times for the successful runs
        :rtype: dict
        """
        if key == match_key:
            raise ValueError('match_key cannot have the same value as key')
        session = session or get_session()
        run_times_query = db_utils.model_query(models.Run, session).filter(
            models.Run.fails == 0, models.Run.passes > 0).join(
                models.RunMetadata,
                models.Run.id == models.RunMetadata.run_id).filter(
                    models.RunMetadata.key == key)
        if match_key and match_value:
            subquery = session.query(models.RunMetadata.run_id).filter(
                models.RunMetadata.key == match_key,
                models.RunMetadata.value == match_value).subquery()
            run_times_query = run_times_query.filter(models.Run.id.in_(subquery))

        run_times_query = _filter_runs_by_date(run_times_query, start_date,
                                               stop_date)
        run_times = run_times_query.values(models.Run.run_at, models.Run.run_time,
                                           models.RunMetadata.value)
        result = {}
        for run in run_times:
            if run.run_at in result:
                if run.value not in result[run.run_at]:
                    result[run.run_at][run.value] = run.run_time
                # NOTE(mtreinish) if there is more than one run with the metadata
                # value and the same start time, just average the results.
                else:
                    old = result[run.run_at][run.value]
                    result[run.run_at][run.value] = float(old + run.run_time) / 2.0
            else:
                result[run.run_at] = {run.value: run.run_time}
        return result

    def get_test_counts_in_date_range(test_id, start_date=None, stop_date=None,
                                      session=None):
        """Return the number of successes, failures, and skips for a single test.

        Optionally you can provide a date to filter the results to be within a
        certain date range

        :param str test_id: The test_id's ID(big integer) to identify the test
        :param datetime start_date: The date to use as the start for counting. A
                                    str in the datetime str format "%b %d %Y" was
                                    the previous format here and will still work
                                    but is deprecated in favor of passing in a
                                    datetime object.
        :param datetime stop_date: The date to use as the cutoff for counting. A
                                   str in the datetime str format "%b %d %Y" was
                                   the previous format here and will still work but
                                   is deprecated in favor of passing in a datetime.
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :return: A dict containing the number of successes, failures, and skips
        :rtype: dict
        """
        if isinstance(start_date, str):
            start_date = datetime.datetime.strptime(start_date, '%b %d %Y')
        if isinstance(stop_date, str):
            stop_date = datetime.datetime.strptime(stop_date, '%b %d %Y')
        session = session or get_session()
        count_dict = {}
        success_query = db_utils.model_query(models.TestRun, session).filter_by(
            test_id=test_id).filter(models.TestRun.status == 'success')
        fail_query = db_utils.model_query(models.TestRun, session).filter_by(
            test_id=test_id).filter(models.TestRun.status == 'fail')
        skip_query = db_utils.model_query(models.TestRun, session).filter_by(
            test_id=test_id).filter(models.TestRun.status == 'skip')

        if start_date:
            success_query = success_query.filter(
                models.TestRun.start_time > start_date)
            fail_query = fail_query.filter(
                models.TestRun.start_time > start_date)
            skip_query = skip_query.filter(
                models.TestRun.start_time > start_date)

        if stop_date:
            success_query = success_query.filter(
                models.TestRun.stop_time < stop_date)
            fail_query = fail_query.filter(
                models.TestRun.stop_time < stop_date)
            skip_query = skip_query.filter(
                models.TestRun.stop_time < stop_date)

        count_dict['success'] = success_query.count()
        count_dict['failure'] = fail_query.count()
        count_dict['skips'] = skip_query.count()
        return count_dict

    def get_failing_test_ids_from_runs_by_key_value(key, value, session=None):
        """Get a list of failing test_ids from runs with run_metadata.

        This method gets a distinct list of test_ids (the test_id column not the id
        column) from all runs that match a run metadata key value pair.

        :param str key: The key to use to match runs from in run_metadata
        :param str value: The value of the key in run_metadata to match runs
                          against
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return: A list of test_ids that failed from runs that match the provided
                 key value run_metadata pair
        :rtype: list
        """

        session = session or get_session()
        test_ids = db_utils.model_query(models.TestRun, session).join(
            models.Test, models.TestRun.test_id == models.Test.id).join(
                models.RunMetadata,
                models.TestRun.run_id == models.RunMetadata.run_id).filter(
                    models.RunMetadata.key == key,
                    models.RunMetadata.value == value,
                    models.TestRun.status == 'fail').values(
                        sqlalchemy.distinct(models.Test.test_id))
        return [test_id[0] for test_id in test_ids]

    def get_test_run_dict_by_run_meta_key_value(key, value, start_date=None,
                                                stop_date=None, session=None):
        """Get a list of test run dicts from runs with a run metadata key value pair

        :param str key: The key to use to match runs from in run_metadata
        :param str value: The value of the key in run_metadata to match runs
                          against
        :param start_date: Optional start date to filter results on
        :param stop_date: Optional stop date to filter results on
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return test_runs: The dictionary of all the tests run on any run that had
                           metadata matching the provided key value pair.
        :rtype: dict
        """
        session = session or get_session()
        query = db_utils.model_query(models.RunMetadata, session).filter(
            models.RunMetadata.key == key,
            models.RunMetadata.value == value).join(
                models.TestRun,
                models.RunMetadata.run_id == models.TestRun.run_id).join(
                    models.Test, models.TestRun.test_id == models.Test.id)
        query = _filter_test_runs_by_date(query, start_date=start_date,
                                          stop_date=stop_date)
        query = query.values(models.Test.test_id,
                             models.TestRun.status,
                             models.TestRun.start_time,
                             models.TestRun.start_time_microsecond,
                             models.TestRun.stop_time,
                             models.TestRun.stop_time_microsecond)
        tests = []
        for test in query:
            if test.start_time:
                start_time = test.start_time
                start_time = start_time.replace(
                    microsecond=test.start_time_microsecond)
            else:
                start_time = None
            if test.stop_time:
                stop_time = test.stop_time
                stop_time = stop_time.replace(
                    microsecond=test.stop_time_microsecond)
            else:
                stop_time = None
            test_run_dict = {
                'test_id': test.test_id,
                'status': test.status,
                'start_time': start_time,
                'stop_time': stop_time,
            }
            tests.append(test_run_dict)
        return tests

    def get_all_runs_time_series_by_key(key, start_date=None,
                                        stop_date=None, session=None):
        """Get a time series of run summaries grouped by a key

        This method will get a time series dictionary of run summary views which
        are grouped by the values of the specified key

        :param str key: The key to use for grouping the run summaries
        :param str start_date: Optional start date to filter results on
        :param str stop_date: Optional stop date to filter results on
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :return runs: A time series dictionary of runs grouped by values of the
                      specified key
        :rtype: dict
        """
        session = session or get_session()
        runs_query = db_utils.model_query(models.Run, session).join(
            models.RunMetadata,
            models.Run.id == models.RunMetadata.run_id).filter(
                models.RunMetadata.key == key)
        runs_query = _filter_runs_by_date(runs_query, start_date, stop_date)
        runs_query = runs_query.values(models.Run.run_at,
                                       models.Run.passes,
                                       models.Run.fails,
                                       models.Run.skips,
                                       models.RunMetadata.value)
        runs = {}
        for run in runs_query:
            if run.run_at not in runs:
                runs[run.run_at] = {run.value: [{
                    'pass': run.passes,
                    'fail': run.fails,
                    'skip': run.skips,
                }]}
            else:
                if run.value not in list(runs[run.run_at].keys()):
                    runs[run.run_at][run.value] = [{
                        'pass': run.passes,
                        'fail': run.fails,
                        'skip': run.skips,
                    }]
                else:
                    runs[run.run_at][run.value].append({
                        'pass': run.passes,
                        'fail': run.fails,
                        'skip': run.skips,
                    })
        return runs

    def get_time_series_runs_by_key_value(key, value, start_date=None,
                                          stop_date=None, session=None):
        """Get a time series of runs with meta for all runs with a key value pai

        :param str key: The metadata key to use for matching the runs
        :param str value: The metadata value to use for matching the runs
        :param start_date: Optional start date to filter results on
        :param str stop_date: Optional stop date to filter results on
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return runs: A time series dictionary (where the top level key is a
                      timestamp) that contains all the runs which
        :rtype: dict
        """
        session = session or get_session()
        sub_query = session.query(models.RunMetadata.run_id).filter(
            models.RunMetadata.key == key,
            models.RunMetadata.value == value).subquery()
        run_query = db_utils.model_query(models.Run, session).join(
            models.RunMetadata,
            models.Run.id == models.RunMetadata.run_id).filter(
                models.Run.id.in_(sub_query))
        run_query = _filter_runs_by_date(run_query, start_date, stop_date)
        run_query = run_query.values(models.Run.uuid,
                                     models.Run.passes,
                                     models.Run.fails,
                                     models.Run.skips,
                                     models.Run.run_time,
                                     models.Run.run_at,
                                     models.RunMetadata.key,
                                     models.RunMetadata.value)
        runs = {}
        for run in run_query:
            run_at = run.run_at
            run_id = run.uuid
            if run_at not in runs:
                # We have hit a new time stamp so we need to add a top level key
                # for the timestamp and populate the run list with a new dict for
                # the run
                runs[run_at] = []
                run_dict = {
                    'id': run_id,
                    'pass': run.passes,
                    'fail': run.fails,
                    'skip': run.skips,
                    'run_time': run.run_time,
                    'metadata': {run.key: run.value}
                }
                runs[run_at].append(run_dict)
            else:
                if run_id not in [loc_run["id"] for loc_run in runs[run_at]]:
                    # We have hit a new run for an existing timestamp, we need to
                    # append a new run dict to the list of runs for that timestamp
                    run_dict = {
                        'id': run_id,
                        'pass': run.passes,
                        'fail': run.fails,
                        'skip': run.skips,
                        'run_time': run.run_time,
                        'metadata': {run.key: run.value}
                    }
                    runs[run_at].append(run_dict)
                else:
                    # The run dictionary has already been added for this timestamp
                    # this means we've hit a new metadata entry, so we need to
                    # update the metadata dictionary with the additional info
                    update_index = None
                    for index, run_dict in list(enumerate(runs[run_at])):
                        if run_dict['id'] == run_id:
                            update_index = index
                    runs[run_at][update_index]['metadata'][run.key] = run.value
        return runs

    def get_run_failure_rate_by_key_value_metadata(key, value, start_date=None,
                                                   stop_date=None, session=None):
        """Return the failure percentage of runs with a set of run metadata

        :param str key: The metadata key to use for matching the runs
        :param str value: The metadata value to use for matching the runs
        :param start_date: Optional start date to filter results on
        :param str stop_date: Optional stop date to filter results on
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return failure_rate: The percentage of runs that failed, will be None if
                              no runs are found
        :rtype: float
        """
        session = session or get_session()
        base_query = db_utils.model_query(models.Run, session).join(
            models.RunMetadata,
            models.Run.id == models.RunMetadata.run_id).filter(
                models.RunMetadata.key == key,
                models.RunMetadata.value == value)
        base_query = _filter_runs_by_date(base_query, start_date, stop_date)
        fails = base_query.filter(models.Run.fails >= 1).count()
        successes = base_query.filter(models.Run.passes > 0,
                                      models.Run.fails == 0).count()
        if fails == 0 and successes == 0:
            return None
        return (float(fails) / float(successes + fails)) * 100

    def add_test_run_attachments(attach_dict, test_run_id, session=None):
        """Add attachments a specific test run.

        This method will take a dictionary and store key blob pair attachments in
        the DB associated with the specified test_run.

        :param dict attachments_dict: A dictionary which will generate a separate
                                      key blob pair row associated with the
                                      test_run_id
        :param str test_run_id: The uuid of the test_run to update. (value of the
                                id column for the row to be updated)
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of created attachment objects
        :rtype: subunit2sql.models.Attachments
        """

        session = session or get_session()
        attachments = []
        for label, attach in attach_dict.items():
            attachment = models.Attachments()
            attachment.label = label
            attachment.attachment = attach
            attachment.test_run_id = test_run_id
            with session.begin():
                session.add(attachment)
            attachments.append(attachment)
        return attachments

    def get_recent_failed_runs_by_run_metadata(key, value, num_runs=10,
                                               start_date=None, session=None):
        """Get a list of recent failed runs for a given run metadata pair

        :param str key: The run_metadata key to get failed runs
        :param str value: The run_metadata value to get failed runs
        :param int num_runs: The number of results to fetch, defaults to 10
        :param datetime start_date: The optional starting dates to get runs from.
                                    Nothing older than this date will be returned
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of recent failed Run objects
        :rtype: subunit2sql.models.Run
        """
        session = session or get_session()
        query = db_utils.model_query(models.Run, session).join(
            models.RunMetadata, models.Run.id == models.RunMetadata.run_id).filter(
                models.RunMetadata.key == key,
                models.RunMetadata.value == value)
        query = _filter_runs_by_date(query, start_date)
        return query.filter(models.Run.fails > 0).order_by(
            models.Run.run_at.desc()).limit(num_runs).all()

    def get_runs_count(session=None):
        """Get the number of runs currently in the database

        :return count: The number of runs in the DB
        :rtype: int
        """
        session = session or get_session()
        query = db_utils.model_query(models.Run, session)
        return query.count()

    def get_runs_counts_by_run_metadata(key, value, start_date=None, session=None):
        """Check runs for a given run metadata pair

        :param str key: The run_metadata key to check runs
        :param str value: The run_metadata value to check runs
        :param datetime start_date: The optional starting dates to get runs from.
                                    Nothing older than this date will be returned
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return count: A count for a given run metadata key-value pair
        :rtype: int
        """
        session = session or get_session()
        query = db_utils.model_query(models.Run, session).join(
            models.RunMetadata, models.Run.id == models.RunMetadata.run_id).filter(
                models.RunMetadata.key == key,
                models.RunMetadata.value == value)
        query = _filter_runs_by_date(query, start_date)
        return query.count()

    def get_runs_by_status_grouped_by_run_metadata(key, start_date=None,
                                                   stop_date=None, session=None):
        session = session or get_session()
        val = models.RunMetadata.value
        run_pass_query = session.query(
            sqlalchemy.func.count(models.Run.id), val).filter(
                models.Run.fails == 0, models.Run.passes > 0).join(
                    models.RunMetadata,
                    models.Run.id == models.RunMetadata.run_id).group_by(
                        val).filter(models.RunMetadata.key == key)
        run_fail_query = session.query(
            sqlalchemy.func.count(models.Run.id), val).filter(
                models.Run.fails > 0, models.Run.passes > 0).join(
                    models.RunMetadata,
                    models.Run.id == models.RunMetadata.run_id).group_by(
                        val).filter(
                            models.RunMetadata.key == key)

        run_pass_query = _filter_runs_by_date(run_pass_query, start_date,
                                              stop_date)
        run_fail_query = _filter_runs_by_date(run_fail_query, start_date,
                                              stop_date)
        rows = run_pass_query.all()
        result = {}
        for row in rows:
            result[row[1]] = {'pass': row[0]}
        rows = run_fail_query.all()
        for row in rows:
            if row[1] in result:
                result[row[1]]['fail'] = row[0]
            else:
                result[row[1]] = {'fail': row[0]}
        return result

    def get_test_runs_by_status_for_run_ids(status, run_ids, key=None,
                                            session=None, include_run_id=False):
        """Get a list of test run dicts by status for all the specified runs

        :param str status: The test status to filter the returned test runs on
        :param list run_ids: A list of run ids (the uuid column from the runs
                             table) to get the test runs from
        :param str key: An optional run_metadata key to add the values for a run
                        to the output dict for each test_run
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation
        :param bool include_run_id: boolean flag to enable including the run uuid
                        in the test run dicts returned

        :return test_runs: A list of dicts for the test_runs and associated data
        :rtype: list
        """
        session = session or get_session()
        query = db_utils.model_query(models.TestRun, session).filter(
            models.TestRun.status == status).join(
                models.Test, models.TestRun.test_id == models.Test.id).join(
                    models.Run, models.TestRun.run_id == models.Run.id).filter(
                        models.Run.uuid.in_(run_ids))

        if key:
            query = query.join(
                models.RunMetadata,
                models.TestRun.run_id == models.RunMetadata.run_id).filter(
                    models.RunMetadata.key == key)
            results = query.values(models.Test.test_id, models.Run.artifacts,
                                   models.TestRun.start_time,
                                   models.TestRun.start_time_microsecond,
                                   models.TestRun.stop_time,
                                   models.TestRun.stop_time_microsecond,
                                   models.RunMetadata.value,
                                   models.Run.uuid)
        else:
            results = query.values(models.Test.test_id, models.Run.artifacts,
                                   models.TestRun.start_time,
                                   models.TestRun.start_time_microsecond,
                                   models.TestRun.stop_time,
                                   models.TestRun.stop_time_microsecond,
                                   models.Run.uuid)
        test_runs = []
        for result in results:
            test_run = {
                'test_id': result.test_id,
                'link': result.artifacts,
                'start_time': result.start_time,
                'stop_time': result.stop_time,
            }
            if include_run_id:
                test_run['uuid'] = result.uuid
            if result.start_time_microsecond is not None:
                test_run['start_time'] = test_run['start_time'].replace(
                    microsecond=result.start_time_microsecond)
            if result.stop_time_microsecond is not None:
                test_run['stop_time'] = test_run['stop_time'].replace(
                    microsecond=result.stop_time_microsecond)
            if hasattr(result, "value"):
                test_run[key] = result.value
            test_runs.append(test_run)
        return test_runs

    def get_all_run_metadata_keys(session=None):
        """Get a list of all the keys used in the run_metadata table

        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return keys: A list of all keys used in the run_metadata table
        :rtype: list
        """
        session = session or get_session()
        keys = session.query(models.RunMetadata.key).distinct().all()
        return [key[0] for key in keys]

    def get_all_test_metadata_keys(session=None):
        """Get a list of all the keys used in the test_metadata table

        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return keys: A list of all keys used in the test_metadata table
        :rtype: list
        """
        session = session or get_session()
        keys = session.query(models.TestMetadata.key).distinct().all()
        return [key[0] for key in keys]

    def get_all_test_run_metadata_keys(session=None):
        """Get a list of all the keys used in the test_run_metadata table

        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return keys: A list of all keys used in the test_run_metadata table
        :rtype: list
        """
        session = session or get_session()
        keys = session.query(models.TestRunMetadata.key).distinct().all()
        return [key[0] for key in keys]

    def get_recent_successful_runs_by_run_metadata(key, value, num_runs=10,
                                                   start_date=None, session=None):
        """Get a list of recent successful runs for a given run metadata pair

        :param str key: The run_metadata key to get successful runs
        :param str value: The run_metadata value to get successful runs
        :param int num_runs: The number of results to fetch, defaults to 10
        :param datetime start_date: The optional starting dates to get runs from.
                                    Nothing older than this date will be returned
        :param session: Optional session object if one isn't provided a new session
                        will be acquired for the duration of this operation

        :return list: The list of recent failed Run objects
        :rtype: subunit2sql.models.Run
        """
        session = session or get_session()
        query = db_utils.model_query(models.Run, session).join(
            models.RunMetadata, models.Run.id == models.RunMetadata.run_id).filter(
                models.RunMetadata.key == key,
                models.RunMetadata.value == value)
        query = _filter_runs_by_date(query, start_date)
        return query.filter(models.Run.passes >= 1).filter(
            models.Run.fails == 0).order_by(
                models.Run.run_at.desc()).limit(num_runs).all()

this page last updated: 2019-07-16 22:32:18

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug questions?