import pytest
-from preload.environment import PreloadEnvironment
+from preload.environment import EnvironmentFileDataSource
from preload.model import Vnf, get_heat_templates
from preload_grapi import GrApiPreloadGenerator
from tests.helpers import first
pytestconfig.getoption = fake_getoption
templates = get_heat_templates(pytestconfig)
- env = PreloadEnvironment(THIS_DIR / "sample_env")
vnf = Vnf(templates)
- generator = GrApiPreloadGenerator(vnf, session_dir, env)
+ datasource = EnvironmentFileDataSource(THIS_DIR / "sample_env")
+ generator = GrApiPreloadGenerator(vnf, session_dir, datasource)
generator.generate()
return session_dir
@pytest.fixture(scope="session")
def base(preload):
- return load_module(preload, "base_incomplete.json")
+ return load_module(preload, "base.json")
@pytest.fixture(scope="session")
def incremental(preload):
- return load_module(preload, "incremental_incomplete.json")
+ return load_module(preload, "incremental.json")
+
+
+def test_incomplete_filenames(preload):
+ base = THIS_DIR / "sample_env/preloads/grapi/base_incomplete.json"
+ inc = THIS_DIR / "sample_env/preloads/grapi/incremental_incomplete.json"
+ assert base.exists()
+ assert inc.exists()
def test_base_fields(base):
assert oam == {
"network-role": "oam",
"network-name": "VALUE FOR: network name of oam_net_id",
- "subnets-data": {"subnet-data": [{"subnet-id": "VALUE FOR: oam_subnet_id"}]},
+ "subnets-data": {
+ "subnet-data": [{"subnet-name": "VALUE FOR: name of oam_subnet_id"}]
+ },
}
"vm-network": [
{
"network-role": "oam",
+ "network-role-tag": "oam",
"network-information-items": {
"network-information-item": [
{
},
{
"network-role": "ha",
+ "network-role-tag": "ha",
"network-information-items": {
"network-information-item": [
{
"ip-version": "4",
- "use-dhcp": "N",
+ "use-dhcp": "Y",
"ip-count": 0,
"network-ips": {"network-ip": []},
},
{
"ip-version": "6",
- "use-dhcp": "N",
+ "use-dhcp": "Y",
"ip-count": 0,
"network-ips": {"network-ip": []},
},
params = base["input"]["preload-vf-module-topology-information"][
"vf-module-topology"
]["vf-module-parameters"]["param"]
- assert params == [
- {"name": "db_vol0_id", "value": "VALUE FOR: db_vol0_id"},
- {"name": "db_vol1_id", "value": "VALUE FOR: db_vol1_id"},
- ]
+ assert params == []
def test_incremental(incremental):
import pytest
from app_tests.preload_tests.test_grapi import load_json
-from preload.environment import PreloadEnvironment
+from preload.environment import EnvironmentFileDataSource
from preload.model import Vnf, get_heat_templates
from preload_vnfapi import VnfApiPreloadGenerator
from tests.helpers import load_yaml, first
pytestconfig.getoption = fake_getoption
templates = get_heat_templates(pytestconfig)
vnf = Vnf(templates)
- preload_env = PreloadEnvironment(THIS_DIR / "sample_env")
- generator = VnfApiPreloadGenerator(vnf, session_dir, preload_env)
+ datasource = EnvironmentFileDataSource(THIS_DIR / "sample_env")
+ generator = VnfApiPreloadGenerator(vnf, session_dir, datasource)
generator.generate()
return session_dir
@pytest.fixture(scope="session")
def base(preload):
- return load_module(preload, "base_incomplete.json")
+ return load_module(preload, "base.json")
@pytest.fixture(scope="session")
def incremental(preload):
- return load_module(preload, "incremental_incomplete.json")
+ return load_module(preload, "incremental.json")
def test_base_azs(base):
{
"network-role": "oam",
"network-name": "VALUE FOR: network name for oam_net_id",
- "subnet-id": "oam_subnet_id",
+ "subnet-name": "VALUE FOR: name for oam_subnet_id",
},
{"network-role": "ha", "network-name": "VALUE FOR: network name for ha_net_id"},
{
"network-role": "ctrl",
"network-name": "VALUE FOR: network name for ctrl_net_id",
- "subnet-id": "ctrl_subnet_id",
+ "subnet-name": "VALUE FOR: name for ctrl_subnet_id",
},
]
"network-ips-v6": [],
"network-macs": [],
"interface-route-prefixes": [],
- "use-dhcp": "N",
+ "use-dhcp": "Y",
},
],
}
def test_base_parameters(base):
params = base["input"]["vnf-topology-information"]["vnf-parameters"]
- assert params == [
- {
- "vnf-parameter-name": "db_vol0_id",
- "vnf-parameter-value": "VALUE FOR: db_vol0_id",
- },
- {
- "vnf-parameter-name": "db_vol1_id",
- "vnf-parameter-value": "VALUE FOR: db_vol1_id",
- },
- ]
+ assert params == []
def test_incremental(incremental):
import pytest
import yaml
-from config import Config, get_generator_plugin_names, to_uri
+from config import Config, to_uri
import vvp
-
+from preload.engine import PLUGIN_MGR
DEFAULT_CONFIG = """
namespace: {namespace}
def test_get_generator_plugin_names(config):
- names = get_generator_plugin_names()
+ names = [g.format_name() for g in PLUGIN_MGR.preload_generators]
assert "VNF-API" in names
assert "GR-API" in names
import importlib
-import inspect
import multiprocessing
import os
-import pkgutil
import queue
from configparser import ConfigParser
-from itertools import chain
from pathlib import Path
from typing import MutableMapping, Iterator, List, Optional, Dict
import yaml
from cached_property import cached_property
+from preload.engine import PLUGIN_MGR
from version import VERSION
-from preload.generator import AbstractPreloadGenerator
from tests.test_environment_file_parameters import ENV_PARAMETER_SPEC
PATH = os.path.dirname(os.path.realpath(__file__))
@property
def preload_formats(self):
excluded = self._config.get("excluded-preloads", [])
- formats = (cls.format_name() for cls in get_generator_plugins())
+ formats = [cls.format_name() for cls in PLUGIN_MGR.preload_generators]
return [f for f in formats if f not in excluded]
+ @property
+ def preload_source_types(self):
+ return [s.get_name() for s in PLUGIN_MGR.preload_sources]
+
@property
def default_preload_format(self):
default = self._user_settings.get("preload_format")
else:
return self.preload_formats[0]
+ @property
+ def default_preload_source(self):
+ default = self._user_settings.get("preload_source")
+ if default and default in self.preload_source_types:
+ return default
+ else:
+ return self.preload_source_types[0]
+
@staticmethod
def get_subdir_for_preload(preload_format):
- for gen in get_generator_plugins():
+ for gen in PLUGIN_MGR.preload_generators:
if gen.format_name() == preload_format:
return gen.output_sub_dir()
return ""
def flush(self):
"""No operation method to satisfy file-like behavior"""
pass
-
-
-def is_preload_generator(class_):
- """
- Returns True if the class is an implementation of AbstractPreloadGenerator
- """
- return (
- inspect.isclass(class_)
- and not inspect.isabstract(class_)
- and issubclass(class_, AbstractPreloadGenerator)
- )
-
-
-def get_generator_plugins():
- """
- Scan the system path for modules that are preload plugins and discover
- and return the classes that implement AbstractPreloadGenerator in those
- modules
- """
- preload_plugins = (
- importlib.import_module(name)
- for finder, name, ispkg in pkgutil.iter_modules()
- if name.startswith("preload_")
- )
- members = chain.from_iterable(
- inspect.getmembers(mod, is_preload_generator) for mod in preload_plugins
- )
- return [m[1] for m in members]
-
-
-def get_generator_plugin_names():
- return [g.format_name() for g in get_generator_plugins()]
{
- "created": "2019-10-08T14:31:11.404157",
+ "created": "2019-12-03T06:33:16.165894",
"current_version": "el alto",
"project": "",
"versions": {
"needs_amount": 813
},
"el alto": {
- "created": "2019-10-08T14:31:11.404078",
+ "created": "2019-12-03T06:33:16.165821",
"filters": {},
"filters_amount": 0,
"needs": {
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Testing",
"sections": [
"Testing",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"sections": [
"Configuration Management via Chef",
"Resource Configuration",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Licensing Requirements",
"sections": [
"Licensing Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"validation_mode": ""
},
"R-146092": {
- "description": "If one or more non-MANO artifact(s) is included in the VNF or PNF TOSCA CSAR\npackage, the Manifest file in this CSAR package **MUST** contain: non-MANO\nartifact set which MAY contain following ONAP public tag.\n\n - onap_ves_events: contains VES registration files\n\n - onap_pm_dictionary: contains the PM dictionary files\n\n - onap_yang_modules: contains Yang module files for configurations\n\n - onap_ansible_playbooks: contains any ansible_playbooks\n\n - onap_others: contains any other non_MANO artifacts, e.g. informational\n documents",
+ "description": "If one or more non-MANO artifact(s) is included in the VNF or PNF CSAR\npackage, the Manifest file in this CSAR package **MUST** contain one or more\nof the following ONAP non-MANO artifact set identifier(s):\n\n - onap_ves_events: contains VES registration files\n\n - onap_pm_dictionary: contains the PM dictionary files\n\n - onap_yang_modules: contains Yang module files for configurations\n\n - onap_ansible_playbooks: contains any ansible_playbooks\n\n - onap_pnf_sw_information: contains the PNF software information file\n\n - onap_others: contains any other non_MANO artifacts, e.g. informational\n documents\n\n *NOTE: According to ETSI SOL004 v.2.6.1, every non-MANO artifact set shall be\n identified by a non-MANO artifact set identifier which shall be registered in\n the ETSI registry. Approved ONAP non-MANO artifact set identifiers are documented\n in the following page* https://wiki.onap.org/display/DW/ONAP+Non-MANO+Artifacts+Set+Identifiers",
"docname": "Chapter5/Tosca/ONAP VNF or PNF CSAR Package",
"full_title": "",
"hide_links": "",
],
"status": null,
"tags": [],
- "target": "VNF or PNF TOSCA PACKAGE",
+ "target": "VNF or PNF CSAR PACKAGE",
"test": "",
"test_case": "",
"test_file": "",
"title_from_content": "",
"type": "req",
"type_name": "Requirement",
- "updated": "",
+ "updated": "frankfurt",
"validated_by": "",
"validation_mode": ""
},
"sections": [
"Configuration Management via Ansible",
"Resource Configuration",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"sections": [
"Configuration Management via Ansible",
"Resource Configuration",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"sections": [
"Configuration Management via Chef",
"Resource Configuration",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Compute, Network, and Storage Requirements",
"sections": [
"Compute, Network, and Storage Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"validated_by": "",
"validation_mode": ""
},
+ "R-225891": {
+ "description": "A VNF's Heat Orchestration Template parameter declaration\n**MAY** contain the attribute ``tags:``.",
+ "docname": "Chapter5/Heat/ONAP Heat Orchestration Template Format",
+ "full_title": "",
+ "hide_links": "",
+ "id": "R-225891",
+ "id_complete": "R-225891",
+ "id_parent": "R-225891",
+ "impacts": "",
+ "introduced": "el alto",
+ "is_need": true,
+ "is_part": false,
+ "keyword": "MAY",
+ "links": [],
+ "notes": "",
+ "parts": {},
+ "section_name": "tags",
+ "sections": [
+ "tags",
+ "parameters",
+ "Heat Orchestration Template Structure",
+ "ONAP Heat Orchestration Template Format"
+ ],
+ "status": null,
+ "tags": [],
+ "target": "VNF",
+ "test": "",
+ "test_case": "",
+ "test_file": "",
+ "title": "",
+ "title_from_content": "",
+ "type": "req",
+ "type_name": "Requirement",
+ "updated": "",
+ "validated_by": "",
+ "validation_mode": ""
+ },
"R-22608": {
"description": "When a VNF's Heat Orchestration Template's Base Module's output\nparameter is declared as an input parameter in an Incremental Module,\nthe parameter attribute ``constraints:`` **SHOULD NOT** be declared.",
"docname": "Chapter5/Heat/ONAP Heat Orchestration Templates Overview",
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"validation_mode": "static"
},
"R-23664": {
- "description": "A VNF's Heat Orchestration template **MUST**\ncontain the section ``resources:``.",
+ "description": "A VNF's Heat Orchestration template's base module, incremental\nmodule, and volume module **MUST**\ncontain the section ``resources:``.",
"docname": "Chapter5/Heat/ONAP Heat Orchestration Template Format",
"full_title": "",
"hide_links": "",
"title_from_content": "",
"type": "req",
"type_name": "Requirement",
- "updated": "",
+ "updated": "frankfurt",
"validated_by": "",
"validation_mode": "static"
},
"section_name": "Compute, Network, and Storage Requirements",
"sections": [
"Compute, Network, and Storage Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Licensing Requirements",
"sections": [
"Licensing Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"sections": [
"Configuration Management via NETCONF/YANG",
"Resource Configuration",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"type_name": "Requirement",
"updated": "",
"validated_by": "",
- "validation_mode": ""
+ "validation_mode": "none"
},
"R-32155": {
"description": "The VNFD provided by VNF vendor may use the below described TOSCA\ninterface types. An on-boarding entity (ONAP SDC) **MUST** support them.\n\n **tosca.interfaces.nfv.vnf.lifecycle.Nfv** supports LCM operations",
"type_name": "Requirement",
"updated": "",
"validated_by": "",
- "validation_mode": ""
+ "validation_mode": "none"
},
"R-32636": {
"description": "The VNF **MUST** support API-based monitoring to take care of\nthe scenarios where the control interfaces are not exposed, or are\noptimized and proprietary in nature.",
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Compute, Network, and Storage Requirements",
"sections": [
"Compute, Network, and Storage Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
],
"status": null,
"tags": [],
- "target": "",
+ "target": "VNF",
"test": "",
"test_case": "",
"test_file": "",
"section_name": "Licensing Requirements",
"sections": [
"Licensing Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Testing",
"sections": [
"Testing",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Licensing Requirements",
"sections": [
"Licensing Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Licensing Requirements",
"sections": [
"Licensing Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Compute, Network, and Storage Requirements",
"sections": [
"Compute, Network, and Storage Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"sections": [
"Configuration Management via Ansible",
"Resource Configuration",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Licensing Requirements",
"sections": [
"Licensing Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Compute, Network, and Storage Requirements",
"sections": [
"Compute, Network, and Storage Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Testing",
"sections": [
"Testing",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"sections": [
"Configuration Management via Ansible",
"Resource Configuration",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Licensing Requirements",
"sections": [
"Licensing Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Licensing Requirements",
"sections": [
"Licensing Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"type_name": "Requirement",
"updated": "dublin",
"validated_by": "",
- "validation_mode": "static"
+ "validation_mode": "none"
},
"R-88899": {
"description": "The VNF or PNF **MUST** support simultaneous <commit> operations\nwithin the context of this locking requirements framework.",
"section_name": "Resource Configuration",
"sections": [
"Resource Configuration",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Control Loop",
"sections": [
"Resource Control Loop",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"type_name": "Requirement",
"updated": "casablanca",
"validated_by": "",
- "validation_mode": ""
+ "validation_mode": "none"
},
"R-91342": {
"description": "A VNF Heat Orchestration Template's Base Module's Environment File\n**MUST** be named identical to the VNF Heat Orchestration Template's\nBase Module with ``.y[a]ml`` replaced with ``.env``.",
],
"status": null,
"tags": [],
- "target": "",
+ "target": "VNF",
"test": "",
"test_case": "",
"test_file": "",
"section_name": "Compute, Network, and Storage Requirements",
"sections": [
"Compute, Network, and Storage Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Compute, Network, and Storage Requirements",
"sections": [
"Compute, Network, and Storage Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"validated_by": "",
"validation_mode": "static"
},
+ "R-972082": {
+ "description": "If the Manifest file in the PNF CSAR package includes \"onap_pnf_sw_information\"\nas a non-MANO artifact set identifiers, then the PNF software information file is\nincluded in the package and it **MUST** be compliant to:\n\n- The file extension which contains the PNF software version must be .yaml\n\n- The PNF software version information must be specified as following:\n\n pnf_software_information:\n\n - pnf_software_version: \"<version>\"",
+ "docname": "Chapter5/Tosca/ONAP VNF or PNF CSAR Package",
+ "full_title": "",
+ "hide_links": "",
+ "id": "R-972082",
+ "id_complete": "R-972082",
+ "id_parent": "R-972082",
+ "impacts": "",
+ "introduced": "frankfurt",
+ "is_need": true,
+ "is_part": false,
+ "keyword": "MUST",
+ "links": [],
+ "notes": "",
+ "parts": {},
+ "section_name": "VNF Package Contents",
+ "sections": [
+ "VNF Package Contents",
+ "VNF or PNF CSAR Package"
+ ],
+ "status": null,
+ "tags": [],
+ "target": "PNF CSAR PACKAGE",
+ "test": "",
+ "test_case": "",
+ "test_file": "",
+ "title": "",
+ "title_from_content": "",
+ "type": "req",
+ "type_name": "Requirement",
+ "updated": "",
+ "validated_by": "",
+ "validation_mode": ""
+ },
"R-97293": {
"description": "The VNF or PNF provider **MUST NOT** require audits\nof Service Provider's business.",
"docname": "Chapter7/VNF-On-boarding-and-package-management",
"section_name": "Licensing Requirements",
"sections": [
"Licensing Requirements",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"section_name": "Resource Description",
"sections": [
"Resource Description",
- "VNF On-boarding and package management"
+ "VNF and PNF On-boarding and package management"
],
"status": null,
"tags": [],
"validation_mode": "static"
}
},
- "needs_amount": 819
+ "needs_amount": 821
}
}
}
\ No newline at end of file
# limitations under the License.
#
# ============LICENSE_END============================================
+
+from preload.environment import EnvironmentFileDataSource
+
+__all__ = ["EnvironmentFileDataSource"]
--- /dev/null
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Iterable, Any, Optional, Mapping
+
+from preload.model import VnfModule
+
+
+class AbstractPreloadInstance(ABC):
+ """
+ Represents the data source for a single instance of a preload for
+ any format. The implementation of AbstractPreloadGenerator will
+ call the methods of this class to retrieve the necessary data
+ to populate the preload. If a data element is not available,
+ then simply return ``None`` and a suitable placeholder will be
+ placed in the preload.
+ """
+
+ @property
+ @abstractmethod
+ def output_dir(self) -> Path:
+ """
+ Base output directory where the preload will be generated. Please
+ note, that the generator may create nested directories under this
+ directory for the preload.
+
+ :return: Path to the desired output directory. This directory
+ and its parents will be created by the generator if
+ it is not already present.
+ """
+ raise NotImplementedError()
+
+ @property
+ @abstractmethod
+ def module_label(self) -> str:
+ """
+ Identifier of the module. This must match the base name of the
+ heat module (ex: if the Heat file name is base.yaml, then the label
+ is 'base'.
+
+ :return: string name of the module
+ """
+ raise NotImplementedError()
+
+ @property
+ @abstractmethod
+ def vf_module_name(self) -> Optional[str]:
+ """
+ :return: module name to populate in the preload if available
+ """
+ raise NotImplementedError()
+
+ @property
+ @abstractmethod
+ def flag_incompletes(self) -> bool:
+ """
+ If True, then the generator will modify the file name of any
+ generated preload to end with _incomplete.<ext> if any preload
+ value was not satisfied by the data source. If False, then
+ the file name will be the same regardless of the completeness
+ of the preload.
+
+ :return: True if file names should denote preload incompleteness
+ """
+ raise NotImplementedError()
+
+ @property
+ @abstractmethod
+ def preload_basename(self) -> str:
+ """
+ Base name of the preload that will be used by the generator to create
+ the file name.
+ """
+ raise NotImplementedError()
+
+ @property
+ @abstractmethod
+ def vnf_name(self) -> Optional[str]:
+ """
+ :return: the VNF name to populate in the prelad if available
+ """
+ raise NotImplementedError()
+
+ @property
+ @abstractmethod
+ def vnf_type(self) -> Optional[str]:
+ """
+ The VNF Type must be match the values in SDC. It is a concatenation
+ of <Service Instance Name>/<Resource Instance Name>.
+
+ :return: VNF Type to populate in the preload if available
+ """
+ raise NotImplementedError()
+
+ @property
+ @abstractmethod
+ def vf_module_model_name(self) -> Optional[str]:
+ """
+ :return: Module model name if available
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_availability_zone(self, index: int, param_name: str) -> Optional[str]:
+ """
+ Retrieve the value for the availability zone at requested zero-based
+ index (i.e. 0, 1, 2, etc.)
+
+ :param index: index of availability zone (0, 1, etc.)
+ :param param_name: Name of the parameter from Heat
+ :return: value for the AZ if available
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_network_name(self, network_role: str, name_param: str) -> Optional[str]:
+ """
+ Retrieve the OpenStack name of the network for the given network role.
+
+ :param network_role: Network role from Heat template
+ :param name_param: Network name parameter from Heat
+ :return: Name of the network if available
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_subnet_id(
+ self, network_role: str, ip_version: int, param_name: str
+ ) -> Optional[str]:
+ """
+ Retrieve the subnet's UUID for the given network and IP version (4 or 6).
+
+ :param network_role: Network role from Heat template
+ :param ip_version: IP Version (4 or 6)
+ :param param_name: Parameter name from Heat
+ :return: UUID of the subnet if available
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_subnet_name(
+ self, network_role: str, ip_version: int, param_name: str
+ ) -> Optional[str]:
+ """
+ Retrieve the OpenStack Subnet name for the given network role and IP version
+
+ :param network_role: Network role from Heat template
+ :param ip_version: IP Version (4 or 6)
+ :param param_name: Parameter name from Heat
+ :return: Name of the subnet if available
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_vm_name(self, vm_type: str, index: int, param_name: str) -> Optional[str]:
+ """
+ Retrieve the vm name for the given VM type and index.
+
+ :param vm_type: VM Type from Heat template
+ :param index: Zero-based index of the VM for the vm-type
+ :param param_name: Parameter name from Heat
+ :return: VM Name if available
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_floating_ip(
+ self, vm_type: str, network_role: str, ip_version: int, param_name: str
+ ) -> Optional[str]:
+ """
+ Retreive the floating IP for the VM and Port identified by VM Type,
+ Network Role, and IP Version.
+
+ :param vm_type: VM Type from Heat template
+ :param network_role: Network Role from Heat template
+ :param ip_version: IP Version (4 or 6)
+ :param param_name: Parameter name from Heat
+ :return: floating IP address if available
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_fixed_ip(
+ self, vm_type: str, network_role: str, ip_version: int, index: int, param: str
+ ) -> Optional[str]:
+ """
+ Retreive the fixed IP for the VM and Port identified by VM Type,
+ Network Role, IP Version, and index.
+
+ :param vm_type: VM Type from Heat template
+ :param network_role: Network Role from Heat template
+ :param ip_version: IP Version (4 or 6)
+ :param index: zero-based index for the IP for the given
+ VM Type, Network Role, IP Version combo
+ :param param_name: Parameter name from Heat
+ :return: floating IP address if available
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_vnf_parameter(self, key: str, value: Any) -> Optional[Any]:
+ """
+ Retrieve the value for the given key. These will be placed in the
+ tag-values/vnf parameters in the preload. If a value was specified in
+ the environment packaged in the Heat for for the VNF module, then
+ that value will be passed in ``value``. This class can return
+ the value or ``None`` if it does not have a value for the given key.
+
+ :param key: parameter name from Heat
+ :param value: Value from Heat env file if it was assigned there;
+ None otherwise
+ :return: Returns the value for the object. This should
+ be a str, dict, or list. The generator will
+ format it properly based on the selected output format
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_additional_parameters(self) -> Mapping[str, Any]:
+ """
+ Return any additional parameters that should be added to the VNF parameters.
+
+ This can be useful if you want to duplicate paramters in tag values that are
+ also in the other sections (ex: VM names).
+
+ :return: dict of str to object mappings that the generator must add to
+ the vnf_parameters/tag values
+ """
+ raise NotImplementedError()
+
+
+class AbstractPreloadDataSource(ABC):
+ """
+ Represents a data source for a VNF preload data. Implementations of this
+ class can be dynamically discovered if they are in a preload plugin module.
+ A module is considered a preload plugin module if it starts with
+ prelaod_ and is available as a top level module on Python's sys.path.
+
+ The ``get_module_preloads`` will be invoked for each module in
+ the VNF. An instance of AbstractPreloadInstance must be returned for
+ each instance of the preload module that is to be created.
+
+ Parameters:
+ :param path: The path to the configuration source selected
+ in either the VVP GUI or command-line. This
+ may be a file or directory depending upon
+ the source_type defined by this data source
+ """
+
+ def __init__(self, path: Path):
+ self.path = path
+
+ @classmethod
+ @abstractmethod
+ def get_source_type(cls) -> str:
+ """
+ If 'FILE' returned, then the config source will be a specific
+ file; If 'DIR', then the config source will be a directory
+ :return:
+ """
+ raise NotImplementedError()
+
+ @classmethod
+ @abstractmethod
+ def get_identifier(cls) -> str:
+ """
+ Identifier for the given data source. This is the value that
+ can be passed via --preload-source-type.
+
+ :return: short identifier for this data source type
+ """
+
+ @classmethod
+ @abstractmethod
+ def get_name(self) -> str:
+ """
+ Human readable name to describe the preload data source. It is
+ recommended not to exceed 50 characters.
+
+ :return: human readable name of the preload data source (ex: Environment Files)
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def get_module_preloads(
+ self, module: VnfModule
+ ) -> Iterable[AbstractPreloadInstance]:
+ """
+ For the requested module, return an instance of AbstractPreloadInstance
+ for every preload module you wish to be created.
+
+ :param module: Module of the VNF
+ :return: iterable of preloads to create for the given module
+ """
+ raise NotImplementedError()
+
+
+class BlankPreloadInstance(AbstractPreloadInstance):
+ """
+ Used to create blank preload templates. VVP will always create
+ a template of a preload in the requested format with no data provided.
+ """
+
+ def __init__(self, output_dir: Path, module_name: str):
+ self._output_dir = output_dir
+ self._module_name = module_name
+
+ @property
+ def flag_incompletes(self) -> bool:
+ return False
+
+ @property
+ def preload_basename(self) -> str:
+ return self._module_name
+
+ @property
+ def vf_module_name(self) -> Optional[str]:
+ return None
+
+ def get_vm_name(self, vm_type: str, index: int, param_name: str) -> Optional[str]:
+ return None
+
+ def get_availability_zone(self, index: int, param_name: str) -> Optional[str]:
+ return None
+
+ @property
+ def output_dir(self) -> Path:
+ return self._output_dir
+
+ @property
+ def module_label(self) -> str:
+ return self._module_name
+
+ @property
+ def vnf_name(self) -> Optional[str]:
+ return None
+
+ @property
+ def vnf_type(self) -> Optional[str]:
+ return None
+
+ @property
+ def vf_module_model_name(self) -> Optional[str]:
+ return None
+
+ def get_network_name(self, network_role: str, name_param: str) -> Optional[str]:
+ return None
+
+ def get_subnet_id(
+ self, network_role: str, ip_version: int, param_name: str
+ ) -> Optional[str]:
+ return None
+
+ def get_subnet_name(
+ self, network_role: str, ip_version: int, param_name: str
+ ) -> Optional[str]:
+ return None
+
+ def get_floating_ip(
+ self, vm_type: str, network_role: str, ip_version: int, param_name: str
+ ) -> Optional[str]:
+ return None
+
+ def get_fixed_ip(
+ self, vm_type: str, network_role: str, ip_version: int, index: int, param: str
+ ) -> Optional[str]:
+ return None
+
+ def get_vnf_parameter(self, key: str, value: Any) -> Optional[Any]:
+ return None
+
+ def get_additional_parameters(self) -> Mapping[str, Any]:
+ return {}
--- /dev/null
+import importlib
+import inspect
+import os
+import pkgutil
+import shutil
+from itertools import chain
+from pathlib import Path
+from typing import List, Type
+
+from preload.data import AbstractPreloadDataSource
+from preload.generator import AbstractPreloadGenerator
+from preload.model import get_heat_templates, Vnf
+from tests.helpers import get_output_dir
+
+
+def create_preloads(config, exitstatus):
+ """
+ Create preloads in every format that can be discovered by get_generator_plugins
+ """
+ if config.getoption("self_test"):
+ return
+ print("+===================================================================+")
+ print("| Preload Template Generation |")
+ print("+===================================================================+")
+
+ preload_dir = os.path.join(get_output_dir(config), "preloads")
+ if os.path.exists(preload_dir):
+ shutil.rmtree(preload_dir)
+ plugins = PluginManager()
+ available_formats = [p.format_name() for p in plugins.preload_generators]
+ selected_formats = config.getoption("preload_formats") or available_formats
+ preload_source = None
+ if config.getoption("preload_source"):
+ preload_source_path = Path(config.getoption("preload_source"))
+ source_class = plugins.get_source_for_id(
+ config.getoption("preload_source_type")
+ )
+ preload_source = source_class(preload_source_path)
+
+ heat_templates = get_heat_templates(config)
+ vnf = None
+ for plugin_class in plugins.preload_generators:
+ if plugin_class.format_name() not in selected_formats:
+ continue
+ vnf = Vnf(heat_templates)
+ generator = plugin_class(vnf, preload_dir, preload_source)
+ generator.generate()
+ if vnf and vnf.uses_contrail:
+ print(
+ "\nWARNING: Preload template generation does not support Contrail\n"
+ "at this time, but Contrail resources were detected. The preload \n"
+ "template may be incomplete."
+ )
+ if exitstatus != 0:
+ print(
+ "\nWARNING: Heat violations detected. Preload templates may be\n"
+ "incomplete or have errors."
+ )
+
+
+def is_implementation_of(class_, base_class):
+ """
+ Returns True if the class is an implementation of AbstractPreloadGenerator
+ """
+ return (
+ inspect.isclass(class_)
+ and not inspect.isabstract(class_)
+ and issubclass(class_, base_class)
+ )
+
+
+def get_implementations_of(class_, modules):
+ """
+ Returns all classes that implement ``class_`` from modules
+ """
+ members = list(
+ chain.from_iterable(
+ inspect.getmembers(mod, lambda c: is_implementation_of(c, class_))
+ for mod in modules
+ )
+ )
+ return [m[1] for m in members]
+
+
+class PluginManager:
+ def __init__(self):
+ self.preload_plugins = [
+ importlib.import_module(name)
+ for finder, name, ispkg in pkgutil.iter_modules()
+ if name.startswith("preload_") or name == "preload"
+ ]
+ self.preload_generators: List[
+ Type[AbstractPreloadGenerator]
+ ] = get_implementations_of(AbstractPreloadGenerator, self.preload_plugins)
+ self.preload_sources: List[
+ Type[AbstractPreloadDataSource]
+ ] = get_implementations_of(AbstractPreloadDataSource, self.preload_plugins)
+
+ def get_source_for_id(self, identifier: str) -> Type[AbstractPreloadDataSource]:
+ for source in self.preload_sources:
+ if identifier == source.get_identifier():
+ return source
+ raise RuntimeError(
+ "Unable to find preload source for identifier {}".format(identifier)
+ )
+
+ def get_source_for_name(self, name: str) -> Type[AbstractPreloadDataSource]:
+ for source in self.preload_sources:
+ if name == source.get_name():
+ return source
+ raise RuntimeError("Unable to find preload source for name {}".format(name))
+
+
+PLUGIN_MGR = PluginManager()
import re
import tempfile
from pathlib import Path
+from typing import Any, Optional, Mapping
from cached_property import cached_property
+from preload.data import AbstractPreloadInstance, AbstractPreloadDataSource
+from preload.model import VnfModule
from tests.helpers import check, first, unzip, load_yaml
SERVICE_TEMPLATE_PATTERN = re.compile(r".*service-.*?-template.yml")
RESOURCE_TEMPLATE_PATTERN = re.compile(r".*resource-(.*?)-template.yml")
+ZONE_PARAMS = ("availability_zone_0", "availability_zone_1", "availability_zone_2")
+
def yaml_files(path):
"""
def __repr__(self):
return "PreloadEnvironment(name={})".format(self.name)
+
+
+class EnvironmentFilePreloadInstance(AbstractPreloadInstance):
+
+ def __init__(self, env: PreloadEnvironment, module_label: str, module_params: dict):
+ self.module_params = module_params
+ self._module_label = module_label
+ self.env = env
+ self.env_cache = {}
+
+ @property
+ def flag_incompletes(self) -> bool:
+ return True
+
+ @property
+ def preload_basename(self) -> str:
+ return self.module_label
+
+ @property
+ def output_dir(self) -> Path:
+ return self.env.base_dir.joinpath("preloads")
+
+ @property
+ def module_label(self) -> str:
+ return self._module_label
+
+ @property
+ def vf_module_name(self) -> str:
+ return self.get_param("vf_module_name")
+
+ @property
+ def vnf_name(self) -> Optional[str]:
+ return self.get_param("vnf_name")
+
+ @property
+ def vnf_type(self) -> Optional[str]:
+ return self.get_param("vnf-type")
+
+ @property
+ def vf_module_model_name(self) -> Optional[str]:
+ return self.get_param("vf-module-model-name")
+
+ def get_availability_zone(self, index: int, param_name: str) -> Optional[str]:
+ return self.get_param(param_name)
+
+ def get_network_name(self, network_role: str, name_param: str) -> Optional[str]:
+ return self.get_param(name_param)
+
+ def get_subnet_id(
+ self, network_role: str, ip_version: int, param_name: str
+ ) -> Optional[str]:
+ return self.get_param(param_name)
+
+ def get_subnet_name(
+ self, network_role: str, ip_version: int, param_name: str
+ ) -> Optional[str]:
+ # Not supported with env files
+ return None
+
+ def get_vm_name(self, vm_type: str, index: int, param_name: str) -> Optional[str]:
+ return self.get_param(param_name, single=True)
+
+ def get_floating_ip(
+ self, vm_type: str, network_role: str, ip_version: int, param_name: str
+ ) -> Optional[str]:
+ return self.get_param(param_name)
+
+ def get_fixed_ip(
+ self, vm_type: str, network_role: str, ip_version: int, index: int, param: str
+ ) -> Optional[str]:
+ return self.get_param(param, single=True)
+
+ def get_vnf_parameter(self, key: str, value: Any) -> Optional[str]:
+ module_value = self.get_param(key)
+ return module_value or value
+
+ def get_additional_parameters(self) -> Mapping[str, Any]:
+ return {}
+
+ def get_param(self, param_name, single=False):
+ """
+ Retrieves the value for the given param if it exists. If requesting a
+ single item, and the parameter is tied to a list then only one item from
+ the list will be returned. For each subsequent call with the same parameter
+ it will iterate/rotate through the values in that list. If single is False
+ then the full list will be returned.
+
+ :param param_name: name of the parameter
+ :param single: If True returns single value from lists otherwises the full
+ list. This has no effect on non-list values
+ """
+ value = self.env_cache.get(param_name)
+ if not value:
+ value = self.module_params.get(param_name)
+ if isinstance(value, list):
+ value = value.copy()
+ value.reverse()
+ self.env_cache[param_name] = value
+
+ if value and single and isinstance(value, list):
+ result = value.pop()
+ else:
+ result = value
+ return result if result != "CHANGEME" else None
+
+
+class EnvironmentFileDataSource(AbstractPreloadDataSource):
+
+ def __init__(self, path: Path):
+ super().__init__(path)
+ check(path.is_dir(), f"{path} must be an existing directory")
+ self.path = path
+ self.env = PreloadEnvironment(path)
+
+ @classmethod
+ def get_source_type(cls) -> str:
+ return "DIR"
+
+ @classmethod
+ def get_identifier(self) -> str:
+ return "envfiles"
+
+ @classmethod
+ def get_name(self) -> str:
+ return "Environment Files"
+
+ def get_module_preloads(self, module: VnfModule):
+ for env in self.env.environments:
+ module_params = env.get_module(module.label)
+ yield EnvironmentFilePreloadInstance(env, module.label, module_params)
import os
from abc import ABC, abstractmethod
from collections import OrderedDict
+from pathlib import Path
import yaml
+from preload.data import (
+ AbstractPreloadDataSource,
+ AbstractPreloadInstance,
+ BlankPreloadInstance,
+)
+from preload.model import VnfModule, Vnf
+
def represent_ordered_dict(dumper, data):
value = []
return new_template
-def yield_by_count(sequence):
- """
- Iterates through sequence and yields each item according to its __count__
- attribute. If an item has a __count__ of it will be returned 3 times
- before advancing to the next item in the sequence.
-
- :param sequence: sequence of dicts (must contain __count__)
- :returns: generator of tuple key, value pairs
- """
- for key, value in sequence.items():
- for i in range(value["__count__"]):
- yield (key, value)
-
-
-def replace(param):
+def replace(param, index=None):
"""
Optionally used by the preload generator to wrap items in the preload
that need to be replaced by end users
- :param param: p
+ :param param: parameter name
+ :param index: optional index (int or str) of the parameter
"""
+ if (param.endswith("_names") or param.endswith("_ips")) and index is not None:
+ param = "{}[{}]".format(param, index)
return "VALUE FOR: {}".format(param) if param else ""
:param vnf: Instance of Vnf that contains the preload data
:param base_output_dir: Base directory to house the preloads. All preloads
must be written to a subdirectory under this directory
+ :param data_source: Source data for preload population
"""
- def __init__(self, vnf, base_output_dir, preload_env):
- self.preload_env = preload_env
+ def __init__(
+ self, vnf: Vnf, base_output_dir: Path, data_source: AbstractPreloadDataSource
+ ):
+ self.data_source = data_source
self.vnf = vnf
- self.current_module = None
- self.current_module_env = {}
self.base_output_dir = base_output_dir
- self.env_cache = {}
self.module_incomplete = False
@classmethod
raise NotImplementedError()
@abstractmethod
- def generate_module(self, module, output_dir):
+ def generate_module(self, module: VnfModule, preload: AbstractPreloadInstance, output_dir: Path):
"""
- Create the preloads and write them to ``output_dir``. This
- method is responsible for generating the content of the preload and
- writing the file to disk.
+ Create the preloads. This method is responsible for generating the
+ content of the preload and writing the file to disk.
"""
raise NotImplementedError()
# handle the base module first
print("\nGenerating {} preloads".format(self.format_name()))
if self.vnf.base_module:
- self.generate_environments(self.vnf.base_module)
+ self.generate_preloads(self.vnf.base_module)
if self.supports_output_passing():
self.vnf.filter_base_outputs()
for mod in self.vnf.incremental_modules:
- self.generate_environments(mod)
+ self.generate_preloads(mod)
- 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
- else:
- self.module_incomplete = True
- return alt_message or replace(param_name)
-
- def start_module(self, module, env):
+ def start_module(self):
"""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):
+ def generate_preloads(self, module):
"""
Generate a preload for the given module in all available environments
in the ``self.preload_env``. This will invoke the abstract
print("\nGenerating Preloads for {}".format(module))
print("-" * 50)
print("... generating blank template")
- 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)
- if self.preload_env:
- for env in self.preload_env.environments:
- output_dir = self.make_preload_dir(env.base_dir / "preloads")
+ self.start_module()
+ preload = BlankPreloadInstance(Path(self.base_output_dir), module.label)
+ blank_preload_dir = self.make_preload_dir(preload)
+ self.generate_module(module, preload, blank_preload_dir)
+ self.generate_preload_env(module, preload)
+
+ if self.data_source:
+ preloads = self.data_source.get_module_preloads(module)
+ for preload in preloads:
+ output_dir = self.make_preload_dir(preload)
print(
- "... generating preload for env ({}) to {}".format(
- env.name, output_dir
+ "... generating preload for {} to {}".format(
+ preload.module_label, output_dir
)
)
- self.start_module(module, env.get_module(module.label))
- self.generate_module(module, output_dir)
+ self.start_module()
+ self.generate_module(module, preload, output_dir)
- def make_preload_dir(self, base_dir):
- path = os.path.join(base_dir, self.output_sub_dir())
- if not os.path.exists(path):
- os.makedirs(path, exist_ok=True)
- return path
+ def make_preload_dir(self, preload: AbstractPreloadInstance):
+ preload_dir = preload.output_dir.joinpath(self.output_sub_dir())
+ preload_dir.mkdir(parents=True, exist_ok=True)
+ return preload_dir
@staticmethod
- def generate_preload_env(module, blank_preload_dir):
+ def generate_preload_env(module: VnfModule, preload: AbstractPreloadInstance):
"""
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")
- 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(env_file, "w") as f:
+ output_dir = preload.output_dir.joinpath("preload_env")
+ env_file = output_dir.joinpath("{}.env".format(module.label))
+ defaults_file = output_dir.joinpath("defaults.yaml")
+ output_dir.mkdir(parents=True, exist_ok=True)
+ with env_file.open("w") as f:
yaml.dump(module.env_template, f)
- if not os.path.exists(defaults_file):
- with open(defaults_file, "w") as f:
+ if not defaults_file.exists():
+ with defaults_file.open("w") as f:
yaml.dump({"vnf_name": "CHANGEME"}, f)
- def get_param(self, param_name, single):
- """
- Retrieves the value for the given param if it exists. If requesting a
- single item, and the parameter is tied to a list then only one item from
- the list will be returned. For each subsequent call with the same parameter
- it will iterate/rotate through the values in that list. If single is False
- then the full list will be returned.
-
- :param param_name: name of the parameter
- :param single: If True returns single value from lists otherwises the full
- list. This has no effect on non-list values
- """
- value = self.env_cache.get(param_name)
- if not value:
- value = self.current_module_env.get(param_name)
- if isinstance(value, list):
- value = value.copy()
- value.reverse()
- self.env_cache[param_name] = value
- if value and single and isinstance(value, list):
- return value.pop()
+ def normalize(self, preload_value, param_name, alt_message=None, index=None):
+ preload_value = None if preload_value == "CHANGEME" else preload_value
+ if preload_value:
+ return preload_value
else:
- return value
+ self.module_incomplete = True
+ return alt_message or replace(param_name, index)
#
# ============LICENSE_END============================================
import os
-import shutil
from abc import ABC, abstractmethod
from collections import OrderedDict
+from itertools import chain
+from typing import Tuple, List
-from preload.generator import yield_by_count
-from preload.environment import PreloadEnvironment
from tests.helpers import (
get_param,
get_environment_pair,
prop_iterator,
- get_output_dir,
is_base_module,
remove,
)
from tests.test_environment_file_parameters import get_preload_excluded_parameters
from tests.utils import nested_dict
from tests.utils.vm_types import get_vm_type_for_nova_server
-from config import Config, get_generator_plugins
from tests.test_environment_file_parameters import ENV_PARAMETER_SPEC
self.subnet_params = set()
def filter_output_params(self, base_outputs):
- self.subnet_params = remove(self.subnet_params, base_outputs)
+ self.subnet_params = remove(
+ self.subnet_params, base_outputs, key=lambda s: s.param_name
+ )
def __hash__(self):
return hash(self.network_role)
return hash(self) == hash(other)
+class Subnet:
+ def __init__(self, param_name: str):
+ self.param_name = param_name
+
+ @property
+ def ip_version(self):
+ return 6 if "_v6_" in self.param_name else 4
+
+ def __hash__(self):
+ return hash(self.param_name)
+
+ def __eq__(self, other):
+ return hash(self) == hash(other)
+
+
class Port(FilterBaseOutputs):
def __init__(self, vm, network):
self.vm = vm
self.network = network
self.fixed_ips = []
- self.floating_ips = []
+ self.floating_ips = set()
self.uses_dhcp = True
def add_ips(self, props):
self.uses_dhcp = False
self.fixed_ips.append(IpParam(ip_address, self))
if subnet:
- self.network.subnet_params.add(subnet)
+ self.network.subnet_params.add(Subnet(subnet))
for ip in prop_iterator(props, "allowed_address_pairs", "ip_address"):
- self.uses_dhcp = False
param = get_param(ip) if ip else ""
if param:
- self.floating_ips.append(IpParam(param, self))
+ self.floating_ips.add(IpParam(param, self))
+
+ @property
+ def ipv6_fixed_ips(self):
+ return list(
+ sorted(
+ (ip for ip in self.fixed_ips if ip.ip_version == 6),
+ key=lambda ip: ip.param,
+ )
+ )
+
+ @property
+ def ipv4_fixed_ips(self):
+ return list(
+ sorted(
+ (ip for ip in self.fixed_ips if ip.ip_version == 4),
+ key=lambda ip: ip.param,
+ )
+ )
+
+ @property
+ def fixed_ips_with_index(self) -> List[Tuple[int, IpParam]]:
+ ipv4s = enumerate(self.ipv4_fixed_ips)
+ ipv6s = enumerate(self.ipv6_fixed_ips)
+ return list(chain(ipv4s, ipv6s))
def filter_output_params(self, base_outputs):
self.fixed_ips = remove(self.fixed_ips, base_outputs, key=lambda ip: ip.param)
class Vnf:
- def __init__(self, templates):
- self.modules = [VnfModule(t, self) for t in templates]
+ def __init__(self, templates, config=None):
+ self.modules = [VnfModule(t, self, config) for t in templates]
self.uses_contrail = self._uses_contrail()
+ self.config = config
self.base_module = next(
(mod for mod in self.modules if mod.is_base_module), None
)
class VnfModule(FilterBaseOutputs):
- def __init__(self, template_file, vnf):
+ def __init__(self, template_file, vnf, config):
self.vnf = vnf
+ self.config = config
self.vnf_name = os.path.splitext(os.path.basename(template_file))[0]
self.template_file = template_file
self.heat = Heat(filepath=template_file, envpath=env_path(template_file))
env_yaml = env_pair.get("eyml") if env_pair else {}
self.parameters = {key: "" for key in self.heat.parameters}
self.parameters.update(env_yaml.get("parameters") or {})
+ # Filter out any parameters passed from the volume module's outputs
+ self.parameters = {
+ key: value
+ for key, value in self.parameters.items()
+ if key not in self.volume_module_outputs
+ }
self.networks = []
self.virtual_machine_types = self._create_vm_types()
self._add_networks()
self.outputs_filtered = False
+ @property
+ def volume_module_outputs(self):
+ heat_dir = os.path.dirname(self.template_file)
+ heat_filename = os.path.basename(self.template_file)
+ basename, ext = os.path.splitext(heat_filename)
+ volume_template_name = "{}_volume{}".format(basename, ext)
+ volume_path = os.path.join(heat_dir, volume_template_name)
+ if os.path.exists(volume_path):
+ volume_mod = Heat(filepath=volume_path)
+ return volume_mod.outputs
+ else:
+ return {}
+
def filter_output_params(self, base_outputs):
for vm in self.virtual_machine_types:
vm.filter_output_params(base_outputs)
@property
def env_specs(self):
"""Return available Environment Spec definitions"""
- try:
- return Config().env_specs
- except FileNotFoundError:
- return [ENV_PARAMETER_SPEC]
+ return [ENV_PARAMETER_SPEC] if not self.config else self.config.env_specs
@property
def platform_provided_params(self):
params[az] = CHANGE
for network in self.networks:
params[network.name_param] = CHANGE
- for param in set(network.subnet_params):
+ for param in set(s.param_name for s in network.subnet_params):
params[param] = CHANGE
for vm in self.virtual_machine_types:
for name in set(vm.names):
return hash(self) == hash(other)
-def create_preloads(config, exitstatus):
+def yield_by_count(sequence):
"""
- Create preloads in every format that can be discovered by get_generator_plugins
+ Iterates through sequence and yields each item according to its __count__
+ attribute. If an item has a __count__ of it will be returned 3 times
+ before advancing to the next item in the sequence.
+
+ :param sequence: sequence of dicts (must contain __count__)
+ :returns: generator of tuple key, value pairs
"""
- if config.getoption("self_test"):
- return
- print("+===================================================================+")
- print("| Preload Template Generation |")
- print("+===================================================================+")
-
- preload_dir = os.path.join(get_output_dir(config), "preloads")
- if os.path.exists(preload_dir):
- shutil.rmtree(preload_dir)
- env_directory = config.getoption("env_dir")
- preload_env = PreloadEnvironment(env_directory) if env_directory else None
- plugins = get_generator_plugins()
- available_formats = [p.format_name() for p in plugins]
- selected_formats = config.getoption("preload_formats") or available_formats
- heat_templates = get_heat_templates(config)
- vnf = None
- for plugin_class in plugins:
- if plugin_class.format_name() not in selected_formats:
- continue
- vnf = Vnf(heat_templates)
- generator = plugin_class(vnf, preload_dir, preload_env)
- generator.generate()
- if vnf and vnf.uses_contrail:
- print(
- "\nWARNING: Preload template generation does not support Contrail\n"
- "at this time, but Contrail resources were detected. The preload \n"
- "template may be incomplete."
- )
- if exitstatus != 0:
- print(
- "\nWARNING: Heat violations detected. Preload templates may be\n"
- "incomplete."
- )
+ for key, value in sequence.items():
+ for i in range(value["__count__"]):
+ yield (key, value)
# ============LICENSE_END============================================
import json
import os
+from pathlib import Path
+from typing import Mapping
+from preload.data import AbstractPreloadInstance
from preload.generator import (
get_json_template,
get_or_create_template,
AbstractPreloadGenerator,
)
+from preload.model import VnfModule, Port
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(THIS_DIR, "grapi_data")
def output_sub_dir(cls):
return "grapi"
- def generate_module(self, vnf_module, output_dir):
+ def generate_module(
+ self,
+ vnf_module: VnfModule,
+ preload_data: AbstractPreloadInstance,
+ output_dir: Path,
+ ):
+ self.module_incomplete = False
template = get_json_template(DATA_DIR, "preload_template")
- self._populate(template, vnf_module)
- vnf_name = vnf_module.vnf_name
- incomplete = "_incomplete" if self.module_incomplete else ""
- outfile = "{}/{}{}.json".format(output_dir, vnf_name, incomplete)
- with open(outfile, "w") as f:
+ self._populate(template, preload_data, vnf_module)
+ incomplete = (
+ "_incomplete"
+ if preload_data.flag_incompletes and self.module_incomplete
+ else ""
+ )
+ filename = "{}{}.json".format(preload_data.preload_basename, incomplete)
+ outfile = output_dir.joinpath(filename)
+ with outfile.open("w") as f:
json.dump(template, f, indent=4)
- def add_floating_ips(self, network_template, floating_ips):
- for ip in floating_ips:
+ def _populate(
+ self,
+ template: Mapping,
+ preload_data: AbstractPreloadInstance,
+ vnf_module: VnfModule,
+ ):
+ self._add_vnf_metadata(template, preload_data)
+ self._add_availability_zones(template, preload_data, vnf_module)
+ self._add_vnf_networks(template, preload_data, vnf_module)
+ self._add_vms(template, preload_data, vnf_module)
+ self._add_parameters(template, preload_data, vnf_module)
+
+ def _add_vnf_metadata(self, template: Mapping, preload: AbstractPreloadInstance):
+ topology = template["input"]["preload-vf-module-topology-information"]
+ vnf_meta = topology["vnf-topology-identifier-structure"]
+ vnf_meta["vnf-name"] = self.normalize(preload.vnf_name, "vnf_name")
+ vnf_meta["vnf-type"] = self.normalize(
+ preload.vnf_type,
+ "vnf-type",
+ "VALUE FOR: Concatenation of <Service Name>/"
+ "<VF Instance Name> MUST MATCH SDC",
+ )
+ module_meta = topology["vf-module-topology"]["vf-module-topology-identifier"]
+ module_meta["vf-module-name"] = self.normalize(
+ preload.vf_module_name, "vf_module_name"
+ )
+ module_meta["vf-module-type"] = self.normalize(
+ preload.vf_module_model_name,
+ "vf-module-model-name",
+ "VALUE FOR: <vfModuleModelName> from CSAR or SDC",
+ )
+
+ def _add_availability_zones(
+ self, template: Mapping, preload: AbstractPreloadInstance, vnf_module: VnfModule
+ ):
+ zones = template["input"]["preload-vf-module-topology-information"][
+ "vnf-resource-assignments"
+ ]["availability-zones"]["availability-zone"]
+ for i, zone_param in enumerate(vnf_module.availability_zones):
+ zone = preload.get_availability_zone(i, zone_param)
+ zones.append(self.normalize(zone, zone_param, index=i))
+
+ def _add_vnf_networks(
+ self, template: Mapping, preload: AbstractPreloadInstance, vnf_module: VnfModule
+ ):
+ networks = template["input"]["preload-vf-module-topology-information"][
+ "vnf-resource-assignments"
+ ]["vnf-networks"]["vnf-network"]
+ for network in vnf_module.networks:
+ network_data = {
+ "network-role": network.network_role,
+ "network-name": self.normalize(
+ preload.get_network_name(network.network_role, network.name_param),
+ network.name_param,
+ "VALUE FOR: network name of {}".format(network.name_param),
+ ),
+ }
+ if network.subnet_params:
+ network_data["subnets-data"] = {"subnet-data": []}
+ subnet_data = network_data["subnets-data"]["subnet-data"]
+ for subnet in network.subnet_params:
+ data = {}
+ subnet_id = preload.get_subnet_id(
+ network.network_role, subnet.ip_version, subnet.param_name
+ )
+ if subnet_id:
+ data["subnet-id"] = self.normalize(subnet_id, subnet.param_name)
+ else:
+ subnet_name = preload.get_subnet_name(
+ network.network_role, subnet.ip_version, ""
+ )
+ data["subnet-name"] = self.normalize(
+ subnet_name,
+ subnet.param_name,
+ alt_message="VALUE FOR: name of {}".format(
+ subnet.param_name
+ ),
+ )
+ subnet_data.append(data)
+ networks.append(network_data)
+
+ def add_floating_ips(
+ self, network_template: dict, port: Port, preload: AbstractPreloadInstance
+ ):
+ for ip in port.floating_ips:
key = "floating-ip-v4" if ip.ip_version == 4 else "floating-ip-v6"
ips = network_template["floating-ips"][key]
- value = self.replace(ip.param, single=True)
- if value not in ips:
- ips.append(value)
+ value = self.normalize(
+ preload.get_floating_ip(
+ port.vm.vm_type, port.network.network_role, ip.ip_version, ip.param
+ ),
+ ip.param,
+ )
+ ips.append(value)
- def add_fixed_ips(self, network_template, fixed_ips, uses_dhcp):
+ def add_fixed_ips(
+ self, network_template: dict, port: Port, preload: AbstractPreloadInstance
+ ):
items = network_template["network-information-items"][
"network-information-item"
]
ipv4s = next(item for item in items if item["ip-version"] == "4")
ipv6s = next(item for item in items if item["ip-version"] == "6")
- if uses_dhcp:
+ if port.uses_dhcp:
ipv4s["use-dhcp"] = "Y"
ipv6s["use-dhcp"] = "Y"
- for ip in fixed_ips:
+ for index, ip in port.fixed_ips_with_index:
target = ipv4s if ip.ip_version == 4 else ipv6s
ips = target["network-ips"]["network-ip"]
if ip.param not in ips:
- ips.append(self.replace(ip.param, single=True))
+ ips.append(
+ self.normalize(
+ preload.get_fixed_ip(
+ port.vm.vm_type,
+ port.network.network_role,
+ ip.ip_version,
+ index,
+ ip.param,
+ ),
+ ip.param,
+ index=index
+ )
+ )
target["ip-count"] += 1
- def _populate(self, preload, vnf_module):
- self._add_vnf_metadata(preload)
- self._add_vms(preload, vnf_module)
- self._add_availability_zones(preload, vnf_module)
- self._add_parameters(preload, vnf_module)
- self._add_vnf_networks(preload, vnf_module)
-
- def _add_vms(self, preload, vnf_module):
- vms = preload["input"]["preload-vf-module-topology-information"][
+ def _add_vms(
+ self, template: Mapping, preload: AbstractPreloadInstance, vnf_module: VnfModule
+ ):
+ vms = template["input"]["preload-vf-module-topology-information"][
"vf-module-topology"
]["vf-module-assignments"]["vms"]["vm"]
for vm in vnf_module.virtual_machine_types:
vm_template = get_json_template(DATA_DIR, "vm")
vms.append(vm_template)
vm_template["vm-type"] = vm.vm_type
- for name in vm.names:
- value = self.replace(name, single=True)
+ for i, param in enumerate(sorted(vm.names)):
+ name = preload.get_vm_name(vm.vm_type, i, param)
+ value = self.normalize(name, param, index=i)
vm_template["vm-names"]["vm-name"].append(value)
vm_template["vm-count"] = vm.vm_count
vm_networks = vm_template["vm-networks"]["vm-network"]
role = port.network.network_role
network_template = get_or_create_network_template(role, vm_networks)
network_template["network-role"] = role
- self.add_fixed_ips(network_template, port.fixed_ips, port.uses_dhcp)
- self.add_floating_ips(network_template, port.floating_ips)
-
- def _add_availability_zones(self, preload, vnf_module):
- zones = preload["input"]["preload-vf-module-topology-information"][
- "vnf-resource-assignments"
- ]["availability-zones"]["availability-zone"]
- for zone in vnf_module.availability_zones:
- value = self.replace(zone, single=True)
- zones.append(value)
+ network_template["network-role-tag"] = role
+ self.add_fixed_ips(network_template, port, preload)
+ self.add_floating_ips(network_template, port, preload)
- def _add_parameters(self, preload, vnf_module):
+ def _add_parameters(
+ self, template: Mapping, preload: AbstractPreloadInstance, vnf_module: VnfModule
+ ):
params = [
- {"name": key, "value": self.replace(key, value)}
+ {
+ "name": key,
+ "value": self.normalize(preload.get_vnf_parameter(key, value), key),
+ }
for key, value in vnf_module.preload_parameters.items()
]
- preload["input"]["preload-vf-module-topology-information"][
+ for key, value in preload.get_additional_parameters().items():
+ params.append(
+ {
+ "name": key,
+ "value": value,
+ }
+ )
+
+ template["input"]["preload-vf-module-topology-information"][
"vf-module-topology"
]["vf-module-parameters"]["param"].extend(params)
-
- def _add_vnf_networks(self, preload, vnf_module):
- networks = preload["input"]["preload-vf-module-topology-information"][
- "vnf-resource-assignments"
- ]["vnf-networks"]["vnf-network"]
- for network in vnf_module.networks:
- network_data = {
- "network-role": network.network_role,
- "network-name": self.replace(
- network.name_param,
- "VALUE FOR: network name of {}".format(network.name_param),
- ),
- }
- if network.subnet_params:
- network_data["subnets-data"] = {"subnet-data": []}
- subnet_data = network_data["subnets-data"]["subnet-data"]
- for subnet_param in network.subnet_params:
- subnet_data.append(
- {"subnet-id": self.replace(subnet_param, single=True)}
- )
- networks.append(network_data)
-
- def _add_vnf_metadata(self, preload):
- topology = preload["input"]["preload-vf-module-topology-information"]
- vnf_meta = topology["vnf-topology-identifier-structure"]
- vnf_meta["vnf-name"] = self.replace("vnf_name")
- vnf_meta["vnf-type"] = self.replace(
- "vnf-type",
- "VALUE FOR: Concatenation of <Service Name>/"
- "<VF Instance Name> MUST MATCH SDC",
- )
- module_meta = topology["vf-module-topology"]["vf-module-topology-identifier"]
- module_meta["vf-module-name"] = self.replace("vf_module_name")
- module_meta["vf-module-type"] = self.replace(
- "vf-module-model-name", "VALUE FOR: <vfModuleModelName> from CSAR or SDC"
- )
# limitations under the License.
#
# ============LICENSE_END============================================
-#
-#
import json
import os
+from pathlib import Path
+from typing import Mapping
+from preload.data import AbstractPreloadInstance
from preload.generator import (
get_json_template,
get_or_create_template,
AbstractPreloadGenerator,
)
+from preload.model import VnfModule, Port
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(THIS_DIR, "vnfapi_data")
def output_sub_dir(cls):
return "vnfapi"
- def generate_module(self, vnf_module, output_dir):
- preload = get_json_template(DATA_DIR, "preload_template")
- self._populate(preload, vnf_module)
- 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"][
+ def generate_module(
+ self,
+ vnf_module: VnfModule,
+ preload_data: AbstractPreloadInstance,
+ output_dir: Path,
+ ):
+ self.module_incomplete = False
+ template = get_json_template(DATA_DIR, "preload_template")
+ self._populate(template, preload_data, vnf_module)
+ incomplete = (
+ "_incomplete"
+ if preload_data.flag_incompletes and self.module_incomplete
+ else ""
+ )
+ filename = "{}{}.json".format(preload_data.preload_basename, incomplete)
+ outfile = output_dir.joinpath(filename)
+ with outfile.open("w") as f:
+ json.dump(template, f, indent=4)
+
+ def _populate(
+ self,
+ template: Mapping,
+ preload_data: AbstractPreloadInstance,
+ vnf_module: VnfModule,
+ ):
+ self._add_vnf_metadata(template, preload_data)
+ self._add_availability_zones(template, preload_data, vnf_module)
+ self._add_vnf_networks(template, preload_data, vnf_module)
+ self._add_vms(template, preload_data, vnf_module)
+ self._add_parameters(template, preload_data, vnf_module)
+
+ def _add_vnf_metadata(self, template: Mapping, preload: AbstractPreloadInstance):
+ vnf_meta = template["input"]["vnf-topology-information"][
"vnf-topology-identifier"
]
- vnf_meta["vnf-name"] = self.replace("vnf_name")
- vnf_meta["generic-vnf-type"] = self.replace(
+
+ vnf_meta["vnf-name"] = self.normalize(preload.vnf_name, "vnf_name")
+ vnf_meta["generic-vnf-type"] = self.normalize(
+ preload.vnf_type,
"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"
+ vnf_meta["vnf-type"] = self.normalize(
+ preload.vf_module_model_name,
+ "vf-module-model-name",
+ "VALUE FOR: <vfModuleModelName> from CSAR or SDC",
)
- def add_floating_ips(self, network_template, network):
+ def _add_availability_zones(
+ self, template: Mapping, preload: AbstractPreloadInstance, vnf_module: VnfModule
+ ):
+ zones = template["input"]["vnf-topology-information"]["vnf-assignments"][
+ "availability-zones"
+ ]
+ for i, zone_param in enumerate(vnf_module.availability_zones):
+ zone = preload.get_availability_zone(i, zone_param)
+ zones.append({"availability-zone": self.normalize(zone, zone_param, index=i)})
+
+ def add_floating_ips(
+ self, network_template: dict, port: Port, preload: AbstractPreloadInstance
+ ):
# 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
# and if multiple floating params exist, then come up with an
# approach to pick just one
- for ip in network.floating_ips:
+ for ip in port.floating_ips:
+ ip_value = preload.get_floating_ip(
+ port.vm.vm_type, port.network.network_role, ip.ip_version, ip.param
+ )
key = "floating-ip" if ip.ip_version == 4 else "floating-ip-v6"
- network_template[key] = self.replace(ip.param, single=True)
-
- def add_fixed_ips(self, network_template, port):
- for ip in port.fixed_ips:
+ network_template[key] = self.normalize(ip_value, ip.param)
+
+ def add_fixed_ips(
+ self, network_template: dict, port: Port, preload: AbstractPreloadInstance
+ ):
+ for index, ip in port.fixed_ips_with_index:
+ ip_value = preload.get_fixed_ip(
+ port.vm.vm_type,
+ port.network.network_role,
+ ip.ip_version,
+ index,
+ ip.param,
+ )
+ ip_value = self.normalize(ip_value, ip.param, index=index)
if ip.ip_version == 4:
- network_template["network-ips"].append(
- {"ip-address": self.replace(ip.param, single=True)}
- )
+ network_template["network-ips"].append({"ip-address": ip_value})
network_template["ip-count"] += 1
else:
- network_template["network-ips-v6"].append(
- {"ip-address": self.replace(ip.param, single=True)}
- )
+ network_template["network-ips-v6"].append({"ip-address": ip_value})
network_template["ip-count-ipv6"] += 1
- def _add_availability_zones(self, preload, vnf_module):
- zones = preload["input"]["vnf-topology-information"]["vnf-assignments"][
- "availability-zones"
- ]
- for zone in vnf_module.availability_zones:
- zones.append({"availability-zone": self.replace(zone, single=True)})
-
- def _add_vnf_networks(self, preload, vnf_module):
- networks = preload["input"]["vnf-topology-information"]["vnf-assignments"][
+ def _add_vnf_networks(
+ self, template: Mapping, preload: AbstractPreloadInstance, vnf_module: VnfModule
+ ):
+ networks = template["input"]["vnf-topology-information"]["vnf-assignments"][
"vnf-networks"
]
for network in vnf_module.networks:
network_data = {
"network-role": network.network_role,
- "network-name": self.replace(
+ "network-name": self.normalize(
+ preload.get_network_name(network.network_role, network.name_param),
network.name_param,
"VALUE FOR: network name for {}".format(network.name_param),
),
}
for subnet in network.subnet_params:
- key = "ipv6-subnet-id" if "_v6_" in subnet else "subnet-id"
- network_data[key] = subnet
+ subnet_id = preload.get_subnet_id(
+ network.network_role, subnet.ip_version, subnet.param_name
+ )
+ if subnet_id:
+ key = (
+ "ipv6-subnet-id" if "_v6_" in subnet.param_name else "subnet-id"
+ )
+ network_data[key] = self.normalize(subnet_id, subnet.param_name)
+ else:
+ subnet_name = preload.get_subnet_name(
+ network.network_role, subnet.ip_version, ""
+ )
+ key = (
+ "ipv6-subnet-name"
+ if "_v6_" in subnet.param_name
+ else "subnet-name"
+ )
+ msg = "VALUE FOR: name for {}".format(subnet.param_name)
+ value = self.normalize(
+ subnet_name, subnet.param_name, alt_message=msg
+ )
+ network_data[key] = value
networks.append(network_data)
- def _add_vms(self, preload, vnf_module):
- vm_list = preload["input"]["vnf-topology-information"]["vnf-assignments"][
+ def _add_vms(
+ self, template: Mapping, preload: AbstractPreloadInstance, vnf_module: VnfModule
+ ):
+ vm_list = template["input"]["vnf-topology-information"]["vnf-assignments"][
"vnf-vms"
]
for vm in vnf_module.virtual_machine_types:
vm_template = get_json_template(DATA_DIR, "vm")
vm_template["vm-type"] = vm.vm_type
vm_template["vm-count"] = vm.vm_count
- for name in vm.names:
- value = self.replace(name, single=True)
- vm_template["vm-names"]["vm-name"].append(value)
+ for i, param in enumerate(sorted(vm.names)):
+ name = preload.get_vm_name(vm.vm_type, i, param)
+ vm_template["vm-names"]["vm-name"].append(self.normalize(name, param, index=i))
vm_list.append(vm_template)
vm_networks = vm_template["vm-networks"]
for port in vm.ports:
network_template["network-role"] = role
network_template["network-role-tag"] = role
network_template["use-dhcp"] = "Y" if port.uses_dhcp else "N"
- self.add_fixed_ips(network_template, port)
- self.add_floating_ips(network_template, port)
+ self.add_fixed_ips(network_template, port, preload)
+ self.add_floating_ips(network_template, port, preload)
- def _add_parameters(self, preload, vnf_module):
- params = preload["input"]["vnf-topology-information"]["vnf-parameters"]
+ def _add_parameters(
+ self, template: Mapping, preload: AbstractPreloadInstance, vnf_module: VnfModule
+ ):
+ params = template["input"]["vnf-topology-information"]["vnf-parameters"]
for key, value in vnf_module.preload_parameters.items():
+ preload_value = preload.get_vnf_parameter(key, value)
+ value = preload_value or value
+ params.append(
+ {
+ "vnf-parameter-name": key,
+ "vnf-parameter-value": self.normalize(value, key),
+ }
+ )
+ for key, value in preload.get_additional_parameters().items():
params.append(
{
"vnf-parameter-name": key,
- "vnf-parameter-value": self.replace(key, value),
+ "vnf-parameter-value": self.normalize(value, key),
}
)
import re
import time
-from preload.model import create_preloads
-from config import get_generator_plugin_names
+from preload.engine import PLUGIN_MGR, create_preloads
from tests.helpers import get_output_dir
try:
help="optional category of test to execute",
)
- parser.addoption(
- "--env-directory",
- dest="env_dir",
- action="store",
- help="optional directory of .env files for preload generation",
- )
-
parser.addoption(
"--preload-format",
dest="preload_formats",
help=(
"Preload format to create (multiple allowed). If not provided "
"then all available formats will be created: {}"
- ).format(", ".join(get_generator_plugin_names())),
+ ).format(", ".join(g.format_name() for g in PLUGIN_MGR.preload_generators)),
+ )
+
+ parser.addoption(
+ "--preload-source-type",
+ dest="preload_source_type",
+ action="store",
+ default="envfiles",
+ help=(
+ "Preload source type to create (multiple allowed): {}"
+ ).format(", ".join(s.get_identifier() for s in PLUGIN_MGR.preload_sources)),
+ )
+
+ parser.addoption(
+ "--preload-source",
+ dest="preload_source",
+ action="store",
+ help="File or directory containing the source dat for the preloads",
)
or config.getoption("self_test")
or config.getoption("help")
):
- raise Exception('One of "--template-dir" or' ' "--self-test" must be specified')
+ raise Exception('One of "--template-directory" or'
+ ' "--self-test" must be specified')
def pytest_generate_tests(metafunc):
@validates("R-01455")
def test_vm_class_has_unique_type(yaml_files):
"""
- When a VNF’s Heat Orchestration Template creates a Virtual
- Machine (i.e., OS::Nova::Server), each “class” of VMs MUST be
- assigned a VNF unique vm-type; where “class” defines VMs that
+ When a VNF's Heat Orchestration Template creates a Virtual
+ Machine (i.e., OS::Nova::Server), each "class" of VMs MUST be
+ assigned a VNF unique vm-type; where "class" defines VMs that
MUST have the following identical characteristics:
1. OS::Nova::Server resource property flavor value
2. OS::Nova::Server resource property image value
3. Cinder Volume attachments
- Each VM in the “class” MUST have the identical Cinder
+ Each VM in the "class" MUST have the identical Cinder
Volume configuration
4. Network attachments and IP address requirements
- Each VM in the “class” MUST have the the identical number of
+ Each VM in the "class" MUST have the the identical number of
ports connecting to the identical networks and requiring the
identical IP address configuration
"""
errors.append(yaml_file)
msg = (
"Volume modules detected, but they do not follow the expected "
- + " naming convention {{module_name}}_volume.[yaml|yml]: {}"
+ + " naming convention {{module_label}}_volume.[yaml|yml]: {}"
).format(", ".join(errors))
assert not errors, msg
from typing import Optional, TextIO, Callable
from config import Config
+from preload.engine import PLUGIN_MGR
VERSION = version.VERSION
PATH = os.path.dirname(os.path.realpath(__file__))
report_format: str,
halt_on_failure: bool,
template_source: str,
- env_dir: str,
+ preload_config: str,
preload_format: list,
+ preload_source: str,
):
"""Runs pytest using the given ``profile`` in a background process. All
``stdout`` and ``stderr`` are redirected to ``log``. The result of the job
prevent a large number of errors from flooding the
report.
:param template_source: The path or name of the template to show on the report
- :param env_dir: Optional directory of env files that can be used
- to generate populated preload templates
- :param preload_format: Selected preload format
+ :param preload_config: Optional directory or file that is input to preload
+ data source
+ :param preload_format: Selected preload format
+ :param preload_source: Name of selected preload data source plugin
"""
out_path = "{}/{}".format(PATH, OUT_DIR)
if os.path.exists(out_path):
"--report-format={}".format(report_format),
"--template-source={}".format(template_source),
]
- if env_dir:
- args.append("--env-directory={}".format(env_dir))
+ if preload_config:
+ args.append("--preload-source={}".format(preload_config))
+ args.append(
+ "--preload-source-type={}".format(
+ PLUGIN_MGR.get_source_for_name(preload_source).get_identifier()
+ )
+ )
if categories:
for category in categories:
args.extend(("--category", category))
args.append("--continue-on-failure")
if preload_format:
args.append("--preload-format={}".format(preload_format))
+ print("args: ", " ".join(args))
pytest.main(args=args)
result_queue.put((True, None))
except Exception:
settings_frame.grid(row=3, column=1, columnspan=3, pady=10, sticky="we")
if self.config.preload_formats:
- preload_format_label = Label(settings_frame, text="Preload Template:")
+ preload_format_label = Label(settings_frame, text="Preload Format:")
preload_format_label.grid(row=settings_row, column=1, sticky=W)
self.preload_format = StringVar(self._root, name="preload_format")
self.preload_format.set(self.config.default_preload_format)
)
settings_row += 1
+ preload_source_label = Label(settings_frame, text="Preload Source:")
+ preload_source_label.grid(row=settings_row, column=1, sticky=W)
+ self.preload_source = StringVar(self._root, name="preload_source")
+ self.preload_source.set(self.config.default_preload_source)
+ preload_source_menu = OptionMenu(
+ settings_frame, self.preload_source, *self.config.preload_source_types
+ )
+ preload_source_menu.config(width=25)
+ preload_source_menu.grid(
+ row=settings_row, column=2, columnspan=3, sticky=E, pady=5
+ )
+ settings_row += 1
+
report_format_label = Label(settings_frame, text="Report Format:")
report_format_label.grid(row=settings_row, column=1, sticky=W)
self.report_format = StringVar(self._root, name="report_format")
self.create_preloads.set(self.config.default_create_preloads)
create_preloads_label = Label(
settings_frame,
- text="Create Preload from Env Files:",
+ text="Create Preload from Datasource:",
anchor=W,
justify=LEFT,
)
directory_browse = Button(actions, text="...", command=self.ask_template_source)
directory_browse.grid(row=4, column=3, pady=5, sticky=W)
- env_dir_label = Label(actions, text="Env Files:")
- env_dir_label.grid(row=5, column=1, pady=5, sticky=W)
- self.env_dir = StringVar(self._root, name="env_dir")
- env_dir_state = NORMAL if self.create_preloads.get() else DISABLED
- self.env_dir_entry = Entry(
- actions, width=40, textvariable=self.env_dir, state=env_dir_state
+ preload_config_label = Label(actions, text="Preload Datasource:")
+ preload_config_label.grid(row=5, column=1, pady=5, sticky=W)
+ self.preload_config = StringVar(self._root, name="preload_config")
+ preload_config_state = NORMAL if self.create_preloads.get() else DISABLED
+ self.preload_config_entry = Entry(
+ actions,
+ width=40,
+ textvariable=self.preload_config,
+ state=preload_config_state,
)
- self.env_dir_entry.grid(row=5, column=2, pady=5, sticky=W)
- env_dir_browse = Button(actions, text="...", command=self.ask_env_dir_source)
- env_dir_browse.grid(row=5, column=3, pady=5, sticky=W)
+ self.preload_config_entry.grid(row=5, column=2, pady=5, sticky=W)
+ preload_config_browse = Button(
+ actions, text="...", command=self.ask_preload_source
+ )
+ preload_config_browse.grid(row=5, column=3, pady=5, sticky=W)
validate_button = Button(
actions, text="Process Templates", command=self.validate
self.halt_on_failure,
self.preload_format,
self.create_preloads,
+ self.preload_source,
)
self.schedule(self.execute_pollers)
if self.config.terms_link_text and not self.config.are_terms_accepted:
def set_env_dir_state(self):
state = NORMAL if self.create_preloads.get() else DISABLED
- self.env_dir_entry.config(state=state)
+ if state == DISABLED:
+ self.preload_config.set("")
+ self.preload_config_entry.config(state=state)
def ask_template_source(self):
if self.input_format.get() == "ZIP File":
template_source = filedialog.askdirectory()
self.template_source.set(template_source)
- def ask_env_dir_source(self):
- self.env_dir.set(filedialog.askdirectory())
+ def ask_preload_source(self):
+ input_type = "DIR"
+ for source in PLUGIN_MGR.preload_sources:
+ if source.get_name() == self.preload_source.get():
+ input_type = source.get_source_type()
+
+ if input_type == "DIR":
+ self.preload_config.set(filedialog.askdirectory())
+ else:
+ self.preload_config.set(
+ filedialog.askopenfilename(
+ title="Select Preload Datasource File",
+ filetypes=(("All Files", "*"),),
+ )
+ )
def validate(self):
"""Run the pytest validations in a background process"""
self.report_format.get().lower(),
self.halt_on_failure.get(),
self.template_source.get(),
- self.env_dir.get(),
+ self.preload_config.get(),
self.preload_format.get(),
+ self.preload_source.get(),
),
)
self.task.daemon = True
Code Review / vvp / validation-scripts.git / commitdiff