Skip to content
GitLab
Projects
Groups
Topics
Snippets
/
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Sign in
Toggle navigation
Menu
TFS
controller
Compare revisions
3320ef8965a5951f07141378f3d91b7cb1307bf5...b1c6ae62ec0142dd2341a7eff8833f856e20ed66
Hide whitespace changes
Inline
Side-by-side
src/tests/benchmark/automation/tests/Objects.py
0 → 100644
View file @
b1c6ae62
# 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
typing
import
Dict
,
List
,
Tuple
from
common.Constants
import
DEFAULT_CONTEXT_UUID
,
DEFAULT_TOPOLOGY_UUID
from
common.tools.object_factory.Context
import
json_context
,
json_context_id
from
common.tools.object_factory.Device
import
(
json_device_connect_rules
,
json_device_emulated_connect_rules
,
json_device_emulated_packet_router_disabled
,
json_device_emulated_tapi_disabled
,
json_device_id
,
json_device_packetrouter_disabled
,
json_device_tapi_disabled
)
from
common.tools.object_factory.EndPoint
import
json_endpoint
,
json_endpoint_id
from
common.tools.object_factory.Link
import
json_link
,
json_link_id
from
common.tools.object_factory.Topology
import
json_topology
,
json_topology_id
from
common.proto.kpi_sample_types_pb2
import
KpiSampleType
# ----- Context --------------------------------------------------------------------------------------------------------
CONTEXT_ID
=
json_context_id
(
DEFAULT_CONTEXT_UUID
)
CONTEXT
=
json_context
(
DEFAULT_CONTEXT_UUID
)
# ----- Topology -------------------------------------------------------------------------------------------------------
TOPOLOGY_ID
=
json_topology_id
(
DEFAULT_TOPOLOGY_UUID
,
context_id
=
CONTEXT_ID
)
TOPOLOGY
=
json_topology
(
DEFAULT_TOPOLOGY_UUID
,
context_id
=
CONTEXT_ID
)
# ----- Monitoring Samples ---------------------------------------------------------------------------------------------
PACKET_PORT_SAMPLE_TYPES
=
[
KpiSampleType
.
KPISAMPLETYPE_PACKETS_TRANSMITTED
,
KpiSampleType
.
KPISAMPLETYPE_PACKETS_RECEIVED
,
KpiSampleType
.
KPISAMPLETYPE_BYTES_TRANSMITTED
,
KpiSampleType
.
KPISAMPLETYPE_BYTES_RECEIVED
,
]
# ----- Devices --------------------------------------------------------------------------------------------------------
DEVICE_ENDPOINT_DEFS
=
[(
'13/0/0'
,
'optical'
,
[]),
(
'13/1/2'
,
'copper'
,
PACKET_PORT_SAMPLE_TYPES
)]
DEVICE_CONNECT_RULES
=
json_device_emulated_connect_rules
(
DEVICE_ENDPOINT_DEFS
)
# ----- Object Collections ---------------------------------------------------------------------------------------------
CONTEXTS
=
[
CONTEXT
]
TOPOLOGIES
=
[
TOPOLOGY
]
DEVICES
=
[]
for
x
in
range
(
1
,
1000
):
DEVICE_UUID
=
'EMU-'
+
str
(
x
)
DEVICE
=
json_device_emulated_packet_router_disabled
(
DEVICE_UUID
)
DEVICES
.
append
((
DEVICE
,
DEVICE_CONNECT_RULES
))
src/tests/benchmark/automation/tests/__init__.py
0 → 100644
View file @
b1c6ae62
# 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.
src/tests/benchmark/automation/tests/test_functional_bootstrap.py
0 → 100644
View file @
b1c6ae62
# 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.
import
copy
,
logging
,
pytest
from
common.Settings
import
get_setting
from
common.proto.monitoring_pb2
import
KpiDescriptorList
from
common.tests.EventTools
import
EVENT_CREATE
,
EVENT_UPDATE
,
check_events
from
common.tools.object_factory.Context
import
json_context_id
from
common.tools.object_factory.Device
import
json_device_id
from
common.tools.object_factory.Link
import
json_link_id
from
common.tools.object_factory.Topology
import
json_topology_id
from
context.client.ContextClient
import
ContextClient
from
monitoring.client.MonitoringClient
import
MonitoringClient
from
context.client.EventsCollector
import
EventsCollector
from
common.proto.context_pb2
import
Context
,
ContextId
,
Device
,
Empty
,
Topology
from
device.client.DeviceClient
import
DeviceClient
from
.Objects
import
CONTEXT_ID
,
CONTEXTS
,
DEVICES
,
TOPOLOGIES
from
tests.Fixtures
import
context_client
,
device_client
,
monitoring_client
LOGGER
=
logging
.
getLogger
(
__name__
)
LOGGER
.
setLevel
(
logging
.
DEBUG
)
def
test_scenario_empty
(
context_client
:
ContextClient
):
# pylint: disable=redefined-outer-name
# ----- List entities - Ensure database is empty -------------------------------------------------------------------
response
=
context_client
.
ListContexts
(
Empty
())
assert
len
(
response
.
contexts
)
==
0
response
=
context_client
.
ListDevices
(
Empty
())
assert
len
(
response
.
devices
)
==
0
def
test_prepare_scenario
(
context_client
:
ContextClient
):
# pylint: disable=redefined-outer-name
# ----- Create Contexts and Topologies -----------------------------------------------------------------------------
for
context
in
CONTEXTS
:
context_uuid
=
context
[
'context_id'
][
'context_uuid'
][
'uuid'
]
LOGGER
.
info
(
'Adding Context {:s}'
.
format
(
context_uuid
))
response
=
context_client
.
SetContext
(
Context
(
**
context
))
assert
response
.
context_uuid
.
uuid
==
context_uuid
for
topology
in
TOPOLOGIES
:
context_uuid
=
topology
[
'topology_id'
][
'context_id'
][
'context_uuid'
][
'uuid'
]
topology_uuid
=
topology
[
'topology_id'
][
'topology_uuid'
][
'uuid'
]
LOGGER
.
info
(
'Adding Topology {:s}/{:s}'
.
format
(
context_uuid
,
topology_uuid
))
response
=
context_client
.
SetTopology
(
Topology
(
**
topology
))
assert
response
.
context_id
.
context_uuid
.
uuid
==
context_uuid
assert
response
.
topology_uuid
.
uuid
==
topology_uuid
context_id
=
json_context_id
(
context_uuid
)
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
)
response
=
context_client
.
ListTopologies
(
ContextId
(
**
CONTEXT_ID
))
assert
len
(
response
.
topologies
)
==
len
(
TOPOLOGIES
)
response
=
context_client
.
ListDevices
(
Empty
())
assert
len
(
response
.
devices
)
==
0
def
test_devices_bootstraping
(
context_client
:
ContextClient
,
device_client
:
DeviceClient
):
# pylint: disable=redefined-outer-name
# ----- Create Devices and Validate Collected Events ---------------------------------------------------------------
for
device
,
connect_rules
in
DEVICES
:
device_uuid
=
device
[
'device_id'
][
'device_uuid'
][
'uuid'
]
LOGGER
.
info
(
'Adding Device {:s}'
.
format
(
device_uuid
))
device_with_connect_rules
=
copy
.
deepcopy
(
device
)
device_with_connect_rules
[
'device_config'
][
'config_rules'
].
extend
(
connect_rules
)
response
=
device_client
.
AddDevice
(
Device
(
**
device_with_connect_rules
))
assert
response
.
device_uuid
.
uuid
==
device_uuid
def
test_devices_bootstrapped
(
context_client
:
ContextClient
):
# pylint: disable=redefined-outer-name
# ----- List entities - Ensure bevices are created -----------------------------------------------------------------
response
=
context_client
.
ListContexts
(
Empty
())
assert
len
(
response
.
contexts
)
==
len
(
CONTEXTS
)
response
=
context_client
.
ListTopologies
(
ContextId
(
**
CONTEXT_ID
))
assert
len
(
response
.
topologies
)
==
len
(
TOPOLOGIES
)
response
=
context_client
.
ListDevices
(
Empty
())
assert
len
(
response
.
devices
)
==
len
(
DEVICES
)
def
test_links_created
(
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
)
response
=
context_client
.
ListTopologies
(
ContextId
(
**
CONTEXT_ID
))
assert
len
(
response
.
topologies
)
==
len
(
TOPOLOGIES
)
response
=
context_client
.
ListDevices
(
Empty
())
assert
len
(
response
.
devices
)
==
len
(
DEVICES
)
\ No newline at end of file
src/tests/benchmark/automation/tests/test_functional_cleanup.py
0 → 100644
View file @
b1c6ae62
# 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.
import
logging
,
pytest
from
common.Settings
import
get_setting
from
common.tests.EventTools
import
EVENT_REMOVE
,
check_events
from
common.tools.object_factory.Context
import
json_context_id
from
common.tools.object_factory.Device
import
json_device_id
from
common.tools.object_factory.Link
import
json_link_id
from
common.tools.object_factory.Topology
import
json_topology_id
from
context.client.ContextClient
import
ContextClient
from
context.client.EventsCollector
import
EventsCollector
from
common.proto.context_pb2
import
ContextId
,
DeviceId
,
Empty
,
LinkId
,
TopologyId
from
device.client.DeviceClient
import
DeviceClient
from
tests.Fixtures
import
context_client
,
device_client
from
.Objects
import
CONTEXT_ID
,
CONTEXTS
,
DEVICES
,
TOPOLOGIES
LOGGER
=
logging
.
getLogger
(
__name__
)
LOGGER
.
setLevel
(
logging
.
DEBUG
)
def
test_scenario_cleanup
(
context_client
:
ContextClient
,
device_client
:
DeviceClient
):
# pylint: disable=redefined-outer-name
# ----- Delete Devices and Validate Collected Events ---------------------------------------------------------------
for
device
,
_
in
DEVICES
:
device_id
=
device
[
'device_id'
]
device_uuid
=
device_id
[
'device_uuid'
][
'uuid'
]
LOGGER
.
info
(
'Deleting Device {:s}'
.
format
(
device_uuid
))
device_client
.
DeleteDevice
(
DeviceId
(
**
device_id
))
#expected_events.append(('DeviceEvent', EVENT_REMOVE, json_device_id(device_uuid)))
# ----- Delete Topologies and Validate Collected Events ------------------------------------------------------------
for
topology
in
TOPOLOGIES
:
topology_id
=
topology
[
'topology_id'
]
context_uuid
=
topology_id
[
'context_id'
][
'context_uuid'
][
'uuid'
]
topology_uuid
=
topology_id
[
'topology_uuid'
][
'uuid'
]
LOGGER
.
info
(
'Deleting Topology {:s}/{:s}'
.
format
(
context_uuid
,
topology_uuid
))
context_client
.
RemoveTopology
(
TopologyId
(
**
topology_id
))
context_id
=
json_context_id
(
context_uuid
)
#expected_events.append(('TopologyEvent', EVENT_REMOVE, json_topology_id(topology_uuid, context_id=context_id)))
# ----- Delete Contexts and Validate Collected Events --------------------------------------------------------------
for
context
in
CONTEXTS
:
context_id
=
context
[
'context_id'
]
context_uuid
=
context_id
[
'context_uuid'
][
'uuid'
]
LOGGER
.
info
(
'Deleting Context {:s}'
.
format
(
context_uuid
))
context_client
.
RemoveContext
(
ContextId
(
**
context_id
))
#expected_events.append(('ContextEvent', EVENT_REMOVE, json_context_id(context_uuid)))
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
response
=
context_client
.
ListDevices
(
Empty
())
assert
len
(
response
.
devices
)
==
0
Prev
1
2
Next