net, dhcpd: move helpers to a dedicated module, add type annotation (RedHatQE#1900)

* net, dhcpd: move helpers to a dedicated module, add type annotation

Improve code organization and re-usability with separate dhcpd module
Signed-off-by: Asia Khromov <azhivovk@redhat.com>

* net, dhcpd: move dhcp ip related constants to dhcp module

Signed-off-by: Asia Khromov <azhivovk@redhat.com>

* net, dhcpd: edit DHCP_IP_SUBNET in dhcpd constants
Signed-off-by: Asia Khromov <azhivovk@redhat.com>

---------

Signed-off-by: Asia Khromov <azhivovk@redhat.com>

Author: azhivovk

tests/network/constants.py

@@ -1,7 +1,5 @@
 from utilities.constants import CLUSTER_NETWORK_ADDONS_OPERATOR
 
-DHCP_IP_RANGE_START = "10.200.3.3"
-DHCP_IP_RANGE_END = "10.200.3.10"
 IPV4_ADDRESS_SUBNET_PREFIX = "10.200.0"
 EXPECTED_CNAO_COMP_NAMES = [
     "multus",

tests/network/l2_bridge/conftest.py

@@ -6,19 +6,21 @@
 import pytest
 from pyhelper_utils.shell import run_ssh_commands
 
-from tests.network.constants import DHCP_IP_RANGE_END, DHCP_IP_RANGE_START
-from tests.network.utils import (
+from tests.network.libs.dhcpd import (
+    DHCP_IP_RANGE_END,
+    DHCP_IP_RANGE_START,
+    DHCP_IP_SUBNET,
     DHCP_SERVER_CONF_FILE,
     DHCP_SERVICE_RESTART,
-    update_cloud_init_extra_user_data,
+    verify_dhcpd_activated,
 )
+from tests.network.utils import update_cloud_init_extra_user_data
 from utilities.infra import get_node_selector_dict, name_prefix
 from utilities.network import (
     cloud_init_network_data,
     get_vmi_mac_address_by_iface_name,
     network_device,
     network_nad,
-    verify_dhcpd_activated,
 )
 from utilities.virt import (
     VirtualMachineForTests,
@@ -27,8 +29,8 @@
 )
 
 #: Test setup
-#       .........                                                                                      ..........
-#       |       |---eth1:10.200.0.1:                                              10.200.0.2:---eth1:|        |
+#       .........                                                                                    ..........
+#       |       |---eth1:10.200.0.1:                                               10.200.0.2:eth1---|        |
 #       | VM-A  |---eth2:10.200.2.1    : multicast(ICMP), custom eth type test:    10.200.2.2:eth2---|  VM-B  |
 #       |       |---eth3:10.200.3.1    : DHCP test :                               10.200.3.2:eth3---|        |
 #       |.......|---eth4:10.200.4.1    : mpls test :                               10.200.4.2:eth4---|........|
@@ -297,7 +299,7 @@ def l2_bridge_running_vm_a(
     namespace, worker_node1, l2_bridge_all_nads, dhcp_nad, unprivileged_client, l2_bridge_running_vm_b
 ):
     dhcpd_data = DHCP_SERVER_CONF_FILE.format(
-        DHCP_IP_SUBNET="10.200.3",
+        DHCP_IP_SUBNET=DHCP_IP_SUBNET,
         DHCP_IP_RANGE_START=DHCP_IP_RANGE_START,
         DHCP_IP_RANGE_END=DHCP_IP_RANGE_END,
         CLIENT_MAC_ADDRESS=get_vmi_mac_address_by_iface_name(vmi=l2_bridge_running_vm_b.vmi, iface_name=dhcp_nad.name),

tests/network/l2_bridge/test_l2_ovs_linux_bridge.py

@@ -6,7 +6,7 @@
 from pyhelper_utils.shell import run_ssh_commands
 from timeout_sampler import TimeoutSampler
 
-from tests.network.constants import DHCP_IP_RANGE_START
+from tests.network.libs.dhcpd import DHCP_IP_RANGE_START
 from utilities.constants import TIMEOUT_2MIN
 from utilities.network import assert_ping_successful, get_vmi_ip_v4_by_name, ping
 

tests/network/libs/dhcpd.py

@@ -0,0 +1,57 @@
+import shlex
+
+from pyhelper_utils.shell import run_ssh_commands
+from timeout_sampler import TimeoutExpiredError, TimeoutSampler
+
+from utilities.constants import TIMEOUT_5SEC, TIMEOUT_30SEC
+from utilities.network import LOGGER
+from utilities.virt import VirtualMachineForTests
+
+DHCP_IP_SUBNET = "10.200.3"
+DHCP_IP_RANGE_START = f"{DHCP_IP_SUBNET}.3"
+DHCP_IP_RANGE_END = f"{DHCP_IP_SUBNET}.10"
+DHCP_SERVICE_RESTART = "sudo systemctl restart dhcpd"
+DHCP_SERVER_CONF_FILE = """
+cat <<EOF >> /etc/dhcp/dhcpd.conf
+default-lease-time 3600;
+max-lease-time 7200;
+authoritative;
+subnet {DHCP_IP_SUBNET}.0 netmask 255.255.255.0 {{
+  option subnet-mask 255.255.255.0;
+  option routers {DHCP_IP_SUBNET}.1;
+  option domain-name-servers {DHCP_IP_SUBNET}.1;
+
+  pool {{
+    range {DHCP_IP_RANGE_START} {DHCP_IP_RANGE_END};
+    allow known-clients;
+    deny  unknown-clients;
+  }}
+}}
+host intended_client_vm {{
+  hardware ethernet {CLIENT_MAC_ADDRESS};
+}}
+EOF
+"""
+
+
+def verify_dhcpd_activated(vm: VirtualMachineForTests) -> bool:
+    active = "active"
+    dhcpd = "dhcpd"
+    sample = None
+    sampler = TimeoutSampler(
+        wait_timeout=TIMEOUT_30SEC,
+        sleep=TIMEOUT_5SEC,
+        func=run_ssh_commands,
+        host=vm.ssh_exec,
+        commands=[shlex.split(f"sudo systemctl is-{active} {dhcpd}")],
+    )
+    try:
+        for sample in sampler:
+            if sample[0].strip() == active:
+                return True
+
+    except TimeoutExpiredError:
+        LOGGER.error(f"{dhcpd} status is not '{active}' but rather '{sample}'")
+        raise
+
+    return False

tests/network/utils.py

@@ -27,28 +27,6 @@
 from utilities.virt import VirtualMachineForTests, fedora_vm_body, running_vm
 
 LOGGER = logging.getLogger(__name__)
-DHCP_SERVICE_RESTART = "sudo systemctl restart dhcpd"
-DHCP_SERVER_CONF_FILE = """
-cat <<EOF >> /etc/dhcp/dhcpd.conf
-default-lease-time 3600;
-max-lease-time 7200;
-authoritative;
-subnet {DHCP_IP_SUBNET}.0 netmask 255.255.255.0 {{
-  option subnet-mask 255.255.255.0;
-  option routers {DHCP_IP_SUBNET}.1;
-  option domain-name-servers {DHCP_IP_SUBNET}.1;
-
-  pool {{
-    range {DHCP_IP_RANGE_START} {DHCP_IP_RANGE_END};
-    allow known-clients;
-    deny  unknown-clients;
-  }}
-}}
-host intended_client_vm {{
-  hardware ethernet {CLIENT_MAC_ADDRESS};
-}}
-EOF
-"""
 SERVICE_MESH_INJECT_ANNOTATION = "sidecar.istio.io/inject"
 
 

utilities/network.py

@@ -23,7 +23,6 @@
 from ocp_resources.pod import Pod
 from ocp_resources.sriov_network import SriovNetwork
 from ocp_resources.sriov_network_node_policy import SriovNetworkNodePolicy
-from pyhelper_utils.shell import run_ssh_commands
 from pytest_testconfig import config as py_config
 from timeout_sampler import TimeoutExpiredError, TimeoutSampler
 
@@ -40,9 +39,7 @@
     SRIOV,
     TIMEOUT_2MIN,
     TIMEOUT_3MIN,
-    TIMEOUT_5SEC,
     TIMEOUT_8MIN,
-    TIMEOUT_30SEC,
     TIMEOUT_90SEC,
     WORKERS_TYPE,
 )
@@ -1083,27 +1080,6 @@ def create_sriov_node_policy(
     wait_for_ready_sriov_nodes(snns=sriov_nodes_states)
 
 
-def verify_dhcpd_activated(vm):
-    active = "active"
-    dhcpd = "dhcpd"
-    sample = None
-    sampler = TimeoutSampler(
-        wait_timeout=TIMEOUT_30SEC,
-        sleep=TIMEOUT_5SEC,
-        func=run_ssh_commands,
-        host=vm.ssh_exec,
-        commands=[shlex.split(f"sudo systemctl is-{active} {dhcpd}")],
-    )
-    try:
-        for sample in sampler:
-            if sample[0].strip() == active:
-                return True
-
-    except TimeoutExpiredError:
-        LOGGER.error(f"{dhcpd} status is not '{active}' but rather '{sample}'")
-        raise
-
-
 def wait_for_node_marked_by_bridge(bridge_nad: LinuxBridgeNetworkAttachmentDefinition, node: Node) -> None:
     bridge_annotation = bridge_nad.resource_name
     sampler = TimeoutSampler(