[6.15.z] Add a test case for pulp tasks purge (#15638)

Add a test case for pulp tasks purge

robottelo/hosts.py

@@ -2418,6 +2418,22 @@ def register_contenthost(
             org=org, lce=lce, username=username, password=password, enable_proxy=enable_proxy
         )
 
+    def run_orphan_cleanup(self, smart_proxy_id=None):
+        """Run orphan cleanup task for all or given smart proxy."""
+        timestamp = datetime.utcnow().replace(microsecond=0)
+        rake_command = 'foreman-rake katello:delete_orphaned_content RAILS_ENV=production'
+        if smart_proxy_id:
+            rake_command = f'{rake_command} SMART_PROXY_ID={smart_proxy_id}'
+        self.execute(rake_command)
+        self.wait_for_tasks(
+            search_query=(
+                'label = Actions::Katello::OrphanCleanup::RemoveOrphans'
+                f' and started_at >= "{timestamp}"'
+            ),
+            search_rate=5,
+            max_tries=10,
+        )
+
 
 class SSOHost(Host):
     """Class for RHSSO functions and setup"""

tests/foreman/cli/test_repositories.py

@@ -125,3 +125,46 @@ def test_positive_disable_rh_repo_with_basearch(module_target_sat, module_entitl
         }
     )
     assert 'Repository disabled' in disabled_repo[0]['message']
+
+
+@pytest.mark.parametrize('setting_update', ['completed_pulp_task_protection_days'], indirect=True)
+def test_purge_pulp_tasks(module_target_sat, module_org, module_repository, setting_update):
+    """Verify that orphan cleanup purges the pulp tasks too.
+
+    :id: c605b42c-9547-4444-bdca-06e7138299b5
+
+    :parametrized: yes
+
+    :setup:
+        1. Enabled and synced custom repository to ensure we have some pulp tasks buffered.
+
+    :steps:
+        1. Read the current pulp tasks count, run orphan cleanup and read the tasks count again.
+        2. Set completed_pulp_task_protection_days to zero.
+        3. Read the current pulp tasks count, run orphan cleanup and read the tasks count again.
+
+    :expectedresults:
+        1. For the default protection time the tasks count should increase since the purge task
+           is a pulp task too and no previous task should have been purged.
+        2. For zero protection time the tasks count should decrease since all successfully
+           completed tasks should have been purged.
+
+    :CaseImportance: Medium
+
+    :Verifies: SAT-25155
+
+    :customerscenario: true
+
+    """
+    original_ptc = int(module_target_sat.execute('pulp task list | jq length').stdout)
+    module_target_sat.run_orphan_cleanup(smart_proxy_id=1)
+    new_ptc = int(module_target_sat.execute('pulp task list | jq length').stdout)
+    assert new_ptc > original_ptc, 'Pulp tasks were unexpectedly purged'
+
+    setting_update.value = 0
+    setting_update.update({'value'})
+
+    original_ptc = int(module_target_sat.execute('pulp task list | jq length').stdout)
+    module_target_sat.run_orphan_cleanup(smart_proxy_id=1)
+    new_ptc = int(module_target_sat.execute('pulp task list | jq length').stdout)
+    assert new_ptc < original_ptc, 'Pulp tasks were not purged'