Merge pull request #73 from prashant1221/master

 pytests: Add conftest.py and other tests

pytest/configs/all.yaml

@@ -2,6 +2,7 @@ system:
     name: minimal
     type: vmx-14 vmx-20
     os_vmw: vmwarePhoton64Guest
+    firmware: efi
 
 networks:
     vm_network:
@@ -16,11 +17,15 @@ hardware:
     memory:
         type: memory
         size: 4096
+    scsi1:
+        type: scsi_controller
+        subtype: lsilogic
     sata1:
         type: sata_controller
     cdrom1:
         type: cd_drive
         parent: sata1
+        connected: true
     rootdisk:
         type: hard_disk
         parent: sata1
@@ -31,6 +36,7 @@ hardware:
         type: ethernet
         subtype: VmxNet3
         network: vm_network
+        connected: false
     ethernet2:
         type: ethernet
         subtype: VmxNet3
@@ -76,6 +82,17 @@ product_sections:
               category: some
               password: true
 
+configurations:
+    tall:
+        label: Tall
+        description: too little for the money
+    grande:
+        label: Grande
+        description: just right
+    venti:
+        label: Venti
+        description: too much
+
 annotation:
     text: the password is top secret
     info: Any information

pytest/configs/raw-image.yaml

@@ -22,6 +22,7 @@ hardware:
         type: hard_disk
         parent: sata1
         raw_image: dummy.img
+        file_id: file1
     usb1:
         type: usb_controller
     ethernet1:

pytest/test_configs.py → pytest/conftest.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2023 VMware, Inc.  All Rights Reserved.
+# Copyright (c) 2024 Broadcom.  All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the “License”); you may not
 # use this file except in compliance with the License.  You may obtain a copy of
@@ -11,6 +11,7 @@
 # conditions of any kind, EITHER EXPRESS OR IMPLIED.  See the License for the
 # specific language governing permissions and limitations under the License.
 
+
 import glob
 import os
 import pytest
@@ -25,13 +26,15 @@
 
 VMDK_CONVERT=os.path.join(THIS_DIR, "..", "build", "vmdk", "vmdk-convert")
 
-CONFIG_DIR=os.path.join(THIS_DIR, "configs")
-
 WORK_DIR=os.path.join(os.getcwd(), "pytest-configs")
 
 
-@pytest.fixture(scope='module', autouse=True)
-def setup_test():
+@pytest.fixture(scope='module')
+def setup_test(request):
+    global WORK_DIR
+    if hasattr(request, 'param'):
+        WORK_DIR = request.param
+
     os.makedirs(WORK_DIR, exist_ok=True)
 
     process = subprocess.run(["dd", "if=/dev/zero", "of=dummy.img", "bs=1024", "count=1024"], cwd=WORK_DIR)
@@ -42,6 +45,7 @@ def setup_test():
 
     yield
     shutil.rmtree(WORK_DIR)
+    WORK_DIR = os.path.join(os.getcwd(), "pytest-configs")
 
 
 def yaml_param(loader, node):
@@ -61,8 +65,9 @@ def yaml_param(loader, node):
     return value
 
 
-@pytest.mark.parametrize("in_yaml", glob.glob(os.path.join(CONFIG_DIR, "*.yaml")))
-def test_configs(in_yaml):
+@pytest.fixture(scope='module')
+def get_configs(request):
+    in_yaml = request.param
     basename = os.path.basename(in_yaml.rsplit(".", 1)[0])
     out_ovf = os.path.join(WORK_DIR, f"{basename}.ovf")
 
@@ -78,18 +83,4 @@ def test_configs(in_yaml):
     with open(out_ovf) as f:
         ovf = xmltodict.parse(f.read())
 
-    cfg_system = config['system']
-    assert cfg_system['name'] == ovf['Envelope']['VirtualSystem']['Name']
-    assert cfg_system['os_vmw'] == ovf['Envelope']['VirtualSystem']['OperatingSystemSection']['@vmw:osType']
-    assert cfg_system['type'] == ovf['Envelope']['VirtualSystem']['VirtualHardwareSection']['System']['vssd:VirtualSystemType']
-
-    vmw_configs = ovf['Envelope']['VirtualSystem']['VirtualHardwareSection']['vmw:Config']
-
-    # TODO: check if default is set to bios unless no_default_configs is set
-    if 'firmware' in cfg_system:
-        firmware = None
-        if type(vmw_configs) == list:
-            for vmw_config in vmw_configs:
-                if vmw_config['@vmw:key'] == "firmware":
-                    firmware = vmw_config['@vmw:value']
-        assert cfg_system['firmware'] == firmware
+    yield config, ovf

pytest/test_all_configs.py

@@ -11,169 +11,141 @@
 # conditions of any kind, EITHER EXPRESS OR IMPLIED.  See the License for the
 # specific language governing permissions and limitations under the License.
 
