less some changes to api_factory

robottelo/host_helpers/api_factory.py

@@ -651,7 +651,6 @@ def wait_for_errata_applicability_task(
         self,
         host_id,
         from_when,
-        look_only=False,
         search_rate=1,
         max_tries=10,
         poll_rate=None,
@@ -661,7 +660,6 @@ def wait_for_errata_applicability_task(
 
         :param int host_id: Content host ID of the host where we are regenerating applicability.
         :param int from_when: Epoch Time (seconds in UTC) to limit number of returned tasks to investigate.
-        :param bool look_only: If True, return an empty list for no found task(s).
         :param int search_rate: Delay between searches.
         :param int max_tries: How many times search should be executed.
         :param int poll_rate: Delay between the end of one task check-up and
@@ -671,22 +669,23 @@ def wait_for_errata_applicability_task(
                 Parameter for ``nailgun.entities.ForemanTask.poll()`` method.
         :return: Relevant errata applicability task.
         :raises: ``AssertionError``. If not tasks were found for given host until timeout.
-                Unless look_only == True; return empty result, without raising ``AssertionError``.
         """
         assert isinstance(host_id, int), 'Param host_id have to be int'
         assert isinstance(from_when, int), 'Param from_when have to be int'
         now = int(time.time())
         assert from_when <= now, 'Param from_when have to be epoch time in the past'
         for _ in range(max_tries):
             now = int(time.time())
-            # Format epoch time as MDY-HMS for search, one second prior margin of safety
-            from_when_timestamp = datetime.fromtimestamp(from_when - 1).strftime(
-                "%m-%d-%Y %H:%M:%S"
+            # Format epoch time for search, one second prior margin of safety
+            timestamp = datetime.fromtimestamp(from_when - 1).strftime("%m-%d-%Y %H:%M:%S")
+            # Long format to match search: ex. 'January 03, 2024 at 03:08:08 PM'
+            long_format = datetime.strptime(timestamp, "%m-%d-%Y %H:%M:%S").strftime(
+                "%B %d, %Y at %I:%M:%S %p"
             )
             search_query = (
                 '( label = Actions::Katello::Applicability::Hosts::BulkGenerate OR label = '
                 'Actions::Katello::Host::UploadPackageProfile ) AND started_at >= "%s"'
-                % str(from_when_timestamp)
+                % str(long_format)
             )
             tasks = self._satellite.api.ForemanTask().search(query={'search': search_query})
             tasks_finished = 0
@@ -707,12 +706,9 @@ def wait_for_errata_applicability_task(
                 break
             time.sleep(search_rate)
         else:
-            if look_only:
-                return []
-            else:
-                raise AssertionError(
-                    f'No task was found using query " {search_query} " for host id: {host_id}'
-                )
+            raise AssertionError(
+                f'No task was found using query " {search_query} " for host id: {host_id}'
+            )
 
     def wait_for_syncplan_tasks(self, repo_backend_id=None, timeout=10, repo_name=None):
         """Search the pulp tasks and identify repositories sync tasks with

tests/foreman/api/test_errata.py

@@ -154,29 +154,40 @@ def _fetch_available_errata(host, expected_amount, timeout=120):
         )
 
 
-def cv_publish_promote(sat, org, cv, lce, needs_publish=True):
+def cv_publish_promote(sat, org, cv, lce=None, needs_publish=True):
     """Publish & promote Content View Version with all content visible in org.
-
+    :param lce: if None, default to 'Library',
+        pass a single instance of lce, or list of instances.
     :param bool needs_publish: if False, skip publish of a new version
     :return dictionary:
         'content-view': instance of updated cv
         'content-view-version': instance of newest cv version
     """
-    # Take a list of lifecycle-environments in order, or just a single instance
-    if not isinstance(lce, list):
-        lce_ids = [lce.id]
-    else:
-        lce_ids = [_lce.id for _lce in lce]
+    # Default to 'Library' lce, if None passed
+    # Take a single instance of lce, or a list
+    # of instances
+    lce_ids = 'Library'
+    if lce is not None:
+        if not isinstance(lce, list):
+            lce_ids = [lce.id]
+        else:
+            lce_ids = [_lce.id for _lce in lce]
 
     if needs_publish is True:
         _publish_and_wait(sat, org, cv)
     # Content-view must have at least one published version
     cv = sat.api.ContentView(id=cv.id).read()
     assert cv.version, f'No version(s) are published to the Content-View: {cv.id}'
-    # Sort Content-view-versions by id, collect latest version.id
+    # Find highest version id, will be the latest
     cvv_id = max(cvv.id for cvv in cv.version)
     # Promote to lifecycle-environment(s)
-    sat.api.ContentViewVersion(id=cvv_id).promote(data={'environment_ids': lce_ids})
+    if lce_ids == 'Library':
+        library_lce = cv.environment[0].read()
+        sat.api.ContentViewVersion(id=cvv_id).promote(
+            data={'environment_ids': library_lce.id, 'force': 'True'}
+        )
+    else:
+        sat.api.ContentViewVersion(id=cvv_id).promote(data={'environment_ids': lce_ids})
     _result = {
         'content-view': sat.api.ContentView(id=cv.id).read(),
         'content-view-version': sat.api.ContentViewVersion(id=cvv_id).read(),
@@ -399,12 +410,15 @@ def test_positive_install_multiple_in_host(
         rhel_contenthost.run(f'yum remove -y {pkg_name}')
         assert rhel_contenthost.run(f'rpm -q {pkg_name}').status == 1
     # Wait for any recalculate task(s), possibly invoked by yum remove,
-    # look_only in case of no task(s), if these packages were already not present.
-    target_sat.api_factory.wait_for_errata_applicability_task(
-        host_id=rhel_contenthost.nailgun_host.id,
-        from_when=epoch_timestamp,
-        look_only=True,
-    )
+    try:
+        target_sat.api_factory.wait_for_errata_applicability_task(
+            host_id=rhel_contenthost.nailgun_host.id,
+            from_when=epoch_timestamp,
+        )
+    except AssertionError:
+        # Packages removal did not trigger errata recalculate task,
+        # packages were already not present, continue
+        pass
     # No applicable errata to start
     assert rhel_contenthost.applicable_errata_count == 0
     # Installing all YUM_9 outdated custom packages
@@ -422,7 +436,8 @@ def test_positive_install_multiple_in_host(
         assert rhel_contenthost.run(f'rpm -q {package}').status == 0
 
     # Some applicable errata now expected for outdated packages
-    assert rhel_contenthost.applicable_errata_count > 0
+    assert (all_app_errata := rhel_contenthost.applicable_errata_count) > 0
+    assert (all_app_pkgs := rhel_contenthost.applicable_package_count) > 0
 
     """Installing all YUM_9 security errata sequentially, if applicable.
        after each install, applicable-errata-count should drop by one,
@@ -506,10 +521,18 @@ def test_positive_install_multiple_in_host(
     """
     assert (
         installed_errata > 0
-    ), f'No applicable or installed erratum from the given list: {FAKE_9_YUM_SECURITY_ERRATUM}.'
-    assert updated_packages > 0, (
-        f'No updated packages from the list: {FAKE_9_YUM_UPDATED_PACKAGES}, '
-        f' were found after attemping install of these erratum: {FAKE_9_YUM_SECURITY_ERRATUM}.'
+    ), f'No applicable errata were found or installed from list: {FAKE_9_YUM_SECURITY_ERRATUM}.'
+    assert (
+        updated_packages > 0
+    ), f'No applicable packages were found or installed from list: {FAKE_9_YUM_UPDATED_PACKAGES}.'
+    assert installed_errata == all_app_errata, (
+        f'Expected to install {all_app_errata} errata from list: {FAKE_9_YUM_SECURITY_ERRATUM},'
+        f' but installed: {installed_errata}.'
+    )
+    assert updated_packages == all_app_pkgs, (
+        f'Expected to install {all_app_pkgs} packages, but installed {updated_packages}.'
+        f' Available packages: {FAKE_9_YUM_UPDATED_PACKAGES},'
+        f' using errata from list: {FAKE_9_YUM_SECURITY_ERRATUM}.'
     )
 
 
@@ -596,13 +619,13 @@ def setup_content_rhel8(
     Using RH SAT-TOOLS RHEL8 for sat-tools, and FAKE_YUM_9 as custom-repo.
     Published to content-view and promoted to lifecycle-environment.
 
-    :return: A dictionary with the satellite instances of
-        :activaton-key:
-        :organization:
-        :content-view:
-        :lifecycle-environment:
-        :rh_repo:
-        :custom_repo:
+    Raises `AssertionError` if one or more of the setup
+        components read are empty.
+
+    :return: if return_result is True:
+        A dictionary (_result) with the satellite instances
+        of activaton-key, organization, content-view,
+        lifecycle-environment, rh_repo, custom_repo.
     """
     org = module_sca_manifest_org
     # Setup Custom and RH repos
@@ -1061,12 +1084,12 @@ def rh_repo_module_manifest(module_sca_manifest_org, module_target_sat):
 
 
 @pytest.fixture(scope='module')
-def rhel8_custom_repo_cv(module_sca_manifest_org, module_target_sat):
+def rhel8_custom_repo_cv(module_entitlement_manifest_org, module_target_sat):
     """Create repo and publish CV so that packages are in Library"""
     return module_target_sat.cli_factory.setup_org_for_a_custom_repo(
         {
             'url': settings.repos.module_stream_1.url,
-            'organization-id': module_sca_manifest_org.id,
+            'organization-id': module_entitlement_manifest_org.id,
         }
     )