Improve class instantiation and resource deployment.

ci/aws.sh

@@ -263,7 +263,9 @@ tee "${TEMPDIR}/resource-file.json" <<EOF
             "instance_type": "t3.medium",
             "username": "$SSH_USER",
             "name": "civ-pkg-testing-image",
-            "spot_instance": true
+            "custom_vpc_name": "$CLOUDX_AWS_INTERNAL_VPC_NAME",
+            "custom_subnet_name": "$CLOUDX_AWS_INTERNAL_SUBNET_NAME",
+            "custom_security_group_name": "$CLOUDX_AWS_INTERNAL_SECURITY_GROUP_NAME"
         }
     ]
 }

cloud-image-val.py

@@ -69,7 +69,7 @@
     os.environ['PYTHONPATH'] = ':'.join(
         [f'{os.path.dirname(__file__)}', os.environ['PYTHONPATH']])
 
-    config_manager = CIVConfig(args)
+    config_manager = CIVConfig(args.__dict__)
 
     config_manager.update_config()
     config_manager.validate_config()

lib/config_lib.py

@@ -9,11 +9,17 @@ class CIVConfig:
 
     config_file_arg_name = 'config_file'
 
-    def __init__(self, args=None):
-        if args and args.config_file:
-            self.config_path = args.config_file
+    def __init__(self, args_dict: dict = None):
+        if args_dict is None:
+            return
+
+        if self.config_file_arg_name in args_dict and \
+           args_dict[self.config_file_arg_name] is not None:
+            self.config_path = args_dict[self.config_file_arg_name]
+        elif os.path.exists(self.config_path):
+            os.system(f'rm -f {self.config_path}')
 
-        self.command_line_args = args.__dict__
+        self.command_line_args = args_dict
 
     def validate_config(self):
         with open(self.config_path) as config_file:

pytest.ini

@@ -1,5 +1,5 @@
 [pytest]
-render_collapsed = True
+render_collapsed = all
 markers =
     pub: Tests that have to be run on images that are already published into cloud providers marketplaces.
     wait: Number of seconds the test should wait before it starts.

test_suite/conftest.py

@@ -216,14 +216,23 @@ def pytest_html_results_table_row(report, cells):
     cells.insert(3, html.td(getattr(report, 'image', '')))
 
 
+def pytest_html_duration_format(duration):
+    """
+    Format the duration of tests in the HTML report.
+    The duration is given in seconds as a float.
+    You can customize the output format here.
+    """
+    if duration < 1:
+        return f"{duration * 1000:.2f} ms"  # Display in milliseconds
+    else:
+        return f"{duration:.2f} s"  # Display in seconds
+
+
 @pytest.hookimpl(hookwrapper=True)
 def pytest_runtest_makereport(item, call):
     outcome = yield
     report = outcome.get_result()
 
-    # Set the test cases 'Duration' format
-    setattr(report, 'duration_formatter', '%S.%f sec')
-
     # Fill 'Test Case' column
     report.test_case = f'{str(item.parent.name)}::{str(item.function.__name__)}'
 

test_suite/generic/test_generic.py

@@ -491,11 +491,6 @@ def test_subscription_manager_auto(self, host, instance_data):
         BugZilla 7.9: 2077086, 2077085
         """
 
-        expected_config = [
-            'auto_registration = 1',
-            'manage_repos = 0'
-        ]
-
         if instance_data['cloud'] == 'aws':
             region = instance_data['availability_zone'][:-1]
 
@@ -516,10 +511,6 @@ def test_subscription_manager_auto(self, host, instance_data):
                 pytest.skip(f'The {region} AWS region is not supported for auto-registration yet.')
 
         with host.sudo():
-            for config in expected_config:
-                assert config in host.check_output('subscription-manager config'), \
-                    f'Expected "{config}" not found in subscription manager configuration'
-
             assert host.service(
                 'rhsmcertd').is_enabled, 'rhsmcertd service must be enabled'
 

test_suite/package/test_awscli2.py

@@ -1,8 +1,8 @@
 import json
-
 import pytest
 
 from lib import test_lib
+from packaging import version
 
 
 @pytest.mark.package
@@ -32,7 +32,13 @@ def import_aws_credentials(self, host):
                 f'Could not configure temporary AWS credentials. Error: {result.stderr}'
 
     def test_awscli2_version(self, host):
-        expected_version = '2.22.9'
+        expected_version_rhel_9 = '2.15.31'
+        expected_version_rhel_10 = '2.22.9'
+
+        if version.parse(host.system_info.release).major == 10:
+            expected_version = expected_version_rhel_10
+        else:
+            expected_version = expected_version_rhel_9
 
         result = test_lib.print_host_command_output(host,
                                                     'aws --version',

test_suite/rhel_devel/cut/test_cut.py

@@ -1,8 +1,8 @@
 import pytest
 from packaging import version
-from lib import test_lib, console_lib
+from lib import test_lib
 from test_suite.generic.test_generic import TestsSubscriptionManager as sub_man
-from test_suite.rhel_devel.test_cloudx_components import test_cloudx_components
+from test_suite.rhel_devel.test_cloudx_components import test_cloudx_components_aws
 
 """
 CUT (Components Upgrade Testing) refers to the RHEL testing phase
@@ -14,53 +14,59 @@
 @pytest.mark.cut
 @pytest.mark.run_on(['rhel9.6', 'rhel10.0'])
 class TestsRhel96to100:
+    @pytest.mark.order(2)
     def test_components_upgrade(self, host, instance_data):
-        console_lib.print_divider('Testing components before major upgrade...')
-        test_cloudx_components(host)
+        print('Testing components BEFORE major upgrade...')
+        test_cloudx_components_aws()
 
