10080664191

Committed 24 Jul 2024 04:44PM UTC coverage: 79.076% (-0.1%) from 79.205%

Build # 10080664191

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #132 from atlanticwave-sdx/update/topo_spec_2

Update datamodel topology files according to Spec 2.0.0

Run Details

690 of 900 branches covered (76.67%)

Branch coverage included in aggregate %.

12 of 14 new or added lines in 1 file covered. (85.71%)

1 existing line in 1 file now uncovered.

1313 of 1633 relevant lines covered (80.4%)

3.21 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.7

/src/sdx_datamodel/parsing/porthandler.py

import json
import logging
import pathlib
from os import PathLike
from typing import List, Union

from sdx_datamodel.models.port import Port
from sdx_datamodel.models.service import Service
from sdx_datamodel.parsing.exceptions import (
    InvalidVlanRangeException,
    MissingAttributeException,
)


class PortHandler:
    """
    Handler for parsing the connection request descritpion in JSON.
    """

    def __init__(self):
        self._logger = logging.getLogger(__name__)

    def import_port_data(self, data: dict) -> Port:
        try:
            id = data["id"]
            name = data["name"]

            # node, short_name, label_range, and private_attributes
            # are allowed to be None.
            node = data.get("node")
            short_name = data.get("short_name")
            label_range = data.get("label_range")
            status = data.get("status")
            state = data.get("state")
            private_attributes = data.get("private_attributes")
            nni = data.get("nni")
            # L2VPN services are optional.
            #
            # TODO: actually use the services value.  We might want to
            # pass it on to Port instance we create, so Port instance
            # may need a services field.  Port class is generated from
            # SDX-LC's OpenAPI spec, so the spec will have to be
            # updated.
            services = self._validate_l2vpn_services(
                services=data.get("services"), port_id=id
            )

        except KeyError as e:
            raise MissingAttributeException(data, e.args[0])

        return Port(
            id=id,
            name=name,
            short_name=short_name,
            node=node,
            label_range=label_range,
            status=status,
            state=state,
            nni=nni,
            services=services,
            private_attributes=private_attributes,
        )

    def _validate_l2vpn_services(self, services: Union[dict, None], port_id):
        """
        Validate any "service" attached to a port definition.

        This method is intended to cover L2VPN services specified in
        SDX Topology Data Model spec 2.0.0.

        https://docs.google.com/document/d/1lgxjIT144EFu1G_OVcU19hN1cSUT_v2-tE0Z-7UlkNg/view#

        There's an existing "domain service" also (see models.Service
        and SDX-LC OpenAPI spec), which seems to conflict with L2VPN
        service.
        """
        if not services:
            self._logger.warning(f"No services defined in port '{port_id}'")
            return None

        if not isinstance(services, dict):
            self._logger.warning(
                f"Service {services} is not a dict in {port_id}"
            )
            return None

        l2vpn_ptp = {}
        l2vpn_ptmp = {}
        if services and services.get("l2vpn-ptp"):
            vlan_range = services.get("l2vpn-ptp").get("vlan_range")
            l2vpn_ptp_vlan_range = self._validate_vlan_range(vlan_range)
            l2vpn_ptp["vlan_range"] = l2vpn_ptp_vlan_range
        else:
            l2vpn_ptp_vlan_range = None

        if isinstance(services, dict) and services.get("l2vpn-ptmp"):
            vlan_range = services.get("l2vpn-ptmp").get("vlan_range")
            l2vpn_ptmp_vlan_range = self._validate_vlan_range(vlan_range)
            l2vpn_ptmp["vlan_range"] = l2vpn_ptmp_vlan_range
        else:
            l2vpn_ptmp_vlan_range = None

        self._logger.info(
            f"Found l2vpn_ptp_vlan_range: {l2vpn_ptp_vlan_range}, "
            f"l2vpn_ptmp_vlan_range: {l2vpn_ptmp_vlan_range} "
            f"in port '{port_id}'"
        )

        # TODO: Perhaps return a Service, or maybe a L2VPN Service?
        # The models.Service class seems to refer to domain services
        # specifically, and it is possibly generated from SDX-LC's
        # OpenAPI spec.  We need to find a way to clear up things.

        return Service(l2vpn_ptp=l2vpn_ptp, l2vpn_ptmp=l2vpn_ptmp)

        return l2vpn_ptp, l2vpn_ptmp

    def _validate_vlan_range(self, vlan_range: list) -> List[List[int]]:
        """
        Parse VLAN ranges.

        VLAN range is of the format [[1, 100], [300, 305], 200], ie.,
        it is a list whose elements are either numbers or number
        pairs.  Raise an exception when they are not of that form.
        """
        if not vlan_range or not isinstance(vlan_range, list):
            raise InvalidVlanRangeException(
                f"VLAN range ({vlan_range}) must be a list, but is {type(vlan_range)}"
            )

        for item in vlan_range:
            if not isinstance(item, list):
                if not isinstance(item, int):
                    raise InvalidVlanRangeException(
                        f"VLAN range must be a list or a number, but {item} is not"
                    )

            if isinstance(item, list):
                if len(item) != 2:
                    raise InvalidVlanRangeException(
                        f"VLAN range {item} is not a list of 2 numbers"
                    )

                if not all(isinstance(vlan, int) for vlan in item):
                    raise InvalidVlanRangeException(
                        f"VLAN ranges in {item} must be numbers, but it is not"
                    )

                if item[0] >= item[1]:
                    raise InvalidVlanRangeException(
                        f"VLAN range {item} is invalid: {item[0]} >= {item[1]}"
                    )

        return vlan_range

    def import_port(
        self, path: Union[str, bytes, PathLike]
    ) -> Union[Port, None]:
        """
        Import port data from a JSON file.
        """
        data = json.loads(pathlib.Path(path).read_text())
        return self.import_port_data(data)

