[POC] New-Style Upgrade Tests

SharedResource:
- Added the ability to validate the result of a given action function via an action_validator function.
- Made an improvement to exiting under error conditions that improved
  tracking file cleanup.

upgrades/conftest:
- Removed the requirement for all upgrade tests to be marked as pre/post
- Introduced fixtures that coordinate checkout/checkin actions between
  multiple xdist workers.
- Introduced a fixture that performs an upgrade on a target satellite
- Introduced a fixture that is used for two test conversions in
  different modules.

test conversions:
- test_cv_upgrade_scenario and test_scenario_custom_repo_check converted
- pre-upgrade tests are now pre-upgrade fixtures that perform setup and
  yield their data in Box objects instead of saving to disk
- post-upgrade tests can now directly access the setup objects by
  inheriting the pre-upgrade fixture results

settings:
- Added SATELLITE_DEPLOY_WORKFLOW and SATELLITE_UPGRADE_JOB_TEMPLATE to
  upgrade.yaml

conf/upgrade.yaml.template

@@ -5,6 +5,10 @@ UPGRADE:
   TO_VERSION: "6.9"
   # Satellite, Capsule hosts RHEL operating system version.
   OS: "rhel7"
+  # The workflow Broker should use to checkout a to-be-upgraded Satellite
+  SATELLITE_DEPLOY_WORKFLOW: deploy-satellite-upgrade
+  # The job template Broker should use to upgrade a Satellite
+  SATELLITE_UPGRADE_JOB_TEMPLATE: satellite-upgrade
   # Capsule's activation key will only be available when we spawn the VM using upgrade template.
   CAPSULE_AK:
     RHEL6:

robottelo/utils/shared_resource.py

@@ -29,6 +29,10 @@
 from broker.helpers import FileLock
 
 
+class SharedResourceError(Exception):
+    """An exception class for SharedResource errors."""
+
+
 class SharedResource:
     """A class representing a shared resource.
 
@@ -43,19 +47,21 @@ class SharedResource:
         is_recovering (bool): Whether the current instance is recovering from an error or not.
     """
 
-    def __init__(self, resource_name, action, *action_args, **action_kwargs):
+    def __init__(self, resource_name, action, *action_args, action_validator=None, **action_kwargs):
         """Initializes a new instance of the SharedResource class.
 
         Args:
             resource_name (str): The name of the shared resource.
             action (function): The function to be executed when the resource is ready.
             action_args (tuple): The arguments to be passed to the action function.
+            action_validator (function): The function to validate the action results.
             action_kwargs (dict): The keyword arguments to be passed to the action function.
         """
         self.resource_file = Path(f"/tmp/{resource_name}.shared")
         self.lock_file = FileLock(self.resource_file)
         self.id = str(uuid4().fields[-1])
         self.action = action
+        self.action_validator = action_validator
         self.action_is_recoverable = action_kwargs.pop("action_is_recoverable", False)
         self.action_args = action_args
         self.action_kwargs = action_kwargs
@@ -151,6 +157,14 @@ def register(self):
             curr_data["statuses"][self.id] = "pending"
             self.resource_file.write_text(json.dumps(curr_data, indent=4))
 
+    def unregister(self):
+        """Unregisters the current process as a watcher."""
+        with self.lock_file:
+            curr_data = json.loads(self.resource_file.read_text())
+            curr_data["watchers"].remove(self.id)
+            del curr_data["statuses"][self.id]
+            self.resource_file.write_text(json.dumps(curr_data, indent=4))
+
     def ready(self):
         """Marks the current process as ready to perform the action."""
         self._update_status("ready")
@@ -163,10 +177,16 @@ def done(self):
     def act(self):
         """Attempt to perform the action."""
         try:
-            self.action(*self.action_args, **self.action_kwargs)
+            result = self.action(*self.action_args, **self.action_kwargs)
         except Exception as err:
             self._update_main_status("error")
-            raise err
+            raise SharedResourceError("Main worker failed during action") from err
+        # If the action_validator is a callable, use it to validate the result
+        if callable(self.action_validator):
+            if not self.action_validator(result):
+                raise SharedResourceError(
+                    f"Action validation failed for {self.action} with {result=}"
+                )
 
     def wait(self):
         """Top-level wait function, separating behavior between main and non-main watchers."""
@@ -189,11 +209,16 @@ def __exit__(self, exc_type, exc_value, traceback):
             raise exc_value
         if exc_type is None:
             self.done()
+            self.unregister()
             if self.is_main:
                 self._wait_for_status("done")
                 self.resource_file.unlink()
         else:
             self._update_status("error")
             if self.is_main:
-                self._update_main_status("error")
+                if self._check_all_status("error"):
+                    # All have failed, delete the file
+                    self.resource_file.unlink()
+                else:
+                    self._update_main_status("error")
             raise exc_value

tests/robottelo/test_shared_resource.py

@@ -11,6 +11,7 @@ def upgrade_action(*args, **kwargs):
     print(f"Upgrading satellite with {args=} and {kwargs=}")
     time.sleep(1)
     print("Satellite upgraded!")
+    return True
 
 
 def run_resource(resource_name):
@@ -24,7 +25,14 @@ def run_resource(resource_name):
 
 def test_shared_resource():
     """Test the SharedResource class."""
-    with SharedResource("test_resource", upgrade_action, 1, 2, 3, foo="bar") as resource:
+
+    def action_validator(result):
+        return result is True
+
+    shared_args = (1, 2, 3)
+    with SharedResource(
+        "test_resource", upgrade_action, *shared_args, action_validator=action_validator, foo="bar"
+    ) as resource:
         assert Path("/tmp/test_resource.shared").exists()
         assert resource.is_main
         assert not resource.is_recovering

tests/upgrades/conftest.py

@@ -87,10 +87,14 @@ def test_capsule_post_upgrade_skipped(pre_upgrade_data):
 import os
 
 from box import Box
+from broker import Broker
 import pytest
 
+from robottelo.config import settings
+from robottelo.hosts import Satellite
 from robottelo.logging import logger
 from robottelo.utils.decorators.func_locker import lock_function
+from robottelo.utils.shared_resource import SharedResource
 
 pre_upgrade_failed_tests = []
 
@@ -334,17 +338,17 @@ def __initiate(config):
     global POST_UPGRADE
     global PRE_UPGRADE_TESTS_FILE_PATH
     PRE_UPGRADE_TESTS_FILE_PATH = getattr(config.option, PRE_UPGRADE_TESTS_FILE_OPTION)
-    if not [
-        upgrade_mark
-        for upgrade_mark in (PRE_UPGRADE_MARK, POST_UPGRADE_MARK)
-        if upgrade_mark in config.option.markexpr
-    ]:
-        # Raise only if the `tests/upgrades` directory is selected
-        if 'upgrades' in config.args[0]:
-            pytest.fail(
-                f'For upgrade scenarios either {PRE_UPGRADE_MARK} or {POST_UPGRADE_MARK} mark '
-                'must be provided'
-            )
+    # if not [
+    #     upgrade_mark
+    #     for upgrade_mark in (PRE_UPGRADE_MARK, POST_UPGRADE_MARK)
+    #     if upgrade_mark in config.option.markexpr
+    # ]:
+    #     # Raise only if the `tests/upgrades` directory is selected
+    #     if 'upgrades' in config.args[0]:
+    #         pytest.fail(
+    #             f'For upgrade scenarios either {PRE_UPGRADE_MARK} or {POST_UPGRADE_MARK} mark '
+    #             'must be provided'
+    #         )
     if PRE_UPGRADE_MARK in config.option.markexpr:
         pre_upgrade_failed_tests = []
         PRE_UPGRADE = True
@@ -418,8 +422,6 @@ def pytest_collection_modifyitems(items, config):
 class DependentTestFailed(Exception):
     """Raise when the dependent test fails"""
 
-    pass
-
 
 @pytest.fixture(autouse=True)
 def post_upgrade_dependant_fail(request):
@@ -428,3 +430,56 @@ def post_upgrade_dependant_fail(request):
     if fail:
         reason = fail.kwargs.get('reason')
         raise DependentTestFailed(reason)
+
+
+def shared_checkout(shared_name):
+    Satellite(hostname="blank")._swap_nailgun(f"{settings.UPGRADE.FROM_VERSION}.z")
+    bx_inst = Broker(
+        workflow=settings.UPGRADE.SATELLITE_DEPLOY_WORKFLOW,
+        deploy_sat_version=settings.UPGRADE.FROM_VERSION,
+        host_class=Satellite,
+        upgrade_group=f"{shared_name}_shared_checkout",
+    )
+    with SharedResource(
+        resource_name=f"{shared_name}_sat_checkout",
+        action=bx_inst.checkout,
+        action_validator=lambda result: isinstance(result, Satellite),
+    ) as sat_checkout:
+        sat_checkout.ready()
+        sat_instance = bx_inst.from_inventory(
+            filter=f'@inv._broker_args.upgrade_group == "{shared_name}_shared_checkout"'
+        )[0]
+        sat_instance.setup()
+    return sat_instance
+
+
+def shared_checkin(sat_instance):
+    sat_instance.teardown()
+    with SharedResource(
+        resource_name=sat_instance.hostname + "_checkin",
+        action=Broker(hosts=[sat_instance]).checkin,
+    ) as sat_checkin:
+        sat_checkin.ready()
+
+
+@pytest.fixture(scope='session')
+def upgrade_action():
+    def _upgrade_action(target_sat):
+        Broker(
+            job_template=settings.UPGRADE.SATELLITE_UPGRADE_JOB_TEMPLATE,
+            target_vm=target_sat.name,
+            sat_version=settings.UPGRADE.TO_VERSION,
+            tower_inventory=target_sat.tower_inventory,
+        ).execute()
+
+    return _upgrade_action
+
+
+@pytest.fixture
+def content_upgrade_shared_satellite():
+    sat_instance = shared_checkout("content_upgrade")
+    with SharedResource(
+        "content_upgrade_tests", shared_checkin, sat_instance=sat_instance
+    ) as test_duration:
+        yield sat_instance
+        test_duration.ready()