Skip to content
Snippets Groups Projects
test_topology.py 9.54 KiB
Newer Older
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
# 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.

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
import copy, grpc, pytest, time
from common.proto.context_pb2 import Context, ContextEvent, ContextId, EventTypeEnum, Topology, TopologyEvent, TopologyId
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
from context.client.ContextClient import ContextClient
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
from context.service.database.uuids.Topology import topology_get_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
from context.client.EventsCollector import EventsCollector
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
from .Objects import CONTEXT, CONTEXT_ID, CONTEXT_NAME, TOPOLOGY, TOPOLOGY_ID, TOPOLOGY_NAME
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
@pytest.mark.depends(on=['context/tests/test_context.py::test_context'])
def test_topology(context_client : ContextClient) -> None:
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- Initialize the EventsCollector -----------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    events_collector = EventsCollector(
        context_client, log_events_received=True,
        activate_context_collector = False, activate_topology_collector = True, activate_device_collector = False,
        activate_link_collector = False, activate_service_collector = False, activate_slice_collector = False,
        activate_connection_collector = False)
    events_collector.start()
    time.sleep(3) # wait for the events collector to start
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- Prepare dependencies for the test and capture related events -----------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.SetContext(Context(**CONTEXT))
    context_uuid = response.context_uuid.uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    event = events_collector.get_event(block=True, timeout=1.0)
    assert isinstance(event, ContextEvent)
    assert event.event.event_type == EventTypeEnum.EVENTTYPE_CREATE
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert event.context_id.context_uuid.uuid == context_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- Get when the object does not exist -------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    topology_id = TopologyId(**TOPOLOGY_ID)
    context_uuid,topology_uuid = topology_get_uuid(topology_id, allow_random=False)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    with pytest.raises(grpc.RpcError) as e:
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
        context_client.GetTopology(topology_id)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert e.value.code() == grpc.StatusCode.NOT_FOUND
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    MSG = 'Topology({:s}/{:s}) not found; context_uuid generated was: {:s}; topology_uuid generated was: {:s}'
    assert e.value.details() == MSG.format(CONTEXT_NAME, TOPOLOGY_NAME, context_uuid, topology_uuid)
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- List when the object does not exist ------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.GetContext(ContextId(**CONTEXT_ID))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topology_ids) == 0
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.service_ids) == 0
    assert len(response.slice_ids) == 0
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.ListTopologyIds(ContextId(**CONTEXT_ID))
    assert len(response.topology_ids) == 0

    response = context_client.ListTopologies(ContextId(**CONTEXT_ID))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topologies) == 0

    # ----- Create the object ------------------------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.SetTopology(Topology(**TOPOLOGY))
    assert response.context_id.context_uuid.uuid == context_uuid
    assert response.topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- Check create event -----------------------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    event = events_collector.get_event(block=True, timeout=1.0)
    assert isinstance(event, TopologyEvent)
    assert event.event.event_type == EventTypeEnum.EVENTTYPE_CREATE
    assert event.topology_id.context_id.context_uuid.uuid == context_uuid
    assert event.topology_id.topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- Get when the object exists ---------------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.GetContext(ContextId(**CONTEXT_ID))
    assert response.context_id.context_uuid.uuid == context_uuid
    assert response.name == CONTEXT_NAME
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topology_ids) == 1
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert response.topology_ids[0].context_id.context_uuid.uuid == context_uuid
    assert response.topology_ids[0].topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.service_ids) == 0
    assert len(response.slice_ids) == 0

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.GetTopology(TopologyId(**TOPOLOGY_ID))
    assert response.topology_id.context_id.context_uuid.uuid == context_uuid
    assert response.topology_id.topology_uuid.uuid == topology_uuid
    assert response.name == TOPOLOGY_NAME
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.device_ids) == 0
    assert len(response.link_ids) == 0

    # ----- List when the object exists --------------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.ListTopologyIds(ContextId(**CONTEXT_ID))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topology_ids) == 1
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert response.topology_ids[0].context_id.context_uuid.uuid == context_uuid
    assert response.topology_ids[0].topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.ListTopologies(ContextId(**CONTEXT_ID))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topologies) == 1
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert response.topologies[0].topology_id.context_id.context_uuid.uuid == context_uuid
    assert response.topologies[0].topology_id.topology_uuid.uuid == topology_uuid
    assert response.topologies[0].name == TOPOLOGY_NAME
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topologies[0].device_ids) == 0
    assert len(response.topologies[0].link_ids) == 0

    # ----- Update the object ------------------------------------------------------------------------------------------
    new_topology_name = 'new'
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    TOPOLOGY_UPDATED = copy.deepcopy(TOPOLOGY)
    TOPOLOGY_UPDATED['name'] = new_topology_name
    response = context_client.SetTopology(Topology(**TOPOLOGY_UPDATED))
    assert response.context_id.context_uuid.uuid == context_uuid
    assert response.topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- Check update event -----------------------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    event = events_collector.get_event(block=True, timeout=1.0)
    assert isinstance(event, TopologyEvent)
    assert event.event.event_type == EventTypeEnum.EVENTTYPE_UPDATE
    assert event.topology_id.context_id.context_uuid.uuid == context_uuid
    assert event.topology_id.topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- Get when the object is modified ----------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.GetTopology(TopologyId(**TOPOLOGY_ID))
    assert response.topology_id.context_id.context_uuid.uuid == context_uuid
    assert response.topology_id.topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert response.name == new_topology_name
    assert len(response.device_ids) == 0
    assert len(response.link_ids) == 0

    # ----- List when the object is modified ---------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.ListTopologyIds(ContextId(**CONTEXT_ID))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topology_ids) == 1
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert response.topology_ids[0].context_id.context_uuid.uuid == context_uuid
    assert response.topology_ids[0].topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.ListTopologies(ContextId(**CONTEXT_ID))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topologies) == 1
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert response.topologies[0].topology_id.context_id.context_uuid.uuid == context_uuid
    assert response.topologies[0].topology_id.topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert response.topologies[0].name == new_topology_name
    assert len(response.topologies[0].device_ids) == 0
    assert len(response.topologies[0].link_ids) == 0

    # ----- Remove the object ------------------------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    context_client.RemoveTopology(TopologyId(**TOPOLOGY_ID))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- Check remove event -----------------------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    event = events_collector.get_event(block=True, timeout=1.0)
    assert isinstance(event, TopologyEvent)
    assert event.event.event_type == EventTypeEnum.EVENTTYPE_REMOVE
    assert event.topology_id.context_id.context_uuid.uuid == context_uuid
    assert event.topology_id.topology_uuid.uuid == topology_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- List after deleting the object -----------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.GetContext(ContextId(**CONTEXT_ID))
    assert len(response.topology_ids) == 0
    assert len(response.service_ids) == 0
    assert len(response.slice_ids) == 0

    response = context_client.ListTopologyIds(ContextId(**CONTEXT_ID))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topology_ids) == 0

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    response = context_client.ListTopologies(ContextId(**CONTEXT_ID))
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    assert len(response.topologies) == 0

    # ----- Clean dependencies used in the test and capture related events ---------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    context_client.RemoveContext(ContextId(**CONTEXT_ID))

Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    event = events_collector.get_event(block=True, timeout=1.0)
    assert isinstance(event, ContextEvent)
    assert event.event.event_type == EventTypeEnum.EVENTTYPE_REMOVE
    assert event.context_id.context_uuid.uuid == context_uuid
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed

    # ----- Stop the EventsCollector -----------------------------------------------------------------------------------
Lluis Gifre Renom's avatar
Lluis Gifre Renom committed
    events_collector.stop()