1	import json	4✔
2	import logging	4✔
3	import pathlib	4✔
4	from os import PathLike	4✔
5	from typing import List, Union	4✔
6
7	from sdx_datamodel.models.port import Port	4✔
8	from sdx_datamodel.models.service import Service	4✔
9	from sdx_datamodel.parsing.exceptions import (	4✔
10	InvalidVlanRangeException,
11	MissingAttributeException,
12	)
13
14
15	class PortHandler:	4✔
16	"""	4✔
17	Handler for parsing the connection request descritpion in JSON.
18	"""
19
20	def __init__(self):	4✔
21	self._logger = logging.getLogger(__name__)	4✔
22
23	def import_port_data(self, data: dict) -> Port:	4✔
24	try:	4✔
25	id = data["id"]	4✔
26	name = data["name"]	4✔
27
28	# node, short_name, label_range, and private_attributes
29	# are allowed to be None.
30	node = data.get("node")	4✔
31	short_name = data.get("short_name")	4✔
32	label_range = data.get("label_range")	4✔
33	status = data.get("status")	4✔
34	state = data.get("state")	4✔
35	private_attributes = data.get("private_attributes")	4✔
36	nni = data.get("nni")	4✔
37	# L2VPN services are optional.
38	#
39	# TODO: actually use the services value. We might want to
40	# pass it on to Port instance we create, so Port instance
41	# may need a services field. Port class is generated from
42	# SDX-LC's OpenAPI spec, so the spec will have to be
43	# updated.
44	services = self._validate_l2vpn_services(	4✔
45	services=data.get("services"), port_id=id
46	)
47
48	except KeyError as e:	4✔
49	raise MissingAttributeException(data, e.args[0])	×
50
51	return Port(	4✔
52	id=id,
53	name=name,
54	short_name=short_name,
55	node=node,
56	label_range=label_range,
57	status=status,
58	state=state,
59	nni=nni,
60	services=services,
61	private_attributes=private_attributes,
62	)
63
64	def _validate_l2vpn_services(self, services: Union[dict, None], port_id):	4✔
65	"""
66	Validate any "service" attached to a port definition.
67
68	This method is intended to cover L2VPN services specified in
69	SDX Topology Data Model spec 2.0.0.
70
71	https://docs.google.com/document/d/1lgxjIT144EFu1G_OVcU19hN1cSUT_v2-tE0Z-7UlkNg/view#
72
73	There's an existing "domain service" also (see models.Service
74	and SDX-LC OpenAPI spec), which seems to conflict with L2VPN
75	service.
76	"""
77	if not services:	4✔
78	self._logger.warning(f"No services defined in port '{port_id}'")	4✔
79	return None	4✔
80
81	if not isinstance(services, dict):	4✔
82	self._logger.warning(	4✔
83	f"Service {services} is not a dict in {port_id}"
84	)
85	return None	4✔
86
87	l2vpn_ptp = {}	4✔
88	l2vpn_ptmp = {}	4✔
89	if services and services.get("l2vpn-ptp"):	4✔
90	vlan_range = services.get("l2vpn-ptp").get("vlan_range")	4✔
91	l2vpn_ptp_vlan_range = self._validate_vlan_range(vlan_range)	4✔
92	l2vpn_ptp["vlan_range"] = l2vpn_ptp_vlan_range	4✔
93	else:
94	l2vpn_ptp_vlan_range = None	×
95
96	if isinstance(services, dict) and services.get("l2vpn-ptmp"):	4✔
97	vlan_range = services.get("l2vpn-ptmp").get("vlan_range")	4✔
98	l2vpn_ptmp_vlan_range = self._validate_vlan_range(vlan_range)	4✔
99	l2vpn_ptmp["vlan_range"] = l2vpn_ptmp_vlan_range	4✔
100	else:
101	l2vpn_ptmp_vlan_range = None	4✔
102
103	self._logger.info(	4✔
104	f"Found l2vpn_ptp_vlan_range: {l2vpn_ptp_vlan_range}, "
105	f"l2vpn_ptmp_vlan_range: {l2vpn_ptmp_vlan_range} "
106	f"in port '{port_id}'"
107	)
108
109	# TODO: Perhaps return a Service, or maybe a L2VPN Service?
110	# The models.Service class seems to refer to domain services
111	# specifically, and it is possibly generated from SDX-LC's
112	# OpenAPI spec. We need to find a way to clear up things.
113
114	return Service(l2vpn_ptp=l2vpn_ptp, l2vpn_ptmp=l2vpn_ptmp)	4✔
115
116	return l2vpn_ptp, l2vpn_ptmp
117
118	def _validate_vlan_range(self, vlan_range: list) -> List[List[int]]:	4✔
119	"""
120	Parse VLAN ranges.
121
122	VLAN range is of the format [[1, 100], [300, 305], 200], ie.,
123	it is a list whose elements are either numbers or number
124	pairs. Raise an exception when they are not of that form.
125	"""
126	if not vlan_range or not isinstance(vlan_range, list):	4✔
127	raise InvalidVlanRangeException(	×
128	f"VLAN range ({vlan_range}) must be a list, but is {type(vlan_range)}"
129	)
130
131	for item in vlan_range:	4!
132	if not isinstance(item, list):	4✔
UNCOV 133	if not isinstance(item, int):	×
134	raise InvalidVlanRangeException(	×
135	f"VLAN range must be a list or a number, but {item} is not"
136	)
137
138	if isinstance(item, list):	4!
139	if len(item) != 2:	4✔
140	raise InvalidVlanRangeException(	×
141	f"VLAN range {item} is not a list of 2 numbers"
142	)
143
144	if not all(isinstance(vlan, int) for vlan in item):	4✔
145	raise InvalidVlanRangeException(	×
146	f"VLAN ranges in {item} must be numbers, but it is not"
147	)
148
149	if item[0] >= item[1]:	4✔
150	raise InvalidVlanRangeException(	4✔
151	f"VLAN range {item} is invalid: {item[0]} >= {item[1]}"
152	)
153
154	return vlan_range	4✔
155
156	def import_port(	4✔
157	self, path: Union[str, bytes, PathLike]
158	) -> Union[Port, None]:
159	"""
160	Import port data from a JSON file.
161	"""
162	data = json.loads(pathlib.Path(path).read_text())	4✔
163	return self.import_port_data(data)	4✔

atlanticwave-sdx / datamodel / 10080664191

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous