Integrate review and fix tests

Signed-off-by: Anja Strunk <anja.strunk@cloudandheat.com>

cli.py

@@ -12,25 +12,14 @@
 
 import json
 import sys
-import warnings
-
-
-from typing import List
 
 import click
 import openstack as os
 import yaml
 
-import generator.common.const as const
 import generator.common.json_ld as json_ld
 
-from pyshacl import validate
-
-from generator.common.json_ld import JsonLdObject
-from generator.discovery.openstack.opentack_discovery import OsCloud
-#from generator.wallet.file_wallet import FileSystemWallet
-#from generator.wallet.wallet import WalletConnector
-#from generator.wallet.xfsc_wallet import XFSCWallet
+from generator.discovery.openstack.openstack_discovery import OsCloud
 
 import rdflib
 
@@ -44,16 +33,6 @@ def cli():
 
 
 @click.command()
-#@click.option(
-#    "--wallet",
-#    is_flag=True,
-#    help="Use '--wallet' to store generated credentials in all wallets. Wallets can be configured in configuration file.",
-#)
-#@click.option(
-#    "--no-print",
-#    is_flag=True,
-#    help="Use '--no-print' to omit json-ld print on screen.",
-#)
 @click.option(
     "--config",
     default="config/config.yaml",
@@ -62,7 +41,6 @@ def cli():
 @click.option("--timeout", default=12, help="Timeout for API calls in seconds")
 @click.argument("cloud")
 def openstack(cloud, timeout, config):
-#def openstack(cloud, timeout, config, no_print, wallet):
     """Generates Gaia-X Credentials for openstack cloud CLOUD.
     CLOUD MUST refer to a name defined in Openstack's configuration file clouds.yaml."""
 
@@ -90,81 +68,26 @@ def openstack(cloud, timeout, config):
         # run discovery
         creds = os_cloud.discover_properties()
 
-        # store creds in wallets
-        #if wallet:
-        #    wallets = init_wallets(config_dict)
-        #    store_creds_in_wallet(wallets, creds)
-
-        # print on screen
-        #if not no_print:
         props = json_ld.get_json_ld_context()
         props["@graph"] = creds
         print(json.dumps(props, indent=4, default=json_ld.to_json_ld))
 
-        #if not wallet and no_print:
-        #    warnings.warn("--no-print is set, but not --wallet. Generated credential(s) will be stored/printed nowhere.")
-
 
 @click.command()
 def kubernetes():
     """Generates Gaia-X Credentials for kubernetes."""
     pass
 
 
-#def init_wallets(config: dict) -> List[WalletConnector]:
-#    wallets = list()
-#    try:
-#        for wallet in config[const.CONFIG_WALLETS]:
-#            if wallet == const.CONFIG_FILESYSTEM_WALLET:
-#                wallets.append(
-#                    FileSystemWallet(
-#                        config[const.CONFIG_WALLETS][const.CONFIG_FILESYSTEM_WALLET]["path"]
-#                    )
-#                )
-#            elif wallet == const.CONFIG_XFSC_WALLET:
-#                wallets.append(XFSCWallet())
-#    except KeyError:
-#        pass
-#    return wallets
-
-
-#def store_creds_in_wallet(
-#        wallets: List[WalletConnector], creds: List[JsonLdObject]
-#) -> None:
-#    for w in wallets:
-#        for c in creds:
-#            w.store_credential(c)
-
-
 def load_file(filepath, file_format=DATA_FILE_FORMAT):
     """Load file in a given format"""
     graph = rdflib.Graph()
     graph.parse(filepath, format=file_format)
     return graph
 
 
-#@click.command()
-#@click.argument("credential", help=f"Filepath of GX Credential to validate. Should have {DATA_FILE_FORMAT} format", )
-#@click.argument("shapes",
-#                help=f"Filepath of shacl schema to be used for validation. Should have {SHAPES_FILE_FORMAT} format", )
-#def validate(credential, shapes):
-#    """Validate SD in jsonld format against given schema in turtle format"""
-#    conforms, results_graph, results_text = validate(
-#        load_file(credential),
-#        shacl_graph=load_file(shapes, file_format=SHAPES_FILE_FORMAT),
-#        data_graph_format=DATA_FILE_FORMAT,
-#        shacl_graph_format=SHAPES_FILE_FORMAT,
-#        inference="rdfs",
-#        debug=False,
-#        serialize_report_graph=True,
-#    )
-#    print(results_text)
-
-
 cli.add_command(openstack)
 cli.add_command(kubernetes)
-#cli.add_command(validate)
-
 
 if __name__ == "__main__":
     cli()

generator/common/json_ld.py

@@ -79,7 +79,7 @@ def to_json_ld(obj) -> dict:
         # call to_json_ld for gx_object
         json_ld.update(to_json_ld(gx_object))
         return json_ld
-    elif isinstance(obj, YAMLRoot):
+    if isinstance(obj, YAMLRoot):
         # if YAMLRoot (= all top level classes in Gaia-X Credential Schema)
         # Using one type is sufficient
         # json_ld['@type'] = get_types(obj.__class__)

generator/discovery/openstack/opentack_discovery.py → generator/discovery/openstack/openstack_discovery.py

@@ -23,7 +23,6 @@ class OsCloud:
     def __init__(self, conn: Connection, config: Dict) -> None:
         # import copy
         self.conn = conn
-        self.auth = conn.auth
         self.regions = list(conn.identity.regions())
         self.config = config
 

generator/discovery/openstack/vm_images_discovery.py

@@ -29,6 +29,9 @@
     Signature,
     SignatureAlgorithm,
     UpdateStrategy,
+    VMDiskType,
+    FirmType,
+    RNGTypes
 )
 from generator.common.gx_schema import VMImage as GX_Image
 from generator.common.json_ld import JsonLdObject
@@ -60,26 +63,22 @@ def _get_cpu_arch(os_image_arch: str) -> str:
 def _add_disk_format(os_image: OS_Image, gx_image: GX_Image) -> None:
     try:
         if os_image.disk_format.lower() == "raw":
-            gx_image.vmImageDiskFormat = "RAW"
+            gx_image.vmImageDiskFormat = VMDiskType("RAW")
         if os_image.disk_format.lower() == "qcow2":
-            gx_image.vmImageDiskFormat = "QCOW2"
+            gx_image.vmImageDiskFormat = VMDiskType("QCOW2")
         if os_image.disk_format.lower() == "vhd":
-            gx_image.vmImageDiskFormat = "VHD"
+            gx_image.vmImageDiskFormat = VMDiskType("VHD")
         if os_image.disk_format.lower() == "iso":
-            gx_image.vmImageDiskFormat = "ISO"
+            gx_image.vmImageDiskFormat = VMDiskType("ISO")
         if os_image.disk_format.lower() == "cvf":
-            gx_image.vmImageDiskFormat = "CVF"
+            gx_image.vmImageDiskFormat = VMDiskType("CVF")
         if os_image.disk_format.lower() == "cva":
-            gx_image.vmImageDiskFormat = "CVA"
+            gx_image.vmImageDiskFormat = VMDiskType("CVA")
     except AttributeError:
         pass
 
 
 class VmDiscovery:
-    # def __init__(self) -> None:
-    #    with open("config/config.yaml", "r") as config_file:
-    #        self.config = yaml.safe_load(config_file)
-
     def __init__(self, conn: Connection, config: Dict) -> None:
         self.conn = conn
         self.config = config
@@ -539,9 +538,9 @@ def _add_firmeware_type(self, os_image: OS_Image, gx_image: GX_Image) -> None:
         try:
             if not os_image.hw_firmware_type:
                 return
-            gx_image.firmwareType = os_image.hw_firmware_type
+            gx_image.firmwareType = FirmType(os_image.hw_firmware_type)
         except AttributeError:
-            gx_image.firmwareType = const.DEFAULT_FIRMWARE_TYPE
+            gx_image.firmwareType = FirmType(const.DEFAULT_FIRMWARE_TYPE)
 
     def _add_watchdog_action(self, os_image: OS_Image, gx_image: GX_Image) -> None:
         try:
@@ -697,15 +696,15 @@ def _add_signature(self, os_image: OS_Image, gx_image: GX_Image) -> None:
     def _add_hypervisor(self, os_image: OS_Image, gx_image: GX_Image) -> None:
         try:
             if os_image.hypervisor_type == "xen":
-                gx_image.hypervisorType = HypervisorType.Xen
+                gx_image.hypervisorType = HypervisorType("Xen")
             elif os_image.hypervisor_type == "quemu":
-                gx_image.hypervisorType = "quemu"
+                gx_image.hypervisorType = HypervisorType("quemu")
             elif os_image.hypervisor_type == "hyperv":
-                gx_image.hypervisorType = "Hyper-V"
+                gx_image.hypervisorType = HypervisorType("Hyper-V")
             else:
-                gx_image.hypervisorType = HypervisorType.other.text
+                gx_image.hypervisorType = HypervisorType("other")
         except AttributeError as e:
-            gx_image.hypervisorType = HypervisorType.other.text
+            gx_image.hypervisorType = HypervisorType("other")
             #raise MissingMandatoryAttribute(e.args)
 
     def _get_signature_algo(self, algo: str) -> str:
@@ -722,4 +721,4 @@ def _add_aggregation_of(self, os_image: OS_Image, gx_image: GX_Image) -> None:
             pass
 
     def _add_rng_model(self, os_image: OS_Image, gx_image: GX_Image) -> None:
-        gx_image.hwRngTypeOfImage = "None"
+        gx_image.hwRngTypeOfImage = RNGTypes("None")

tests/common.py

@@ -164,13 +164,12 @@ def check_image(self, ob_1: Image, ob_2: Image):
         self.assertEqual(ob_1.licenseIncluded, ob_2.licenseIncluded, "Image.licenseIncluded")
         self.assertEqual(ob_1.maintenance, ob_2.maintenance, "Image.maintenance")
 
-    def check_vm_image(self, ob_1: VMImage, ob_2: VMImage):
-        self.check_image(ob_1, ob_2)
-        self.assertEqual(ob_1.vmImageDiskFormat, str(ob_2.vmImageDiskFormat), "VM_Image.vmImageDiskFormat")
-        self.assertEqual(ob_1.hypervisorType, str(ob_2.hypervisorType), "VM_Image.hypervisorType")
-        self.assertEqual(ob_1.firmwareType, str(ob_2.firmwareType), "VM_Image.firmwareType")
-        self.assertEqual(ob_1.hwRngTypeOfImage, str(ob_2.hwRngTypeOfImage), "VM_Image.hwRngTypeOfImage")
-        self.assertEqual(ob_1.watchDogAction, str(ob_2.watchDogAction), "VM_Image.watchDogAction")
+    def check_vm_image(self, exp: VMImage, act: VMImage):
+        self.check_image(exp, act)
+        self.assertEqual(exp.vmImageDiskFormat, act.vmImageDiskFormat, "VM_Image.vmImageDiskFormat")
+        self.assertEqual(exp.hypervisorType, act.hypervisorType, "VM_Image.hypervisorType")
+        self.assertEqual(exp.firmwareType, act.firmwareType, "VM_Image.firmwareType")
+        self.assertEqual(exp.hwRngTypeOfImage, act.hwRngTypeOfImage, "VM_Image.hwRngTypeOfImage")
 
 
 class MockConnection: