Sanity Testing fixes for Ipv6

conf/dynaconf_hooks.py

@@ -7,7 +7,7 @@
 
 from robottelo.logging import logger
 from robottelo.utils.ohsnap import dogfood_repository
-from robottelo.utils.url import is_url
+from robottelo.utils.url import ipv6_hostname_translation, is_url
 
 
 def post(settings):
@@ -26,6 +26,7 @@ def post(settings):
             )
             data = get_repos_config(settings)
             write_cache(settings_cache_path, data)
+    ipv6_hostname_translation(settings, data)
     config_migrations(settings, data)
     data['dynaconf_merge'] = True
     return data

pytest_fixtures/core/broker.py

@@ -13,10 +13,7 @@ def _default_sat(align_to_satellite):
     """Returns a Satellite object for settings.server.hostname"""
     if settings.server.hostname:
         try:
-            sat = Satellite.get_host_by_hostname(settings.server.hostname)
-            http_proxy = sat.enable_ipv6_http_proxy()
-            yield sat
-            sat.disable_ipv6_http_proxy(http_proxy)
+            return Satellite.get_host_by_hostname(settings.server.hostname)
         except ContentHostError:
             return Satellite()
     return None
@@ -37,7 +34,9 @@ def _target_sat_imp(request, _default_sat, satellite_factory):
         settings.set('server.hostname', installer_sat.hostname)
         yield installer_sat
     else:
+        _default_sat.enable_ipv6_http_proxy()
         yield _default_sat
+        _default_sat.disable_ipv6_http_proxy()
 
 
 @pytest.fixture

pytest_fixtures/core/sat_cap_factory.py

@@ -365,9 +365,8 @@ def installer_satellite(request):
     else:
         sat = lru_sat_ready_rhel(getattr(request, 'param', None))
     sat.setup_firewall()
-    http_proxy = sat.enable_ipv6_http_proxy()
     # # Register for RHEL8 repos, get Ohsnap repofile, and enable and download satellite
-    sat.register_to_cdn()
+    sat.register_to_cdn(enable_proxy=True)
     sat.download_repofile(
         product='satellite',
         release=settings.server.version.release,
@@ -386,14 +385,12 @@ def installer_satellite(request):
         ).get_command(),
         timeout='30m',
     )
+    sat.enable_ipv6_http_proxy()
     if 'sanity' in request.config.option.markexpr:
         configure_nailgun()
         configure_airgun()
     yield sat
     if 'sanity' not in request.config.option.markexpr:
         sat = Satellite.get_host_by_hostname(sat.hostname)
         sat.unregister()
-        sat.disable_ipv6_http_proxy(http_proxy)
         Broker(hosts=[sat]).checkin()
-    else:
-        sat.disable_ipv6_http_proxy(http_proxy)

pytest_fixtures/core/xdist.py

@@ -18,8 +18,9 @@ def align_to_satellite(request, worker_id, satellite_factory):
         if settings.server.hostname:
             sanity_sat = Satellite(settings.server.hostname)
             sanity_sat.unregister()
-            broker_sat = Satellite.get_host_by_hostname(sanity_sat.hostname)
-            Broker(hosts=[broker_sat]).checkin()
+            if settings.server.auto_checkin:
+                broker_sat = Satellite.get_host_by_hostname(sanity_sat.hostname)
+                Broker(hosts=[broker_sat]).checkin()
     else:
         # clear any hostname that may have been previously set
         settings.set("server.hostname", None)

robottelo/config/__init__.py

@@ -10,7 +10,6 @@
 
 from robottelo.config.validators import VALIDATORS
 from robottelo.logging import logger, robottelo_root_dir
-from robottelo.utils.url import ipv6_hostname_translation
 
 if not os.getenv('ROBOTTELO_DIR'):
     # dynaconf robottelo file uses ROBOTELLO_DIR for screenshots
@@ -45,8 +44,6 @@ def get_settings():
                 f'Dynaconf validation failed, continuing for the sake of unit tests\n{err}'
             )
 
-        ipv6_hostname_translation(settings)
-
         return settings
 
 

robottelo/host_helpers/contenthost_mixins.py

@@ -90,11 +90,16 @@ def _dogfood_helper(self, product, release, repo=None):
             )
         return product, release, v_major, repo
 
-    def download_repofile(self, product=None, release=None, snap=''):
+    def download_repofile(self, product=None, release=None, snap='', proxy=None):
         """Downloads the tools/client, capsule, or satellite repos on the machine"""
         product, release, v_major, _ = self._dogfood_helper(product, release)