-        console_lib.print_divider('Registering system with subscription-manager...')
+        print('Registering system with subscription-manager...')
         sub_man.test_subscription_manager_auto(self, host, instance_data)
 
-        console_lib.print_divider('Installing leapp package...')
+        print('Installing leapp package...')
         result = test_lib.print_host_command_output(host, 'dnf install leapp-upgrade-el9toel10 -y', capture_result=True)
 
         assert result.succeeded, 'Failed to install leapp-upgrade-el9toel10'
 
-        console_lib.print_divider('Adding RHEL-10 repos...')
+        # compose_url = os.environ['COMPOSE_URL']
+        compose_url = "http://download.eng.pek2.redhat.com/rhel-10/nightly/RHEL-10/latest-RHEL-10.0"
+        basearch = host.system_info.arch
+
+        print('Adding RHEL-10 repos...')
         repo_file_name = '/etc/yum.repos.d/rhel10.repo'
-        rhel_10_repo_file = """
+        rhel_10_repo_file = f"""
 [AppStream10]
 name=AppStream for RHEL-10
-baseurl=$COMPOSE_URL/compose/AppStream/$basearch/os/
+baseurl={compose_url}/compose/AppStream/{basearch}/os/
 enabled=0
 gpgcheck=0
 
 [BaseOS10]
 name=BaseOS for RHEL-10
-baseurl=$COMPOSE_URL/compose/BaseOS/$basearch/os/
+baseurl={compose_url}/compose/BaseOS/{basearch}/os/
 enabled=0
 gpgcheck=0
 """
         test_lib.print_host_command_output(host, f'echo "{rhel_10_repo_file}" > {repo_file_name}')
 
-        console_lib.print_divider('Running leapp upgrade...')
+        print('Running leapp upgrade...')
         result = test_lib.print_host_command_output(
             host,
             'leapp upgrade --no-rhsm --enablerepo AppStream10 --enablerepo BaseOS10',
             capture_result=True)
 
-        console_lib.print_divider('Running leapp upgrade...')
+        print('Running leapp upgrade...')
         if result.failed:
             reapp_report_file = '/var/log/leapp/leapp-report.txt'
             if host.file(reapp_report_file).exists:
                 print('Leapp Report:\n', host.file(reapp_report_file).content_string)
 
             pytest.fail('RHEL major upgrade failed. Please check leapp-report.txt for more details.')
 
-        console_lib.print_divider('Rebooting host...')
+        print('Rebooting host...')
         host = test_lib.reboot_host(host)
 
         assert version.parse(host.system_info.release).major == 10, \
             'Failed to upgrade from RHEL-9.6 to RHEL-10.0 even after reboot.'
 
-        test_cloudx_components(host)
+        print('Testing components AFTER major upgrade...')
+        test_cloudx_components_aws()

test_suite/rhel_devel/test_cloudx_components.py

@@ -1,39 +1,29 @@
 import pytest
-import importlib
-from lib import console_lib
+import json
+import os
+from test_suite.suite_runner import SuiteRunner
 
 
 @pytest.mark.run_on(['all'])
-def test_cloudx_components(host):
-    cloudx_components_test_suites = {
-        'awscli2': 'test_suite.package.test_awscli2',
-        'opentelemetry-collector': 'test_suite.package.otel_package.test_otel'
-    }
+def test_cloudx_components_aws():
+    cloudx_components_test_suites = [
+        "test_suite/package/test_awscli2.py",
+        "test_suite/package/otel_package/test_otel.py",
+    ]
 
-    failures_count = 0
+    # For this test, we asssume one instance has been deployed at a time.
+    with open(os.environ['CIV_INSTANCES_JSON'], 'r') as f:
+        inst = json.load(f)
 
-    for component, module_name in cloudx_components_test_suites.items():
-        try:
-            module = importlib.import_module(module_name)
-            test_classes = [cls for cls in vars(module).values() if isinstance(cls, type) and issubclass(cls, object)]
+    suite_runner = SuiteRunner(cloud_provider='aws',
+                               instances=inst,
+                               ssh_config=os.environ['CIV_SSH_CONFIG_FILE'],
+                               parallel=False,
+                               debug=True)
 
-            for test_class in test_classes:
-                console_lib.print_debug(f'TEST CLASS: {test_class.__name__}')
+    status = suite_runner.run_tests(test_suite_paths=cloudx_components_test_suites,
+                                    output_filepath=os.environ['CIV_OUTPUT_FILE'])
 
-                if hasattr(test_class, "__pytest_mark__"):
-                    # Run the test methods
-                    for attr in dir(test_class):
-                        if attr.startswith("test_"):
-                            test_func = getattr(test_class, attr)
-                            try:
-                                test_func(host)
-                                print(f'{component}::{attr} PASS')
-                            except AssertionError as e:
-                                print(f'{component}::{attr} FAIL - {e}')
-                                failures_count += 1
+    return_code = status >> 8
 
-        except ModuleNotFoundError:
-            print(f'{component} SKIPPED (Module not found)')
-            failures_count += 1
-
-    assert failures_count == 0, 'One or more components failed. Please check the logs.'
+    assert return_code == 0, "One or more components failed."

test_suite/suite_runner.py

@@ -21,8 +21,8 @@ class SuiteRunner:
 
     def __init__(self,
                  cloud_provider,
-                 instances,
-                 ssh_config,
+                 instances: dict,
+                 ssh_config: str,
                  parallel=True,
                  debug=False):
         self.cloud_provider = cloud_provider
@@ -53,8 +53,8 @@ def run_tests(self,
     def compose_pytest_command(self,
                                test_suite_paths,
                                output_filepath,
-                               test_filter,
-                               include_markers):
+                               test_filter=None,
+                               include_markers=None):
         all_hosts = self.get_all_instances_hosts_with_users()
 
         if not test_suite_paths: