.idea/
ice_validator/output/
+sample_env/grapi/*
CURRENT_NEEDS_PATH = os.path.join(THIS_DIR, "ice_validator/heat_requirements.json")
+def run_pytest(*args, msg="pytest failed"):
+ original_dir = os.getcwd()
+ try:
+ os.chdir(os.path.join(THIS_DIR, "ice_validator"))
+ if pytest.main(list(args)) != 0:
+ return [msg]
+ finally:
+ os.chdir(original_dir)
+
+
class Traceability:
PATH = os.path.join(THIS_DIR, "ice_validator/output/traceability.csv")
return None
+def check_app_tests_pass():
+ return run_pytest("tests", "--self-test",
+ msg="app_tests failed. Run pytest app_tests and fix errors.")
+
+
def check_self_test_pass():
- """
- Run pytest self-test and ensure it passes
- :return:
- """
- original_dir = os.getcwd()
- try:
- os.chdir(os.path.join(THIS_DIR, "ice_validator"))
- if pytest.main(["tests", "--self-test"]) != 0:
- return ["VVP self-test failed. Run pytest --self-test and fix errors."]
- finally:
- os.chdir(original_dir)
+ return run_pytest("tests", "--self-test",
+ msg="self-test failed. Run pytest --self-test and fix errors.")
def check_testable_requirements_are_mapped():
--- /dev/null
+parameters:
+ common: "ABC"
+ my_ip: default
--- /dev/null
+availability_zone_0: az0
--- /dev/null
+parameters:
+ my_ip: 192.168.0.1
--- /dev/null
+parameters:
+ my_ip: 192.168.0.13
--- /dev/null
+availability_zone_0: az0-b
+custom_env_3: default
--- /dev/null
+parameters:
+ my_ip: 192.168.0.2
--- /dev/null
+parameters:
+ inc_property: "global"
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+parameters:
+ availability_zone_0: az0
+ availability_zone_1: az1
--- /dev/null
+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
--- /dev/null
+heat_template_version: 2015-04-30
+
+description: Base Module of Sample VNF
+
+parameters:
+
+ # ONAP Assigned Parameters
+ workload_context:
+ type: string
+ description: Unique ID for this VNF instance
+
+ environment_context:
+ type: string
+ description: Unique ID for this VNF instance
+
+ vnf_id:
+ type: string
+ description: Unique ID for this VNF instance
+
+ vf_module_id:
+ type: string
+ description: Unique ID for this VNF module instance
+
+ vf_module_index:
+ type: number
+ description: Index of this VF Module
+
+ vnf_name:
+ type: string
+ description: Unique name for this VNF instance
+
+
+ # Availability Zones
+ availability_zone_0:
+ type: string
+ description: Primary Availability Zone
+
+ availability_zone_1:
+ type: string
+ description: Secondary Availability Zone
+
+
+ # External Networks
+ oam_net_id:
+ type: string
+ description: Operations, Administration, and Management Network
+
+ oam_subnet_id:
+ type: string
+ description: Subnet for OAM Network
+
+ ha_net_id:
+ type: string
+ description: High Availability Network
+
+ ctrl_net_id:
+ type: string
+ description: Control Plane network
+
+ ctrl_subnet_id:
+ type: string
+ description: Subnet for High Availability Network
+
+
+ # Server Inputs: Database
+ db_name_0:
+ type: string
+ description: Primary DB Server Name
+
+ db_name_1:
+ type: string
+ description: Secondary DB
+
+ db_image_name:
+ type: string
+ description: Database VM Image Name
+
+ db_flavor_name:
+ type: string
+ description: Database VM Flavor Name
+
+ db_ha_floating_v6_ip:
+ type: string
+ description: Database Floating IPv6 Address for HA
+
+ db_ha_floating_ip:
+ type: string
+ description: Database Floating IPv4 Address for HA
+
+ db_oam_ip_0:
+ type: string
+ description: Fixed IPv4 Address for OAM
+
+ db_oam_ip_1:
+ type: string
+ description: Fixed IPv4 Address for OAM
+
+ db_vol0_id:
+ type: string
+ description: Volume ID for DB in AZ 0
+
+ db_vol1_id:
+ type: string
+ description: Volume ID for DB in AZ 1
+
+
+ # Server Inputs: Loadbalancer
+ lb_name_0:
+ type: string
+ description: Load Balancer Name
+
+ lb_image_name:
+ type: string
+ description: Loadbalancer VM Image
+
+ lb_flavor_name:
+ type: string
+ description: Loadbalancer VM Flavor
+
+ lb_ha_floating_ip:
+ type: string
+ description: Floating HA IP for LB
+
+
+ lb_ha_floating_v6_ip:
+ type: string
+ description: Floating HA IP for LB
+
+
+ # Server Inputs: Webservice Controller Plane Interface (mgmt)
+ mgmt_name_0:
+ type: string
+ description: List of Management VM Names
+
+ mgmt_image_name:
+ type: string
+ description: Management VM Image
+
+ mgmt_flavor_name:
+ type: string
+ description: Management VM Flavor
+
+ mgmt_ctrl_ip_0:
+ type: string
+ description: IP to web service for control plane
+
+ mgmt_ctrl_v6_ip_0:
+ type: string
+ description: IP to web service for control plane
+
+
+ # Server Inputs: Services
+ svc_names:
+ type: comma_delimited_list
+ description: Service VM Names
+
+ svc_image_name:
+ type: string
+ description: Service VM Image
+
+ svc_flavor_name:
+ type: string
+ description: Service VM Flavor
+
+ svc_count:
+ type: number
+ description: Number of instances of Service to create
+
+resources:
+
+ int_private_network:
+ type: OS::Neutron::Net
+
+ int_private_subnet:
+ type: OS::Neutron::Subnet
+ properties:
+ name:
+ str_replace:
+ template: $VNF_NAME-private_subnet
+ params:
+ $VNF_NAME: { get_param: vnf_name }
+ network: { get_resource: int_private_network }
+
+ db_server_0:
+ type: OS::Nova::Server
+ properties:
+ image: { get_param: db_image_name }
+ flavor: { get_param: db_flavor_name }
+ name: { get_param: db_name_0 }
+ metadata:
+ vnf_id: { get_param: vnf_id }
+ vf_module_id: { get_param: vf_module_id }
+ vf_module_index: { get_param: vf_module_index }
+ vnf_name: { get_param: vnf_name }
+ workload_context: { get_param: workload_context }
+ environment_context: { get_param: environment_context }
+ networks:
+ - port: { get_resource: db_0_int_private_port_0 }
+ - port: { get_resource: db_0_ha_port_0 }
+ - port: { get_resource: db_0_oam_port_0 }
+ user_data: { get_file: user.data }
+ availability_zone: { get_param: availability_zone_0 }
+
+ db_server_1:
+ type: OS::Nova::Server
+ properties:
+ image: { get_param: db_image_name }
+ flavor: { get_param: db_flavor_name }
+ name: { get_param: db_name_1 }
+ metadata:
+ vnf_id: { get_param: vnf_id}
+ vf_module_id: { get_param: vf_module_id }
+ vnf_name: { get_param: vnf_name }
+ workload_context: { get_param: workload_context }
+ environment_context: { get_param: environment_context }
+ networks:
+ - port: {get_resource: db_1_int_private_port_0}
+ - port: {get_resource: db_1_ha_port_0}
+ - port: { get_resource: db_1_oam_port_0 }
+ availability_zone: { get_param: availability_zone_1 }
+
+ db_0_oam_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_param: oam_net_id }
+ fixed_ips:
+ - subnet_id: { get_param: oam_subnet_id }
+ ip_address: { get_param: db_oam_ip_0 }
+
+ db_0_ha_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_param: ha_net_id }
+ allowed_address_pairs:
+ - ip_address: {get_param: db_ha_floating_ip }
+ - ip_address: {get_param: db_ha_floating_v6_ip }
+
+ db_0_int_private_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_resource: int_private_network }
+ fixed_ips:
+ - subnet_id: { get_resource: int_private_subnet }
+
+ db_1_oam_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_param: oam_net_id }
+ fixed_ips:
+ - subnet_id: { get_param: oam_subnet_id }
+ ip_address: { get_param: db_oam_ip_1 }
+
+ db_1_ha_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_param: ha_net_id }
+ allowed_address_pairs:
+ - ip_address: {get_param: db_ha_floating_ip }
+ - ip_address: {get_param: db_ha_floating_v6_ip }
+
+ db_1_int_private_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_resource: int_private_network }
+ fixed_ips:
+ - subnet_id: { get_resource: int_private_subnet }
+
+
+ db_volume_attachment_0:
+ type: OS::Cinder::VolumeAttachment
+ properties:
+ volume_id: { get_param: db_vol0_id }
+ server: { 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 }
+
+ mgmt_server_0:
+ type: OS::Nova::Server
+ properties:
+ image: { get_param: mgmt_image_name }
+ flavor: { get_param: mgmt_flavor_name }
+ name: { get_param: mgmt_name_0 }
+ metadata:
+ vnf_id: { get_param: vnf_id }
+ vf_module_id: { get_param: vf_module_id }
+ vf_module_index: { get_param: vf_module_index }
+ vnf_name: { get_param: vnf_name }
+ workload_context: { get_param: workload_context }
+ environment_context: { get_param: environment_context }
+ networks:
+ - port: { get_resource: mgmt_0_int_private_port_0 }
+ - port: { get_resource: mgmt_0_ctrl_port_0 }
+ user_data: { get_file: user.data }
+ availability_zone: { get_param: availability_zone_0 }
+
+ mgmt_0_int_private_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_resource: int_private_network }
+ fixed_ips:
+ - subnet_id: { get_resource: int_private_subnet }
+
+ mgmt_0_ctrl_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_param: ctrl_net_id }
+ fixed_ips:
+ - subnet: { get_param: ctrl_subnet_id }
+ - ip_address: { get_param: mgmt_ctrl_ip_0 }
+ - ip_address: { get_param: mgmt_ctrl_v6_ip_0}
+
+ lb_server_0:
+ type: OS::Nova::Server
+ properties:
+ image: { get_param: lb_image_name }
+ flavor: { get_param: lb_flavor_name }
+ name: { get_param: lb_name_0 }
+ metadata:
+ vnf_id: { get_param: vnf_id }
+ vf_module_id: { get_param: vf_module_id }
+ vf_module_index: { get_param: vf_module_index }
+ vnf_name: { get_param: vnf_name }
+ workload_context: { get_param: workload_context }
+ environment_context: { get_param: environment_context }
+ networks:
+ - port: { get_resource: lb_0_int_private_port_0 }
+ - port: { get_resource: lb_0_ha_port_0 }
+ user_data: { get_file: user.data }
+ availability_zone: { get_param: availability_zone_0 }
+
+ lb_0_ha_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_param: ha_net_id }
+ allowed_address_pairs:
+ - ip_address: {get_param: lb_ha_floating_ip }
+ - ip_address: {get_param: lb_ha_floating_v6_ip }
+
+ lb_0_int_private_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_resource: int_private_network }
+ fixed_ips:
+ - subnet_id: { get_resource: int_private_subnet }
+
+ svc_resource_group_0:
+ type: OS::Heat::ResourceGroup
+ properties:
+ count: { get_param: svc_count }
+ resource_def:
+ type: nested_svc.yaml
+ properties:
+ workload_context: {get_param: workload_context}
+ environment_context: {get_param: environment_context}
+ vnf_id: {get_param: vnf_id}
+ vf_module_id: {get_param: vf_module_id}
+ vnf_name: {get_param: vnf_name}
+ availability_zone_0: {get_param: availability_zone_0}
+ svc_names: {get_param: svc_names}
+ svc_image_name: {get_param: svc_image_name}
+ svc_flavor_name: {get_param: svc_flavor_name}
+ index: "%index%"
+ int_private_net_id: {get_resource: int_private_network}
+ int_private_subnet_id: {get_resource: int_private_subnet}
+
+outputs:
+
+ int_private_subnet_id:
+ value: { get_resource: int_private_subnet }
+
+ int_private_net_id:
+ value: { get_resource: int_private_network }
--- /dev/null
+parameters:
+ volume_size: 10
\ No newline at end of file
--- /dev/null
+heat_template_version: '2013-05-23'
+
+description: nsadkfl
+
+parameters:
+
+ vnf_name:
+ type: string
+ label: VF name
+ description: Unique name for this VF instance.
+
+ volume_size:
+ type: number
+ description: Size in GB
+ constraints:
+ - range: { min: 100, max: 400 }
+
+resources:
+
+ db_vol0:
+ type: OS::Cinder::Volume
+ properties:
+ name:
+ str_replace:
+ template: VNF_NAME_db_vol0
+ params:
+ VNF_NAME: {get_param: vnf_name}
+ volume_type: "solidfire"
+ volume_size: { get_param: volume_size }
+
+ db_vol1:
+ type: OS::Cinder::Volume
+ properties:
+ name:
+ str_replace:
+ template: VNF_NAME_db_vol1
+ params:
+ VNF_NAME: {get_param: vnf_name}
+ volume_type: "solidfire"
+ volume_size: { get_param: volume_size }
+
+outputs:
+ db_vol0_id:
+ value: { get_resource: db_vol0 }
+
+ db_vol1_id:
+ value: { get_resource: db_vol1 }
--- /dev/null
+parameters:
+
+ 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
--- /dev/null
+heat_template_version: 2015-04-30
+
+description: Base Module of Sample VNF
+
+parameters:
+
+ # ONAP Assigned Parameters
+ workload_context:
+ type: string
+ description: Unique ID for this VNF instance
+
+ environment_context:
+ type: string
+ description: Unique ID for this VNF instance
+
+ vnf_id:
+ type: string
+ description: Unique ID for this VNF instance
+
+ vf_module_id:
+ type: string
+ description: Unique ID for this VNF module instance
+
+ vf_module_index:
+ type: number
+ description: Index of this VF Module
+
+ vnf_name:
+ type: string
+ description: Unique name for this VNF instance
+
+
+ # Availability Zones
+ availability_zone_0:
+ type: string
+ description: Primary Availability Zone
+
+
+ # External Networks
+ ha_net_id:
+ type: string
+ description: High Availability Network
+
+ int_private_net_id:
+ type: string
+ description: Private network
+
+ int_private_subnet_id:
+ type: string
+ description: Private network subnet
+
+ # Server Inputs: Loadbalancer
+ lb_names:
+ type: comma_delimited_list
+ description: Load Balancer Names
+
+ lb_image_name:
+ type: string
+ description: Loadbalancer VM Image
+
+ lb_flavor_name:
+ type: string
+ description: Loadbalancer VM Flavor
+
+ lb_ha_floating_ip:
+ type: string
+ description: Floating HA IP for LB
+
+ lb_ha_floating_v6_ip:
+ type: string
+ description: Floating HA IP for LB
+
+ # Server Inputs: Services
+ svc_0_names:
+ type: comma_delimited_list
+ description: Service VM Names
+
+ svc_1_names:
+ type: comma_delimited_list
+ description: Service VM Names
+
+ svc_2_names:
+ type: comma_delimited_list
+ description: Service VM Names
+
+ svc_image_name:
+ type: string
+ description: Service VM Image
+
+ svc_flavor_name:
+ type: string
+ description: Service VM Flavor
+
+ svc_count:
+ type: number
+ description: Number of instances of Service to create
+
+resources:
+
+
+ lb_server_1:
+ type: OS::Nova::Server
+ properties:
+ image: { get_param: lb_image_name }
+ flavor: { get_param: lb_flavor_name }
+ name: { get_param: [lb_names, {get_param: vf_module_index}] }
+ metadata:
+ vnf_id: { get_param: vnf_id }
+ vf_module_id: { get_param: vf_module_id }
+ vf_module_index: { get_param: vf_module_index }
+ vnf_name: { get_param: vnf_name }
+ workload_context: { get_param: workload_context }
+ environment_context: { get_param: environment_context }
+ networks:
+ - port: { get_resource: lb_1_int_private_port_0 }
+ - port: { get_resource: lb_1_ha_port_0 }
+ user_data: { get_file: user.data }
+ availability_zone: { get_param: availability_zone_0 }
+
+ lb_1_ha_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_param: ha_net_id }
+ allowed_address_pairs:
+ - ip_address: {get_param: lb_ha_floating_ip }
+ - ip_address: {get_param: lb_ha_floating_v6_ip }
+
+ lb_1_int_private_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_param: int_private_net_id }
+ fixed_ips:
+ - subnet_id: { get_param: int_private_subnet_id }
+
+ svc_resource_group_1:
+ type: OS::Heat::ResourceGroup
+ properties:
+ count: { get_param: svc_count }
+ resource_def:
+ type: nested_svc.yaml
+ properties:
+ workload_context: {get_param: workload_context}
+ environment_context: {get_param: environment_context}
+ vnf_id: {get_param: vnf_id}
+ vf_module_id: {get_param: vf_module_id}
+ vnf_name: {get_param: vnf_name}
+ availability_zone_0: {get_param: availability_zone_0}
+ svc_names:
+ - {get_param: [svc_0_names, {get_param: vf_module_index}]}
+ - {get_param: [svc_1_names, {get_param: vf_module_index}]}
+ - {get_param: [svc_2_names, {get_param: vf_module_index}]}
+ svc_image_name: {get_param: svc_image_name}
+ svc_flavor_name: {get_param: svc_flavor_name}
+ int_private_net_id: {get_param: int_private_net_id}
+ int_private_subnet_id: {get_param: int_private_subnet_id}
+ index: "%index%"
--- /dev/null
+heat_template_version: 2015-04-30
+
+description: Base Module of Sample VNF
+
+parameters:
+
+ # ONAP Assigned Parameters
+ workload_context:
+ type: string
+ description: Unique ID for this VNF instance
+
+ environment_context:
+ type: string
+ description: Unique ID for this VNF instance
+
+ vnf_id:
+ type: string
+ description: Unique ID for this VNF instance
+
+ vf_module_id:
+ type: string
+ description: Unique ID for this VNF module instance
+
+
+ vnf_name:
+ type: string
+ description: Unique name for this VNF instance
+
+
+ # Availability Zones
+ availability_zone_0:
+ type: string
+ description: Primary Availability Zone
+
+
+ # Server Inputs: Services
+ svc_names:
+ type: comma_delimited_list
+ description: Service VM Names
+
+ svc_image_name:
+ type: string
+ description: Service VM Image
+
+ svc_flavor_name:
+ type: string
+ description: Service VM Flavor
+
+ index:
+ type: number
+ description: Number of services to create
+
+ int_private_net_id:
+ type: string
+ description: Network ID of internal private network
+
+ int_private_subnet_id:
+ type: string
+ description: Subnet ID of internal private network
+
+resources:
+
+ svc_server_0:
+ type: OS::Nova::Server
+ properties:
+ image: { get_param: svc_image_name }
+ flavor: { get_param: svc_flavor_name }
+ name: { get_param: [svc_names, {get_param: index}] }
+ metadata:
+ vnf_id: { get_param: vnf_id }
+ vf_module_id: { get_param: vf_module_id }
+ vnf_name: { get_param: vnf_name }
+ workload_context: { get_param: workload_context }
+ environment_context: { get_param: environment_context }
+ networks:
+ - port: {get_resource: svc_0_int_private_port_0}
+ availability_zone: { get_param: availability_zone_0 }
+
+ svc_0_int_private_port_0:
+ type: OS::Neutron::Port
+ properties:
+ network: { get_param: int_private_net_id }
+ fixed_ips:
+ - subnet: { get_param: int_private_subnet_id }
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+from pathlib import Path
+
+import pytest
+
+from preload.environment import CloudServiceArchive, PreloadEnvironment
+
+THIS_DIR = Path(__file__).parent
+PRELOAD_ENV_DIR = THIS_DIR / "preload_envs"
+
+
+@pytest.fixture(scope="session")
+def csar():
+ return CloudServiceArchive(PRELOAD_ENV_DIR / "test.csar")
+
+
+@pytest.fixture(scope="session")
+def env():
+ return PreloadEnvironment(PRELOAD_ENV_DIR)
+
+
+def test_csar_service_name(csar):
+ assert csar.service_name == "stark_vccf_svc"
+
+
+def test_csar_str_and_repr(csar):
+ assert str(csar) == "CSAR (path=test.csar, name=stark_vccf_svc)"
+ assert repr(csar) == "CSAR (path=test.csar, name=stark_vccf_svc)"
+
+
+def test_csar_vf_module_model_name(csar):
+ assert (
+ csar.get_vf_module_model_name("base_vIECCF")
+ == "StarkVccfVf..base_vIECCF..module-0"
+ )
+
+
+def test_csar_get_vf_module_resource_name(csar):
+ assert csar.get_vf_module_resource_name("base_vIECCF") == "stark_vccf_vf"
+
+
+def test_csar_get_vf_module_resource_name_not_found(csar):
+ assert csar.get_vf_module_resource_name("unknown") is None
+
+
+def test_preload_environment_global_csar(env):
+ assert env.csar.service_name == "stark_vccf_svc"
+
+
+def test_preload_environment_nest_env_csar_inherit(env):
+ env_two = env.get_environment("env_two")
+ assert env_two.csar.service_name == "stark_vccf_svc"
+
+
+def test_preload_environment_nest_env_csar_override(env):
+ sub_env = env.get_environment("env_three")
+ assert sub_env.csar.service_name == "StarkMultiModule2_43550"
+
+
+def test_preload_environment_environments(env):
+ names = {e.name for e in env.environments}
+ assert names == {"env_two", "env_three", "env_one_a"}
+
+
+def test_preload_environment_environments_nested(env):
+ env_one = env.get_environment("env_one")
+ names = {e.name for e in env_one.environments}
+ assert names == {"env_one_a"}
+
+
+def test_preload_environment_get_module_global_base(env):
+ module = env.get_module("base")
+ assert module["my_ip"] == "default"
+
+
+def test_preload_environment_get_module_global_not_found(env):
+ module = env.get_module("unknown")
+ assert module == {}
+
+
+def test_preload_environment_get_module_sub_env(env):
+ env_two = env.get_environment("env_two")
+ module = env_two.get_module("base")
+ assert module["my_ip"] == "192.168.0.2"
+ assert module["common"] == "ABC"
+
+
+def test_preload_environment_module_names(env):
+ expected = {"base.env", "incremental.env"}
+ assert env.module_names == expected
+ # check a nested env with inherits all modules
+ assert env.get_environment("env_three").module_names == expected
+
+
+def test_preload_environment_modules(env):
+ modules = env.modules
+ assert isinstance(modules, dict)
+ assert modules.keys() == {"base.env", "incremental.env"}
+ assert all(isinstance(val, dict) for val in modules.values())
+
+
+def test_preload_environment_is_base(env):
+ assert env.is_base
+ assert not env.get_environment("env_one").is_base
+
+
+def test_preload_environment_is_leaf(env):
+ assert not env.is_leaf
+ assert env.get_environment("env_two").is_leaf
+ assert not env.get_environment("env_one").is_leaf
+ assert env.get_environment("env_one_a").is_leaf
+
+
+def test_preload_environment_str_repr(env):
+ assert str(env) == "PreloadEnvironment(name=preload_envs)"
+ assert repr(env) == "PreloadEnvironment(name=preload_envs)"
+
+
+def test_preload_environment_defaults(env):
+ expected = {"availability_zone_0": "az0"}
+ assert env.defaults == expected
+ assert env.get_environment("env_one_a").defaults == expected
+
+
+def test_preload_environment_defaults_merging_and_override(env):
+ assert env.get_environment("env_three").defaults == {
+ "availability_zone_0": "az0-b",
+ "custom_env_3": "default",
+ }
+
+
+def test_preload_environment_defaults_in_module_env(env):
+ mod = env.get_environment("env_three").get_module("base")
+ assert mod == {
+ "availability_zone_0": "az0-b",
+ "common": "ABC",
+ "custom_env_3": "default",
+ "my_ip": "default",
+ }
+ mod = env.get_environment("env_one").get_module("base")
+ assert mod == {
+ "availability_zone_0": "az0",
+ "common": "ABC",
+ "my_ip": "192.168.0.1",
+ }
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+import json
+import tempfile
+from pathlib import Path
+from shutil import rmtree
+
+import pytest
+
+from preload.environment import PreloadEnvironment
+from preload.model import Vnf, get_heat_templates
+from preload_grapi import GrApiPreloadGenerator
+from tests.helpers import first
+
+THIS_DIR = Path(__file__).parent
+SAMPLE_HEAT_DIR = THIS_DIR / "sample_heat"
+
+
+def load_json(path):
+ with path.open("r") as f:
+ return json.load(f)
+
+
+def load_module(base_dir, name):
+ path = Path(str(base_dir / "grapi" / name))
+ assert path.exists(), "{} does not exist".format(path)
+ return load_json(path)
+
+
+@pytest.fixture(scope="session")
+def session_dir(request):
+ # Temporary directory that gets deleted at the session
+ # pytest tmpdir doesn't support a non-function scoped temporary directory
+ session_dir = Path(tempfile.mkdtemp())
+ request.addfinalizer(lambda: rmtree(session_dir))
+ return session_dir
+
+
+@pytest.fixture(scope="session")
+def preload(pytestconfig, session_dir):
+ # Generate the preloads for testing
+ def fake_getoption(opt, default=None):
+ return [SAMPLE_HEAT_DIR.as_posix()] if opt == "template_dir" else None
+
+ pytestconfig.getoption = fake_getoption
+ templates = get_heat_templates(pytestconfig)
+ env = PreloadEnvironment(THIS_DIR / "sample_env")
+ vnf = Vnf(templates)
+ generator = GrApiPreloadGenerator(vnf, session_dir, env)
+ generator.generate()
+ return session_dir
+
+
+@pytest.fixture(scope="session")
+def base(preload):
+ return load_module(preload, "base.json")
+
+
+@pytest.fixture(scope="session")
+def incremental(preload):
+ return load_module(preload, "incremental.json")
+
+
+def test_base_fields(base):
+ data = base["input"]["preload-vf-module-topology-information"][
+ "vnf-topology-identifier-structure"
+ ]
+ assert data["vnf-name"] == "VALUE FOR: vnf_name"
+ assert "<Service Name>/<VF Instance Name>" in data["vnf-type"]
+
+
+def test_base_azs(base):
+ az = base["input"]["preload-vf-module-topology-information"][
+ "vnf-resource-assignments"
+ ]["availability-zones"]["availability-zone"]
+ assert isinstance(az, list)
+ assert len(az) == 2
+ assert az[0] == "VALUE FOR: availability_zone_0"
+
+
+def test_base_networks(base):
+ nets = base["input"]["preload-vf-module-topology-information"][
+ "vnf-resource-assignments"
+ ]["vnf-networks"]["vnf-network"]
+ assert isinstance(nets, list)
+ assert len(nets) == 3
+ oam = first(nets, lambda n: n["network-role"] == "oam")
+ 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"}]},
+ }
+
+
+def test_base_vm_types(base):
+ vms = base["input"]["preload-vf-module-topology-information"]["vf-module-topology"][
+ "vf-module-assignments"
+ ]["vms"]["vm"]
+ vm_types = {vm["vm-type"] for vm in vms}
+ assert vm_types == {"db", "svc", "mgmt", "lb"}
+ db = first(vms, lambda v: v["vm-type"] == "db")
+ assert db == {
+ "vm-type": "db",
+ "vm-count": 2,
+ "vm-names": {"vm-name": ["VALUE FOR: db_name_0", "VALUE FOR: db_name_1"]},
+ "vm-networks": {
+ "vm-network": [
+ {
+ "network-role": "oam",
+ "network-information-items": {
+ "network-information-item": [
+ {
+ "ip-version": "4",
+ "use-dhcp": "N",
+ "ip-count": 2,
+ "network-ips": {
+ "network-ip": [
+ "VALUE FOR: db_oam_ip_0",
+ "VALUE FOR: db_oam_ip_1",
+ ]
+ },
+ },
+ {
+ "ip-version": "6",
+ "use-dhcp": "N",
+ "ip-count": 0,
+ "network-ips": {"network-ip": []},
+ },
+ ]
+ },
+ "mac-addresses": {"mac-address": []},
+ "floating-ips": {"floating-ip-v4": [], "floating-ip-v6": []},
+ "interface-route-prefixes": {"interface-route-prefix": []},
+ },
+ {
+ "network-role": "ha",
+ "network-information-items": {
+ "network-information-item": [
+ {
+ "ip-version": "4",
+ "use-dhcp": "N",
+ "ip-count": 0,
+ "network-ips": {"network-ip": []},
+ },
+ {
+ "ip-version": "6",
+ "use-dhcp": "N",
+ "ip-count": 0,
+ "network-ips": {"network-ip": []},
+ },
+ ]
+ },
+ "mac-addresses": {"mac-address": []},
+ "floating-ips": {
+ "floating-ip-v4": ["VALUE FOR: db_ha_floating_ip"],
+ "floating-ip-v6": ["VALUE FOR: db_ha_floating_v6_ip"],
+ },
+ "interface-route-prefixes": {"interface-route-prefix": []},
+ },
+ ]
+ },
+ }
+
+
+def test_base_general(base):
+ general = base["input"]["preload-vf-module-topology-information"][
+ "vf-module-topology"
+ ]["vf-module-topology-identifier"]
+ assert (
+ general["vf-module-type"] == "VALUE FOR: <vfModuleModelName> from CSAR or SDC"
+ )
+ assert general["vf-module-name"] == "VALUE FOR: vf_module_name"
+
+
+def test_base_parameters(base):
+ params = base["input"]["preload-vf-module-topology-information"][
+ "vf-module-topology"
+ ]["vf-module-parameters"]["param"]
+ assert params == [
+ {"name": "svc_image_name", "value": "svc_image"},
+ {"name": "svc_flavor_name", "value": "svc_flavor"},
+ ]
+
+
+def test_incremental(incremental):
+ az = incremental["input"]["preload-vf-module-topology-information"][
+ "vnf-resource-assignments"
+ ]["availability-zones"]["availability-zone"]
+ assert isinstance(az, list)
+ assert len(az) == 1
+ assert az[0] == "VALUE FOR: availability_zone_0"
+
+
+def test_incremental_networks(incremental):
+ nets = incremental["input"]["preload-vf-module-topology-information"][
+ "vnf-resource-assignments"
+ ]["vnf-networks"]["vnf-network"]
+ assert isinstance(nets, list)
+ assert len(nets) == 1
+ assert nets[0]["network-role"] == "ha"
+
+
+def test_preload_env_population(preload):
+ base_path = THIS_DIR / "sample_env/preloads/grapi/base.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"]
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+import tempfile
+from pathlib import Path
+from shutil import rmtree
+
+import pytest
+
+from app_tests.preload_tests.test_grapi import load_json
+from preload.environment import PreloadEnvironment
+from preload.model import Vnf, get_heat_templates
+from preload_vnfapi import VnfApiPreloadGenerator
+from tests.helpers import load_yaml, first
+
+THIS_DIR = Path(__file__).parent
+SAMPLE_HEAT_DIR = THIS_DIR / "sample_heat"
+
+
+def load_module(base_dir, name):
+ path = Path(str(base_dir / "vnfapi" / name))
+ assert path.exists(), "{} does not exist".format(path)
+ return load_yaml(str(path))
+
+
+@pytest.fixture(scope="session")
+def session_dir(request):
+ # Temporary directory that gets deleted at the session
+ # pytest tmpdir doesn't support a non-function scoped temporary directory
+ session_dir = Path(tempfile.mkdtemp())
+ request.addfinalizer(lambda: rmtree(session_dir))
+ return session_dir
+
+
+@pytest.fixture(scope="session")
+def preload(pytestconfig, session_dir):
+ # Generate the preloads for testing
+ def fake_getoption(opt, default=None):
+ return [SAMPLE_HEAT_DIR.as_posix()] if opt == "template_dir" else None
+
+ 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)
+ generator.generate()
+ return session_dir
+
+
+@pytest.fixture(scope="session")
+def base(preload):
+ return load_module(preload, "base.json")
+
+
+@pytest.fixture(scope="session")
+def incremental(preload):
+ return load_module(preload, "incremental.json")
+
+
+def test_base_azs(base):
+ az = base["input"]["vnf-topology-information"]["vnf-assignments"][
+ "availability-zones"
+ ]
+ assert az == [
+ {"availability-zone": "VALUE FOR: availability_zone_0"},
+ {"availability-zone": "VALUE FOR: availability_zone_1"},
+ ]
+
+
+def test_base_networks(base):
+ nets = base["input"]["vnf-topology-information"]["vnf-assignments"]["vnf-networks"]
+ assert nets == [
+ {
+ "network-role": "oam",
+ "network-name": "VALUE FOR: network name for oam_net_id",
+ "subnet-id": "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",
+ },
+ ]
+
+
+def test_base_vm_types(base):
+ vms = base["input"]["vnf-topology-information"]["vnf-assignments"]["vnf-vms"]
+ vm_types = {vm["vm-type"] for vm in vms}
+ assert vm_types == {"db", "svc", "mgmt", "lb"}
+ db = first(vms, lambda v: v["vm-type"] == "db")
+ assert db == {
+ "vm-type": "db",
+ "vm-count": 2,
+ "vm-names": {"vm-name": ["VALUE FOR: db_name_0", "VALUE FOR: db_name_1"]},
+ "vm-networks": [
+ {
+ "network-role": "oam",
+ "network-role-tag": "oam",
+ "ip-count": 2,
+ "ip-count-ipv6": 0,
+ "floating-ip": "",
+ "floating-ip-v6": "",
+ "network-ips": [
+ {"ip-address": "VALUE FOR: db_oam_ip_0"},
+ {"ip-address": "VALUE FOR: db_oam_ip_1"},
+ ],
+ "network-ips-v6": [],
+ "network-macs": [],
+ "interface-route-prefixes": [],
+ "use-dhcp": "N",
+ },
+ {
+ "network-role": "ha",
+ "network-role-tag": "ha",
+ "ip-count": 0,
+ "ip-count-ipv6": 0,
+ "floating-ip": "VALUE FOR: db_ha_floating_ip",
+ "floating-ip-v6": "VALUE FOR: db_ha_floating_v6_ip",
+ "network-ips": [],
+ "network-ips-v6": [],
+ "network-macs": [],
+ "interface-route-prefixes": [],
+ "use-dhcp": "N",
+ },
+ ],
+ }
+
+
+def test_base_parameters(base):
+ params = base["input"]["vnf-topology-information"]["vnf-parameters"]
+ assert params == [
+ {"vnf-parameter-name": "svc_image_name", "vnf-parameter-value": "svc_image"},
+ {"vnf-parameter-name": "svc_flavor_name", "vnf-parameter-value": "svc_flavor"},
+ ]
+
+
+def test_incremental(incremental):
+ az = incremental["input"]["vnf-topology-information"]["vnf-assignments"][
+ "availability-zones"
+ ]
+ assert isinstance(az, list)
+ assert len(az) == 1
+ assert az[0] == {"availability-zone": "VALUE FOR: availability_zone_0"}
+
+
+def test_incremental_networks(incremental):
+ nets = incremental["input"]["vnf-topology-information"]["vnf-assignments"][
+ "vnf-networks"
+ ]
+ assert isinstance(nets, list)
+ assert len(nets) == 1
+ assert nets[0]["network-role"] == "ha"
+
+
+def test_preload_env_population(preload):
+ base_path = THIS_DIR / "sample_env/preloads/vnfapi/base.json"
+ data = load_json(base_path)
+ azs = data["input"]["vnf-topology-information"]["vnf-assignments"][
+ "availability-zones"
+ ]
+ assert azs == [{"availability-zone": "az0"}, {"availability-zone": "az1"}]
# limitations under the License.
#
# ============LICENSE_END============================================
-#
-#
+import uuid
from io import StringIO
import pytest
import yaml
+from config import Config, get_generator_plugin_names, to_uri
import vvp
+
DEFAULT_CONFIG = """
+namespace: {namespace}
+owner: onap-test
ui:
app-name: VNF Validation Tool
+ requirement-link-url: http://requirement.url.com
categories:
- name: Environment File Compliance. (Required to Onboard)
category: environment_file
Required for ASDC onboarding, not needed for manual Openstack testing.
settings:
polling-freqency: 1000
- default-verbosity: Standard
+ env-specs:
+ - tests.test_environment_file_parameters.ENV_PARAMETER_SPEC
+terms:
+ version: 1.0.0
+ path: path/to/terms.txt
+ popup-title: Terms and Conditions
+ popup-link-text: View Terms and Conditions
+ popup-msg-text: Review and Accept the Terms
"""
# noinspection PyShadowingNames
-@pytest.fixture(scope="module")
+@pytest.fixture()
def config():
- return vvp.Config(yaml.safe_load(StringIO(DEFAULT_CONFIG)))
+ unique = str(uuid.uuid4())
+ data = DEFAULT_CONFIG.format(namespace=unique)
+ return Config(yaml.safe_load(StringIO(data)))
def test_app_name(config):
)
-def test_default_verbosity(config):
- assert config.default_verbosity(vvp.ValidatorApp.VERBOSITY_LEVELS) == "Standard (-v)"
-
-
def test_queues(config):
assert config.log_queue.empty(), "Log should start empty"
config.log_file.write("Test")
MISSING_CATEGORY_FIELD = """
+namespace: org.onap.test
+owner: onap-test
ui:
app-name: VNF Validation Tool
categories:
def test_missing_category_fields():
settings = yaml.safe_load(StringIO(MISSING_CATEGORY_FIELD))
with pytest.raises(RuntimeError) as e:
- vvp.Config(settings)
+ Config(settings)
assert "Missing: name" in str(e)
def test_input_formats(config):
assert "Directory (Uncompressed)" in config.input_formats
assert "ZIP File" in config.input_formats
+
+
+def test_env_specs(config):
+ specs = config.env_specs
+ assert len(specs) == 1
+ assert "ALL" in specs[0]
+
+
+def test_get_generator_plugin_names(config):
+ names = get_generator_plugin_names()
+ assert "VNF-API" in names
+ assert "GR-API" in names
+
+
+def test_preload_formats(config):
+ formats = config.preload_formats
+ assert all(format in formats for format in ("VNF-API", "GR-API"))
+
+
+def test_requirement_link_http(config):
+ assert config.requirement_link_url == "http://requirement.url.com"
+
+
+def test_to_uri_relative_path():
+ assert to_uri("path/").startswith("file://")
+ assert to_uri("path/").endswith("/path")
+
+
+def test_to_uri_relative_http():
+ assert to_uri("http://url.com") == "http://url.com"
+
+
+def test_to_uri_absolute_path():
+ assert to_uri("/path/one").startswith("file:///")
+ assert to_uri("/path/one").endswith("/path/one")
+
+
+def test_requirement_link_path(config):
+ config._config["ui"]["requirement-link-url"] = "path/to/reqs.txt"
+ url = config.requirement_link_url
+ assert url.startswith("file://")
+ assert "path/to/reqs.txt" in url
+
+
+def test_terms_version(config):
+ assert config.terms_version == "1.0.0"
+
+
+def test_terms_popup_title(config):
+ assert config.terms_popup_title == "Terms and Conditions"
+
+
+def test_terms_popup_message(config):
+ assert config.terms_popup_message == "Review and Accept the Terms"
+
+
+def test_terms_link_url_default(config):
+ config._config["terms"]["path"] = None
+ assert config.terms_link_url is None
+
+
+def test_terms_acceptance(config):
+ assert not config.are_terms_accepted
+ config.set_terms_accepted()
+ assert config.are_terms_accepted
+
+
+def test_terms_link_url_path(config):
+ assert config.terms_link_url.startswith("file://")
+ assert config.terms_link_url.endswith("/path/to/terms.txt")
+
+
+def test_terms_link_text(config):
+ assert config.terms_link_text == "View Terms and Conditions"
+
+
+def test_default_halt_on_failure(config):
+ assert config.default_halt_on_failure
+
+
+def test_get_subdir_for_preload(config):
+ assert config.get_subdir_for_preload("VNF-API") == "vnfapi"
+
+
+def test_default_preload_format(config):
+ assert config.default_preload_format in ("VNF-API", "GR-API", "Excel")
+
+
+def test_category_description(config):
+ assert "Checks certain parameters" in config.get_description(
+ "Environment File Compliance. (Required to Onboard)"
+ )
+
+
+def test_get_category_by_name(config):
+ assert (
+ config.get_category("Environment File Compliance. (Required to Onboard)")
+ == "environment_file"
+ )
+
+
+def test_cached_category_setting(config):
+ assert (
+ config.get_category_value("Environment File Compliance. (Required to Onboard)")
+ == 0
+ )
+
+
+def test_disclaimer_text(config):
+ assert config.disclaimer_text == ""
+
+
+def test_requirement_link_text(config):
+ url_text = "Requirement URL"
+ config._config["ui"]["requirement-link-text"] = url_text
+ assert config.requirement_link_text == url_text
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+from pathlib import Path
+
+import pytest
+
+from tests.helpers import check, first, unzip, remove
+
+THIS_DIR = Path(__file__).parent
+
+
+def test_check_fail():
+ with pytest.raises(RuntimeError, match="pre-condition failed"):
+ check(False, "pre-condition failed")
+
+
+def test_check_pass():
+ check(True, "pre-condition failed")
+
+
+def test_first_found():
+ result = first(range(1, 10), lambda x: x % 4 == 0)
+ assert result == 4
+
+
+def test_first_not_found():
+ result = first(range(1, 3), lambda x: x % 4 == 0)
+ assert result is None
+
+
+def test_first_custom_default():
+ result = first(range(1, 3), lambda x: x % 4 == 0, default="not found")
+ assert result == "not found"
+
+
+def test_unzip_success(tmpdir):
+ test_zip = THIS_DIR / "test_data.zip"
+ target_dir = tmpdir.join("sub-dir")
+ unzip(test_zip, target_dir)
+ assert "data.txt" in (p.basename for p in target_dir.listdir())
+
+
+def test_unzip_not_found(tmpdir):
+ test_zip = THIS_DIR / "test_data1.zip"
+ with pytest.raises(RuntimeError, match="not a valid zipfile"):
+ unzip(test_zip, tmpdir)
+
+
+def test_remove_with_no_key():
+ assert remove([1, 2, 3, 4], [3]) == [1, 2, 4]
+
+
+def test_remove_with_key():
+ assert remove(["a", "b", "c", "d"], ["A"], lambda s: s.upper()) == ["b", "c", "d"]
# ============LICENSE_END============================================
#
#
-
+namespace: org.onap.test
+owner: onap-test
ui:
app-name: VNF Validation Tool
categories:
settings:
polling-freqency: 1000
default-verbosity: Standard
+ env-specs:
+ - tests.test_environment_file_parameters.ENV_PARAMETER_SPEC
--- /dev/null
+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 appdirs
+import yaml
+from cached_property import cached_property
+
+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__))
+PROTOCOLS = ("http:", "https:", "file:")
+
+
+def to_uri(path):
+ if any(path.startswith(p) for p in PROTOCOLS):
+ return path
+ return Path(path).absolute().as_uri()
+
+
+class UserSettings(MutableMapping):
+ FILE_NAME = "UserSettings.ini"
+
+ def __init__(self, namespace, owner):
+ user_config_dir = appdirs.AppDirs(namespace, owner).user_config_dir
+ if not os.path.exists(user_config_dir):
+ os.makedirs(user_config_dir, exist_ok=True)
+ self._settings_path = os.path.join(user_config_dir, self.FILE_NAME)
+ self._config = ConfigParser()
+ self._config.read(self._settings_path)
+
+ def __getitem__(self, k):
+ return self._config["DEFAULT"][k]
+
+ def __setitem__(self, k, v) -> None:
+ self._config["DEFAULT"][k] = v
+
+ def __delitem__(self, v) -> None:
+ del self._config["DEFAULT"][v]
+
+ def __len__(self) -> int:
+ return len(self._config["DEFAULT"])
+
+ def __iter__(self) -> Iterator:
+ return iter(self._config["DEFAULT"])
+
+ def save(self):
+ with open(self._settings_path, "w") as f:
+ self._config.write(f)
+
+
+class Config:
+ """
+ Configuration for the Validation GUI Application
+
+ Attributes
+ ----------
+ ``log_queue`` Queue for the ``stdout`` and ``stderr` of
+ the background job
+ ``log_file`` File-like object (write only!) that writes to
+ the ``log_queue``
+ ``status_queue`` Job completion status of the background job is
+ posted here as a tuple of (bool, Exception).
+ The first parameter is True if the job completed
+ successfully, and False otherwise. If the job
+ failed, then an Exception will be provided as the
+ second element.
+ ``command_queue`` Used to send commands to the GUI. Currently only
+ used to send shutdown commands in tests.
+ """
+
+ DEFAULT_FILENAME = "vvp-config.yaml"
+ DEFAULT_POLLING_FREQUENCY = "1000"
+
+ def __init__(self, config: dict = None):
+ """Creates instance of application configuration.
+
+ :param config: override default configuration if provided."""
+ if config:
+ self._config = config
+ else:
+ with open(self.DEFAULT_FILENAME, "r") as f:
+ self._config = yaml.safe_load(f)
+ self._user_settings = UserSettings(
+ self._config["namespace"], self._config["owner"]
+ )
+ self._watched_variables = []
+ self._validate()
+
+ @cached_property
+ def manager(self):
+ return multiprocessing.Manager()
+
+ @cached_property
+ def log_queue(self):
+ return self.manager.Queue()
+
+ @cached_property
+ def status_queue(self):
+ return self.manager.Queue()
+
+ @cached_property
+ def log_file(self):
+ return QueueWriter(self.log_queue)
+
+ @cached_property
+ def command_queue(self):
+ return self.manager.Queue()
+
+ def watch(self, *variables):
+ """Traces the variables and saves their settings for the user. The
+ last settings will be used where available"""
+ self._watched_variables = variables
+ for var in self._watched_variables:
+ var.trace_add("write", self.save_settings)
+
+ # noinspection PyProtectedMember,PyUnusedLocal
+ def save_settings(self, *args):
+ """Save the value of all watched variables to user settings"""
+ for var in self._watched_variables:
+ self._user_settings[var._name] = str(var.get())
+ self._user_settings.save()
+
+ @property
+ def app_name(self) -> str:
+ """Name of the application (displayed in title bar)"""
+ app_name = self._config["ui"].get("app-name", "VNF Validation Tool")
+ return "{} - {}".format(app_name, VERSION)
+
+ @property
+ def category_names(self) -> List[str]:
+ """List of validation profile names for display in the UI"""
+ return [category["name"] for category in self._config["categories"]]
+
+ @property
+ def polling_frequency(self) -> int:
+ """Returns the frequency (in ms) the UI polls the queue communicating
+ with any background job"""
+ return int(
+ self._config["settings"].get(
+ "polling-frequency", self.DEFAULT_POLLING_FREQUENCY
+ )
+ )
+
+ @property
+ def disclaimer_text(self) -> str:
+ return self._config["ui"].get("disclaimer-text", "")
+
+ @property
+ def requirement_link_text(self) -> str:
+ return self._config["ui"].get("requirement-link-text", "")
+
+ @property
+ def requirement_link_url(self) -> str:
+ path = self._config["ui"].get("requirement-link-url", "")
+ return to_uri(path)
+
+ @property
+ def terms(self) -> dict:
+ return self._config.get("terms", {})
+
+ @property
+ def terms_link_url(self) -> Optional[str]:
+ path = self.terms.get("path")
+ return to_uri(path) if path else None
+
+ @property
+ def terms_link_text(self):
+ return self.terms.get("popup-link-text")
+
+ @property
+ def terms_version(self) -> Optional[str]:
+ return self.terms.get("version")
+
+ @property
+ def terms_popup_title(self) -> Optional[str]:
+ return self.terms.get("popup-title")
+
+ @property
+ def terms_popup_message(self) -> Optional[str]:
+ return self.terms.get("popup-msg-text")
+
+ @property
+ def are_terms_accepted(self) -> bool:
+ version = "terms-{}".format(self.terms_version)
+ return self._user_settings.get(version, "False") == "True"
+
+ def set_terms_accepted(self):
+ version = "terms-{}".format(self.terms_version)
+ self._user_settings[version] = "True"
+ self._user_settings.save()
+
+ def get_description(self, category_name: str) -> str:
+ """Returns the description associated with the category name"""
+ return self._get_category(category_name)["description"]
+
+ def get_category(self, category_name: str) -> str:
+ """Returns the category associated with the category name"""
+ return self._get_category(category_name).get("category", "")
+
+ def get_category_value(self, category_name: str) -> str:
+ """Returns the saved value for a category name"""
+ return self._user_settings.get(category_name, 0)
+
+ def _get_category(self, category_name: str) -> Dict[str, str]:
+ """Returns the profile definition"""
+ for category in self._config["categories"]:
+ if category["name"] == category_name:
+ return category
+ raise RuntimeError(
+ "Unexpected error: No category found in vvp-config.yaml "
+ "with a name of " + category_name
+ )
+
+ @property
+ def default_report_format(self):
+ return self._user_settings.get("report_format", "HTML")
+
+ @property
+ def report_formats(self):
+ return ["CSV", "Excel", "HTML"]
+
+ @property
+ def preload_formats(self):
+ excluded = self._config.get("excluded-preloads", [])
+ formats = (cls.format_name() for cls in get_generator_plugins())
+ return [f for f in formats if f not in excluded]
+
+ @property
+ def default_preload_format(self):
+ default = self._user_settings.get("preload_format")
+ if default and default in self.preload_formats:
+ return default
+ else:
+ return self.preload_formats[0]
+
+ @staticmethod
+ def get_subdir_for_preload(preload_format):
+ for gen in get_generator_plugins():
+ if gen.format_name() == preload_format:
+ return gen.output_sub_dir()
+ return ""
+
+ @property
+ def default_input_format(self):
+ requested_default = self._user_settings.get("input_format") or self._config[
+ "settings"
+ ].get("default-input-format")
+ if requested_default in self.input_formats:
+ return requested_default
+ else:
+ return self.input_formats[0]
+
+ @property
+ def input_formats(self):
+ return ["Directory (Uncompressed)", "ZIP File"]
+
+ @property
+ def default_halt_on_failure(self):
+ setting = self._user_settings.get("halt_on_failure", "True")
+ return setting.lower() == "true"
+
+ @property
+ def env_specs(self):
+ env_specs = self._config["settings"].get("env-specs")
+ specs = []
+ if not env_specs:
+ return [ENV_PARAMETER_SPEC]
+ for mod_path, attr in (s.rsplit(".", 1) for s in env_specs):
+ module = importlib.import_module(mod_path)
+ specs.append(getattr(module, attr))
+ return specs
+
+ def _validate(self):
+ """Ensures the config file is properly formatted"""
+ categories = self._config["categories"]
+
+ # All profiles have required keys
+ expected_keys = {"name", "description"}
+ for category in categories:
+ actual_keys = set(category.keys())
+ missing_keys = expected_keys.difference(actual_keys)
+ if missing_keys:
+ raise RuntimeError(
+ "Error in vvp-config.yaml file: "
+ "Required field missing in category. "
+ "Missing: {} "
+ "Categories: {}".format(",".join(missing_keys), category)
+ )
+
+
+class QueueWriter:
+ """``stdout`` and ``stderr`` will be written to this queue by pytest, and
+ pulled into the main GUI application"""
+
+ def __init__(self, log_queue: queue.Queue):
+ """Writes data to the provided queue.
+
+ :param log_queue: the queue instance to write to.
+ """
+ self.queue = log_queue
+
+ def write(self, data: str):
+ """Writes ``data`` to the queue """
+ self.queue.put(data)
+
+ # noinspection PyMethodMayBeStatic
+ def isatty(self) -> bool:
+ """Always returns ``False``"""
+ return False
+
+ 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()]
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
--- /dev/null
+import re
+import tempfile
+from pathlib import Path
+
+from cached_property import cached_property
+
+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")
+
+
+def yaml_files(path):
+ """
+ Return files that are YAML (end with .yml or .yaml)
+
+ :param path: Directory path object
+ :return: list of paths to YAML files
+ """
+ return [
+ p
+ for p in path.iterdir()
+ if p.is_file() and p.suffix.lower() in (".yml", ".yaml")
+ ]
+
+
+class CloudServiceArchive:
+ """
+ Wrapper to extract information from a CSAR file.
+ """
+
+ def __init__(self, csar_path):
+ self.csar_path = Path(csar_path)
+ with tempfile.TemporaryDirectory() as csar_dir:
+ csar_dir = Path(csar_dir)
+ unzip(self.csar_path, csar_dir)
+ self._service = self._get_service_template(csar_dir)
+ self._resources = self._get_vf_module_resource_templates(csar_dir)
+
+ def get_vf_module(self, vf_module):
+ """
+ Retrieve the VF Module definition from the CSAR for the given heat
+ module name (should not include the file extension - ex: base)
+
+ :param vf_module: name of Heat module (no path or file extension)
+ :return: The definition of the module as a dict or None if not found
+ """
+ groups = self._service.get("topology_template", {}).get("groups", {})
+ for props in groups.values():
+ module_label = props.get("properties", {}).get("vf_module_label", "")
+ if module_label.lower() == vf_module.lower():
+ return props
+ return None
+
+ def get_vf_module_model_name(self, vf_module):
+ """
+ Retrieves the vfModuleModelName of the module or None if vf_module is not
+ found (see get_vf_module)
+
+ :param vf_module: name of Heat module (no path or file extension)
+ :return: The value if vfModuleModelName as string or None if not found
+ """
+ module = self.get_vf_module(vf_module)
+ return module.get("metadata", {}).get("vfModuleModelName") if module else None
+
+ @property
+ def topology_template(self):
+ """
+ Return dict representing the topology_template node of the service
+ template
+ """
+ return self._service.get("topology_template") or {}
+
+ @property
+ def groups(self):
+ """
+ Return dict representing the groups node of the service
+ template
+ """
+ return self.topology_template.get("groups") or {}
+
+ @property
+ def vf_modules(self):
+ """
+ Returns mapping of group ID to VfModule present in the service template
+ """
+ return {
+ group_id: props
+ for group_id, props in self.groups.items()
+ if props.get("type") == "org.openecomp.groups.VfModule"
+ }
+
+ @property
+ def vf_module_resource_names(self):
+ """
+ Returns the resource names for all VfModules (these can be used
+ to find the resource templates as they will be part of the filename)
+ """
+ names = (
+ module.get("metadata", {}).get("vfModuleModelName")
+ for module in self.vf_modules.values()
+ )
+ return [name.split(".")[0] for name in names if name]
+
+ def get_vf_module_resource_name(self, vf_module):
+ """
+ Retrieves the resource name of the module or None if vf_module is not
+ found (see get_vf_module)
+
+ :param vf_module: name of Heat module (no path or file extension)
+ :return: The value if resource nae as string or None if not found
+ """
+ vf_model_name = self.get_vf_module_model_name(vf_module)
+ if not vf_model_name:
+ return None
+ resource_name = vf_model_name.split(".")[0]
+ resource = self._resources.get(resource_name, {})
+ return resource.get("metadata", {}).get("name")
+
+ @staticmethod
+ def _get_definition_files(csar_dir):
+ """
+ Returns a list of all files in the CSAR's Definitions directory
+ """
+ def_dir = csar_dir / "Definitions"
+ check(
+ def_dir.exists(),
+ f"CSAR is invalid. {csar_dir.as_posix()} does not contain a "
+ f"Definitions directory.",
+ )
+ return yaml_files(def_dir)
+
+ def _get_service_template(self, csar_dir):
+ """
+ Returns the service template as a dict. Assumes there is only one.
+ """
+ files = map(str, self._get_definition_files(csar_dir))
+ service_template = first(files, SERVICE_TEMPLATE_PATTERN.match)
+ return load_yaml(service_template) if service_template else {}
+
+ def _get_vf_module_resource_templates(self, csar_dir):
+ """
+ Returns a mapping of resource name to resource definition (as a dict)
+ (Only loads resource templates that correspond to VF Modules
+ """
+ def_dir = csar_dir / "Definitions"
+ mapping = (
+ (name, def_dir / "resource-{}-template.yml".format(name))
+ for name in self.vf_module_resource_names
+ )
+ return {name: load_yaml(path) for name, path in mapping if path.exists()}
+
+ @property
+ def service_name(self):
+ """
+ Name of the service (extracted from the service template
+ """
+ return self._service.get("metadata", {}).get("name")
+
+ def __repr__(self):
+ return f"CSAR (path={self.csar_path.name}, name={self.service_name})"
+
+ def __str__(self):
+ return repr(self)
+
+
+class PreloadEnvironment:
+ """
+ A
+ """
+
+ def __init__(self, env_dir, parent=None):
+ self.base_dir = Path(env_dir)
+ self.parent = parent
+ self._modules = self._load_modules()
+ self._sub_env = self._load_envs()
+ self._defaults = self._load_defaults()
+
+ def _load_defaults(self):
+ defaults = self.base_dir / "defaults.yaml"
+ return load_yaml(defaults) if defaults.exists() else {}
+
+ def _load_modules(self):
+ files = [
+ p
+ for p in self.base_dir.iterdir()
+ if p.is_file() and p.suffix.lower().endswith(".env")
+ ]
+ return {f.name.lower(): load_yaml(f).get("parameters", {}) for f in files}
+
+ def _load_envs(self):
+ env_dirs = [
+ p for p in self.base_dir.iterdir() if p.is_dir() and p.name != "preloads"
+ ]
+ return {d.name: PreloadEnvironment(d, self) for d in env_dirs}
+
+ @cached_property
+ def csar(self):
+ csar_path = first(self.base_dir.iterdir(), lambda p: p.suffix == ".csar")
+ if csar_path:
+ return CloudServiceArchive(csar_path)
+ else:
+ return self.parent.csar if self.parent else None
+
+ @property
+ def defaults(self):
+ result = {}
+ if self.parent:
+ result.update(self.parent.defaults)
+ result.update(self._defaults)
+ return result
+
+ @property
+ def environments(self):
+ all_envs = [self]
+ for env in self._sub_env.values():
+ all_envs.append(env)
+ all_envs.extend(env.environments)
+ return [e for e in all_envs if e.is_leaf]
+
+ def get_module(self, name):
+ name = name if name.lower().endswith(".env") else f"{name}.env".lower()
+ if name not in self.module_names:
+ return {}
+ result = {}
+ parent_module = self.parent.get_module(name) if self.parent else None
+ module = self._modules.get(name)
+ for m in (parent_module, self.defaults, module):
+ if m:
+ result.update(m)
+ return result
+
+ @property
+ def module_names(self):
+ parent_modules = self.parent.module_names if self.parent else set()
+ result = set()
+ result.update(self._modules.keys())
+ result.update(parent_modules)
+ return result
+
+ @property
+ def modules(self):
+ return {name: self.get_module(name) for name in self.module_names}
+
+ def get_environment(self, env_name):
+ for name, env in self._sub_env.items():
+ if name == env_name:
+ return env
+ result = env.get_environment(env_name)
+ if result:
+ return result
+ return None
+
+ @property
+ def is_base(self):
+ return self.parent is None
+
+ @property
+ def is_leaf(self):
+ return not self._sub_env
+
+ @property
+ def name(self):
+ return self.base_dir.name
+
+ def __repr__(self):
+ return f"PreloadEnvironment(name={self.name})"
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+
+import json
+import os
+from abc import ABC, abstractmethod
+
+import yaml
+
+
+def get_json_template(template_dir, template_name):
+ template_name = template_name + ".json"
+ with open(os.path.join(template_dir, template_name)) as f:
+ return json.loads(f.read())
+
+
+def get_or_create_template(template_dir, key, value, sequence, template_name):
+ """
+ Search a sequence of dicts where a given key matches value. If
+ found, then it returns that item. If not, then it loads the
+ template identified by template_name, adds it ot the sequence, and
+ returns the template
+ """
+ for item in sequence:
+ if item[key] == value:
+ return item
+ new_template = get_json_template(template_dir, template_name)
+ sequence.append(new_template)
+ 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):
+ """
+ Optionally used by the preload generator to wrap items in the preload
+ that need to be replaced by end users
+ :param param: p
+ """
+ return "VALUE FOR: {}".format(param) if param else ""
+
+
+class AbstractPreloadGenerator(ABC):
+ """
+ All preload generators must inherit from this class and implement the
+ abstract methods.
+
+ Preload generators are automatically discovered at runtime via a plugin
+ architecture. The system path is scanned looking for modules with the name
+ preload_*, then all non-abstract classes that inherit from AbstractPreloadGenerator
+ are registered as preload plugins
+
+ Attributes:
+ :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
+ """
+
+ def __init__(self, vnf, base_output_dir, preload_env):
+ self.preload_env = preload_env
+ self.vnf = vnf
+ self.current_module = None
+ self.current_module_env = {}
+ self.base_output_dir = base_output_dir
+ self.env_cache = {}
+
+ @classmethod
+ @abstractmethod
+ def format_name(cls):
+ """
+ String name to identify the format (ex: VN-API, GR-API)
+ """
+ raise NotImplementedError()
+
+ @classmethod
+ @abstractmethod
+ def output_sub_dir(cls):
+ """
+ String sub-directory name that will appear under ``base_output_dir``
+ """
+ raise NotImplementedError()
+
+ @classmethod
+ @abstractmethod
+ def supports_output_passing(cls):
+ """
+ Some preload methods allow automatically mapping output parameters in the
+ base module to the input parameter of other modules. This means these
+ that the incremental modules do not need these base module outputs in their
+ preloads.
+
+ At this time, VNF-API does not support output parameter passing, but
+ GR-API does.
+
+ If this is true, then the generator will call Vnf#filter_output_params
+ after the preload module for the base module has been created
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def generate_module(self, module, output_dir):
+ """
+ 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.
+ """
+ raise NotImplementedError()
+
+ def generate(self):
+ # handle the base module first
+ print("\nGenerating {} preloads".format(self.format_name()))
+ self.generate_environments(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)
+
+ def replace(self, param_name, alt_message=None, single=False):
+ value = self.get_param(param_name, single)
+ if value:
+ return value
+ return alt_message or replace(param_name)
+
+ def generate_environments(self, module):
+ """
+ Generate a preload for the given module in all available environments
+ in the ``self.preload_env``. This will invoke the abstract
+ generate_module once for each available environment **and** an
+ empty environment to create a blank template.
+
+ :param module: module to generate for
+ """
+ print("\nGenerating Preloads for {}".format(module))
+ print("-" * 50)
+ print("... generating blank template")
+ self.current_module = module
+ self.current_module_env = {}
+ self.env_cache = {}
+ 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")
+ print(
+ "... generating preload for env ({}) to {}".format(
+ env.name, output_dir
+ )
+ )
+ self.env_cache = {}
+ self.current_module = module
+ self.current_module_env = 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())
+ if not os.path.exists(path):
+ os.makedirs(path, exist_ok=True)
+ return path
+
+ def generate_preload_env(self, module, blank_preload_dir):
+ """
+ Create a .env template suitable for completing and using for
+ preload generation from env files.
+ """
+ output_dir = os.path.join(blank_preload_dir, "preload_env")
+ output_file = os.path.join(output_dir, "{}.env".format(module.vnf_name))
+ if not os.path.exists(output_dir):
+ os.makedirs(output_dir, exist_ok=True)
+ with open(output_file, "w") as f:
+ yaml.dump(module.env_template, 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.reverse()
+ self.env_cache[param_name] = value
+ if value and single and isinstance(value, list):
+ return value.pop()
+ else:
+ return value
# limitations under the License.
#
# ============LICENSE_END============================================
-import importlib
-import inspect
-import json
import os
-import pkgutil
import shutil
from abc import ABC, abstractmethod
-from itertools import chain
-from typing import Set
+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.parametrizers import parametrize_heat_templates
from tests.structures import NeutronPortProcessor, Heat
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
+
+CHANGE = "CHANGEME"
# This is only used to fake out parametrizers
return heat_templates
-def get_json_template(template_dir, template_name):
- template_name = template_name + ".json"
- with open(os.path.join(template_dir, template_name)) as f:
- return json.loads(f.read())
-
-
-def remove(sequence, exclude, key=None):
- """
- Remove a copy of sequence that items occur in exclude.
-
- :param sequence: sequence of objects
- :param exclude: objects to excluded (must support ``in`` check)
- :param key: optional function to extract key from item in sequence
- :return: list of items not in the excluded
- """
- key_func = key if key else lambda x: x
- result = (s for s in sequence if key_func(s) not in exclude)
- return set(result) if isinstance(sequence, Set) else list(result)
-
-
-def get_or_create_template(template_dir, key, value, sequence, template_name):
- """
- Search a sequence of dicts where a given key matches value. If
- found, then it returns that item. If not, then it loads the
- template identified by template_name, adds it ot the sequence, and
- returns the template
- """
- for item in sequence:
- if item[key] == value:
- return item
- new_template = get_json_template(template_dir, template_name)
- sequence.append(new_template)
- return new_template
-
-
-def replace(param):
- """
- Optionally used by the preload generator to wrap items in the preload
- that need to be replaced by end users
- :param param: p
- """
- return "VALUE FOR: {}".format(param) if param else ""
-
-
-class AbstractPreloadGenerator(ABC):
- """
- All preload generators must inherit from this class and implement the
- abstract methods.
-
- Preload generators are automatically discovered at runtime via a plugin
- architecture. The system path is scanned looking for modules with the name
- preload_*, then all non-abstract classes that inherit from AbstractPreloadGenerator
- are registered as preload plugins
-
- Attributes:
- :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
- """
-
- def __init__(self, vnf, base_output_dir):
- self.vnf = vnf
- self.base_output_dir = base_output_dir
- os.makedirs(self.output_dir, exist_ok=True)
-
- @classmethod
- @abstractmethod
- def format_name(cls):
- """
- String name to identify the format (ex: VN-API, GR-API)
- """
- raise NotImplementedError()
-
- @classmethod
- @abstractmethod
- def output_sub_dir(cls):
- """
- String sub-directory name that will appear under ``base_output_dir``
- """
- raise NotImplementedError()
-
- @classmethod
- @abstractmethod
- def supports_output_passing(cls):
- """
- Some preload methods allow automatically mapping output parameters in the
- base module to the input parameter of other modules. This means these
- that the incremental modules do not need these base module outputs in their
- preloads.
-
- At this time, VNF-API does not support output parameter passing, but
- GR-API does.
-
- If this is true, then the generator will call Vnf#filter_output_params
- after the preload module for the base module has been created
- """
- raise NotImplementedError()
-
- @abstractmethod
- def generate_module(self, module):
- """
- Create the preloads and write them to ``self.output_dir``. This
- method is responsible for generating the content of the preload and
- writing the file to disk.
- """
- raise NotImplementedError()
-
- @property
- def output_dir(self):
- return os.path.join(self.base_output_dir, self.output_sub_dir())
-
- def generate(self):
- # handle the base module first
- print("\nGenerating {} preloads".format(self.format_name()))
- self.generate_module(self.vnf.base_module)
- print("... generated template for {}".format(self.vnf.base_module))
- if self.supports_output_passing():
- self.vnf.filter_base_outputs()
- for mod in self.vnf.incremental_modules:
- self.generate_module(mod)
- print("... generated for {}".format(mod))
-
-
class FilterBaseOutputs(ABC):
"""
Invoked to remove parameters in an object that appear in the base module.
mod.filter_output_params(self.base_output_params)
-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 env_path(heat_path):
"""
Create the path to the env file for the give heat path.
p for p in self.heat.parameters if p.startswith("availability_zone")
)
+ @property
+ def label(self):
+ """
+ Label for the VF module that will appear in the CSAR
+ """
+ return self.vnf_name
+
+ @property
+ def env_specs(self):
+ """Return available Environment Spec definitions"""
+ return Config().env_specs
+
+ @property
+ def env_template(self):
+ """
+ Returns a a template .env file that can be completed to enable
+ preload generation.
+ """
+ params = {}
+ params["vnf-name"] = CHANGE
+ params["vnf-type"] = CHANGE
+ params["vf-module-model-name"] = CHANGE
+ params["vf_module_name"] = CHANGE
+ for network in self.networks:
+ params[network.name_param] = CHANGE
+ for param in set(network.subnet_params):
+ params[param] = CHANGE
+ for vm in self.virtual_machine_types:
+ for name in set(vm.names):
+ params[name] = CHANGE
+ for ip in vm.floating_ips:
+ params[ip.param] = CHANGE
+ for ip in vm.fixed_ips:
+ params[ip.param] = CHANGE
+ excluded = get_preload_excluded_parameters(
+ self.template_file, persistent_only=True
+ )
+ for name, value in self.parameters.items():
+ if name in excluded:
+ continue
+ params[name] = value
+ return {"parameters": params}
+
@property
def preload_parameters(self):
"""
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 gen_class in get_generator_plugins():
+ for plugin_class in plugins:
+ if plugin_class.format_name() not in selected_formats:
+ continue
vnf = Vnf(heat_templates)
- generator = gen_class(vnf, preload_dir)
+ generator = plugin_class(vnf, preload_dir, preload_env)
generator.generate()
if vnf and vnf.uses_contrail:
print(
"\nWARNING: Heat violations detected. Preload templates may be\n"
"incomplete."
)
-
-
-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]
import json
import os
-from preload import (
- AbstractPreloadGenerator,
- get_or_create_template,
+from preload.generator import (
get_json_template,
- replace,
+ get_or_create_template,
+ AbstractPreloadGenerator,
)
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
)
-def add_fixed_ips(network_template, fixed_ips, uses_dhcp):
- 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:
- ipv4s["use-dhcp"] = "Y"
- ipv6s["use-dhcp"] = "Y"
- for ip in fixed_ips:
- target = ipv4s if ip.ip_version == 4 else ipv6s
- ips = target["network-ips"]["network-ip"]
- if ip.param not in ips:
- ips.append(replace(ip.param))
- target["ip-count"] += 1
-
-
-def add_floating_ips(network_template, floating_ips):
- for ip in floating_ips:
- key = "floating-ip-v4" if ip.ip_version == 4 else "floating-ip-v6"
- ips = network_template["floating-ips"][key]
- value = replace(ip.param)
- if value not in ips:
- ips.append(value)
-
-
class GrApiPreloadGenerator(AbstractPreloadGenerator):
@classmethod
def supports_output_passing(cls):
def output_sub_dir(cls):
return "grapi"
- def generate_module(self, vnf_module):
+ def generate_module(self, vnf_module, output_dir):
template = get_json_template(DATA_DIR, "preload_template")
self._populate(template, vnf_module)
vnf_name = vnf_module.vnf_name
- outfile = "{}/{}.json".format(self.output_dir, vnf_name)
+ outfile = "{}/{}.json".format(output_dir, vnf_name)
with open(outfile, "w") as f:
json.dump(template, f, indent=4)
+ def add_floating_ips(self, network_template, floating_ips):
+ for ip in 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)
+
+ def add_fixed_ips(self, network_template, fixed_ips, uses_dhcp):
+ 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:
+ ipv4s["use-dhcp"] = "Y"
+ ipv6s["use-dhcp"] = "Y"
+ for ip in fixed_ips:
+ 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))
+ target["ip-count"] += 1
+
def _populate(self, preload, vnf_module):
self._add_vnf_metadata(preload)
self._add_vms(preload, vnf_module)
self._add_parameters(preload, vnf_module)
self._add_vnf_networks(preload, vnf_module)
- @staticmethod
- def _add_vms(preload, vnf_module):
+ def _add_vms(self, preload, vnf_module):
vms = preload["input"]["preload-vf-module-topology-information"][
"vf-module-topology"
]["vf-module-assignments"]["vms"]["vm"]
vm_template = get_json_template(DATA_DIR, "vm")
vms.append(vm_template)
vm_template["vm-type"] = vm.vm_type
- vm_template["vm-names"]["vm-name"].extend(map(replace, vm.names))
+ for name in vm.names:
+ value = self.replace(name, single=True)
+ vm_template["vm-names"]["vm-name"].append(value)
vm_template["vm-count"] = vm.vm_count
vm_networks = vm_template["vm-networks"]["vm-network"]
for port in vm.ports:
role = port.network.network_role
network_template = get_or_create_network_template(role, vm_networks)
network_template["network-role"] = role
- add_fixed_ips(network_template, port.fixed_ips, port.uses_dhcp)
- add_floating_ips(network_template, port.floating_ips)
+ self.add_fixed_ips(network_template, port.fixed_ips, port.uses_dhcp)
+ self.add_floating_ips(network_template, port.floating_ips)
- @staticmethod
- def _add_availability_zones(preload, vnf_module):
+ def _add_availability_zones(self, preload, vnf_module):
zones = preload["input"]["preload-vf-module-topology-information"][
"vnf-resource-assignments"
]["availability-zones"]["availability-zone"]
- zones.extend(map(replace, vnf_module.availability_zones))
+ for zone in vnf_module.availability_zones:
+ value = self.replace(zone, single=True)
+ zones.append(value)
- @staticmethod
- def _add_parameters(preload, vnf_module):
+ def _add_parameters(self, preload, vnf_module):
params = [
- {"name": key, "value": value}
+ {"name": key, "value": self.replace(key, value)}
for key, value in vnf_module.preload_parameters.items()
]
preload["input"]["preload-vf-module-topology-information"][
"vf-module-topology"
]["vf-module-parameters"]["param"].extend(params)
- @staticmethod
- def _add_vnf_networks(preload, vnf_module):
+ 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": replace("network name of {}".format(network.name_param)),
+ "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": replace(subnet_param)})
+ subnet_data.append(
+ {"subnet-id": self.replace(subnet_param, single=True)}
+ )
networks.append(network_data)
- @staticmethod
- def _add_vnf_metadata(preload):
+ 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"] = replace("vnf_name")
- vnf_meta["vnf-type"] = replace("Concatenation of "
- "<Service Name>/<VF Instance Name> "
- "MUST MATCH SDC")
+ 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"] = replace("vf_module_name")
- module_meta["vf-module-type"] = replace("<vfModuleModelName> from CSAR or SDC")
+ 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"
+ )
import json
import os
-from preload import (
- AbstractPreloadGenerator,
+from preload.generator import (
get_json_template,
get_or_create_template,
- replace,
+ AbstractPreloadGenerator,
)
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(THIS_DIR, "vnfapi_data")
-def add_fixed_ips(network_template, port):
- for ip in port.fixed_ips:
- if ip.ip_version == 4:
- network_template["network-ips"].append({"ip-address": replace(ip.param)})
- network_template["ip-count"] += 1
- else:
- network_template["network-ips-v6"].append({"ip-address": replace(ip.param)})
- network_template["ip-count-ipv6"] += 1
-
-
-def add_floating_ips(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
- # and if multiple floating params exist, then come up with an
- # approach to pick just one
- for ip in network.floating_ips:
- key = "floating-ip" if ip.ip_version == 4 else "floating-ip-v6"
- network_template[key] = replace(ip.param)
-
-
def get_or_create_network_template(network_role, vm_networks):
"""
If the network role already exists in vm_networks, then
def output_sub_dir(cls):
return "vnfapi"
- def generate_module(self, vnf_module):
+ def generate_module(self, vnf_module, output_dir):
preload = get_json_template(DATA_DIR, "preload_template")
self._populate(preload, vnf_module)
- outfile = "{}/{}.json".format(self.output_dir, vnf_module.vnf_name)
+ outfile = "{}/{}.json".format(output_dir, vnf_module.vnf_name)
with open(outfile, "w") as f:
json.dump(preload, f, indent=4)
+ 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
+ # and if multiple floating params exist, then come up with an
+ # approach to pick just one
+ for ip in network.floating_ips:
+ 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:
+ if ip.ip_version == 4:
+ network_template["network-ips"].append(
+ {"ip-address": self.replace(ip.param, single=True)}
+ )
+ network_template["ip-count"] += 1
+ else:
+ network_template["network-ips-v6"].append(
+ {"ip-address": self.replace(ip.param, single=True)}
+ )
+ 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)
- @staticmethod
- def _add_availability_zones(preload, vnf_module):
+ 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": replace(zone)})
+ zones.append({"availability-zone": self.replace(zone, single=True)})
- @staticmethod
- def _add_vnf_networks(preload, vnf_module):
+ def _add_vnf_networks(self, preload, vnf_module):
networks = preload["input"]["vnf-topology-information"]["vnf-assignments"][
"vnf-networks"
]
for network in vnf_module.networks:
network_data = {
"network-role": network.network_role,
- "network-name": replace(
- "network name for {}".format(network.name_param)
+ "network-name": self.replace(
+ network.name_param,
+ "VALUE FOR: network name for {}".format(network.name_param),
),
}
for subnet in network.subnet_params:
network_data[key] = subnet
networks.append(network_data)
- @staticmethod
- def _add_vms(preload, vnf_module):
+ def _add_vms(self, preload, vnf_module):
vm_list = preload["input"]["vnf-topology-information"]["vnf-assignments"][
"vnf-vms"
]
vm_template = get_json_template(DATA_DIR, "vm")
vm_template["vm-type"] = vm.vm_type
vm_template["vm-count"] = vm.vm_count
- vm_template["vm-names"]["vm-name"].extend(map(replace, vm.names))
+ for name in vm.names:
+ value = self.replace(name, single=True)
+ vm_template["vm-names"]["vm-name"].append(value)
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"
- add_fixed_ips(network_template, port)
- add_floating_ips(network_template, port)
+ self.add_fixed_ips(network_template, port)
+ self.add_floating_ips(network_template, port)
- @staticmethod
- def _add_parameters(preload, vnf_module):
+ def _add_parameters(self, preload, vnf_module):
params = preload["input"]["vnf-topology-information"]["vnf-parameters"]
for key, value in vnf_module.preload_parameters.items():
- params.append({"vnf-parameter-name": key, "vnf-parameter-value": value})
+ params.append(
+ {
+ "vnf-parameter-name": key,
+ "vnf-parameter-value": self.replace(key, value),
+ }
+ )
import re
import time
-from preload import create_preloads
+from preload.model import create_preloads
+from config import get_generator_plugin_names
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",
+ action="append",
+ help=(
+ "Preload format to create (multiple allowed). If not provided "
+ "then all available formats will be created: {}"
+ ).format(", ".join(get_generator_plugin_names()))
+ )
+
def pytest_configure(config):
"""
import os
import re
+import zipfile
from collections import defaultdict
+from typing import Set
from boltons import funcutils
from tests import cached_yaml as yaml
if not os.path.exists(output_dir):
os.makedirs(output_dir, exist_ok=True)
return output_dir
+
+
+def first(seq, predicate, default=None):
+ """
+ Return the first item in sequence that satisfies the callable, predicate, or
+ returns the default if not found.
+
+ :param seq: iterable sequence of objects
+ :param predicate: callable that accepts one item from the sequence
+ :param default: value to return if not found (default is None)
+ :return: default value if no item satisfies the predicate
+ """
+ return next((i for i in seq if predicate(i)), default)
+
+
+def check(predicate, message):
+ """
+ Raise a RuntimeError with the provided message if predicate is False.
+
+ Example:
+ check(path.is_file(), "{} must be a file".format(path.as_posix()))
+
+ :param predicate: boolean condition
+ :param message: message
+ """
+ if not predicate:
+ raise RuntimeError(message)
+
+
+def unzip(zip_path, target_dir):
+ """
+ Extracts a Zip archive located at zip_path to target_dir (which will be
+ created if it already exists)
+
+ :param zip_path: path to valid zip file
+ :param target_dir: directory to unzip zip_path
+ """
+ check(zipfile.is_zipfile(zip_path), "{} is not a valid zipfile or does not exist".format(zip_path))
+ archive = zipfile.ZipFile(zip_path)
+ if not os.path.exists(target_dir):
+ os.makedirs(target_dir, exist_ok=True)
+ archive.extractall(path=target_dir)
+
+
+def remove(sequence, exclude, key=None):
+ """
+ Remove a copy of sequence that items occur in exclude.
+
+ :param sequence: sequence of objects
+ :param exclude: objects to excluded (must support ``in`` check)
+ :param key: optional function to extract key from item in sequence
+ :return: list of items not in the excluded
+ """
+ key_func = key if key else lambda x: x
+ result = (s for s in sequence if key_func(s) not in exclude)
+ return set(result) if isinstance(sequence, Set) else list(result)
or, during --self-test, the directory whos name matches
the current tests module name
"""
- if metafunc.config.getoption("template_dir") is None:
+ if metafunc.config.getoption("template_dir", None) is None:
return path.join(
path.dirname(metafunc.module.__file__),
"fixtures",
"""
extensions = [".yaml", ".yml", ".env"] if extensions is None else extensions
filenames_lists = []
- if metafunc.config.getoption("self_test"):
+ if metafunc.config.getoption("self_test", None):
filenames_lists.append(
list_template_dir(
metafunc, extensions, exclude_nested, template_type, ["pass"]
assert not invalid_parameters, "\n".join(invalid_parameters)
-def get_preload_excluded_parameters(yaml_file):
+def get_preload_excluded_parameters(yaml_file, persistent_only=False, env_spec=None):
"""
Returns set of all parameters that should not be included in the preload's
vnf parameters/tag-values section.
+
+ if persistent_only only parameters that are marked as persistent will
+ be excluded
"""
+ env_spec = env_spec or ENV_PARAMETER_SPEC
results = []
- for resource_type, specs in ENV_PARAMETER_SPEC.items():
+ for resource_type, specs in env_spec.items():
# apply to all resources if not in the format of an OpenStack resource
all_resources = "::" not in resource_type
for spec in specs:
+ if persistent_only and not spec.get("persistent"):
+ continue
results.extend(get_template_parameters(yaml_file, resource_type,
spec, all_resources))
return {item["param"] for item in results}
heat template-validate from the command line.
settings:
polling-freqency: 1000
- default-verbosity: Standard
NOTE: This script does require Python 3.6+
"""
-import appdirs
+
import os
+import traceback
+
import pytest
import version
-import yaml
import contextlib
import multiprocessing
import queue
import platform
import subprocess # nosec
-from collections import MutableMapping
-from configparser import ConfigParser
from multiprocessing import Queue
from pathlib import Path
from shutil import rmtree
NORMAL,
)
from tkinter.scrolledtext import ScrolledText
-from typing import Optional, List, Dict, TextIO, Callable, Iterator
+from typing import Optional, TextIO, Callable
-import preload
+from config import Config
VERSION = version.VERSION
PATH = os.path.dirname(os.path.realpath(__file__))
return
-class QueueWriter:
- """``stdout`` and ``stderr`` will be written to this queue by pytest, and
- pulled into the main GUI application"""
-
- def __init__(self, log_queue: queue.Queue):
- """Writes data to the provided queue.
-
- :param log_queue: the queue instance to write to.
- """
- self.queue = log_queue
-
- def write(self, data: str):
- """Writes ``data`` to the queue """
- self.queue.put(data)
-
- # noinspection PyMethodMayBeStatic
- def isatty(self) -> bool:
- """Always returns ``False``"""
- return False
-
- def flush(self):
- """No operation method to satisfy file-like behavior"""
- pass
-
-
def run_pytest(
template_dir: str,
log: TextIO,
result_queue: Queue,
categories: Optional[list],
- verbosity: str,
report_format: str,
halt_on_failure: bool,
template_source: str,
+ env_dir: str,
+ preload_format: list,
):
"""Runs pytest using the given ``profile`` in a background process. All
``stdout`` and ``stderr`` are redirected to ``log``. The result of the job
will collect and execute all tests that are
decorated with any of the passed categories, as
well as tests not decorated with a category.
- :param verbosity: Flag to be passed to pytest to control verbosity.
- Options are '' (empty string), '-v' (verbose),
- '-vv' (more verbose).
:param report_format: Determines the style of report written. Options are
csv, html, or excel
:param halt_on_failure: Determines if validation will halt when basic failures
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
"""
out_path = "{}/{}".format(PATH, OUT_DIR)
if os.path.exists(out_path):
args = [
"--ignore=app_tests",
"--capture=sys",
- verbosity,
"--template-directory={}".format(template_dir),
"--report-format={}".format(report_format),
"--template-source={}".format(template_source),
]
+ if env_dir:
+ args.append("--env-directory={}".format(env_dir))
if categories:
for category in categories:
args.extend(("--category", category))
if not halt_on_failure:
args.append("--continue-on-failure")
+ if preload_format:
+ args.append("--preload-format={}".format(preload_format))
pytest.main(args=args)
result_queue.put((True, None))
- except Exception as e:
- result_queue.put((False, e))
-
-
-class UserSettings(MutableMapping):
- FILE_NAME = "UserSettings.ini"
-
- def __init__(self, namespace, owner):
- user_config_dir = appdirs.AppDirs(namespace, owner).user_config_dir
- if not os.path.exists(user_config_dir):
- os.makedirs(user_config_dir, exist_ok=True)
- self._settings_path = os.path.join(user_config_dir, self.FILE_NAME)
- self._config = ConfigParser()
- self._config.read(self._settings_path)
-
- def __getitem__(self, k):
- return self._config["DEFAULT"][k]
-
- def __setitem__(self, k, v) -> None:
- self._config["DEFAULT"][k] = v
-
- def __delitem__(self, v) -> None:
- del self._config["DEFAULT"][v]
-
- def __len__(self) -> int:
- return len(self._config["DEFAULT"])
-
- def __iter__(self) -> Iterator:
- return iter(self._config["DEFAULT"])
-
- def save(self):
- with open(self._settings_path, "w") as f:
- self._config.write(f)
-
-
-class Config:
- """
- Configuration for the Validation GUI Application
-
- Attributes
- ----------
- ``log_queue`` Queue for the ``stdout`` and ``stderr` of
- the background job
- ``log_file`` File-like object (write only!) that writes to
- the ``log_queue``
- ``status_queue`` Job completion status of the background job is
- posted here as a tuple of (bool, Exception).
- The first parameter is True if the job completed
- successfully, and False otherwise. If the job
- failed, then an Exception will be provided as the
- second element.
- ``command_queue`` Used to send commands to the GUI. Currently only
- used to send shutdown commands in tests.
- """
-
- DEFAULT_FILENAME = "vvp-config.yaml"
- DEFAULT_POLLING_FREQUENCY = "1000"
-
- def __init__(self, config: dict = None):
- """Creates instance of application configuration.
-
- :param config: override default configuration if provided."""
- if config:
- self._config = config
- else:
- with open(self.DEFAULT_FILENAME, "r") as f:
- self._config = yaml.safe_load(f)
- self._user_settings = UserSettings(
- self._config["namespace"], self._config["owner"]
- )
- self._watched_variables = []
- self._validate()
- self._manager = multiprocessing.Manager()
- self.log_queue = self._manager.Queue()
- self.status_queue = self._manager.Queue()
- self.log_file = QueueWriter(self.log_queue)
- self.command_queue = self._manager.Queue()
-
- def watch(self, *variables):
- """Traces the variables and saves their settings for the user. The
- last settings will be used where available"""
- self._watched_variables = variables
- for var in self._watched_variables:
- var.trace_add("write", self.save_settings)
-
- # noinspection PyProtectedMember,PyUnusedLocal
- def save_settings(self, *args):
- """Save the value of all watched variables to user settings"""
- for var in self._watched_variables:
- self._user_settings[var._name] = str(var.get())
- self._user_settings.save()
-
- @property
- def app_name(self) -> str:
- """Name of the application (displayed in title bar)"""
- app_name = self._config["ui"].get("app-name", "VNF Validation Tool")
- return "{} - {}".format(app_name, VERSION)
-
- @property
- def category_names(self) -> List[str]:
- """List of validation profile names for display in the UI"""
- return [category["name"] for category in self._config["categories"]]
-
- @property
- def polling_frequency(self) -> int:
- """Returns the frequency (in ms) the UI polls the queue communicating
- with any background job"""
- return int(
- self._config["settings"].get(
- "polling-frequency", self.DEFAULT_POLLING_FREQUENCY
- )
- )
-
- @property
- def disclaimer_text(self) -> str:
- return self._config["ui"].get("disclaimer-text", "")
-
- @property
- def requirement_link_text(self) -> str:
- return self._config["ui"].get("requirement-link-text", "")
-
- @property
- def requirement_link_url(self) -> str:
- path = self._config["ui"].get("requirement-link-url", "")
- if not path.startswith("http"):
- path = "file://{}".format(os.path.join(PATH, path))
- return path
-
- @property
- def terms(self) -> dict:
- return self._config.get("terms", {})
-
- @property
- def terms_link_url(self) -> Optional[str]:
- return self.terms.get("path")
-
- @property
- def terms_link_text(self):
- return self.terms.get("popup-link-text")
-
- @property
- def terms_version(self) -> Optional[str]:
- return self.terms.get("version")
-
- @property
- def terms_popup_title(self) -> Optional[str]:
- return self.terms.get("popup-title")
-
- @property
- def terms_popup_message(self) -> Optional[str]:
- return self.terms.get("popup-msg-text")
-
- @property
- def are_terms_accepted(self) -> bool:
- version = "terms-{}".format(self.terms_version)
- return self._user_settings.get(version, "False") == "True"
-
- def set_terms_accepted(self):
- version = "terms-{}".format(self.terms_version)
- self._user_settings[version] = "True"
- self._user_settings.save()
-
- def default_verbosity(self, levels: Dict[str, str]) -> str:
- requested_level = self._user_settings.get("verbosity") or self._config[
- "settings"
- ].get("default-verbosity", "Standard")
- keys = [key for key in levels]
- for key in levels:
- if key.lower().startswith(requested_level.lower()):
- return key
- raise RuntimeError(
- "Invalid default-verbosity level {}. Valid "
- "values are {}".format(requested_level, ", ".join(keys))
- )
-
- def get_description(self, category_name: str) -> str:
- """Returns the description associated with the category name"""
- return self._get_category(category_name)["description"]
-
- def get_category(self, category_name: str) -> str:
- """Returns the category associated with the category name"""
- return self._get_category(category_name).get("category", "")
-
- def get_category_value(self, category_name: str) -> str:
- """Returns the saved value for a category name"""
- return self._user_settings.get(category_name, 0)
-
- def _get_category(self, category_name: str) -> Dict[str, str]:
- """Returns the profile definition"""
- for category in self._config["categories"]:
- if category["name"] == category_name:
- return category
- raise RuntimeError(
- "Unexpected error: No category found in vvp-config.yaml "
- "with a name of " + category_name
- )
-
- @property
- def default_report_format(self):
- return self._user_settings.get("report_format", "HTML")
-
- @property
- def report_formats(self):
- return ["CSV", "Excel", "HTML"]
-
- @property
- def preload_formats(self):
- excluded = self._config.get("excluded-preloads", [])
- formats = (cls.format_name() for cls in preload.get_generator_plugins())
- return [f for f in formats if f not in excluded]
-
- @property
- def default_preload_format(self):
- default = self._user_settings.get("preload_format")
- if default and default in self.preload_formats:
- return default
- else:
- return self.preload_formats[0]
-
- @staticmethod
- def get_subdir_for_preload(preload_format):
- for gen in preload.get_generator_plugins():
- if gen.format_name() == preload_format:
- return gen.output_sub_dir()
- return ""
-
- @property
- def default_input_format(self):
- requested_default = self._user_settings.get("input_format") or self._config[
- "settings"
- ].get("default-input-format")
- if requested_default in self.input_formats:
- return requested_default
- else:
- return self.input_formats[0]
-
- @property
- def input_formats(self):
- return ["Directory (Uncompressed)", "ZIP File"]
-
- @property
- def default_halt_on_failure(self):
- setting = self._user_settings.get("halt_on_failure", "True")
- return setting.lower() == "true"
-
- def _validate(self):
- """Ensures the config file is properly formatted"""
- categories = self._config["categories"]
-
- # All profiles have required keys
- expected_keys = {"name", "description"}
- for category in categories:
- actual_keys = set(category.keys())
- missing_keys = expected_keys.difference(actual_keys)
- if missing_keys:
- raise RuntimeError(
- "Error in vvp-config.yaml file: "
- "Required field missing in category. "
- "Missing: {} "
- "Categories: {}".format(",".join(missing_keys), category)
- )
-
-
-def validate():
- return True
+ except Exception:
+ result_queue.put((False, traceback.format_exc()))
class Dialog(Toplevel):
# noinspection PyUnusedLocal
def ok(self, event=None):
- if not validate():
- self.initial_focus.focus_set() # put focus back
- return
self.withdraw()
self.update_idletasks()
self.apply()
class ValidatorApp:
- VERBOSITY_LEVELS = {"Less": "", "Standard (-v)": "-v", "More (-vv)": "-vv"}
-
def __init__(self, config: Config = None):
"""Constructs the GUI element of the Validation Tool"""
self.task = None
)
actions = Frame(control_panel)
control_panel.add(actions)
- control_panel.paneconfigure(actions, minsize=250)
+ control_panel.paneconfigure(actions, minsize=350)
if self.config.disclaimer_text or self.config.requirement_link_text:
self.footer = self.create_footer(parent_frame)
settings_frame = LabelFrame(actions, text="Settings")
settings_row = 1
settings_frame.grid(row=3, column=1, columnspan=3, pady=10, sticky="we")
- verbosity_label = Label(settings_frame, text="Verbosity:")
- verbosity_label.grid(row=settings_row, column=1, sticky=W)
- self.verbosity = StringVar(self._root, name="verbosity")
- self.verbosity.set(self.config.default_verbosity(self.VERBOSITY_LEVELS))
- verbosity_menu = OptionMenu(
- settings_frame, self.verbosity, *tuple(self.VERBOSITY_LEVELS.keys())
- )
- verbosity_menu.config(width=25)
- verbosity_menu.grid(row=settings_row, column=2, columnspan=3, sticky=E, pady=5)
- settings_row += 1
if self.config.preload_formats:
preload_format_label = Label(settings_frame, text="Preload Template:")
self.halt_on_failure = BooleanVar(self._root, name="halt_on_failure")
self.halt_on_failure.set(self.config.default_halt_on_failure)
- halt_on_failure_label = Label(settings_frame, text="Halt on Basic Failures:")
- halt_on_failure_label.grid(row=settings_row, column=1, sticky=E, pady=5)
+ halt_on_failure_label = Label(
+ settings_frame, text="Halt on Basic Failures:", anchor=W, justify=LEFT
+ )
+ halt_on_failure_label.grid(row=settings_row, column=1, sticky=W, pady=5)
halt_checkbox = Checkbutton(
settings_frame, offvalue=False, onvalue=True, variable=self.halt_on_failure
)
halt_checkbox.grid(row=settings_row, column=2, columnspan=2, sticky=W, pady=5)
+ settings_row += 1
+
+ self.create_preloads = BooleanVar(self._root, name="create_preloads")
+ self.create_preloads.set(0)
+ create_preloads_label = Label(
+ settings_frame,
+ text="Create Preload from Env Files:",
+ anchor=W,
+ justify=LEFT,
+ )
+ create_preloads_label.grid(row=settings_row, column=1, sticky=W, pady=5)
+ create_preloads_checkbox = Checkbutton(
+ settings_frame,
+ offvalue=False,
+ onvalue=True,
+ variable=self.create_preloads,
+ command=self.set_env_dir_state,
+ )
+ create_preloads_checkbox.grid(
+ row=settings_row, column=2, columnspan=2, sticky=W, pady=5
+ )
directory_label = Label(actions, text="Template Location:")
directory_label.grid(row=4, column=1, pady=5, sticky=W)
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")
+ self.env_dir_entry = Entry(
+ actions, width=40, textvariable=self.env_dir, state=DISABLED
+ )
+ 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)
+
validate_button = Button(
- actions, text="Validate Templates", command=self.validate
+ actions, text="Process Templates", command=self.validate
)
- validate_button.grid(row=5, column=1, columnspan=2, pady=5)
+ validate_button.grid(row=6, column=1, columnspan=2, pady=5)
self.result_panel = Frame(actions)
# We'll add these labels now, and then make them visible when the run completes
self.result_label.bind("<Button-1>", self.open_report)
self.preload_label = Label(
- self.result_panel, text="View Preloads", fg="blue", cursor="hand2"
+ self.result_panel, text="View Preload Templates", fg="blue", cursor="hand2"
)
self.underline(self.preload_label)
self.preload_label.bind("<Button-1>", self.open_preloads)
- self.result_panel.grid(row=6, column=1, columnspan=2)
+ self.result_panel.grid(row=7, column=1, columnspan=2)
control_panel.pack(fill=BOTH, expand=1)
main_window.add(control_panel)
self.config.watch(
*self.categories,
- self.verbosity,
self.input_format,
self.report_format,
self.halt_on_failure,
footer.pack(fill=BOTH, expand=True)
return footer
+ def set_env_dir_state(self):
+ state = NORMAL if self.create_preloads.get() else DISABLED
+ self.env_dir_entry.config(state=state)
+
def ask_template_source(self):
if self.input_format.get() == "ZIP File":
template_source = filedialog.askopenfilename(
template_source = filedialog.askdirectory()
self.template_source.set(template_source)
+ def ask_env_dir_source(self):
+ self.env_dir.set(filedialog.askdirectory())
+
def validate(self):
"""Run the pytest validations in a background process"""
if not self.delete_prior_report():
self.config.log_file,
self.config.status_queue,
self.categories_list(),
- self.VERBOSITY_LEVELS[self.verbosity.get()],
self.report_format.get().lower(),
self.halt_on_failure.get(),
self.template_source.get(),
+ self.env_dir.get(),
+ self.preload_format.get(),
),
)
self.task.daemon = True
# noinspection PyUnusedLocal
def open_report(self, event):
"""Open the report in the user's default browser"""
- webbrowser.open_new("file://{}".format(self.report_file_path))
+ path = Path(self.report_file_path).absolute().resolve().as_uri()
+ webbrowser.open_new(path)
def open_preloads(self, event):
"""Open the report in the user's default browser"""
pyinstaller
mock
openstack-heat
+cached-property>=1.5,<1.6
Code Review / vvp / validation-scripts.git / commitdiff