#

  # Written by Matt Prahl <mprahl@redhat.com

  import os

  from datetime import datetime, timedelta

  from mock import patch

  from module_build_service.utils import get_rpm_release, import_mmd, mmd_to_str

  from module_build_service.config import init_config

  from module_build_service.models import (

  )

  from module_build_service import Modulemd

  def make_module(

      nsvc,

      base_module=None,

      filtered_rpms=None,

      xmd=None,

      virtual_streams=None,

      arches=None,

  ):

      Creates new models.ModuleBuild defined by `nsvc` string with requires

      and buildrequires set according to ``requires_list`` and ``build_requires_list``.

      :param str nsvc: name:stream:version:context of a module.

      :param filtered_rpms: list of filtered RPMs which are added to filter

          section in module metadata.

      :type filtered_rpms: list[str]

      :param dict xmd: a mapping representing XMD section in module metadata. A

          custom xmd could be passed for testing a particular scenario and some

          default key/value pairs are added if not present.

      :param bool store_to_db: whether to store created module metadata to the

      :return: New Module Build if set to store module metadata to database,

          otherwise the module metadata is returned.

      :rtype: ModuleBuild or Modulemd.Module

      """

      name, stream, version, context = nsvc.split(":")

      mmd = Modulemd.ModuleStreamV2.new(name, stream)

      mmd.set_version(int(version))

          for rpm in filtered_rpms:

              mmd.add_rpm_filter(rpm)

      # Caller could pass whole xmd including mbs, but if something is missing,

      # default values are given here.

                  module_build.virtual_streams.append(vs_obj)

                  db_session.commit()

          if not arch_obj:

              db_session.add(arch_obj)

              db_session.commit()

      return module_build

  def module_build_from_modulemd(yaml):

      """

      Create a ModuleBuild object and return. It is not written into database.

  import pytest

  from mock import patch, MagicMock

  from module_build_service.builder.KojiModuleBuilder import KojiModuleBuilder

      @patch("tempfile.mkdtemp")

      @patch("module_build_service.builder.KojiModuleBuilder.execute_cmd")

      def _build_srpm(self, db_session, execute_cmd, mkdtemp):

              "{name}:{stream}:{version}:{context}".format(**self.module_nsvc),

          mkdtemp.return_value = self.tmp_srpm_build_dir

          return KojiModuleBuilder.get_disttag_srpm("disttag", module_build)

  from module_build_service.models import ModuleBuild, ComponentBuild

  from module_build_service.builder.MockModuleBuilder import MockModuleBuilder

  from module_build_service.utils import import_fake_base_module, mmd_to_str, load_mmd

  class TestMockModuleBuilder:

          import_fake_base_module(db_session, "platform:f29:1:000000")

          platform = ModuleBuild.get_last_build_in_stream(db_session, "platform", "f29")

          patched_open.side_effect = [

              mock.mock_open(read_data="[fake]\nrepofile 1\n").return_value,

  from module_build_service import conf

  from module_build_service.models import ComponentBuild, ComponentBuildTrace, ModuleBuild

  from module_build_service.utils.general import mmd_to_str, load_mmd

  @pytest.mark.usefixtures("model_tests_init_data")

          """

          clean_database(False)

          builds = ModuleBuild.get_last_builds_in_stream_version_lte(

              db_session, "platform", 290200)

      def test_get_module_count(self, db_session):

          clean_database(False)

          count = ModuleBuild.get_module_count(db_session, name="platform")

          db_session.commit()

      def test_add_virtual_streams_filter(self, db_session):

          clean_database(False)

          query = db_session.query(ModuleBuild).filter_by(name="platform")

          query = ModuleBuild._add_virtual_streams_filter(db_session, query, ["f28", "f29"])

  from conf.config import TestConfiguration

  from six.moves import reload_module

  num_of_metrics = 18

  @mock.patch("module_build_service.monitor.builder_success_counter.inc")

  def test_monitor_state_changing_success(succ_cnt, failed_cnt, db_session):

      conf = mbs_config.Config(TestConfiguration)

      b.transition(db_session, conf, models.BUILD_STATES["wait"])

      b.transition(db_session, conf, models.BUILD_STATES["build"])

      b.transition(db_session, conf, models.BUILD_STATES["done"])

  def test_monitor_state_changing_failure(succ_cnt, failed_cnt, db_session):

      failure_type = "user"

      conf = mbs_config.Config(TestConfiguration)

      b.transition(db_session, conf, models.BUILD_STATES["wait"])

      b.transition(db_session, conf, models.BUILD_STATES["build"])

      b.transition(db_session, conf, models.BUILD_STATES["failed"], failure_type=failure_type)

          import_mmd(db_session, mmd)

          platform_f300103 = db_session.query(ModuleBuild).filter_by(stream="f30.1.3").one()

          build = ModuleBuild(

              name="testmodule",

              stream="master",

                      "version": 1,

                      "context": "c1",

                      "modulemd": mmd_to_str(

                      ),

                  },

                  {

                      "version": 2,

                      "context": "c1",

                      "modulemd": mmd_to_str(

                      ),

                  },

              ],

  from module_build_service.models import ModuleBuild

  from module_build_service.scheduler import default_modules

  from module_build_service.utils.general import load_mmd, mmd_to_str

  @patch("module_build_service.scheduler.default_modules._handle_collisions")

      Test that default modules present in the database are added, and the others are ignored.

      """

      clean_database()

      mmd = load_mmd(read_staged_data("formatted_testmodule.yaml"))

      xmd_brs = mmd.get_xmd()["mbs"]["buildrequires"]

      assert set(xmd_brs.keys()) == {"platform"}

      Test that an exception is raised when the request to get the default modules failed.

      """

      clean_database()

      mmd = load_mmd(read_staged_data("formatted_testmodule.yaml"))

      xmd_brs = mmd.get_xmd()["mbs"]["buildrequires"]

      assert set(xmd_brs.keys()) == {"platform"}

  from module_build_service.scheduler.consumer import MBSConsumer

  from module_build_service.scheduler.handlers.greenwave import get_corresponding_module_build

  from module_build_service.scheduler.handlers.greenwave import decision_update

  class TestGetCorrespondingModuleBuild:

          clean_database()

          # This build should be queried and transformed to ready state

          module_build.transition(

              db_session, conf, BUILD_STATES["done"], "Move to done directly for running test."

          )

  from mock import patch, Mock

  import pytest

  from module_build_service.utils.greenwave import greenwave

  class TestGreenwaveQuery():

          response.status_code = resp_status

          mock_requests.post.return_value = response

          got_response = greenwave.query_decision(fake_build, prod_version="xxxx-8")

          assert got_response == resp_content

          mock_requests.get.return_value = responses[0]

          mock_requests.post.side_effect = responses[1:]

          result = greenwave.check_gating(fake_build)

          assert result == policies_satisfied

  from module_build_service import conf

  from module_build_service.utils import ursine

  class TestFindModuleKojiTags:

          # From the behavior of following code, the reason of the error is

          # mixing use of db.session and make_session, the latter one is called

          # from function ``get_modulemds_from_ursine_content``.

          koji_tag = "tag"  # It's ok to use arbitrary tag name.

          with patch.object(conf, "koji_external_repo_url_prefix", new="http://example.com/"):

          self, find_stream_collision_modules, db_session

      ):

          xmd = {"mbs": {"buildrequires": {"modulea": {"stream": "master"}}}}

          original_xmd = fake_mmd.get_xmd()

          with patch.object(ursine, "log") as log:

                  }

              }

          }

          expected_xmd = fake_mmd.get_xmd()

          get_modulemds_from_ursine_content.return_value = []

                  }

              }

          }

          def mock_get_ursine_modulemds(db_session, koji_tag):

              if koji_tag == "module-rhel-8.0-build":

                  return [

                      # This is the one

                  ]

              if koji_tag == "module-project-1.0-build":

                  return [

                      # Both of them are the collided modules

                  ]

          get_modulemds_from_ursine_content.side_effect = mock_get_ursine_modulemds

      def test_no_collisions_found(self, get_modulemds_from_ursine_content, db_session):

          xmd_mbs_buildrequires = {"modulea": {"stream": "master"}, "moduleb": {"stream": "10"}}

          get_modulemds_from_ursine_content.return_value = [

          ]

          assert [] == ursine.find_stream_collision_modules(

              db_session, xmd_mbs_buildrequires, "koji_tag")

      def test_collision_modules_are_found(self, get_modulemds_from_ursine_content, db_session):

          xmd_mbs_buildrequires = {"modulea": {"stream": "master"}, "moduleb": {"stream": "10"}}

          fake_modules = [

          ]

          get_modulemds_from_ursine_content.return_value = fake_modules

      clean_database,

      init_data,

      scheduler_init_data,

      read_staged_data, staged_data_filename)

  import mock

  import koji

          build adds new MSE build (it means there are new expanded

          buildrequires).

          """

          assert build.state == models.BUILD_STATES["ready"]

          mmd1 = build.mmd()

  import module_build_service.utils

  from module_build_service import Modulemd, models

  from module_build_service.errors import StreamAmbigous

  class TestUtilsModuleStreamExpansion:

          Generates gtk:1, gtk:2, foo:1 and foo:2 modules requiring the

          platform:f28 and platform:f29 modules.

          """

      def test_generate_expanded_mmds_context(self, db_session):

          self._generate_default_modules(db_session)

          mmds = module_build_service.utils.generate_expanded_mmds(

              db_session, module_build.mmd())

          contexts = set([mmd.get_context() for mmd in mmds])

          assert set(["e1e005fb", "ce132a1e"]) == contexts

      @pytest.mark.parametrize(

          [

              (

                  True,

                  set(

                      [

                  set([frozenset(["gtk:1", "platform:f28"]), frozenset(["gtk:2", "platform:f28"])]),

              ),

              (

                  True,

                  set(

                      [

                  ),

              ),

              (

                  False,

                  set([frozenset(["foo:1:0:c2", "gtk:1:0:c2", "platform:f28:0:c10"])]),

                  set([frozenset(["foo:1", "gtk:1", "platform:f28"])]),

              ),

              (

                  False,

                  set([frozenset(["foo:1:0:c2", "gtk:1:0:c2", "platform:f28:0:c10"])]),

                  set([frozenset(["foo:1", "gtk:1", "platform:f28"])]),

              ),

              (

                  True,

                  set([frozenset(["foo:1:0:c2", "gtk:1:0:c2", "platform:f28:0:c10"])]),

                  set([frozenset(["foo:1", "gtk:1", "platform:f28"])]),

              ),

              (

                  False,

                  set([frozenset(["gtk:1:0:c2", "platform:f28:0:c10"])]),

                  set([frozenset(["gtk:1", "platform:f28"])]),

              ),

              (

                  True,

                  set([frozenset(["gtk:1:0:c2", "platform:f28:0:c10"])]),

                  set([frozenset(["gtk:1", "platform:f28"])]),

              ),

              (

                  False,

                  set([frozenset(["app:1:0:c6", "platform:f29:0:c11"])]),

                  set([frozenset(["app:1", "platform:f29"])]),

          ],

      )

      def test_generate_expanded_mmds_buildrequires(

      ):

          self._generate_default_modules(db_session)

          # Check that generate_expanded_mmds raises an exception if stream is ambigous

          # and also that it does not raise an exception otherwise.

          assert buildrequires_per_mmd_buildrequires == expected_buildrequires

      @pytest.mark.parametrize(

          [

              (

                  set([frozenset(["gtk:1"]), frozenset(["gtk:2"])]),

              ),

              (

                  set([frozenset(["gtk:1", "gtk:2"])]),

              ),

              (

                  set([frozenset(["foo:1", "gtk:1"])]),

              ),

              (

                  set([frozenset(["foo:1", "gtk:1"])]),

              ),

              (

                  set([frozenset([])]),

              ),

          ],

      )

          self._generate_default_modules(db_session)

          mmds = module_build_service.utils.generate_expanded_mmds(db_session, module_build.mmd())

          requires_per_mmd = set()

          assert requires_per_mmd == expected

      @pytest.mark.parametrize(

          [

              (

                  [

                      "platform:f29:0:c11",

                      "gtk:2:0:c4",

                  ],

              ),

              (