Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

pytests: Add conftest.py and other tests #73

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions pytest/configs/all.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ system:
name: minimal
type: vmx-14 vmx-20
os_vmw: vmwarePhoton64Guest
firmware: efi

networks:
vm_network:
Expand All @@ -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
Expand All @@ -31,6 +36,7 @@ hardware:
type: ethernet
subtype: VmxNet3
network: vm_network
connected: false
ethernet2:
type: ethernet
subtype: VmxNet3
Expand Down Expand Up @@ -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
1 change: 1 addition & 0 deletions pytest/configs/raw-image.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ hardware:
type: hard_disk
parent: sata1
raw_image: dummy.img
file_id: file1
usb1:
type: usb_controller
ethernet1:
Expand Down
35 changes: 13 additions & 22 deletions pytest/test_configs.py → pytest/conftest.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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
Expand All @@ -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)
Expand All @@ -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):
Expand All @@ -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")

Expand All @@ -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
214 changes: 93 additions & 121 deletions pytest/test_all_configs.py
Original file line number Diff line number Diff line change
Expand Up @@ -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'])
Loading
Loading