[6.14.z] CP of API-Errata modifications

pytest_fixtures/component/repository.py

@@ -245,3 +245,25 @@ def module_repos_collection_with_manifest(
     )
     _repos_collection.setup_content(module_entitlement_manifest_org.id, module_lce.id)
     return _repos_collection
+
+
+@pytest.fixture
+def function_repos_collection_with_manifest(
+    request, target_sat, function_sca_manifest_org, function_lce
+):
+    """This fixture and its usage is very similar to repos_collection fixture above with extra
+    setup_content and uploaded manifest capabilities using function_lce and
+    function_sca_manifest_org fixtures
+    """
+    repos = getattr(request, 'param', [])
+    repo_distro, repos = _simplify_repos(request, repos)
+    _repos_collection = target_sat.cli_factory.RepositoryCollection(
+        distro=repo_distro,
+        repositories=[
+            getattr(target_sat.cli_factory, repo_name)(**repo_params)
+            for repo in repos
+            for repo_name, repo_params in repo.items()
+        ],
+    )
+    _repos_collection.setup_content(function_sca_manifest_org.id, function_lce.id)
+    return _repos_collection

robottelo/constants/__init__.py

@@ -752,6 +752,9 @@ class Colored(Box):
 REAL_RHEL7_0_1_PACKAGE_FILENAME = 'python-pulp-common-2.21.0.2-1.el7sat.noarch.rpm'
 REAL_RHEL7_0_2_PACKAGE_NAME = 'python2-psutil'  # for RHBA-2021:1314
 REAL_RHEL7_0_2_PACKAGE_FILENAME = 'python2-psutil-5.7.2-2.el7sat.x86_64.rpm'
+REAL_RHEL8_1_PACKAGE_NAME = 'puppet-agent'  # for RHSA-2022:4867
+REAL_RHEL8_1_PACKAGE_FILENAME = 'puppet-agent-6.19.1-1.el8sat.x86_64'
+REAL_RHEL8_2_PACKAGE_FILENAME = 'puppet-agent-6.26.0-1.el8sat.x86_64'
 FAKE_0_CUSTOM_PACKAGE_GROUP_NAME = 'birds'
 FAKE_3_YUM_OUTDATED_PACKAGES = [
     'acme-package-1.0.1-1.noarch',
@@ -806,6 +809,7 @@ class Colored(Box):
 FAKE_2_ERRATA_ID = 'RHSA-2012:0055'  # for FAKE_1_CUSTOM_PACKAGE
 REAL_RHEL7_0_ERRATA_ID = 'RHBA-2020:3615'  # for REAL_RHEL7_0_0_PACKAGE
 REAL_RHEL7_1_ERRATA_ID = 'RHBA-2017:0395'  # tcsh bug fix update
+REAL_RHEL8_1_ERRATA_ID = 'RHSA-2022:4867'  # for REAL_RHEL8_1_PACKAGE
 FAKE_1_YUM_REPOS_COUNT = 32
 FAKE_3_YUM_REPOS_COUNT = 78
 FAKE_9_YUM_SECURITY_ERRATUM = [

robottelo/host_helpers/api_factory.py

@@ -3,6 +3,7 @@
 example: my_satellite.api_factory.api_method()
 """
 from contextlib import contextmanager
+from datetime import datetime
 import time
 
 from fauxfactory import gen_ipaddr, gen_mac, gen_string
@@ -647,12 +648,18 @@ def attach_custom_product_subscription(self, prod_name=None, host_name=None):
         )
 
     def wait_for_errata_applicability_task(
-        self, host_id, from_when, search_rate=1, max_tries=10, poll_rate=None, poll_timeout=15
+        self,
+        host_id,
+        from_when,
+        search_rate=1,
+        max_tries=10,
+        poll_rate=None,
+        poll_timeout=15,
     ):
         """Search the generate applicability task for given host and make sure it finishes
 
         :param int host_id: Content host ID of the host where we are regenerating applicability.
-        :param int from_when: Timestamp (in UTC) to limit number of returned tasks to investigate.
+        :param int from_when: Epoch Time (seconds in UTC) to limit number of returned tasks to investigate.
         :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
@@ -666,23 +673,30 @@ def wait_for_errata_applicability_task(
         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 timestamp in the past'
+        assert from_when <= now, 'Param from_when have to be epoch time in the past'
         for _ in range(max_tries):
             now = int(time.time())
-            max_age = now - from_when + 1
+            # 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::Host::GenerateApplicability OR label = '
-                'Actions::Katello::Host::UploadPackageProfile ) AND started_at > "%s seconds ago"'
-                % max_age
+                '( label = Actions::Katello::Applicability::Hosts::BulkGenerate OR'
+                ' label = Actions::Katello::Host::UploadPackageProfile ) AND'
+                f' started_at >= "{long_format}" '
             )
             tasks = self._satellite.api.ForemanTask().search(query={'search': search_query})
             tasks_finished = 0
             for task in tasks:
                 if (
-                    task.label == 'Actions::Katello::Host::GenerateApplicability'
+                    task.label == 'Actions::Katello::Applicability::Hosts::BulkGenerate'
+                    and 'host_ids' in task.input
                     and host_id in task.input['host_ids']
                 ) or (
                     task.label == 'Actions::Katello::Host::UploadPackageProfile'
+                    and 'host' in task.input
                     and host_id == task.input['host']['id']
                 ):
                     task.poll(poll_rate=poll_rate, timeout=poll_timeout)
@@ -692,7 +706,7 @@ def wait_for_errata_applicability_task(
             time.sleep(search_rate)
         else:
             raise AssertionError(
-                f"No task was found using query '{search_query}' for host '{host_id}'"
+                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):

robottelo/host_helpers/contenthost_mixins.py

@@ -136,6 +136,12 @@ def applicable_errata_count(self):
         """return the applicable errata count for a host"""
         return self.nailgun_host.read().content_facet_attributes['errata_counts']['total']
 
+    @property
+    def applicable_package_count(self):
+        """return the applicable package count for a host"""
+        self.run('subscription-manager repos')
+        return self.nailgun_host.read().content_facet_attributes['applicable_package_count']
+
 
 class SystemFacts:
     """Helpers mixin that enables getting/setting subscription-manager facts on a host"""