-        url = dogfood_repofile_url(settings.ohsnap, product, release, v_major, snap)
-        self.execute(f'curl -o /etc/yum.repos.d/dogfood.repo -L {url}')
+        if not proxy and settings.server.is_ipv6:
+            proxy = settings.server.http_proxy_ipv6_url
+        url = dogfood_repofile_url(settings.ohsnap, product, release, v_major, snap, proxy=proxy)
+        command = f'curl -o /etc/yum.repos.d/dogfood.repo -L {url}'
+        if settings.server.is_ipv6:
+            command += f' -x {settings.server.http_proxy_ipv6_url}'
+        self.execute(command)
 
     def dogfood_repository(self, repo=None, product=None, release=None, snap=''):
         """Returns a repository definition based on the arguments provided"""

robottelo/hosts.py

@@ -824,6 +824,7 @@ def register_contenthost(
         auto_attach=False,
         serverurl=None,
         baseurl=None,
+        enable_proxy=False,
     ):
         """Registers content host on foreman server either by specifying
         organization name and activation key name or by specifying organization
@@ -880,6 +881,12 @@ def register_contenthost(
             cmd += f' --serverurl {serverurl}'
         if baseurl:
             cmd += f' --baseurl {baseurl}'
+
+        # Enabling proxy for Ipv6
+        if enable_proxy and all([settings.server.is_ipv6, settings.server.http_proxy_ipv6_url]):
+            url = urlparse(settings.server.http_proxy_ipv6_url)
+            self.enable_server_proxy(url.hostname, url.port)
+
         return self.execute(cmd)
 
     def unregister(self):
@@ -926,6 +933,12 @@ def put_ssh_key(self, source_key_path, destination_key_name):
         if result.status != 0:
             raise CLIFactoryError(f'Failed to chmod ssh key file:\n{result.stderr}')
 
+    def enable_server_proxy(self, hostname, port=None):
+        cmd = f"subscription-manager config --server.proxy_hostname={hostname}"
+        if port:
+            cmd += f' --server.proxy_port={port}'
+        self.execute(cmd)
+
     def add_authorized_key(self, pub_key):
         """Inject a public key into the authorized keys file
 
@@ -1470,7 +1483,7 @@ def install_tracer(self):
             raise ContentHostError('There was an error installing katello-host-tools-tracer')
         self.execute('katello-tracer-upload')
 
-    def register_to_cdn(self, pool_ids=None):
+    def register_to_cdn(self, pool_ids=None, enable_proxy=False):
         """Subscribe satellite to CDN"""
         if pool_ids is None:
             pool_ids = [settings.subscription.rhn_poolid]
@@ -1480,6 +1493,7 @@ def register_to_cdn(self, pool_ids=None):
             lce=None,
             username=settings.subscription.rhn_username,
             password=settings.subscription.rhn_password,
