Issue/662 default storage class

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

@@ -28,11 +28,11 @@ def check_default_rules(rules, short=False):
         if short is True, the testing mode is set on short for older OpenStack versions
     """
     ingress_rules = egress_rules = 0
-    egress_vars = {'IPv4': {}, 'IPv6': {}}
+    egress_vars = {"IPv4": {}, "IPv6": {}}
     for key, value in egress_vars.items():
-        value['default'] = 0
+        value["default"] = 0
         if not short:
-            value['custom'] = 0
+            value["custom"] = 0
     if not rules:
         logger.info("No default security group rules defined.")
     for rule in rules:
@@ -42,36 +42,48 @@ def check_default_rules(rules, short=False):
             if not short:
                 # we allow ingress from the same security group
                 # but only for the default security group
-                if rule.remote_group_id == "PARENT" and not rule["used_in_non_default_sg"]:
+                if (
+                    rule.remote_group_id == "PARENT"
+                    and not rule["used_in_non_default_sg"]
+                ):
                     continue
             ingress_rules += 1
         elif direction == "egress" and ethertype in egress_vars:
             egress_rules += 1
             if short:
-                egress_vars[ethertype]['default'] += 1
+                egress_vars[ethertype]["default"] += 1
                 continue
             if rule.remote_ip_prefix:
                 # this rule does not allow traffic to all external ips
                 continue
             # note: these two are not mutually exclusive
             if rule["used_in_default_sg"]:
-                egress_vars[ethertype]['default'] += 1
+                egress_vars[ethertype]["default"] += 1
             if rule["used_in_non_default_sg"]:
-                egress_vars[ethertype]['custom'] += 1
+                egress_vars[ethertype]["custom"] += 1
     # test whether there are no unallowed ingress rules
     if ingress_rules:
         logger.error(f"Expected no default ingress rules, found {ingress_rules}.")
     # test whether all expected egress rules are present
-    missing = [(key, key2) for key, val in egress_vars.items() for key2, val2 in val.items() if not val2]
+    missing = [
+        (key, key2)
+        for key, val in egress_vars.items()
+        for key2, val2 in val.items()
+        if not val2
+    ]
     if missing:
         logger.error(
             "Expected rules for egress for IPv4 and IPv6 both for default and custom security groups. "
             f"Missing rule types: {', '.join(str(x) for x in missing)}"
         )
-    logger.info(str({
-        "Unallowed Ingress Rules": ingress_rules,
-        "Egress Rules": egress_rules,
-    }))
+    logger.info(
+        str(
+            {
+                "Unallowed Ingress Rules": ingress_rules,
+                "Egress Rules": egress_rules,
+            }
+        )
+    )
 
 
 def create_security_group(conn, sg_name: str = SG_NAME, description: str = DESCRIPTION):
@@ -139,7 +151,9 @@ def main():
         "to the OS_CLOUD environment variable",
     )
     parser.add_argument(
-        "--debug", action="store_true", help="Enable debug logging",
+        "--debug",
+        action="store_true",
+        help="Enable debug logging",
     )
     args = parser.parse_args()
     openstack.enable_logging(debug=args.debug)
@@ -164,10 +178,17 @@ def main():
         test_rules(conn)
 
     c = counting_handler.bylevel
-    logger.debug(f"Total critical / error / warning: {c[logging.CRITICAL]} / {c[logging.ERROR]} / {c[logging.WARNING]}")
+    logger.debug(
+        f"Total critical / error / warning: {c[logging.CRITICAL]} / {c[logging.ERROR]} / {c[logging.WARNING]}"
+    )
     if not c[logging.CRITICAL]:
-        print("security-groups-default-rules-check: " + ('PASS', 'FAIL')[min(1, c[logging.ERROR])])
-    return min(127, c[logging.CRITICAL] + c[logging.ERROR])  # cap at 127 due to OS restrictions
+        print(
+            "security-groups-default-rules-check: "
+            + ("PASS", "FAIL")[min(1, c[logging.ERROR])]
+        )
+    return min(
+        127, c[logging.CRITICAL] + c[logging.ERROR]
+    )  # cap at 127 due to OS restrictions
 
 
 if __name__ == "__main__":

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

@@ -2,6 +2,7 @@
 import sys
 import logging
 from kubernetes import client, config
+import os
 
 manual_result_file_template = {
     "name": None,
@@ -13,7 +14,7 @@
 
 
 def initialize_logging():
-    logging.basicConfig(format="%(levelname)s: %(message)s", level=logging.DEBUG)
+    logging.basicConfig(format="%(levelname)s: %(message)s", level=logging.INFO)
 
 
 def print_usage(file=sys.stderr):
@@ -34,11 +35,11 @@ class SCSTestException(Exception):
     """Raised when an Specific test did not pass"""
 
     def __init__(self, *args, return_code: int):
+        super().__init__(*args)
         self.return_code = return_code
 
 
 def setup_k8s_client(kubeconfigfile=None):
-
     if kubeconfigfile:
         logger.debug(f"using kubeconfig file '{kubeconfigfile}'")
         config.load_kube_config(kubeconfigfile)
@@ -56,19 +57,15 @@ def setup_k8s_client(kubeconfigfile=None):
 
 
 def gen_sonobuoy_result_file(error_n: int, error_msg: str, test_file_name: str):
-
-    test_name = test_file_name.replace(".py", "")
-
+    test_name = test_file_name.removesuffix(".py")
     test_status = "passed"
-
     if error_n != 0:
         test_status = test_name + "_" + str(error_n)
-
-    result_file = manual_result_file_template
-
-    result_file["name"] = test_name
-    result_file["status"] = test_status
-    result_file["details"]["messages"] = error_msg
-
-    with open(f"./{test_name}.result.yaml", "w") as file:
-        yaml.dump(result_file, file)
+        result_file = manual_result_file_template
+        result_file["name"] = test_name
+        result_file["status"] = test_status
+        result_file["details"]["messages"] = error_msg
+
+        script_directory = os.path.dirname(os.path.abspath(__file__))
+        with open(f"{script_directory}/{test_name}.result.yaml", "w") as file:
+            yaml.dump(result_file, file)