Tests - Tools - SIMAP Server

        node : Dict = self._restconf_client.get(endpoint)

        return node['ietf-network-topology:termination-point'][0]

    def update(self, supporting_termination_point_ids : List[Tuple[str, str, str]] = []) -> None:

        endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id)

        tp = {'tp-id': self._tp_id}

        stps = [

            {'network-ref': snet_id, 'node-ref': snode_id, 'tp-ref': stp_id}

            for snet_id,snode_id,stp_id in supporting_termination_point_ids

        ]

        if len(stps) > 0: tp['supporting-termination-point'] = stps

        node = {'node-id': self._node_id, 'ietf-network-topology:termination-point': [tp]}

        network = {'network-id': self._network_id, 'node': [node]}

        payload  = {'ietf-network:networks': {'network': [network]}}

        self._restconf_client.patch(endpoint, payload)

    def delete(self) -> None:

        endpoint = TerminationPoint.ENDPOINT_ID.format(self._network_id, self._node_id, self._tp_id)

        self._restconf_client.delete(endpoint)

        node : Dict = self._restconf_client.get(endpoint)

        return node['ietf-network:node'][0]

    def update(

        self, termination_point_ids : List[str] = [],

        supporting_node_ids : List[Tuple[str, str]] = []

    ) -> None:

        endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id)

        node = {'node-id': self._node_id}

        tps = [{'tp-id': tp_id} for tp_id in termination_point_ids]

        if len(tps) > 0: node['ietf-network-topology:termination-point'] = tps

        sns = [{'network-ref': snet_id, 'node-ref': snode_id} for snet_id,snode_id in supporting_node_ids]

        if len(sns) > 0: node['supporting-node'] = sns

        network = {'network-id': self._network_id, 'node': [node]}

        payload  = {'ietf-network:networks': {'network': [network]}}

        self._restconf_client.patch(endpoint, payload)

    def delete(self) -> None:

        endpoint = Node.ENDPOINT_ID.format(self._network_id, self._node_id)

        self._restconf_client.delete(endpoint)

        link : Dict = self._restconf_client.get(endpoint)

        return link['ietf-network-topology:link'][0]

    def update(

        self, src_node_id : str, src_tp_id : str, dst_node_id : str, dst_tp_id : str,

        supporting_link_ids : List[Tuple[str, str]] = []

    ) -> None:

        endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id)

        link = {

            'link-id'    : self._link_id,

            'source'     : {'source-node': src_node_id, 'source-tp': src_tp_id},

            'destination': {'dest-node'  : dst_node_id, 'dest-tp'  : dst_tp_id},

        }

        sls = [{'network-ref': snet_id, 'link-ref': slink_id} for snet_id,slink_id in supporting_link_ids]

        if len(sls) > 0: link['supporting-link'] = sls

        network = {'network-id': self._network_id, 'ietf-network-topology:link': [link]}

        payload  = {'ietf-network:networks': {'network': [network]}}

        self._restconf_client.patch(endpoint, payload)

    def delete(self) -> None:

        endpoint = Link.ENDPOINT_ID.format(self._network_id, self._link_id)

        self._restconf_client.delete(endpoint)

        networks : Dict = self._restconf_client.get(endpoint)

        return networks['ietf-network:network'][0]

    def update(self, supporting_network_ids : List[str] = []) -> None:

        endpoint = Network.ENDPOINT_ID.format(self._network_id)

        network = {'network-id': self._network_id}

        sns = [{'network-ref': sn_id} for sn_id in supporting_network_ids]

        if len(sns) > 0: network['supporting-network'] = sns

        payload  = {'ietf-network:networks': {'network': [network]}}

        self._restconf_client.patch(endpoint, payload)

    def delete(self) -> None:

        endpoint = Network.ENDPOINT_ID.format(self._network_id)

        self._restconf_client.delete(endpoint)

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (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.

import logging

from .RestConfClient import RestConfClient

from .SimapClient import SimapClient

logging.basicConfig(level=logging.INFO)

LOGGER = logging.getLogger(__name__)

def main() -> None:

    restconf_client = RestConfClient('127.0.0.1', logger=LOGGER)

    simap_client = SimapClient(restconf_client)

    simap_client.network('topology').create()

    simap_client.network('topology').node('r1').create()

    simap_client.network('topology').node('r1').termination_point('100').create()

    simap_client.network('topology').node('r1').termination_point('101').create()

    simap_client.network('topology').node('r1').termination_point('102').create()

    simap_client.network('topology').node('r2').create(tp_ids=['200', '201', '202'])

    simap_client.network('topology').link('l1').create('r1', '102', 'r2', '201')

    print('networks=', simap_client.networks())

    print('network[topology].nodes=', simap_client.network('topology').nodes())

    print('network[topology].node[r2]=', simap_client.network('topology').node('r2').get())

    print('network[topology].links=', simap_client.network('topology').links())

    print('network[topology].link[l1]=', simap_client.network('topology').link('l1').get())

    simap_client.network('topology').link('l1').delete()

    simap_client.network('topology').node('r2').delete()

    simap_client.network('topology').node('r1').delete()

    print('networks=', simap_client.networks())

if __name__ == '__main__':

    main()