# the version as used in setup.py

  __version__ = "2.1.1"

  # Flask App

  app = Flask(__name__)

      root_logger.setLevel(logging.DEBUG)

      if app.config['STREAM_LOGGING']:

          stream_handler = logging.StreamHandler()

          stream_handler.setLevel(loglevel)

          stream_handler.setFormatter(formatter)

          app.logger.addHandler(stream_handler)

      if app.config['SYSLOG_LOGGING']:

          syslog_handler = logging.handlers.SysLogHandler(address='/dev/log',

                                                          facility=logging.handlers.SysLogHandler.LOG_LOCAL4)

          syslog_handler.setLevel(loglevel)

          app.logger.addHandler(syslog_handler)

      if app.config['FILE_LOGGING'] and app.config['LOGFILE']:

          file_handler = logging.handlers.RotatingFileHandler(

              app.config['LOGFILE'], maxBytes=500000, backupCount=5)

          file_handler.setLevel(loglevel)

  def upgrade_db(*args):

      alembic_cfg = get_alembic_config()

      al_command.upgrade(alembic_cfg, "head")

      current_rev = context.get_current_revision()

      if not current_rev:

          al_command.stamp(alembic_cfg, "15f5eeb9f635")

      else:

  def initialize_db(destructive):

      alembic_cfg = get_alembic_config()

      if destructive:

          db.drop_all()

      # check whether the table 'group' exists

      insp = reflection.Inspector.from_engine(db.engine)

      table_names = insp.get_table_names()

      if 'testcase' not in table_names and 'Testcase' not in table_names:

          db.create_all()

          al_command.stamp(alembic_cfg, "head")

      # check to see if the db has already been initialized by checking for an

      context = MigrationContext.configure(db.engine.connect())

      current_rev = context.get_current_revision()

      if current_rev:

          upgrade_db(destructive)

      else:

  def mock_data(destructive):

      if destructive or not db.session.query(Testcase).count():

          tc1 = Testcase(ref_url="http://example.com/depcheck", name="depcheck")

          tc2 = Testcase(ref_url="http://example.com/rpmlint", name="rpmlint")

          db.session.commit()

      else:

  def main():

      (options, args) = parser.parse_args()

      if len(args) != 1 or args[0] not in possible_commands:

          print

          sys.exit(1)

      command = {

      }[args[0]]

      if not options.destructive:

      command(options.destructive)

  api = Blueprint('api_v1', __name__)

  @app.errorhandler(400)

  def bad_request(error):

      return jsonify({"message": "Bad request"}), 400

      # Filter by result_data

      if result_data is not None:

              try:

                  key, modifier = key.split(':')

              except ValueError:  # no : in key

      #

      req_args = dict(request.args)  # this is important, do not delete ;)

      extra_data = {k: req_args[k] for k in req_args if k not in args}

          for i, s in enumerate(v):

              extra_data[k][i] = s.split(',')

          # flatten the list

  api = Blueprint('api_v2', __name__)

  @app.errorhandler(400)

  def bad_request(error):

      return jsonify({"message": "Bad request"}), 400

      the value required. Or if the value is not yet in the request-parser (which now

      realistically only applies to the `data.` values in result) it is added.

      """

          if key not in RP:

              app.logger.error("Error in config: REQUIRED_DATA contains unknown endpoint %r.", key)

              continue

      # Filter by result_data

      if result_data is not None:

              try:

                  key, modifier = key.split(':')

              except ValueError:  # no : in key

      # req_args is a dict of lists, where keys are param names and values are param values

      #  the value is a list even if only one param value was specified

          for i, value in enumerate(values):

              results_data[param][i] = value.split(',')

          # flatten the list

          return jsonify({'message': "outcome must be one of %r" % (RESULT_OUTCOME,)}), 400

      if args['data']:

          if invalid_keys:

              app.logger.warning("Colon not allowed in key name: %s", invalid_keys)

              return jsonify({'message': "Colon not allowed in key name: %r" % invalid_keys}), 400

  import numbers

  def non_empty(typ, value, *args, **kwargs):

      if args or kwargs:

      Find previous result with the same testcase, item, type, and arch.

      Return None if no result is found.

      logic as is.

      """

      q = db.session.query(Result).filter(Result.id != result.id)

      """

      Publish a fedmsg on the taskotron topic with Taskotron-compatible structure.

      topic instead.

      """

      prev_result = get_prev_result(result)

      if prev_result is not None and prev_result.outcome == result.outcome:

          # a message for this new result.

          # designed.

          log.debug("Skipping Taskotron message for result %d, outcome has not changed", result.id)

          return

  from datetime import date, datetime

  class DBSerialize(object):

      pass

          if isinstance(value, dict):

              ret = {}

                  ret[k] = self.serialize(v, **kwargs)

              return ret

          # convert iterables to list of serialized stuff

          if hasattr(value, '__iter__'):

              ret = []

          if job_load_results:

              rv['results'] = o.results

      def _serialize_Testcase(self, o, **kwargs):

          rv = dict(

              href=self.get_uri(o)

          )

      def _serialize_Result(self, o, **kwargs):

          result_data = {}

              href=self.get_uri(o),

          )

      def _serialize_ResultData(self, o, **kwargs):

          rv = dict(

              value=o.value,

          )

              href=url_for('api_v2.get_group', group_id=o.uuid, _external=True),

          )

      def _serialize_Testcase(self, o, **kwargs):

          rv = dict(

              href=url_for('api_v2.get_testcase', testcase_name=o.name, _external=True),

          )

      def _serialize_Result(self, o, **kwargs):

          result_data = {}

              href=url_for('api_v2.get_result', result_id=o.id, _external=True),

          )

      def _serialize_ResultData(self, o, **kwargs):

          rv = dict(

              value=o.value,

          )

  import resultsdb.messaging

  import resultsdb.controllers.api_v2 as api_v2

  class AboutTime(object):

              'outcome': self.ref_result_outcome,

              'note': self.ref_result_note,

              'ref_url': self.ref_result_ref_url,

              'href': self.ref_url_prefix + '/results/1',

          }

  import resultsdb.lib.helpers as helpers

  import resultsdb.messaging as messaging

  class MyRequest(object):

          self.rq.url = 'URL'

          monkeypatch.setattr(apiv2, 'request', self.rq)

          assert data == [0]

          assert prev is None

          assert next == 'URL?page=1'

          self.rq.url = 'URL?stuff=some'

          monkeypatch.setattr(apiv2, 'request', self.rq)

          assert data == [0]

          assert prev is None

          assert next == 'URL?stuff=some&page=1'

          self.rq.url = 'URL?page=1&limit=1'

          monkeypatch.setattr(apiv2, 'request', self.rq)

          assert data == [0]

          assert prev == 'URL?page=0&limit=1'

          assert next == 'URL?page=2&limit=1'

          self.rq.url = 'URL?limit=1&page=1'

          monkeypatch.setattr(apiv2, 'request', self.rq)

          assert data == [0]

          assert prev == 'URL?limit=1&page=0'

          assert next == 'URL?limit=1&page=2'

          self.rq.url = 'URL&page=1&limit=1'

          monkeypatch.setattr(apiv2, 'request', self.rq)

          assert data == [0]

          assert prev == 'URL&page=0&limit=1'

          assert next == 'URL&page=2&limit=1'

          self.rq.url = 'URL&limit=1&page=1'

          monkeypatch.setattr(apiv2, 'request', self.rq)

          assert data == [0]

          assert prev == 'URL&limit=1&page=0'

          assert next == 'URL&limit=1&page=2'

              plugin = messaging.load_messaging_plugin('fedmsg', {})

          except KeyError as err:

              if "not found" in err.message:

  This exception can be caused by the fact, that you did not run

  `python setup.py develop` before executing the testsuite.

  If you ran `python setup.py develop` and are still seeing this error, then:

   - you might me missing the 'fedmsg' entrypoint in setup.py

              raise

          assert isinstance(plugin, messaging.FedmsgPlugin), "check whether `fedmsg` entrypoint in setup.py points to resultsdb.messaging:FedmsgPlugin"

  # This is a common file where different test suites/linters can be configured.

  [flake8]

  max-line-length=99

  [pep8]

  python_functions=test should

  python_files=test_* functest_*

  addopts=--functional testing/ --cov resultsdb --cov-report=term-missing