parameters:
availability_zone_0: az0
availability_zone_1: az1
+ vnf_name: CHANGEME
parameters:
-
db_image_name: db_image
-
db_flavor_name: db_flavor
-
lb_image_name: lb_image
-
lb_flavor_name: lb_flavor
-
svc_image_name: svc_image
-
svc_flavor_name: svc_flavor
-
- svc_count: 3
\ No newline at end of file
+ svc_count: 3
+ mgmt_image_name: mgmt_image
+ mgmt_flavor_name: mgmt_flavor
type: OS::Cinder::VolumeAttachment
properties:
volume_id: { get_param: db_vol0_id }
- server: { get_resource: db_server_0 }
+ instance_uuid: { get_resource: db_server_0 }
db_volume_attachment_1:
type: OS::Cinder::VolumeAttachment
properties:
volume_id: { get_param: db_vol1_id }
- server: { get_resource: db_server_1 }
+ instance_uuid: { get_resource: db_server_1 }
mgmt_server_0:
type: OS::Nova::Server
params:
VNF_NAME: {get_param: vnf_name}
volume_type: "solidfire"
- volume_size: { get_param: volume_size }
+ size: { get_param: volume_size }
db_vol1:
type: OS::Cinder::Volume
params:
VNF_NAME: {get_param: vnf_name}
volume_type: "solidfire"
- volume_size: { get_param: volume_size }
+ size: { get_param: volume_size }
outputs:
db_vol0_id:
from pathlib import Path
import pytest
+from mock import mock
from preload.environment import CloudServiceArchive, PreloadEnvironment
assert csar.get_vf_module_resource_name("base_vIECCF") == "stark_vccf_vf"
+def test_csar_get_vnf_type(csar):
+ assert csar.get_vnf_type("base_vIECCF") == "stark_vccf_svc/stark_vccf_vf"
+
+
def test_csar_get_vf_module_resource_name_not_found(csar):
assert csar.get_vf_module_resource_name("unknown") is None
"common": "ABC",
"my_ip": "192.168.0.1",
}
+
+
+def test_preload_environment_uses_csar(env, monkeypatch):
+ csar = mock.MagicMock(spec=CloudServiceArchive)
+ csar.get_vnf_type = mock.Mock(return_value="stark_vccf_svc/stark_vccf_vf")
+ csar.get_vf_module_model_name = mock.Mock(return_value="model_name")
+ env = env.get_environment("env_three")
+ monkeypatch.setattr(env, "csar", csar)
+ mod = env.get_module("base")
+ assert mod["vnf-type"] == "stark_vccf_svc/stark_vccf_vf"
+ assert mod["vf-module-model-name"] == "model_name"
@pytest.fixture(scope="session")
def base(preload):
- return load_module(preload, "base.json")
+ return load_module(preload, "base_incomplete.json")
@pytest.fixture(scope="session")
def incremental(preload):
- return load_module(preload, "incremental.json")
+ return load_module(preload, "incremental_incomplete.json")
def test_base_fields(base):
def test_preload_env_population(preload):
- base_path = THIS_DIR / "sample_env/preloads/grapi/base.json"
+ base_path = THIS_DIR / "sample_env/preloads/grapi/base_incomplete.json"
data = load_json(base_path)
azs = data["input"]["preload-vf-module-topology-information"][
"vnf-resource-assignments"
]["availability-zones"]["availability-zone"]
assert azs == ["az0", "az1"]
+
+
+def test_preload_env_population_missing_value(preload):
+ base_path = THIS_DIR / "sample_env/preloads/grapi/base_incomplete.json"
+ data = load_json(base_path)
+ vnf_name = data["input"]["preload-vf-module-topology-information"][
+ "vnf-topology-identifier-structure"]["vnf-name"]
+ assert vnf_name == "VALUE FOR: vnf_name"
@pytest.fixture(scope="session")
def base(preload):
- return load_module(preload, "base.json")
+ return load_module(preload, "base_incomplete.json")
@pytest.fixture(scope="session")
def incremental(preload):
- return load_module(preload, "incremental.json")
+ return load_module(preload, "incremental_incomplete.json")
def test_base_azs(base):
def test_preload_env_population(preload):
- base_path = THIS_DIR / "sample_env/preloads/vnfapi/base.json"
+ base_path = THIS_DIR / "sample_env/preloads/vnfapi/base_incomplete.json"
data = load_json(base_path)
azs = data["input"]["vnf-topology-information"]["vnf-assignments"][
"availability-zones"
if props.get("type") == "org.openecomp.groups.VfModule"
}
+ def get_vnf_type(self, module):
+ """
+ Concatenation of service and VF instance name
+ """
+ service_name = self.service_name
+ instance_name = self.get_vf_module_resource_name(module)
+ if service_name and instance_name:
+ return "{}/{}".format(service_name, instance_name)
+
@property
def vf_module_resource_names(self):
"""
def_dir = csar_dir / "Definitions"
check(
def_dir.exists(),
- f"CSAR is invalid. {csar_dir.as_posix()} does not contain a "
- f"Definitions directory.",
+ "CSAR is invalid. {} does not contain a Definitions directory.".format(
+ csar_dir.as_posix()
+ ),
)
return yaml_files(def_dir)
class PreloadEnvironment:
- """
- A
- """
def __init__(self, env_dir, parent=None):
self.base_dir = Path(env_dir)
for m in (parent_module, self.defaults, module):
if m:
result.update(m)
+ if self.csar:
+ vnf_type = self.csar.get_vnf_type(name)
+ if vnf_type:
+ result["vnf-type"] = vnf_type
+ model_name = self.csar.get_vf_module_model_name(name)
+ if model_name:
+ result["vf-module-model-name"] = model_name
return result
@property
import json
import os
from abc import ABC, abstractmethod
+from collections import OrderedDict
import yaml
+def represent_ordered_dict(dumper, data):
+ value = []
+
+ for item_key, item_value in data.items():
+ node_key = dumper.represent_data(item_key)
+ node_value = dumper.represent_data(item_value)
+
+ value.append((node_key, node_value))
+
+ return yaml.nodes.MappingNode(u'tag:yaml.org,2002:map', value)
+
+
def get_json_template(template_dir, template_name):
template_name = template_name + ".json"
with open(os.path.join(template_dir, template_name)) as f:
self.current_module_env = {}
self.base_output_dir = base_output_dir
self.env_cache = {}
+ self.module_incomplete = False
@classmethod
@abstractmethod
def replace(self, param_name, alt_message=None, single=False):
value = self.get_param(param_name, single)
+ value = None if value == "CHANGEME" else value
if value:
return value
- return alt_message or replace(param_name)
+ else:
+ self.module_incomplete = True
+ return alt_message or replace(param_name)
+
+ def start_module(self, module, env):
+ """Initialize/reset the environment for the module"""
+ self.current_module = module
+ self.current_module_env = env
+ self.module_incomplete = False
+ self.env_cache = {}
def generate_environments(self, module):
"""
print("\nGenerating Preloads for {}".format(module))
print("-" * 50)
print("... generating blank template")
- self.current_module = module
- self.current_module_env = {}
- self.env_cache = {}
+ self.start_module(module, {})
blank_preload_dir = self.make_preload_dir(self.base_output_dir)
self.generate_module(module, blank_preload_dir)
self.generate_preload_env(module, blank_preload_dir)
env.name, output_dir
)
)
- self.env_cache = {}
- self.current_module = module
- self.current_module_env = env.get_module(module.label)
+ self.start_module(module, env.get_module(module.label))
self.generate_module(module, output_dir)
- self.current_module = None
- self.current_module_env = None
def make_preload_dir(self, base_dir):
path = os.path.join(base_dir, self.output_sub_dir())
os.makedirs(path, exist_ok=True)
return path
- def generate_preload_env(self, module, blank_preload_dir):
+ @staticmethod
+ def generate_preload_env(module, blank_preload_dir):
"""
Create a .env template suitable for completing and using for
preload generation from env files.
"""
+ yaml.add_representer(OrderedDict, represent_ordered_dict)
output_dir = os.path.join(blank_preload_dir, "preload_env")
- output_file = os.path.join(output_dir, "{}.env".format(module.vnf_name))
+ env_file = os.path.join(output_dir, "{}.env".format(module.vnf_name))
+ defaults_file = os.path.join(output_dir, "defaults.yaml")
if not os.path.exists(output_dir):
os.makedirs(output_dir, exist_ok=True)
- with open(output_file, "w") as f:
+ with open(env_file, "w") as f:
yaml.dump(module.env_template, f)
+ if not os.path.exists(defaults_file):
+ with open(defaults_file, "w") as f:
+ yaml.dump({"vnf_name": "CHANGEME"}, f)
def get_param(self, param_name, single):
"""
import os
import shutil
from abc import ABC, abstractmethod
+from collections import OrderedDict
from preload.generator import yield_by_count
from preload.environment import PreloadEnvironment
Returns a a template .env file that can be completed to enable
preload generation.
"""
- params = {}
- params["vnf-name"] = CHANGE
+ params = OrderedDict()
+ params["vnf_name"] = CHANGE
params["vnf-type"] = CHANGE
params["vf-module-model-name"] = CHANGE
params["vf_module_name"] = CHANGE
+ for az in self.availability_zones:
+ params[az] = CHANGE
for network in self.networks:
params[network.name_param] = CHANGE
for param in set(network.subnet_params):
template = get_json_template(DATA_DIR, "preload_template")
self._populate(template, vnf_module)
vnf_name = vnf_module.vnf_name
- outfile = "{}/{}.json".format(output_dir, vnf_name)
+ incomplete = "_incomplete" if self.module_incomplete else ""
+ outfile = "{}/{}{}.json".format(output_dir, vnf_name, incomplete)
with open(outfile, "w") as f:
json.dump(template, f, indent=4)
def generate_module(self, vnf_module, output_dir):
preload = get_json_template(DATA_DIR, "preload_template")
self._populate(preload, vnf_module)
- outfile = "{}/{}.json".format(output_dir, vnf_module.vnf_name)
+ incomplete = "_incomplete" if self.module_incomplete else ""
+ outfile = "{}/{}{}.json".format(output_dir, vnf_module.vnf_name, incomplete)
with open(outfile, "w") as f:
json.dump(preload, f, indent=4)
+ def _populate(self, preload, vnf_module):
+ self._add_vnf_metadata(preload)
+ self._add_availability_zones(preload, vnf_module)
+ self._add_vnf_networks(preload, vnf_module)
+ self._add_vms(preload, vnf_module)
+ self._add_parameters(preload, vnf_module)
+
+ def _add_vnf_metadata(self, preload):
+ vnf_meta = preload["input"]["vnf-topology-information"]["vnf-topology-identifier"]
+ vnf_meta["vnf-name"] = self.replace("vnf_name")
+ vnf_meta["generic-vnf-type"] = self.replace(
+ "vnf-type",
+ "VALUE FOR: Concatenation of <Service Name>/"
+ "<VF Instance Name> MUST MATCH SDC",
+ )
+ vnf_meta["vnf-type"] = self.replace(
+ "vf-module-model-name", "VALUE FOR: <vfModuleModelName> from CSAR or SDC"
+ )
+
def add_floating_ips(self, network_template, network):
# only one floating IP is really supported, in the preload model
# so for now we'll just use the last one. We might revisit this
)
network_template["ip-count-ipv6"] += 1
- def _populate(self, preload, vnf_module):
- self._add_availability_zones(preload, vnf_module)
- self._add_vnf_networks(preload, vnf_module)
- self._add_vms(preload, vnf_module)
- self._add_parameters(preload, vnf_module)
-
def _add_availability_zones(self, preload, vnf_module):
zones = preload["input"]["vnf-topology-information"]["vnf-assignments"][
"availability-zones"
mock
openstack-heat
cached-property>=1.5,<1.6
+bandit
Code Review / vvp / validation-scripts.git / commitdiff