-import glob
 import os
 import pytest
-import shutil
-import subprocess
-import yaml
-import xmltodict
+import test_envelope_configs
 
 
 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
-OVA_COMPOSE = os.path.join(THIS_DIR, "..", "ova-compose", "ova-compose.py")
-
-VMDK_CONVERT=os.path.join(THIS_DIR, "..", "build", "vmdk", "vmdk-convert")
-
 CONFIG_DIR=os.path.join(THIS_DIR, "configs")
 
-WORK_DIR=os.path.join(os.getcwd(), "pytest-configs")
-
-
-@pytest.fixture(scope='module', autouse=True)
-def setup_test():
-    os.makedirs(WORK_DIR, exist_ok=True)
-
-    process = subprocess.run(["dd", "if=/dev/zero", "of=dummy.img", "bs=1024", "count=1024"], cwd=WORK_DIR)
-    assert process.returncode == 0
-
-    process = subprocess.run([VMDK_CONVERT, "dummy.img", "dummy.vmdk"], cwd=WORK_DIR)
-    assert process.returncode == 0
-
-    yield
-    shutil.rmtree(WORK_DIR)
 
+@pytest.mark.parametrize('get_configs', [os.path.join(CONFIG_DIR, "all.yaml")], indirect=True)
+class TestAllConfigs:
+    ''' Test configs under each section '''
 
-def yaml_param(loader, node):
-    params = loader.app_params
-    default = None
-    key = node.value
+    @pytest.fixture(autouse=True)
+    def setup_class(self, setup_test, get_configs):
+        self.config, self.ovf = get_configs
 
-    assert type(key) is str, f"param name must be a string"
 
-    if '=' in key:
-        key, default = [t.strip() for t in key.split('=', maxsplit=1)]
-        default = yaml.safe_load(default)
-    value = params.get(key, default)
+    def assert_values(self, val1, val2):
+        if val1:
+            assert val1 == val2
 
-    assert value is not None, f"no param set for '{key}', and there is no default"
 
-    return value
+    def test_product_sections_configs(self):
+        #ProductSection
 
+        cfg_product_sections = self.config["product_sections"]
+        cfg_vmw_ovf = self.ovf['Envelope']['VirtualSystem']['ProductSection']
 
-def assert_values(val1, val2):
-    if val1:
-        assert val1 == val2
+        transports = []
+        key_dict_product = {'instance': '@ovf:instance', 'class': '@ovf:class', 'vendor': 'Vendor',
+                            'product': 'Product', 'info': 'Info', 'version': 'Version', 'full_version': 'FullVersion'}
 
+        key_dict_properties = {'type': '@ovf:type', 'value': '@ovf:value', 'description': 'Description', 'label': 'Label'}
 
-@pytest.fixture(scope='module')
-def get_configs(setup_test):
-    in_yaml =  os.path.join(CONFIG_DIR, "all.yaml")
-    basename = os.path.basename(in_yaml.rsplit(".", 1)[0])
-    out_ovf = os.path.join(WORK_DIR, f"{basename}.ovf")
+        for i, cfg_product_section in enumerate(cfg_product_sections):
+            for el1, el2 in key_dict_product.items():
+                self.assert_values(cfg_product_section.get(el1), cfg_vmw_ovf[i].get(el2))
 
-    process = subprocess.run([OVA_COMPOSE, "-i", in_yaml, "-o", out_ovf, "--vmdk-convert", VMDK_CONVERT], cwd=WORK_DIR)
-    assert process.returncode == 0
+            if cfg_product_section.get('properties'):
+                properties = cfg_vmw_ovf[i]['Property'] if isinstance(cfg_vmw_ovf[i]['Property'], list) else [cfg_vmw_ovf[i]['Property']]
+                for key in cfg_product_sections[i]['properties']:
+                     j = 0
+                     for j in range(len(properties)):
+                         if key == properties[j]['@ovf:key']:
+                             break
+                     self.assert_values(key, properties[j]['@ovf:key'])
+                     for el1, el2 in key_dict_properties.items():
+                         self.assert_values(cfg_product_section['properties'][key].get(el1), properties[j].get(el2))
+                     self.assert_values(cfg_product_section['properties'][key].get('user_configurable'),  bool(properties[j].get('@ovf:userConfigurable')))
+                     self.assert_values(cfg_product_section['properties'][key].get('password'),  bool(properties[j].get('@ovf:password')))
 
-    with open(in_yaml) as f:
-        yaml_loader = yaml.SafeLoader
-        yaml_loader.app_params = {}
-        yaml.add_constructor("!param", yaml_param, Loader=yaml_loader)
-        config = yaml.load(f, Loader=yaml_loader)
+            if 'categories' in cfg_product_section:
+                self.assert_values(list(cfg_product_section['categories'].values()), list(cfg_vmw_ovf[i].get('Category')))
 
-    with open(out_ovf) as f:
-        ovf = xmltodict.parse(f.read())
+            if 'transports' in cfg_product_section:
+                transports.extend(cfg_product_section['transports'])
 
-    yield config, ovf
+        if transports:
+            self.assert_values(set(transports), set(self.ovf['Envelope']['VirtualSystem']['VirtualHardwareSection']['@ovf:transport'].split()))
 
 
-def test_product_sections_configs(get_configs):
-    #ProductSection
-    config, ovf = get_configs
+    def test_annotation_configs(self):
+        #Annotation Section
 
-    cfg_product_sections = config["product_sections"]
-    cfg_vmw_ovf = ovf['Envelope']['VirtualSystem']['ProductSection']
+        cfg_annotation_section = self.config['annotation']
+        cfg_vmw_ovf =  self.ovf['Envelope']['VirtualSystem']['AnnotationSection']
 
-    transports = []
-    key_dict_product = {'instance': '@ovf:instance', 'class': '@ovf:class', 'vendor': 'Vendor',
-                        'product': 'Product', 'info': 'Info', 'version': 'Version', 'full_version': 'FullVersion'}
+        if cfg_annotation_section.get('file'):
+            with open(cfg_annotation_section['file'], 'r') as f:
+                self.assert_values(f.read()[:-1], cfg_vmw_ovf.get('Annotation'))
+        else:
+            self.assert_values(cfg_annotation_section.get('text'), cfg_vmw_ovf.get('Annotation'))
+        self.assert_values(cfg_annotation_section.get('info'), cfg_vmw_ovf.get('Info'))
 
-    key_dict_properties = {'type': '@ovf:type', 'value': '@ovf:value', 'description': 'Description', 'label': 'Label'}
 
-    for i, cfg_product_section in enumerate(cfg_product_sections):
-        for el1, el2 in key_dict_product.items():
-            assert_values(cfg_product_section.get(el1), cfg_vmw_ovf[i].get(el2))
+    def test_networks_configs(self):
+        #Network Section
 
-        if cfg_product_section.get('properties'):
-            properties = cfg_vmw_ovf[i]['Property'] if isinstance(cfg_vmw_ovf[i]['Property'], list) else [cfg_vmw_ovf[i]['Property']]
-            for key in cfg_product_sections[i]['properties']:
-                 j = 0
-                 for j in range(len(properties)):
-                     if key == properties[j]['@ovf:key']:
-                         break
-                 assert_values(key, properties[j]['@ovf:key'])
-                 for el1, el2 in key_dict_properties.items():
-                     assert_values(cfg_product_section['properties'][key].get(el1), properties[j].get(el2))
-                 assert_values(cfg_product_section['properties'][key].get('user_configurable'),  bool(properties[j].get('@ovf:userConfigurable')))
-                 assert_values(cfg_product_section['properties'][key].get('password'),  bool(properties[j].get('@ovf:password')))
+        cfg_networks_section = self.config['networks']
+        cfg_vmw_ovf = self.ovf['Envelope']['NetworkSection']['Network']
+        cfg_vmw_ovf = cfg_vmw_ovf if isinstance(cfg_vmw_ovf, list) else [cfg_vmw_ovf]
 
-        if 'categories' in cfg_product_section:
-            assert_values(list(cfg_product_section['categories'].values()), list(cfg_vmw_ovf[i].get('Category')))
+        for idx, key in enumerate(cfg_networks_section):
+            self.assert_values(cfg_networks_section[key]['name'], cfg_vmw_ovf[idx]['@ovf:name'])
+            self.assert_values(f"The {cfg_networks_section[key]['name']} Network", cfg_vmw_ovf[idx]['Description'])
 
-        if 'transports' in cfg_product_section:
-            transports.extend(cfg_product_section['transports'])
 
-    if transports:
-        assert_values(set(transports), set(ovf['Envelope']['VirtualSystem']['VirtualHardwareSection']['@ovf:transport'].split()))
+    def test_envelope_configs(self):
+        #Envelope Section
 
+        test_envelope_configs.test_envelope_configs(None, (self.config, self.ovf))
 
-def test_annotation_configs(get_configs):
-    #Annotation Section
-    config, ovf = get_configs
 
-    cfg_annotation_section = config['annotation']
-    cfg_vmw_ovf =  ovf['Envelope']['VirtualSystem']['AnnotationSection']
+    def test_system_configs(self):
+        # Operating System Section
 
-    if cfg_annotation_section.get('file'):
-        with open(cfg_annotation_section['file'], 'r') as f:
-            assert_values(f.read()[:-1], cfg_vmw_ovf.get('Annotation'))
-    else:
-        assert_values(cfg_annotation_section.get('text'), cfg_vmw_ovf.get('Annotation'))
-    assert_values(cfg_annotation_section.get('info'), cfg_vmw_ovf.get('Info'))
+        cfg_system_section = self.config['system']
+        cfg_vmw_ovf =  self.ovf['Envelope']['VirtualSystem']
+        vmw_configs = cfg_vmw_ovf['VirtualHardwareSection']['vmw:Config']
 
+        self.assert_values(cfg_system_section['name'], cfg_vmw_ovf['Name'])
+        self.assert_values(cfg_system_section['type'], cfg_vmw_ovf['VirtualHardwareSection']['System']['vssd:VirtualSystemType'])
+        self.assert_values(cfg_system_section['os_vmw'], cfg_vmw_ovf['OperatingSystemSection']['@vmw:osType'])
 
-def test_networks_configs(get_configs):
-    #Network Section
-    config, ovf = get_configs
+        for vmw_config in vmw_configs:
+            if vmw_config['@vmw:key'] == "firmware":
+                self.assert_values(cfg_system_section.get('firmware', "bios"), vmw_config['@vmw:value'])
 
-    cfg_networks_section = config['networks']
-    cfg_vmw_ovf = ovf['Envelope']['NetworkSection']['Network']
-    cfg_vmw_ovf = cfg_vmw_ovf if isinstance(ovf['Envelope']['NetworkSection']['Network'], list) else [ovf['Envelope']['NetworkSection']['Network']]
 
-    for idx, key in enumerate(cfg_networks_section):
-        assert_values(cfg_networks_section[key]['name'], cfg_vmw_ovf[idx]['@ovf:name'])
-        assert_values(f"The {cfg_networks_section[key]['name']} Network", cfg_vmw_ovf[idx]['Description'])
+    def test_virtual_hardware_configs(self):
+        #Hardware Section
 
+        cfg_hardware_section = self.config['hardware']
+        cfg_vmw_ovf = self.ovf['Envelope']['VirtualSystem']['VirtualHardwareSection']['Item']
 
-def test_envelope_configs(get_configs):
-    #Envelope Section
-    config, ovf = get_configs
+        for idx, key in enumerate(cfg_hardware_section):
+            self.assert_values(key, cfg_vmw_ovf[idx]['rasd:ElementName'])
+            if key == 'cpus':
+                self.assert_values(cfg_hardware_section[key], int(cfg_vmw_ovf[idx]['rasd:VirtualQuantity']))
+            elif key == 'memory':
+                self.assert_values(cfg_hardware_section[key]['size'], int(cfg_vmw_ovf[idx]['rasd:VirtualQuantity']))
+            else:
+                self.assert_values(cfg_hardware_section[key].get('subtype'), cfg_vmw_ovf[idx].get('rasd:ResourceSubType'))
+                self.assert_values(cfg_hardware_section[key].get('connected'), bool(cfg_vmw_ovf[idx].get('rasd:AutomaticAllocation')))
 
-    cfg_hardware_section =  config['hardware']
-    cfg_vmw_ovf = ovf['Envelope']['References']['File']
-    cfg_vmw_ovf = cfg_vmw_ovf if isinstance(ovf['Envelope']['References']['File'], list) else [ovf['Envelope']['References']['File']]
 
-    j = 0
-    for key in cfg_hardware_section:
-        if isinstance(cfg_hardware_section[key], dict):
-            image_path = cfg_hardware_section[key].get('disk_image') or cfg_hardware_section[key].get('raw_image')
-            if image_path:
-                image_path = image_path if image_path.endswith('.vmdk') else os.path.splitext(image_path)[0] + '.vmdk'
-                assert_values(image_path, cfg_vmw_ovf[j]['@ovf:href'])
-                assert_values(cfg_hardware_section[key].get('file_id'), cfg_vmw_ovf[j]['@ovf:id'])
-                j += 1
+    def test_configuration_configs(self):
+        #Deployment Option Section
 
+        cfg_configuration_section = self.config['configurations']
+        cfg_vmw_ovf = self.ovf['Envelope']['DeploymentOptionSection']['Configuration']
+        cfg_vmw_ovf = cfg_vmw_ovf if isinstance(cfg_vmw_ovf, list) else [cfg_vmw_ovf]
 
-def test_virtual_hardware_configs():
-    # TODO
-    pass
+        for idx, key in enumerate(cfg_configuration_section):
+            self.assert_values(key, cfg_vmw_ovf[idx]['@ovf:id'])
+            self.assert_values(cfg_configuration_section[key]['label'], cfg_vmw_ovf[idx]['Label'])
+            self.assert_values(cfg_configuration_section[key]['description'], cfg_vmw_ovf[idx]['Description'])