Add test_scratch_final_mmd

tests/integration/example.test.env.yaml

@@ -110,3 +110,8 @@ testdata:
     branch: test-highest-version-selection
     stream_module: testmodule
     test_stream: test-stream-highest-version
+  scratch_final_mmd:
+    # build to be reused - debug only
+    build_id: 3345
+    module: testmodule
+    branch: test-scratch-final-mmd

tests/integration/test_scratch_final_mmd.py

@@ -0,0 +1,38 @@
+
+def test_scratch_final_mmd(scenario, repo, mbs, pkg_util):
+    """Test that scratch builds have properly generated final mmds.
+
+    Steps:
+    * Submit a scratch build and wait for it to be ready.
+    * Query MBS API for final mmds.
+
+    Checks:
+    * That there's a mmd for each arch.
+    * Content of the final mmd is roughly as expected.
+    """
+    build = pkg_util.run("--scratch", "--watch", reuse=scenario.get("build_id"))[0]
+
+    original_mmd = repo.modulemd
+    expected_arches = set(build.module_build_data["arches"])
+    expected_rpms = set(original_mmd["data"]["components"]["rpms"].keys())
+
+    # Get the actual final mmds
+    mmds = mbs.get_final_mmds(build)
+
+    assert expected_arches == set(mmds.keys())
+
+    for arch in expected_arches:
+        mmd = mmds.get(arch)["data"]
+        assert arch == mmd["arch"]
+        assert expected_rpms == set(mmd["components"]["rpms"].keys())
+        assert build.module_build_data["context"] == mmd["context"]
+
+        # assert that the submitted rpms are present in the artifacts list
+        actual_artifact_rpms = set(mmd["artifacts"]["rpms"])
+        for rpm in expected_rpms:
+            actual_rpm = build.components(package=rpm)
+            assert actual_rpm
+            n, v, r = actual_rpm[0]["nvr"].split("-")
+            artifact_name = "{}-0:{}-{}.{}"
+            artifacts = {artifact_name.format(n, v, r, a) for a in [arch, "src"]}
+            assert artifacts.issubset(actual_artifact_rpms)

tests/integration/utils.py

@@ -197,7 +197,7 @@ class PackagingUtility:
             instead of triggering and waiting for new one(s) to finish.
             Intended to be used while developing the tests.
         :return: list of Build objects for the MBS builds created
-        :rtype: list of Build objects
+        :rtype: list[Build]
         """
         build_ids = []
 
@@ -321,7 +321,7 @@ class Build:
         :param int batch: the number of the batch the components should be in
         :param string package: name of the component (package)
         :return: List of filtered components
-        :rtype: list of dict
+        :rtype: list[dict]
         """
         filtered = self.component_data["items"]
         if batch is not None:
@@ -484,6 +484,15 @@ class MBS:
     def __init__(self, mbs_api):
         self._mbs_api = mbs_api
 
+    @staticmethod
+    def _get_build_id(build_data):
+        if type(build_data) is Build:
+            return build_data.id
+        elif type(build_data) is int:
+            return build_data
+        else:
+            raise TypeError
+
     def get_builds(self, module, stream, order_desc_by=None):
         """Get list of Builds objects via mbs api.
 
@@ -543,13 +552,14 @@ class MBS:
         r.raise_for_status()
         return [Build(self._mbs_api, build["id"]) for build in r.json()["items"]]
 
-    def get_module_build(self, build_id, **kwargs):
+    def get_module_build(self, build_data, **kwargs):
         """Query MBS API on module-builds endpoint for a specific build
 
-        :attribute build_id (int): build ID
+        :attribute build_data (int|Build): build ID
         :return: module build object
         :rtype: Build
         """
+        build_id = self._get_build_id(build_data)
         url = f"{self._mbs_api}module-builds/{build_id}"
         r = requests.get(url, params=kwargs)
 
@@ -565,11 +575,7 @@ class MBS:
         :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
+        build_id = self._get_build_id(build_data)
 
         while (time.time() - start) < timeout:
             build = self.get_module_build(build_id)
@@ -593,3 +599,21 @@ class MBS:
             else:
                 return False
         return self.wait_for_module_build(build_data, predicate, timeout, interval)
+
+    def get_final_mmds(self, build_data):
+        """Get build's final mmds.
+
+        :param int|Build build_data: build obj or ID
+        :return: dict of parsed mmds (for each arch available)
+        :rtype: dict
+        """
+        build_id = self._get_build_id(build_data)
+
+        url = f"{self._mbs_api}final-modulemd/{build_id}"
+        r = requests.get(url)
+        r.raise_for_status()
+
+        mmds = {}
+        for arch, mmd in r.json().items():
+            mmds[arch] = yaml.safe_load(mmd)
+        return mmds