#1623 Add test_buildonly
Merged 3 years ago by breilly. Opened 3 years ago by jobrauer.
jobrauer/fm-orchestrator test_buildonly  into  master

@@ -92,3 +92,20 @@ 

  def mbs(test_env):

      """MBS instance session."""

      return utils.MBS(test_env["mbs_api"])

+ 

+ 

+ @pytest.fixture(scope="function")

+ def clone_and_start_build(repo, pkg_util):

+     """Shortcut for tests that need a running build/s. Auto clean-up.

+ 

+     :return: repo and list of submitted builds

+     :rtype utils.Repo, list:

+     """

+     repo.bump()

+     builds = pkg_util.run("--optional", "rebuild_strategy=all")

+     yield repo, builds

+     for build in builds:

+         try:

+             pkg_util.cancel(build)

+         except sh.ErrorReturnCode:

+             pass  # we don't need to bother with clean-up errors

@@ -0,0 +1,39 @@ 

+ def test_buildonly(clone_and_start_build, mbs):

+     """Test the buildonly attribute

+ 

+     Preconditions:

+     * modulemd with at least 1 'buildonly = true' RPM component.

+ 

+     Steps:

+     * Start the  module build and wait for it to finish.

+     * Assert that the expected filter field was created in the module metadata.

+     """

+ 

+     repo, builds = clone_and_start_build

+ 

+     # Check the original modulemd

+     modulemd = repo.modulemd

+     assert not modulemd["data"].get("filter")

+     assert modulemd["data"]["components"].get("rpms")

+     rpms = modulemd["data"]["components"].get("rpms")

+ 

+     build_only_rpms = [k for k, v in rpms.items() if bool(v.get("buildonly"))]

+     other_rpms = [k for k, v in rpms.items() if not bool(v.get("buildonly"))]

+ 

+     assert build_only_rpms, "No RPM has 'buildonly' field set to true"

+ 

+     # Wait until our build is ready

+     build_id = builds[0].id

+     mbs.wait_for_module_build_to_succeed(build_id, timeout=(60 * 60), interval=30)

+ 

+     # assert filter field in final modulemd

+     modulemd = mbs.get_module_build(build_id).get_modulemd()

+     assert modulemd["data"]["filter"], "No filters in the build metadata."

+     assert modulemd["data"]["filter"]["rpms"], "No RPM filters in the build metadata."

+     msg = "Expected RPM '{}' not found in the finished build filters."

+     for rpm in build_only_rpms:

+         assert rpm in modulemd["data"]["filter"]["rpms"], msg.format(rpm)

+ 

+     msg = "Not expected RPM '{}' found in the finished build filters."

+     for rpm in other_rpms:

+         assert rpm not in modulemd["data"]["filter"]["rpms"], msg.format(rpm)

@@ -295,6 +295,15 @@ 

          """Name of the state of this module build"""

          return self.data["state_name"]

  

+     def get(self, key):

+         """Get a value from build data dict

+ 

+         :param str key: key

+         :rtype: str

+         :return: value or None

+         """

+         return self.module_build_data.get(key)

+ 

      def components(self, state=None, batch=None, package=None):

          """Components of this module build, optionally filtered based on properties

  
@@ -538,3 +547,41 @@ 

  

          r.raise_for_status()

          return Build(self._mbs_api, r.json()["id"])

+ 

+     def wait_for_module_build(self, build_data, predicate_func, timeout=60, interval=5):

+         """Wait for module build. Wait until the specified function returns True.

+ 

+         :param int|str build_data: build definition (either id or Build object)

+         :param predicate_func: function(Build) -> bool

+         :param int timeout: timeout in seconds

+         :param int interval: scan interval in seconds

+         """

+         start = time.time()

+ 

+         if type(build_data) is not int:

+             build_id = build_data.id

+         else:

+             build_id = build_data

+ 

+         while (time.time() - start) < timeout:

+             build = self.get_module_build(build_id)

+             if predicate_func(build):

+                 return

+             time.sleep(interval)

+         raise TimeoutError("Wait for build timed out after {}s".format(timeout))

+ 

+     def wait_for_module_build_to_succeed(self, build_data, timeout=60, interval=5):

+         """Wait for module build to be 'ready'.

+ 

+         :param int|str build_data: build definition (either id or Build object)

+         :param int timeout: timeout in seconds

+         :param int interval: scan interval in seconds

+         """

+         def predicate(build):

+             if build.get("state") == 5:

+                 return True

+             elif build.get("state") == 4:

+                 pytest.fail("Module build failed!")

+             else:

+                 return False

+         return self.wait_for_module_build(build_data, predicate, timeout, interval)