Update provisioning tests for IPv6 provisoning

Signed-off-by: Shubham Ganar <sganar@redhat.com>

pytest_fixtures/component/provision_pxe.py

@@ -177,37 +177,40 @@ def module_provisioning_sat(
     # TODO: investigate DNS setup issue on Satellite,
     # we might need to set up Sat's DNS server as the primary one on the Sat host
     provisioning_upstream_dns_primary = (
-        broker_data_out.provisioning_upstream_dns.pop()
+        broker_data_out.provisioning_upstream_dns
+        if settings.server.is_ipv6
+        else broker_data_out.provisioning_upstream_dns.pop()
     )  # There should always be at least one upstream DNS
     provisioning_upstream_dns_secondary = (
         broker_data_out.provisioning_upstream_dns.pop()
-        if len(broker_data_out.provisioning_upstream_dns)
+        if len(broker_data_out.provisioning_upstream_dns) and not settings.server.is_ipv6
         else None
     )
 
     domain = sat.api.Domain(
         location=[module_location],
         organization=[module_sca_manifest_org],
-        dns=module_provisioning_capsule.id,
+        dns=None if settings.server.is_ipv6 else module_provisioning_capsule.id,
         name=provisioning_domain_name,
     ).create()
 
     subnet = sat.api.Subnet(
         location=[module_location],
         organization=[module_sca_manifest_org],
         network=str(provisioning_network.network_address),
+        network_type='IPv6' if settings.server.is_ipv6 else 'IPv4',
         mask=str(provisioning_network.netmask),
         gateway=broker_data_out.provisioning_gw_ip,
         from_=broker_data_out.provisioning_host_range_start,
         to=broker_data_out.provisioning_host_range_end,
         dns_primary=provisioning_upstream_dns_primary,
         dns_secondary=provisioning_upstream_dns_secondary,
         boot_mode='DHCP',
-        ipam='DHCP',
-        dhcp=module_provisioning_capsule.id,
+        ipam='None' if settings.server.is_ipv6 else 'DHCP',
+        dhcp=None if settings.server.is_ipv6 else module_provisioning_capsule.id,
         tftp=module_provisioning_capsule.id,
         template=module_provisioning_capsule.id,
-        dns=module_provisioning_capsule.id,
+        dns=None if settings.server.is_ipv6 else module_provisioning_capsule.id,
         httpboot=module_provisioning_capsule.id,
         discovery=module_provisioning_capsule.id,
         remote_execution_proxy=[module_provisioning_capsule.id],
@@ -229,6 +232,8 @@ def module_ssh_key_file():
 @pytest.fixture
 def provisioning_host(module_ssh_key_file, pxe_loader, module_provisioning_sat):
     """Fixture to check out blank VM"""
+    if pxe_loader.vm_firmware == 'bios' and settings.server.is_ipv6:
+        pytest.skip('BIOS is not supported with IPv6')
     vlan_id = settings.provisioning.vlan_id
     cd_iso = (
         ""  # TODO: Make this an optional fixture parameter (update vm_firmware when adding this)
@@ -245,10 +250,25 @@ def provisioning_host(module_ssh_key_file, pxe_loader, module_provisioning_sat):
     ) as prov_host:
         yield prov_host
         # Set host as non-blank to run teardown of the host
-        assert module_provisioning_sat.sat.execute('systemctl restart dhcpd').status == 0
+        if not settings.server.is_ipv6:
+            assert module_provisioning_sat.sat.execute('systemctl restart dhcpd').status == 0
         prov_host.blank = getattr(prov_host, 'blank', False)
 
 
+@pytest.fixture(scope='module')
+def configure_kea_dhcp6_server():
+    if settings.server.is_ipv6:
+        kea_host = Broker(
+            workflow=settings.provisioning.provisioning_kea_workflow,
+            artifacts='last',
+            host_class=ContentHost,
+            blank=True,
+            target_vlan_id=settings.provisioning.vlan_id,
+        ).execute()
+        yield kea_host
+        Broker(workflow='remove-vm', source_vm=kea_host['_broker_facts']['name'][0]).execute()
+
+
 @pytest.fixture
 def provisioning_hostgroup(
     module_provisioning_sat,
@@ -273,7 +293,8 @@ def provisioning_hostgroup(
         root_pass=settings.provisioning.host_root_password,
         operatingsystem=module_provisioning_rhel_content.os,
         ptable=default_partitiontable,
-        subnet=module_provisioning_sat.subnet,
+        subnet=module_provisioning_sat.subnet if not settings.server.is_ipv6 else None,
+        subnet6=module_provisioning_sat.subnet if settings.server.is_ipv6 else None,
         pxe_loader=pxe_loader.pxe_loader,
         group_parameters_attributes=[
             {

pytest_plugins/infra_dependent_markers.py

@@ -5,6 +5,7 @@ def pytest_configure(config):
     """Register custom markers to avoid warnings."""
     markers = [
         "on_premises_provisioning: Tests that runs on on_premises Providers",
+        "ipv6_provisioning: Tests for IPv6 provisioning"
         "libvirt_discovery: Tests depends on Libvirt Provider for discovery",
         "external_auth: External Authentication tests",
         "vlan_networking: Tests depends on static predefined vlan networking etc",

pytest_plugins/marker_deselection.py

@@ -9,6 +9,7 @@ def pytest_addoption(parser):
     """Add options for pytest to collect tests than can run on SatLab infra"""
     infra_options = [
         '--include-onprem-provisioning',
+        '--include-ipv6-provisioning',
         '--include-libvirt',
         '--include-external-auth',
         '--include-vlan-networking',
@@ -34,6 +35,7 @@ def pytest_collection_modifyitems(items, config):
     Collects and modifies tests collection based on pytest option to deselect tests for new infra
     """
     include_onprem_provision = config.getoption('include_onprem_provisioning', False)
+    include_ipv6_provisioning = config.getoption('include_ipv6_provisioning', False)
     include_libvirt = config.getoption('include_libvirt', False)
     include_eauth = config.getoption('include_external_auth', False)
     include_vlan = config.getoption('include_vlan_networking', False)
@@ -53,12 +55,15 @@ def pytest_collection_modifyitems(items, config):
             else:
                 deselected.append(item)
             continue
-
         item_marks = [m.name for m in item.iter_markers()]
         # Include / Exclude On Premises Provisioning Tests
         if 'on_premises_provisioning' in item_marks:
             selected.append(item) if include_onprem_provision else deselected.append(item)
             continue
+        # Include / Exclude IPv6 Provisioning Tests
+        if 'ipv6_provisioning' in item_marks:
+            selected.append(item) if include_ipv6_provisioning else deselected.append(item)
+            continue
         # Include / Exclude External Libvirt based Tests
         if 'libvirt_discovery' in item_marks:
             selected.append(item) if include_libvirt else deselected.append(item)
@@ -67,7 +72,7 @@ def pytest_collection_modifyitems(items, config):
         if 'external_auth' in item_marks:
             selected.append(item) if include_eauth else deselected.append(item)
             continue
-        # Include / Exclude VLAN networking based based Tests
+        # Include / Exclude VLAN networking based Tests
         if 'vlan_networking' in item_marks:
             selected.append(item) if include_vlan else deselected.append(item)
             continue

robottelo/host_helpers/satellite_mixins.py

@@ -383,10 +383,11 @@ def provisioning_cleanup(self, hostname, interface='API'):
                 host[0].delete()
             assert not self.api.Host().search(query={'search': f'name={hostname}'})
         # Workaround SAT-28381
-        assert self.execute('cat /dev/null > /var/lib/dhcpd/dhcpd.leases').status == 0
-        assert self.execute('systemctl restart dhcpd').status == 0
-        # Workaround BZ: 2207698
-        assert self.cli.Service.restart().status == 0
+        if not settings.server.is_ipv6:
+            assert self.execute('cat /dev/null > /var/lib/dhcpd/dhcpd.leases').status == 0
+            assert self.execute('systemctl restart dhcpd').status == 0
+            # Workaround BZ: 2207698
+            assert self.cli.Service.restart().status == 0
 
 
 class Factories:

tests/foreman/api/test_provisioning.py

@@ -65,6 +65,7 @@ def assert_host_logs(channel, pattern):
 @pytest.mark.upgrade
 @pytest.mark.parametrize('pxe_loader', ['bios', 'uefi'], indirect=True)
 @pytest.mark.on_premises_provisioning
+@pytest.mark.ipv6_provisioning
 @pytest.mark.rhel_ver_match(r'^(?!.*fips).*$')
 def test_rhel_pxe_provisioning(
     request,
@@ -77,6 +78,7 @@ def test_rhel_pxe_provisioning(
     provisioning_hostgroup,
     module_lce_library,
     module_default_org_view,
+    configure_kea_dhcp6_server,
 ):
     """Simulate baremetal provisioning of a RHEL system via PXE on RHV provider
 
@@ -98,16 +100,23 @@ def test_rhel_pxe_provisioning(
 
     :parametrized: yes
     """
+    if pxe_loader.vm_firmware == 'bios' and settings.server.is_ipv6:
+        pytest.skip('Test cannot be run on BIOS as its not supported')
     host_mac_addr = provisioning_host.provisioning_nic_mac_addr
     sat = module_provisioning_sat.sat
+    # Configure the grubx64.efi image to setup the interface and use TFTP to load the configuration
+    if settings.server.is_ipv6:
+        sat.execute("echo -e 'net_bootp6\nset root=tftp\nset prefix=(tftp)/grub2' > pre.cfg")
+        sat.execute(
+            'grub2-mkimage -c pre.cfg -o /var/lib/tftpboot/grub2/grubx64.efi -p /grub2/ -O x86_64-efi efinet efi_netfs efienv efifwsetup efi_gop tftp net normal chain configfile loadenv procfs romfs'
+        )
     host = sat.api.Host(
         hostgroup=provisioning_hostgroup,
         organization=module_sca_manifest_org,
         location=module_location,
         name=gen_string('alpha').lower(),
         mac=host_mac_addr,
         operatingsystem=module_provisioning_rhel_content.os,
-        subnet=module_provisioning_sat.subnet,
         host_parameters_attributes=[
             {'name': 'remote_execution_connect_by_ip', 'value': 'true', 'parameter_type': 'boolean'}
         ],
@@ -136,62 +145,66 @@ def test_rhel_pxe_provisioning(
     # Change the hostname of the host as we know it already.
     # In the current infra environment we do not support
     # addressing hosts using FQDNs, falling back to IP.
-    provisioning_host.hostname = host.ip
-    # Host is not blank anymore
-    provisioning_host.blank = False
-
-    # Wait for the host to be rebooted and SSH daemon to be started.
-    provisioning_host.wait_for_connection()
-
-    # Perform version check and check if root password is properly updated
-    host_os = host.operatingsystem.read()
-    expected_rhel_version = f'{host_os.major}.{host_os.minor}'
-
-    if int(host_os.major) >= 9:
-        assert (
-            provisioning_host.execute(
-                'echo -e "\nPermitRootLogin yes" >> /etc/ssh/sshd_config; systemctl restart sshd'
-            ).status
-            == 0
+    if is_open('SAT-30601') and not settings.server.is_ipv6:
+        provisioning_host.hostname = host.ip
+        # Host is not blank anymore
+        provisioning_host.blank = False
+
+        # Wait for the host to be rebooted and SSH daemon to be started.
+        provisioning_host.wait_for_connection()
+
+        # Perform version check and check if root password is properly updated
+        host_os = host.operatingsystem.read()
+        expected_rhel_version = f'{host_os.major}.{host_os.minor}'
+
+        if int(host_os.major) >= 9:
+            assert (
+                provisioning_host.execute(
+                    'echo -e "\nPermitRootLogin yes" >> /etc/ssh/sshd_config; systemctl restart sshd'
+                ).status
+                == 0
+            )
+        host_ssh_os = sat.execute(
+            f'sshpass -p {settings.provisioning.host_root_password} '
+            'ssh -o StrictHostKeyChecking=no -o PubkeyAuthentication=no -o PasswordAuthentication=yes '
+            f'-o UserKnownHostsFile=/dev/null root@{provisioning_host.hostname} cat /etc/redhat-release'
+        )
+        assert host_ssh_os.status == 0
+        assert expected_rhel_version in host_ssh_os.stdout, (
+            'Different than the expected OS version was installed'
         )
-    host_ssh_os = sat.execute(
-        f'sshpass -p {settings.provisioning.host_root_password} '
-        'ssh -o StrictHostKeyChecking=no -o PubkeyAuthentication=no -o PasswordAuthentication=yes '
-        f'-o UserKnownHostsFile=/dev/null root@{provisioning_host.hostname} cat /etc/redhat-release'
-    )
-    assert host_ssh_os.status == 0
-    assert expected_rhel_version in host_ssh_os.stdout, (
-        'Different than the expected OS version was installed'
-    )
 
-    # Verify provisioning log exists on host at correct path
-    assert provisioning_host.execute('test -s /root/install.post.log').status == 0
-    assert provisioning_host.execute('test -s /mnt/sysimage/root/install.post.log').status == 1
+        # Verify provisioning log exists on host at correct path
+        assert provisioning_host.execute('test -s /root/install.post.log').status == 0
+        assert provisioning_host.execute('test -s /mnt/sysimage/root/install.post.log').status == 1
 
-    # Run a command on the host using REX to verify that Satellite's SSH key is present on the host
-    template_id = (
-        sat.api.JobTemplate().search(query={'search': 'name="Run Command - Script Default"'})[0].id
-    )
-    job = sat.api.JobInvocation().run(
-        data={
-            'job_template_id': template_id,
-            'inputs': {
-                'command': f'subscription-manager config | grep "hostname = {sat.hostname}"'
+        # Run a command on the host using REX to verify that Satellite's SSH key is present on the host
+        template_id = (
+            sat.api.JobTemplate()
+            .search(query={'search': 'name="Run Command - Script Default"'})[0]
+            .id
+        )
+        job = sat.api.JobInvocation().run(
+            data={
+                'job_template_id': template_id,
+                'inputs': {
+                    'command': f'subscription-manager config | grep "hostname = {sat.hostname}"'
+                },
+                'search_query': f"name = {host.name}",
+                'targeting_type': 'static_query',
             },
-            'search_query': f"name = {host.name}",
-            'targeting_type': 'static_query',
-        },
-    )
-    assert job['result'] == 'success', 'Job invocation failed'
+        )
+        assert job['result'] == 'success', 'Job invocation failed'
 
-    # check if katello-ca-consumer is not used while host registration
-    assert provisioning_host.execute('rpm -qa |grep katello-ca-consumer').status == 1
-    assert (
-        'katello-ca-consumer' not in provisioning_host.execute('cat /root/install.post.log').stdout
-    )
-    # assert that the host is subscribed and consumes
-    # subsctiption provided by the activation key
-    assert provisioning_host.subscribed, 'Host is not subscribed'
+        # check if katello-ca-consumer is not used while host registration
+        assert provisioning_host.execute('rpm -qa |grep katello-ca-consumer').status == 1
+        assert (
+            'katello-ca-consumer'
+            not in provisioning_host.execute('cat /root/install.post.log').stdout
+        )
+        # assert that the host is subscribed and consumes
+        # subsctiption provided by the activation key
+        assert provisioning_host.subscribed, 'Host is not subscribed'
 
 
 @pytest.mark.e2e