Newer
Older
# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
#
# 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.
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
import logging
from common.Constants import DEFAULT_CONTEXT_NAME
from common.proto.context_pb2 import ContextId
from common.proto.forecaster_pb2 import ForecastLinkCapacityRequest, ForecastTopologyCapacityRequest
from common.tools.descriptor.Loader import DescriptorLoader, check_descriptor_load_results, validate_empty_scenario
from common.tools.object_factory.Context import json_context_id
from context.client.ContextClient import ContextClient
from forecaster.client.ForecasterClient import ForecasterClient
from monitoring.client.MonitoringClient import MonitoringClient
from .PrepareTestScenario import ( # pylint: disable=unused-import
# be careful, order of symbols is important here!
mock_service, forecaster_service, context_client, monitoring_client, forecaster_client)
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME))
DESCRIPTORS_FILE = '' # use dummy descriptor here
def test_prepare_environment(
context_client : ContextClient, # pylint: disable=redefined-outer-name
) -> None:
validate_empty_scenario(context_client)
descriptor_loader = DescriptorLoader(descriptors_file=DESCRIPTORS_FILE, context_client=context_client)
results = descriptor_loader.process()
check_descriptor_load_results(results, descriptor_loader)
descriptor_loader.validate()
# Verify the scenario has no services/slices
response = context_client.GetContext(ADMIN_CONTEXT_ID)
assert len(response.service_ids) == 0
assert len(response.slice_ids) == 0
def test_forecast_link(
context_client : ContextClient,
forecaster_client : ForecasterClient,
): # pylint: disable=redefined-outer-name
# TODO: select link
forecast_request = ForecastLinkCapacityRequest()
forecast_request.forecast_window_seconds = 10 * 24 * 60 * 60 # 10 days in seconds
# TODO: populate request
forecast_reply = forecaster_client.ForecastLinkCapacity(forecast_request)
# TODO: validate reply
def test_forecast_topology(
context_client : ContextClient,
forecaster_client : ForecasterClient,
): # pylint: disable=redefined-outer-name
# TODO: get topology id
forecast_request = ForecastTopologyCapacityRequest()
forecast_request.forecast_window_seconds = 10 * 24 * 60 * 60 # 10 days in seconds
# TODO: populate request
forecast_reply = forecaster_client.ForecastTopologyCapacity(forecast_request)
# TODO: validate reply
def test_cleanup_environment(
context_client : ContextClient, # pylint: disable=redefined-outer-name
) -> None:
# Verify the scenario has no services/slices
response = context_client.GetContext(ADMIN_CONTEXT_ID)
assert len(response.service_ids) == 0
assert len(response.slice_ids) == 0
# Load descriptors and validate the base scenario
descriptor_loader = DescriptorLoader(descriptors_file=DESCRIPTORS_FILE, context_client=context_client)
descriptor_loader.validate()
descriptor_loader.unload()
validate_empty_scenario(context_client)