Commit 6ddf5dc2 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Device component - gNMI OpenConfig Driver: 

- Added test code for L2 VPN configuration
parent 777d7dff

src/device/tests/gnmi_openconfig/Dockerfile.l2vpn

0 → 100644
+85 −0
Original line number Diff line number Diff line 
# 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.



# Minimal reuse of device component Dockerfile steps to run L2VPN tests

FROM python:3.9-slim



# Base deps + libyang build

RUN apt-get update -qq && apt-get install -y -qq wget g++ git build-essential cmake libpcre2-dev python3-dev python3-cffi && rm -rf /var/lib/apt/lists/*

RUN mkdir -p /var/libyang && git clone https://github.com/CESNET/libyang.git /var/libyang

WORKDIR /var/libyang

RUN git fetch && git checkout v2.1.148 && mkdir -p build

WORKDIR /var/libyang/build

RUN cmake -D CMAKE_BUILD_TYPE:String="Release" .. && make && make install && ldconfig



ENV PYTHONUNBUFFERED=0



# Python toolchain

RUN python3 -m pip install --upgrade 'pip==25.2'

RUN python3 -m pip install --upgrade 'setuptools==79.0.0' 'wheel==0.45.1'

RUN python3 -m pip install --upgrade 'pip-tools==7.3.0'



WORKDIR /var/teraflow

COPY common_requirements.in common_requirements.in

RUN pip-compile --quiet --output-file=common_requirements.txt common_requirements.in

RUN python3 -m pip install -r common_requirements.txt



# Common files + proto

WORKDIR /var/teraflow/common

COPY src/common/. ./

RUN rm -rf proto

RUN mkdir -p /var/teraflow/common/proto

WORKDIR /var/teraflow/common/proto

RUN touch __init__.py

COPY proto/*.proto ./

RUN python3 -m grpc_tools.protoc -I=. --python_out=. --grpc_python_out=. *.proto

RUN rm *.proto

RUN find . -type f -exec sed -i -E 's/^(import\ .*)_pb2/from . \1_pb2/g' {} \;



# Device deps

WORKDIR /var/teraflow/device

COPY src/device/requirements.in requirements.in

RUN pip-compile --quiet --output-file=requirements.txt requirements.in

RUN python3 -m pip install -r requirements.txt



# Source tree

WORKDIR /var/teraflow

COPY src/__init__.py ./__init__.py

COPY src/common/*.py ./common/

COPY src/common/tests/. ./common/tests/

COPY src/common/tools/. ./common/tools/

COPY src/context/__init__.py context/__init__.py

COPY src/context/client/. context/client/

COPY src/device/. device/

COPY src/monitoring/__init__.py monitoring/__init__.py

COPY src/monitoring/client/. monitoring/client/

COPY src/service/__init__.py service/__init__.py

COPY src/service/client/. service/client/

COPY src/vnt_manager/__init__.py vnt_manager/__init__.py

COPY src/vnt_manager/client/. vnt_manager/client/



# OpenConfig models as in device Dockerfile

RUN mkdir -p /tmp/openconfig && git clone https://github.com/openconfig/public.git /tmp/openconfig

WORKDIR /tmp/openconfig

RUN git fetch && git checkout v4.4.0

RUN rm -rf /var/teraflow/device/service/drivers/gnmi_openconfig/git

RUN mkdir -p /var/teraflow/device/service/drivers/gnmi_openconfig/git/openconfig/public

RUN mv /tmp/openconfig/release /var/teraflow/device/service/drivers/gnmi_openconfig/git/openconfig/public

RUN mv /tmp/openconfig/third_party /var/teraflow/device/service/drivers/gnmi_openconfig/git/openconfig/public

RUN rm -rf /tmp/openconfig

WORKDIR /var/teraflow



ENV RUN_L2VPN_LAB=1

ENV PYTHONPATH=/var/teraflow

CMD ["pytest", "--log-level=DEBUG", "--verbose", "device/tests/gnmi_openconfig/test_unitary_gnmi_oc_arista_l2vpn.py"]

src/device/tests/gnmi_openconfig/test_unitary_gnmi_oc_arista_l2vpn.py

+122 −543

File changed.

Preview size limit exceeded, changes collapsed.

src/device/tests/gnmi_openconfig/tools/request_composers.py

+82 −3
Original line number Diff line number Diff line
@@ -14,11 +14,20 @@ 


from typing import Dict, Tuple



def interface(if_name, sif_index, ipv4_address, ipv4_prefix, enabled) -> Tuple[str, Dict]:

def interface(if_name, sif_index, ipv4_address=None, ipv4_prefix=None, enabled=True, vlan_id=None) -> Tuple[str, Dict]:

    str_path = '/interface[{:s}]'.format(if_name)

    str_data = {

        'name': if_name, 'enabled': enabled, 'sub_if_index': sif_index, 'sub_if_enabled': enabled,

        'sub_if_ipv4_enabled': enabled, 'sub_if_ipv4_address': ipv4_address, 'sub_if_ipv4_prefix': ipv4_prefix

        'name': if_name,

        'enabled': enabled,

        'index': sif_index,

        'sub_if_index': sif_index,

        'sub_if_enabled': enabled,

        'sub_if_ipv4_enabled': enabled,

        'sub_if_ipv4_address': ipv4_address,

        'sub_if_ipv4_prefix': ipv4_prefix,

        'address_ip': ipv4_address,

        'address_prefix': ipv4_prefix,

        'vlan_id': vlan_id,

    }

    return str_path, str_data
@@ -42,3 +51,73 @@ def network_instance_interface(ni_name, if_name, sif_index) -> Tuple[str, Dict]: 
        'name': ni_name, 'if_name': if_name, 'sif_index': sif_index

    }

    return str_path, str_data



def mpls_global(ldp_router_id: str, hello_interval: int = None, hello_holdtime: int = None) -> Tuple[str, Dict]:

    str_path = '/mpls'

    str_data = {

        'ldp': {

            'lsr_id': ldp_router_id,

            'hello_interval': hello_interval,

            'hello_holdtime': hello_holdtime,

        }

    }

    return str_path, str_data



def mpls_ldp_interface(if_name: str, hello_interval: int = None, hello_holdtime: int = None) -> Tuple[str, Dict]:

    str_path = '/mpls/interface[{:s}]'.format(if_name)

    str_data = {

        'interface': if_name,

        'hello_interval': hello_interval,

        'hello_holdtime': hello_holdtime,

    }

    return str_path, str_data



def connection_point(ni_name: str, cp_id: str) -> Tuple[str, Dict]:

    str_path = '/network_instance[{:s}]/connection_point[{:s}]'.format(ni_name, cp_id)

    str_data = {'name': ni_name, 'connection_point_id': cp_id}

    return str_path, str_data



def connection_point_endpoint_local(

    ni_name: str, cp_id: str, ep_id: str, if_name: str, subif: int = 0, precedence: int = 0, site_id: int = None

) -> Tuple[str, Dict]:

    str_path = '/network_instance[{:s}]/connection_point[{:s}]/endpoint[{:s}]'.format(ni_name, cp_id, ep_id)

    str_data = {

        'name': ni_name,

        'connection_point_id': cp_id,

        'endpoint_id': ep_id,

        'type': 'LOCAL',

        'precedence': precedence,

        'interface': if_name,

        'subinterface': subif,

        'site_id': site_id,

    }

    return str_path, str_data



def connection_point_endpoint_remote(

    ni_name: str, cp_id: str, ep_id: str, remote_system: str, vc_id: int,

    precedence: int = 0, site_id: int = None

) -> Tuple[str, Dict]:

    str_path = '/network_instance[{:s}]/connection_point[{:s}]/endpoint[{:s}]'.format(ni_name, cp_id, ep_id)

    str_data = {

        'name': ni_name,

        'connection_point_id': cp_id,

        'endpoint_id': ep_id,

        'type': 'REMOTE',

        'precedence': precedence,

        'remote_system': remote_system,

        'virtual_circuit_id': vc_id,

        'site_id': site_id,

    }

    return str_path, str_data



def vlan(ni_name: str, vlan_id: int, members=None, vlan_name: str = None) -> Tuple[str, Dict]:

    if members is None:

        members = []

    str_path = '/network_instance[{:s}]/vlan[{:d}]'.format(ni_name, vlan_id)

    str_data = {

        'name': ni_name,

        'vlan_id': vlan_id,

        'vlan_name': vlan_name,

        'members': members,

    }

    return str_path, str_data