default values

Signed-off-by: Katharina Trentau <[email protected]>

Author: fraugabel

Tests/iaas/security-groups/default-security-group-rules.py

@@ -56,7 +56,7 @@ def test_rules(cloud_name: str):
                 # we allow ingress from the same security group
                 # but only for the default security group
                 r_group_id = rule.remote_group_id
-                if (r_group_id == "PARENT" and not r_custom_sg):
+                if r_group_id == "PARENT" and not r_custom_sg:
                     ingress_from_same_sg += 1
             elif direction == "egress" and ethertype == "IPv4":
                 egress_rules += 1
@@ -82,34 +82,39 @@ def test_rules(cloud_name: str):
         f"Expected only ingress rules for default security groups, "
         f"that allow ingress traffic from the same group. "
         f"But there are more - in total {ingress_rules} ingress rules. "
-        f"There should be only {ingress_from_same_sg} ingress rules.")
-    assert egress_rules > 0, (
-        f"Expected to have more than {egress_rules} egress rules present.")
-    var_list = [egress_ipv4_default_sg, egress_ipv4_custom_sg,
-                egress_ipv6_default_sg, egress_ipv6_custom_sg]
+        f"There should be only {ingress_from_same_sg} ingress rules."
+    )
+    assert (
+        egress_rules > 0
+    ), f"Expected to have more than {egress_rules} egress rules present."
+    var_list = [
+        egress_ipv4_default_sg,
+        egress_ipv4_custom_sg,
+        egress_ipv6_default_sg,
+        egress_ipv6_custom_sg,
+    ]
     assert all([var > 0 for var in var_list]), (
         "Not all expected egress rules are present. "
         "Expected rules for egress for IPv4 and IPv6 "
-        "both for default and custom security groups.")
+        "both for default and custom security groups."
+    )
 
-    result_dict = {
-        "Ingress Rules": ingress_rules,
-        "Egress Rules": egress_rules
-    }
+    result_dict = {"Ingress Rules": ingress_rules, "Egress Rules": egress_rules}
     return result_dict
 
 
 def main():
     parser = argparse.ArgumentParser(
-        description="SCS Default Security Group Rules Checker")
+        description="SCS Default Security Group Rules Checker"
+    )
     parser.add_argument(
-        "--os-cloud", type=str,
+        "--os-cloud",
+        type=str,
         help="Name of the cloud from clouds.yaml, alternative "
-        "to the OS_CLOUD environment variable"
+        "to the OS_CLOUD environment variable",
     )
     parser.add_argument(
-        "--debug", action="store_true",
-        help="Enable OpenStack SDK debug logging"
+        "--debug", action="store_true", help="Enable OpenStack SDK debug logging"
     )
     args = parser.parse_args()
     openstack.enable_logging(debug=args.debug)

Tests/kaas/k8s-default-storage-class/k8s-default-storage-class-check.py

@@ -86,15 +86,15 @@ def check_default_storageclass(k8s_client_storage):
     return default_storage_class
 
 
-def create_pvc_pod(k8s_api_instance, storage_class):
+def create_pvc_pod(k8s_api_instance, storage_class, pvc_name = PVC_NAME, pod_name = POD_NAME, pv_name = PV_NAME, namespace = NAMESPACE, num_retries = NUM_RETRIES):
     """
     1. Create PersistantVolumeClaim
     2. Create pod which uses the PersitantVolumeClaim
     """
     # 1. Create PersistantVolumeClaim
-    logger.debug(f"create pvc: {PVC_NAME}")
+    logger.debug(f"create pvc: {pvc_name}")
 
-    pvc_meta = client.V1ObjectMeta(name=PVC_NAME)
+    pvc_meta = client.V1ObjectMeta(name=pvc_name)
     pvc_resources = client.V1ResourceRequirements(
         requests={"storage": "1Gi"},
     )
@@ -112,11 +112,11 @@ def create_pvc_pod(k8s_api_instance, storage_class):
     )
 
     # 2. Create a pod which makes use of the PersitantVolumeClaim
-    logger.debug(f"create pod: {POD_NAME}")
+    logger.debug(f"create pod: {pod_name}")
 
     pod_vol = client.V1Volume(
         name=PV_NAME,
-        persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(PVC_NAME),
+        persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(pvc_name),
     )
     pod_con = client.V1Container(
         name="nginx",
@@ -130,7 +130,7 @@ def create_pvc_pod(k8s_api_instance, storage_class):
     pod_body = client.V1Pod(
         api_version="v1",
         kind="Pod",
-        metadata=client.V1ObjectMeta(name=POD_NAME),
+        metadata=client.V1ObjectMeta(name=pod_name),
         spec=pod_spec,
     )
 
@@ -141,7 +141,7 @@ def create_pvc_pod(k8s_api_instance, storage_class):
     pod_status = pod_info["status"]["phase"]
 
     retries = 0
-    while pod_status != "Running" and retries <= NUM_RETRIES:
+    while pod_status != "Running" and retries <= num_retries:
         api_response = k8s_api_instance.read_namespaced_pod(
             POD_NAME, NAMESPACE, _preload_content=False
         )