+            enable_proxy=enable_proxy,
         )
         if cmd_result.status != 0:
             raise ContentHostError(
@@ -1637,9 +1651,7 @@ def enable_capsule_downstream_repos(self):
     def enable_ipv6_http_proxy(self):
         if all([settings.server.is_ipv6, settings.server.http_proxy_ipv6_url]):
             url = urlparse(settings.server.http_proxy_ipv6_url)
-            self.execute(
-                f'subscription-manager config --server.proxy_hostname={url.hostname} --server.proxy_port={url.port}'
-            )
+            self.enable_server_proxy(url.hostname, url.port)
 
     def disable_ipv6_http_proxy(self):
         if settings.server.is_ipv6:
@@ -1768,6 +1780,10 @@ def enable_satellite_or_capsule_module_for_rhel8(self):
         Note: Make sure required repos are enabled before using this.
         """
         if self.os_version.major == 8:
+            if settings.server.is_ipv6:
+                self.execute(
+                    f"echo -e 'proxy={settings.server.http_proxy_ipv6_url}' >> /etc/dnf/dnf.conf"
+                )
             assert (
                 self.execute(
                     f'dnf -y module enable {self.product_rpm_name}:el{self.os_version.major}'
@@ -1811,7 +1827,7 @@ def _swap_nailgun(self, new_version):
         [sys.modules.pop(k) for k in to_clear]
 
     def enable_ipv6_http_proxy(self):
-        # Execute procedures for enabling IPv6 HTTP Proxy
+        """Execute procedures for enabling IPv6 HTTP Proxy"""
         if all([settings.server.is_ipv6, settings.server.http_proxy_ipv6_url]):
             proxy_name = 'Robottelo IPv6 Automation Proxy'
             if not self.cli.HttpProxy.exists(search=('name', proxy_name)):
@@ -1845,9 +1861,7 @@ def enable_ipv6_http_proxy(self):
         return None
 
     def disable_ipv6_http_proxy(self, http_proxy):
-        """
-        Execute procedures for disabling IPv6 HTTP Proxy
-        """
+        """Execute procedures for disabling IPv6 HTTP Proxy"""
         if http_proxy:
             http_proxy.delete()
             self.cli.Settings.set(

robottelo/utils/ohsnap.py

@@ -21,7 +21,7 @@ def ohsnap_response_hook(r, *args, **kwargs):
     r.raise_for_status()
 
 
-def ohsnap_repo_url(ohsnap, request_type, product, release, os_release, snap=''):
+def ohsnap_repo_url(ohsnap, request_type, product, release, os_release, snap='', proxy=None):
     """Returns a URL pointing to Ohsnap "repo_file" or "repositories" API endpoint"""
     if request_type not in ['repo_file', 'repositories']:
         raise InvalidArgumentError('Type must be one of "repo_file" or "repositories"')
@@ -42,11 +42,14 @@ def ohsnap_repo_url(ohsnap, request_type, product, release, os_release, snap='')
                 f'.z version component not provided in the release ({release}),'
                 f' fetching the recent z-stream from ohsnap'
             )
+            request_query = {
+                'url': f'{ohsnap.host}/api/streams',
+                'hooks': {'response': ohsnap_response_hook},
+            }
+            if proxy:
+                request_query['proxies'] = {'http': proxy}
             res, _ = wait_for(
-                lambda: requests.get(
-                    f'{ohsnap.host}/api/streams',
-                    hooks={'response': ohsnap_response_hook},
-                ),
+                lambda: requests.get(**request_query),
                 handle_exception=True,
                 raise_original=True,
                 timeout=ohsnap.request_retry.timeout,
@@ -69,8 +72,8 @@ def ohsnap_repo_url(ohsnap, request_type, product, release, os_release, snap='')
     )
 
 
-def dogfood_repofile_url(ohsnap, product, release, os_release, snap=''):
-    return ohsnap_repo_url(ohsnap, 'repo_file', product, release, os_release, snap)
+def dogfood_repofile_url(ohsnap, product, release, os_release, snap='', proxy=None):
+    return ohsnap_repo_url(ohsnap, 'repo_file', product, release, os_release, snap, proxy)
 
 
 def dogfood_repository(

robottelo/utils/url.py

@@ -14,17 +14,17 @@ def is_url(url):
 def is_ipv4_url(text):
     # Did not find the better way to filter only URLs so skipping it simple
     # and open for reviewers suggestions
-    if isinstance(text, str) and 'ipv4' in text and 'redhat.com' in text:
+    if isinstance(text, str) and '-ipv4' in text:
         return True
     return False
 
 
-def ipv6_translator(settings_list, setting_major, settings):
+def ipv6_translator(settings_list, setting_major, settings, data):
     """Translates the hostname containing ipv4 to ipv6 and updates the settings object"""
     dotted_settings = '.'.join(setting_major)
     for _key, _val in settings_list.items():
         if is_ipv4_url(_val):
-            settings.set(f'{dotted_settings}.{_key}', str(_val).replace('ipv4', 'ipv6'))
+            data[f'{dotted_settings}.{_key}'] = str(_val).replace('ipv4', 'ipv6')
             logger.debug(f'Setting translated to IPv6, Path: {dotted_settings}.{_key}')
         elif isinstance(_val, list):
             updated = False
@@ -34,20 +34,22 @@ def ipv6_translator(settings_list, setting_major, settings):
                     new_list[i] = new_list[i].replace('ipv4', 'ipv6')
                     updated = True
             if updated:
-                settings.set(f'{dotted_settings}.{_key}', new_list)
+                data[f'{dotted_settings}.{_key}'] = new_list
                 logger.debug(f'Setting translated to IPv6, Path: {dotted_settings}.{_key}')
         elif isinstance(_val, dict):
             new_setting_major = setting_major + [_key]
-            ipv6_translator(settings_list=_val, setting_major=new_setting_major, settings=settings)
+            ipv6_translator(
+                settings_list=_val, setting_major=new_setting_major, settings=settings, data=data
+            )
 
 
-def ipv6_hostname_translation(settings):
+def ipv6_hostname_translation(settings, data):
     """Migrates any ipv4 containing hostname in conf to ipv6 hostname"""
     settings_path = []
     if settings.server.is_ipv6:
         all_settings = settings.loaded_by_loaders.items()
         for loader_name, loader_settings in tuple(all_settings):
             if loader_name.loader == 'yaml':
-                ipv6_translator(loader_settings, settings_path, settings)
+                ipv6_translator(loader_settings, settings_path, settings, data)
     else:
         logger.debug('Ipv6 Hostname dynaconf migration hook is skipped for ipv4 testing')

tests/foreman/installer/test_installer.py

@@ -162,7 +162,7 @@ def sat_default_install(module_sat_ready_rhels):
     install_satellite(module_sat_ready_rhels[0], installer_args)
     sat = module_sat_ready_rhels[0]
     http_proxy = sat.enable_ipv6_http_proxy()
-    yield
+    yield sat
     sat.disable_ipv6_http_proxy(http_proxy)