Skip to content
Snippets Groups Projects
Commit 1e645f9c authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

ECOC'22 functional test: 

- moved old files to separate folder
- updated scripts
- updated descriptor files
- updated gitignore
- updated unitary test code
parent 9d97706f
No related branches found
No related tags found
2 merge requests!54Release 2.0.0,!48Fixes for OFC'22 and ECOC'22 tests
Hide whitespace changes
Inline Side-by-side
Showing
with 259 additions and 127 deletions
src/tests/ecoc22/descriptors_emulated.json
+ 38
− 19
View file @ 1e645f9c
{
"contexts": [
{
"context_id": {"context_uuid": {"uuid": "admin"}},
"topology_ids": [], "service_ids": []
}
{"context_id": {"context_uuid": {"uuid": "admin"}}}
],
"topologies": [
{
"topology_id": {"context_id": {"context_uuid": {"uuid": "admin"}}, "topology_uuid": {"uuid": "admin"}},
"device_ids": [], "link_ids": []
}
{"topology_id": {"context_id": {"context_uuid": {"uuid": "admin"}}, "topology_uuid": {"uuid": "admin"}}}
],
"devices": [
{
......@@ -17,7 +11,11 @@
"device_endpoints": [], "device_operational_status": 1, "device_config": {"config_rules": [
{"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}},
{"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}},
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"eth1\"}, {\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"eth2\"}, {\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"int\"}]}"}}
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [
{"sample_types": [], "type": "copper", "uuid": "eth1"},
{"sample_types": [], "type": "copper", "uuid": "eth2"},
{"sample_types": [], "type": "copper", "uuid": "int"}
]}}}
]}
},
{
......@@ -25,39 +23,55 @@
"device_endpoints": [], "device_operational_status": 1, "device_config": {"config_rules": [
{"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}},
{"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}},
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"eth1\"}, {\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"eth2\"}, {\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"int\"}]}"}}
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [
{"sample_types": [], "type": "copper", "uuid": "eth1"},
{"sample_types": [], "type": "copper", "uuid": "eth2"},
{"sample_types": [], "type": "copper", "uuid": "int"}
]}}}
]}
},
{
"device_id": {"device_uuid": {"uuid": "CS1-GW1"}}, "device_type": "packet-router", "device_drivers": [1],
"device_id": {"device_uuid": {"uuid": "CS1-GW1"}}, "device_type": "emu-packet-router", "device_drivers": [1],
"device_endpoints": [], "device_operational_status": 1, "device_config": {"config_rules": [
{"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}},
{"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}},
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"10/1\"}, {\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"1/1\"}]}"}}
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [
{"sample_types": [101, 102, 201, 202], "type": "copper", "uuid": "10/1"},
{"sample_types": [101, 102, 201, 202], "type": "copper", "uuid": "1/1"}
]}}}
]}
},
{
"device_id": {"device_uuid": {"uuid": "CS1-GW2"}}, "device_type": "packet-router", "device_drivers": [1],
"device_id": {"device_uuid": {"uuid": "CS1-GW2"}}, "device_type": "emu-packet-router", "device_drivers": [1],
"device_endpoints": [], "device_operational_status": 1, "device_config": {"config_rules": [
{"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}},
{"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}},
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"10/1\"}, {\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"1/1\"}]}"}}
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [
{"sample_types": [101, 102, 201, 202], "type": "copper", "uuid": "10/1"},
{"sample_types": [101, 102, 201, 202], "type": "copper", "uuid": "1/1"}
]}}}
]}
},
{
"device_id": {"device_uuid": {"uuid": "CS2-GW1"}}, "device_type": "packet-router", "device_drivers": [1],
"device_id": {"device_uuid": {"uuid": "CS2-GW1"}}, "device_type": "emu-packet-router", "device_drivers": [1],
"device_endpoints": [], "device_operational_status": 1, "device_config": {"config_rules": [
{"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}},
{"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}},
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"10/1\"}, {\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"1/1\"}]}"}}
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [
{"sample_types": [101, 102, 201, 202], "type": "copper", "uuid": "10/1"},
{"sample_types": [101, 102, 201, 202], "type": "copper", "uuid": "1/1"}
]}}}
]}
},
{
"device_id": {"device_uuid": {"uuid": "CS2-GW2"}}, "device_type": "packet-router", "device_drivers": [1],
"device_id": {"device_uuid": {"uuid": "CS2-GW2"}}, "device_type": "emu-packet-router", "device_drivers": [1],
"device_endpoints": [], "device_operational_status": 1, "device_config": {"config_rules": [
{"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}},
{"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}},
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"10/1\"}, {\"sample_types\": [], \"type\": \"copper\", \"uuid\": \"1/1\"}]}"}}
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [
{"sample_types": [101, 102, 201, 202], "type": "copper", "uuid": "10/1"},
{"sample_types": [101, 102, 201, 202], "type": "copper", "uuid": "1/1"}
]}}}
]}
},
{
......@@ -65,7 +79,12 @@
"device_endpoints": [], "device_operational_status": 1, "device_config": {"config_rules": [
{"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": "127.0.0.1"}},
{"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": "0"}},
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": "{\"endpoints\": [{\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"aade6001-f00b-5e2f-a357-6a0a9d3de870\"}, {\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"eb287d83-f05e-53ec-ab5a-adf6bd2b5418\"}, {\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"0ef74f99-1acc-57bd-ab9d-4b958b06c513\"}, {\"sample_types\": [], \"type\": \"optical\", \"uuid\": \"50296d99-58cc-5ce7-82f5-fc8ee4eec2ec\"}]}"}}
{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"endpoints": [
{"sample_types": [], "type": "optical", "uuid": "aade6001-f00b-5e2f-a357-6a0a9d3de870"},
{"sample_types": [], "type": "optical", "uuid": "eb287d83-f05e-53ec-ab5a-adf6bd2b5418"},
{"sample_types": [], "type": "optical", "uuid": "0ef74f99-1acc-57bd-ab9d-4b958b06c513"},
{"sample_types": [], "type": "optical", "uuid": "50296d99-58cc-5ce7-82f5-fc8ee4eec2ec"}
]}}}
]}
}
],
......
src/tests/ecoc22/redeploy.sh
+ 15
− 1
View file @ 1e645f9c
#!/bin/bash
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of 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.
source ecoc22/deploy_specs.sh
./deploy.sh
./deploy/all.sh
source tfs_runtime_env_vars.sh
src/tests/ecoc22/run_test_01_bootstrap.sh
+ 1
− 1
View file @ 1e645f9c
......@@ -14,4 +14,4 @@
# limitations under the License.
source tfs_runtime_env_vars.sh
pytest --verbose src/tests/ecoc22/tests/test_functional_bootstrap.py
pytest --verbose --log-level=INFO src/tests/ecoc22/tests/test_functional_bootstrap.py
src/tests/ecoc22/run_test_02_create_service.sh
+ 1
− 1
View file @ 1e645f9c
......@@ -14,4 +14,4 @@
# limitations under the License.
source tfs_runtime_env_vars.sh
pytest --verbose src/tests/ecoc22/tests/test_functional_create_service.py
pytest --verbose --log-level=INFO src/tests/ecoc22/tests/test_functional_create_service.py
src/tests/ecoc22/run_test_03_delete_service.sh
+ 1
− 1
View file @ 1e645f9c
......@@ -14,4 +14,4 @@
# limitations under the License.
source tfs_runtime_env_vars.sh
pytest --verbose src/tests/ecoc22/tests/test_functional_delete_service.py
pytest --verbose --log-level=INFO src/tests/ecoc22/tests/test_functional_delete_service.py
src/tests/ecoc22/run_test_04_cleanup.sh
+ 1
− 1
View file @ 1e645f9c
......@@ -14,4 +14,4 @@
# limitations under the License.
source tfs_runtime_env_vars.sh
pytest --verbose src/tests/ecoc22/tests/test_functional_cleanup.py
pytest --verbose --log-level=INFO src/tests/ecoc22/tests/test_functional_cleanup.py
src/tests/ecoc22/run_tests_and_coverage.sh → src/tests/ecoc22/run_tests.sh
+ 21
− 0
View file @ 1e645f9c
......@@ -13,32 +13,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
PROJECTDIR=`pwd`
RCFILE=$PROJECTDIR/coverage/.coveragerc
COVERAGEFILE=$PROJECTDIR/coverage/.coverage
# Configure the correct folder on the .coveragerc file
cat $PROJECTDIR/coverage/.coveragerc.template | sed s+~/teraflow/controller+$PROJECTDIR+g > $RCFILE
# Destroy old coverage file
rm -f $COVERAGEFILE
# Run functional tests
source tfs_runtime_env_vars.sh
# Force a flush of Context database
kubectl --namespace $TFS_K8S_NAMESPACE exec -it deployment/contextservice --container redis -- redis-cli FLUSHALL
# Run functional tests and analyze code coverage at the same time
coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
src/tests/ecoc22/tests/test_functional_bootstrap.py
coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
src/tests/ecoc22/tests/test_functional_create_service.py
coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
src/tests/ecoc22/tests/test_functional_delete_service.py
coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
src/tests/ecoc22/tests/test_functional_cleanup.py
pytest --verbose --log-level=INFO src/tests/ecoc22/tests/test_functional_bootstrap.py
pytest --verbose --log-level=INFO src/tests/ecoc22/tests/test_functional_create_service.py
pytest --verbose --log-level=INFO src/tests/ecoc22/tests/test_functional_delete_service.py
pytest --verbose --log-level=INFO src/tests/ecoc22/tests/test_functional_cleanup.py
src/tests/ecoc22/tests/.gitignore
+ 1
− 0
View file @ 1e645f9c
# Add here your files containing confidential testbed details such as IP addresses, ports, usernames, passwords, etc.
Credentials.py
src/tests/ecoc22/tests/Credentials.py deleted 100644 → 0
+ 0
− 0
View file @ 9d97706f
src/tests/ecoc22/tests/Fixtures.py
+ 7
− 5
View file @ 1e645f9c
......@@ -12,15 +12,17 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
import pytest, logging
from common.Settings import get_setting
from compute.tests.mock_osm.MockOSM import MockOSM
#from .Objects_BigNet import WIM_MAPPING, WIM_PASSWORD, WIM_USERNAME
from .Objects_DC_CSGW_TN import WIM_MAPPING, WIM_PASSWORD, WIM_USERNAME
#from .Objects_DC_CSGW_TN_OLS import WIM_MAPPING, WIM_PASSWORD, WIM_USERNAME
from tests.tools.mock_osm.Constants import WIM_PASSWORD, WIM_USERNAME
from tests.tools.mock_osm.MockOSM import MockOSM
from .Objects import WIM_MAPPING
LOGGER = logging.getLogger(__name__)
@pytest.fixture(scope='session')
def osm_wim():
wim_url = 'http://{:s}:{:s}'.format(
get_setting('COMPUTESERVICE_SERVICE_HOST'), str(get_setting('COMPUTESERVICE_SERVICE_PORT_HTTP')))
LOGGER.info('WIM_MAPPING = {:s}'.format(str(WIM_MAPPING)))
return MockOSM(wim_url, WIM_MAPPING, WIM_USERNAME, WIM_PASSWORD)
src/tests/ecoc22/tests/Objects.py 0 → 100644
+ 51
− 0
View file @ 1e645f9c
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of 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.
from common.tools.object_factory.Device import json_device_id
from common.tools.object_factory.EndPoint import json_endpoint_id
from tests.tools.mock_osm.Tools import connection_point, wim_mapping
# ----- WIM Service Settings -------------------------------------------------------------------------------------------
# PRI = primary // BKP = backup
SITE_ID_DC1 = 'DC1'
DEV_ID_DC1 = json_device_id('DC1-GW')
EP_ID_DC1_PRI = json_endpoint_id(DEV_ID_DC1, 'eth1')
EP_ID_DC1_BKP = json_endpoint_id(DEV_ID_DC1, 'eth2')
DEV_ID_CS1GW1 = json_device_id('CS1-GW1')
DEV_ID_CS1GW2 = json_device_id('CS1-GW2')
SITE_ID_DC2 = 'DC2'
DEV_ID_DC2 = json_device_id('DC2-GW')
EP_ID_DC2_PRI = json_endpoint_id(DEV_ID_DC2, 'eth1')
EP_ID_DC2_BKP = json_endpoint_id(DEV_ID_DC2, 'eth2')
DEV_ID_CS2GW1 = json_device_id('CS2-GW1')
DEV_ID_CS2GW2 = json_device_id('CS2-GW2')
WIM_SEP_DC1_PRI, WIM_MAP_DC1_PRI = wim_mapping(SITE_ID_DC1, EP_ID_DC1_PRI, DEV_ID_CS1GW1, priority=10, redundant=['DC1:DC1-GW:eth2'])
WIM_SEP_DC1_BKP, WIM_MAP_DC1_BKP = wim_mapping(SITE_ID_DC1, EP_ID_DC1_BKP, DEV_ID_CS1GW2, priority=20, redundant=['DC1:DC1-GW:eth1'])
WIM_SEP_DC2_PRI, WIM_MAP_DC2_PRI = wim_mapping(SITE_ID_DC2, EP_ID_DC2_PRI, DEV_ID_CS2GW1, priority=10, redundant=['DC2:DC2-GW:eth2'])
WIM_SEP_DC2_BKP, WIM_MAP_DC2_BKP = wim_mapping(SITE_ID_DC2, EP_ID_DC2_BKP, DEV_ID_CS2GW2, priority=20, redundant=['DC2:DC2-GW:eth1'])
WIM_MAPPING = [
WIM_MAP_DC1_PRI, WIM_MAP_DC1_BKP,
WIM_MAP_DC2_PRI, WIM_MAP_DC2_BKP,
]
WIM_SRV_VLAN_ID = 300
WIM_SERVICE_TYPE = 'ELAN'
WIM_SERVICE_CONNECTION_POINTS = [
connection_point(WIM_SEP_DC1_PRI, 'dot1q', WIM_SRV_VLAN_ID),
connection_point(WIM_SEP_DC2_PRI, 'dot1q', WIM_SRV_VLAN_ID),
]
src/tests/ecoc22/tests/test_functional_bootstrap.py
+ 32
− 48
View file @ 1e645f9c
......@@ -13,21 +13,24 @@
# limitations under the License.
import logging
from common.proto.context_pb2 import Context, ContextId, Device, DeviceId, Empty, Link, LinkId, Topology, TopologyId
from common.Constants import DEFAULT_CONTEXT_NAME
from common.proto.context_pb2 import ContextId, Empty
from common.tests.LoadScenario import load_scenario_from_descriptor
from common.tools.object_factory.Context import json_context_id
from context.client.ContextClient import ContextClient
from device.client.DeviceClient import DeviceClient
from tests.Fixtures import context_client, device_client
#from .Objects_BigNet import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES
#from .Objects_DC_CSGW_TN import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, OBJECTS_PER_TOPOLOGY
#from .Objects_DC_CSGW_TN_OLS import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, OBJECTS_PER_TOPOLOGY
from .Objects_DC_CSGW_OLS import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, OBJECTS_PER_TOPOLOGY
from tests.Fixtures import context_client, device_client # pylint: disable=unused-import
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
DESCRIPTOR_FILE = 'ecoc22/descriptors_emulated.json'
ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME))
def test_scenario_empty(context_client : ContextClient): # pylint: disable=redefined-outer-name
def test_scenario_bootstrap(
context_client : ContextClient, # pylint: disable=redefined-outer-name
device_client : DeviceClient, # pylint: disable=redefined-outer-name
) -> None:
# ----- List entities - Ensure database is empty -------------------------------------------------------------------
response = context_client.ListContexts(Empty())
assert len(response.contexts) == 0
......@@ -39,53 +42,34 @@ def test_scenario_empty(context_client : ContextClient): # pylint: disable=rede
assert len(response.links) == 0
def test_prepare_environment(
context_client : ContextClient, # pylint: disable=redefined-outer-name
device_client : DeviceClient): # pylint: disable=redefined-outer-name
for context in CONTEXTS : context_client.SetContext (Context (**context ))
for topology in TOPOLOGIES: context_client.SetTopology(Topology(**topology))
for device in DEVICES : device_client .AddDevice (Device (**device ))
for topology_id, device_ids, _ in OBJECTS_PER_TOPOLOGY:
topology = Topology()
topology.CopyFrom(context_client.GetTopology(TopologyId(**topology_id)))
device_ids_in_topology = {device_id.device_uuid.uuid for device_id in topology.device_ids}
func_device_id_not_added = lambda device_id: device_id['device_uuid']['uuid'] not in device_ids_in_topology
func_device_id_json_to_grpc = lambda device_id: DeviceId(**device_id)
device_ids_to_add = list(map(func_device_id_json_to_grpc, filter(func_device_id_not_added, device_ids)))
topology.device_ids.extend(device_ids_to_add)
# ----- Load Scenario ----------------------------------------------------------------------------------------------
descriptor_loader = load_scenario_from_descriptor(
DESCRIPTOR_FILE, context_client, device_client, None, None)
context_client.SetTopology(topology)
for link in LINKS : context_client.SetLink (Link (**link ))
for topology_id, _, link_ids in OBJECTS_PER_TOPOLOGY:
topology = Topology()
topology.CopyFrom(context_client.GetTopology(TopologyId(**topology_id)))
link_ids_in_topology = {link_id.link_uuid.uuid for link_id in topology.link_ids}
func_link_id_not_added = lambda link_id: link_id['link_uuid']['uuid'] not in link_ids_in_topology
func_link_id_json_to_grpc = lambda link_id: LinkId(**link_id)
link_ids_to_add = list(map(func_link_id_json_to_grpc, filter(func_link_id_not_added, link_ids)))
topology.link_ids.extend(link_ids_to_add)
context_client.SetTopology(topology)
def test_scenario_ready(context_client : ContextClient): # pylint: disable=redefined-outer-name
# ----- List entities - Ensure scenario is ready -------------------------------------------------------------------
response = context_client.ListContexts(Empty())
assert len(response.contexts) == len(CONTEXTS)
assert len(response.contexts) == descriptor_loader.num_contexts
response = context_client.ListTopologies(ContextId(**CONTEXT_ID))
assert len(response.topologies) == len(TOPOLOGIES)
for context_uuid, num_topologies in descriptor_loader.num_topologies.items():
response = context_client.ListTopologies(ContextId(**json_context_id(context_uuid)))
assert len(response.topologies) == num_topologies
response = context_client.ListDevices(Empty())
assert len(response.devices) == len(DEVICES)
assert len(response.devices) == descriptor_loader.num_devices
response = context_client.ListLinks(Empty())
assert len(response.links) == len(LINKS)
assert len(response.links) == descriptor_loader.num_links
for context_uuid, _ in descriptor_loader.num_services.items():
response = context_client.ListServices(ContextId(**json_context_id(context_uuid)))
assert len(response.services) == 0
for context_uuid, _ in descriptor_loader.num_slices.items():
response = context_client.ListSlices(ContextId(**json_context_id(context_uuid)))
assert len(response.slices) == 0
response = context_client.ListServices(ContextId(**CONTEXT_ID))
assert len(response.services) == 0
# This scenario assumes no services are created beforehand
response = context_client.GetContext(ADMIN_CONTEXT_ID)
assert len(response.service_ids) == 0
assert len(response.slice_ids) == 0
src/tests/ecoc22/tests/test_functional_cleanup.py
+ 44
− 21
View file @ 1e645f9c
......@@ -13,49 +13,72 @@
# limitations under the License.
import logging
from common.Constants import DEFAULT_CONTEXT_NAME
from common.proto.context_pb2 import ContextId, DeviceId, Empty, LinkId, TopologyId
from common.tools.descriptor.Loader import DescriptorLoader
from common.tools.object_factory.Context import json_context_id
from context.client.ContextClient import ContextClient
from device.client.DeviceClient import DeviceClient
from tests.Fixtures import context_client, device_client
#from .Objects_BigNet import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES
#from .Objects_DC_CSGW_TN import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES
#from .Objects_DC_CSGW_TN_OLS import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES
from .Objects_DC_CSGW_OLS import CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES
from tests.Fixtures import context_client, device_client # pylint: disable=unused-import
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
DESCRIPTOR_FILE = 'ecoc22/descriptors_emulated.json'
ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME))
def test_services_removed(context_client : ContextClient): # pylint: disable=redefined-outer-name
def test_services_removed(
context_client : ContextClient, # pylint: disable=redefined-outer-name
device_client : DeviceClient, # pylint: disable=redefined-outer-name
) -> None:
# ----- List entities - Ensure service is removed ------------------------------------------------------------------
with open(DESCRIPTOR_FILE, 'r', encoding='UTF-8') as f:
descriptors = f.read()
descriptor_loader = DescriptorLoader(descriptors)
response = context_client.ListContexts(Empty())
assert len(response.contexts) == len(CONTEXTS)
assert len(response.contexts) == descriptor_loader.num_contexts
response = context_client.ListTopologies(ContextId(**CONTEXT_ID))
assert len(response.topologies) == len(TOPOLOGIES)
for context_uuid, num_topologies in descriptor_loader.num_topologies.items():
response = context_client.ListTopologies(ContextId(**json_context_id(context_uuid)))
assert len(response.topologies) == num_topologies
response = context_client.ListDevices(Empty())
assert len(response.devices) == len(DEVICES)
assert len(response.devices) == descriptor_loader.num_devices
response = context_client.ListLinks(Empty())
assert len(response.links) == len(LINKS)
assert len(response.links) == descriptor_loader.num_links
for context_uuid, _ in descriptor_loader.num_services.items():
response = context_client.ListServices(ContextId(**json_context_id(context_uuid)))
assert len(response.services) == 0
for context_uuid, _ in descriptor_loader.num_slices.items():
response = context_client.ListSlices(ContextId(**json_context_id(context_uuid)))
assert len(response.slices) == 0
# This scenario assumes no services are created beforehand
response = context_client.GetContext(ADMIN_CONTEXT_ID)
assert len(response.service_ids) == 0
assert len(response.slice_ids) == 0
response = context_client.ListServices(ContextId(**CONTEXT_ID))
assert len(response.services) == 0
# ----- Delete Links, Devices, Topologies, Contexts ----------------------------------------------------------------
for link in descriptor_loader.links:
context_client.RemoveLink(LinkId(**link['link_id']))
for device in descriptor_loader.devices:
device_client .DeleteDevice(DeviceId(**device['device_id']))
def test_scenario_cleanup(
context_client : ContextClient, device_client : DeviceClient): # pylint: disable=redefined-outer-name
for context_uuid, topology_list in descriptor_loader.topologies.items():
for topology in topology_list:
context_client.RemoveTopology(TopologyId(**topology['topology_id']))
for link in LINKS : context_client.RemoveLink (LinkId (**link ['link_id' ]))
for device in DEVICES : device_client .DeleteDevice (DeviceId (**device ['device_id' ]))
for topology in TOPOLOGIES: context_client.RemoveTopology(TopologyId(**topology['topology_id']))
for context in CONTEXTS : context_client.RemoveContext (ContextId (**context ['context_id' ]))
for context in descriptor_loader.contexts:
context_client.RemoveContext(ContextId(**context['context_id']))
def test_scenario_empty_again(context_client : ContextClient): # pylint: disable=redefined-outer-name
# ----- List entities - Ensure database is empty again -------------------------------------------------------------
response = context_client.ListContexts(Empty())
assert len(response.contexts) == 0
......
src/tests/ecoc22/tests/test_functional_create_service.py
+ 46
− 29
View file @ 1e645f9c
......@@ -13,73 +13,90 @@
# limitations under the License.
import logging
from common.Constants import DEFAULT_CONTEXT_NAME
from common.proto.context_pb2 import ContextId, Empty, ServiceTypeEnum
from common.tools.descriptor.Loader import DescriptorLoader
from common.tools.grpc.Tools import grpc_message_to_json_string
from compute.tests.mock_osm.MockOSM import MockOSM
from common.tools.object_factory.Context import json_context_id
from context.client.ContextClient import ContextClient
from tests.Fixtures import context_client
from .Fixtures import osm_wim
#from .Objects_BigNet import (
# CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, WIM_SERVICE_CONNECTION_POINTS, WIM_SERVICE_TYPE)
#from .Objects_DC_CSGW_TN import (
# CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, WIM_SERVICE_CONNECTION_POINTS, WIM_SERVICE_TYPE)
#from .Objects_DC_CSGW_TN_OLS import (
# CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, WIM_SERVICE_CONNECTION_POINTS, WIM_SERVICE_TYPE)
from .Objects_DC_CSGW_OLS import (
CONTEXT_ID, CONTEXTS, DEVICES, LINKS, TOPOLOGIES, WIM_SERVICE_CONNECTION_POINTS, WIM_SERVICE_TYPE)
from tests.Fixtures import context_client # pylint: disable=unused-import
from tests.tools.mock_osm.MockOSM import MockOSM
from .Fixtures import osm_wim # pylint: disable=unused-import
from .Objects import WIM_SERVICE_CONNECTION_POINTS, WIM_SERVICE_TYPE
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
DESCRIPTOR_FILE = 'ecoc22/descriptors_emulated.json'
ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME))
def test_service_creation(context_client : ContextClient, osm_wim : MockOSM): # pylint: disable=redefined-outer-name
# ----- List entities - Ensure scenario is ready -------------------------------------------------------------------
with open(DESCRIPTOR_FILE, 'r', encoding='UTF-8') as f:
descriptors = f.read()
descriptor_loader = DescriptorLoader(descriptors)
def test_scenario_is_correct(context_client : ContextClient): # pylint: disable=redefined-outer-name
# ----- List entities - Ensure links are created -------------------------------------------------------------------
response = context_client.ListContexts(Empty())
assert len(response.contexts) == len(CONTEXTS)
assert len(response.contexts) == descriptor_loader.num_contexts
response = context_client.ListTopologies(ContextId(**CONTEXT_ID))
assert len(response.topologies) == len(TOPOLOGIES)
for context_uuid, num_topologies in descriptor_loader.num_topologies.items():
response = context_client.ListTopologies(ContextId(**json_context_id(context_uuid)))
assert len(response.topologies) == num_topologies
response = context_client.ListDevices(Empty())
assert len(response.devices) == len(DEVICES)
assert len(response.devices) == descriptor_loader.num_devices
response = context_client.ListLinks(Empty())
assert len(response.links) == len(LINKS)
assert len(response.links) == descriptor_loader.num_links
response = context_client.ListServices(ContextId(**CONTEXT_ID))
assert len(response.services) == 0
for context_uuid, num_services in descriptor_loader.num_services.items():
response = context_client.ListServices(ContextId(**json_context_id(context_uuid)))
assert len(response.services) == num_services
for context_uuid, num_slices in descriptor_loader.num_slices.items():
response = context_client.ListSlices(ContextId(**json_context_id(context_uuid)))
assert len(response.slices) == num_slices
# This scenario assumes no services are created beforehand
response = context_client.GetContext(ADMIN_CONTEXT_ID)
assert len(response.service_ids) == 0
assert len(response.slice_ids) == 0
def test_service_creation(context_client : ContextClient, osm_wim : MockOSM): # pylint: disable=redefined-outer-name
# ----- Create Service ---------------------------------------------------------------------------------------------
service_uuid = osm_wim.create_connectivity_service(WIM_SERVICE_TYPE, WIM_SERVICE_CONNECTION_POINTS)
osm_wim.get_connectivity_service_status(service_uuid)
def test_scenario_service_created(context_client : ContextClient): # pylint: disable=redefined-outer-name
# ----- List entities - Ensure service is created ------------------------------------------------------------------
response = context_client.ListContexts(Empty())
assert len(response.contexts) == len(CONTEXTS)
assert len(response.contexts) == descriptor_loader.num_contexts
response = context_client.ListTopologies(ContextId(**CONTEXT_ID))
assert len(response.topologies) == len(TOPOLOGIES)
for context_uuid, num_topologies in descriptor_loader.num_topologies.items():
response = context_client.ListTopologies(ContextId(**json_context_id(context_uuid)))
assert len(response.topologies) == num_topologies
response = context_client.ListDevices(Empty())
assert len(response.devices) == len(DEVICES)
assert len(response.devices) == descriptor_loader.num_devices
response = context_client.ListLinks(Empty())
assert len(response.links) == len(LINKS)
assert len(response.links) == descriptor_loader.num_links
response = context_client.ListServices(ContextId(**CONTEXT_ID))
response = context_client.ListServices(ADMIN_CONTEXT_ID)
LOGGER.info('Services[{:d}] = {:s}'.format(len(response.services), grpc_message_to_json_string(response)))
assert len(response.services) == 3 # 1xL2NM + 2xTAPI
for service in response.services:
service_id = service.service_id
response = context_client.ListConnections(service_id)
LOGGER.info(' ServiceId[{:s}] => Connections[{:d}] = {:s}'.format(
grpc_message_to_json_string(service_id), len(response.connections), grpc_message_to_json_string(response)))
if service.service_type == ServiceTypeEnum.SERVICETYPE_L2NM:
assert len(response.connections) == 2 # 2 connections per service (primary + backup)
elif service.service_type == ServiceTypeEnum.SERVICETYPE_TAPI_CONNECTIVITY_SERVICE:
assert len(response.connections) == 1 # 1 connection per service
else:
str_service = grpc_message_to_json_string(service)
raise Exception('Unexpected ServiceType: {:s}'.format(str_service))
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment