diff --git a/pyproject.toml b/pyproject.toml index 9e98629..3d3e24c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,10 +23,10 @@ classifiers = [ dependencies = [ "grenml", "networkx", + "pydantic[email]", "transitions", "fastapi", "pytz", - "importlib-resources; python_version < '3.9'", ] [project.urls] diff --git a/src/sdx_datamodel/data/requests/test-l2vpn-p2p-v2.json b/src/sdx_datamodel/data/requests/test-l2vpn-p2p-v2.json deleted file mode 100644 index a593471..0000000 --- a/src/sdx_datamodel/data/requests/test-l2vpn-p2p-v2.json +++ /dev/null @@ -1,34 +0,0 @@ -{ - "name": "new-connection", - "id": "urn:sdx:connection:ampath.net:Ampath3:50-sax.net:Sax01:41", - "endpoints": [ - { - "port_id": "urn:sdx:port:ampath.net:Ampath3:50", - "vlan": "777" - }, - { - "port_id": "urn:sdx:port:sax.net:Sax01:41", - "vlan": "55:90" - } - ], - "description": "a test circuit", - "qos_metrics": { - "min_bw": { - "value": 12, - "strict": true - }, - "max_delay": { - "value": 50, - "strict": false - }, - "max_number_oxps": { - "value": 7, - "strict": true - } - }, - "notifications": [ - { - "email": "muhaziz@fiu.edu" - } - ] - } \ No newline at end of file diff --git a/src/sdx_datamodel/data/requests/test_request-amlight_zaoxi-p2p-v2.json b/src/sdx_datamodel/data/requests/test_request-amlight_zaoxi-p2p-v2.json deleted file mode 100644 index fbb1b48..0000000 --- a/src/sdx_datamodel/data/requests/test_request-amlight_zaoxi-p2p-v2.json +++ /dev/null @@ -1,34 +0,0 @@ -{ - "name": "User port connection", - "id": "urn:sdx:connection:ampath.net:Ampath01:3-zaoxi.net:zaoxi02:1", - "endpoints": [ - { - "port_id": "urn:sdx:port:amlight.net:A1:3", - "vlan": "150" - }, - { - "port_id": "urn:sdx:port:zaoxi:B2:1", - "vlan": "150" - } - ], - "description": "a test circuit", - "qos_metrics": { - "min_bw": { - "value": 12, - "strict": true - }, - "max_delay": { - "value": 50, - "strict": false - }, - "max_number_oxps": { - "value": 7, - "strict": true - } - }, - "notifications": [ - { - "email": "muhaziz@fiu.edu" - } - ] - } \ No newline at end of file diff --git a/src/sdx_datamodel/data/requests/test_request.json b/src/sdx_datamodel/data/requests/v0/test_request.json similarity index 100% rename from src/sdx_datamodel/data/requests/test_request.json rename to src/sdx_datamodel/data/requests/v0/test_request.json diff --git a/src/sdx_datamodel/data/requests/test_request_no_node.json b/src/sdx_datamodel/data/requests/v0/test_request_no_node.json similarity index 100% rename from src/sdx_datamodel/data/requests/test_request_no_node.json rename to src/sdx_datamodel/data/requests/v0/test_request_no_node.json diff --git a/src/sdx_datamodel/data/requests/test_request_p2p.json b/src/sdx_datamodel/data/requests/v0/test_request_p2p.json similarity index 100% rename from src/sdx_datamodel/data/requests/test_request_p2p.json rename to src/sdx_datamodel/data/requests/v0/test_request_p2p.json diff --git a/src/sdx_datamodel/data/requests/v1.0/test-request-amlight-zaoxi-p2p.json b/src/sdx_datamodel/data/requests/v1.0/test-request-amlight-zaoxi-p2p.json new file mode 100644 index 0000000..a760e53 --- /dev/null +++ b/src/sdx_datamodel/data/requests/v1.0/test-request-amlight-zaoxi-p2p.json @@ -0,0 +1,38 @@ +{ + "name": "new-connection", + "id": "urn:sdx:connection:ampath.net:Ampath3:50-sax.net:Sax01:41", + "endpoints": [ + { + "port_id": "urn:sdx:port:amlight.net:A1:3", + "vlan": "101" + }, + { + "port_id": "urn:sdx:port:zaoxi:B2:1", + "vlan": "101" + } + ], + "description": "a test circuit", + "scheduling": { + "start_time": "2024-06-24T01:00:00.000Z", + "end_time": "2024-06-26T01:00:00.000Z" + }, + "qos_metrics": { + "min_bw": { + "value": 12, + "strict": true + }, + "max_delay": { + "value": 50, + "strict": false + }, + "max_number_oxps": { + "value": 7, + "strict": true + } + }, + "notifications": [ + { + "email": "muhaziz@fiu.edu" + } + ] +} diff --git a/src/sdx_datamodel/data/requests/test-request-amlight_sax-p2p-v2.json b/src/sdx_datamodel/data/requests/v1.0/test-request-amlight_sax-p2p-v2.json similarity index 100% rename from src/sdx_datamodel/data/requests/test-request-amlight_sax-p2p-v2.json rename to src/sdx_datamodel/data/requests/v1.0/test-request-amlight_sax-p2p-v2.json diff --git a/src/sdx_datamodel/data/requests/v1.0/test-request-l2vpn-p2p.json b/src/sdx_datamodel/data/requests/v1.0/test-request-l2vpn-p2p.json new file mode 100644 index 0000000..aae0ed1 --- /dev/null +++ b/src/sdx_datamodel/data/requests/v1.0/test-request-l2vpn-p2p.json @@ -0,0 +1,38 @@ +{ + "name": "new-connection", + "id": "urn:sdx:connection:ampath.net:Ampath3:50-sax.net:Sax01:41", + "endpoints": [ + { + "port_id": "urn:sdx:port:ampath.net:Ampath3:50", + "vlan": "100" + }, + { + "port_id": "urn:sdx:port:sax.net:Sax01:41", + "vlan": "100" + } + ], + "description": "a test circuit", + "scheduling": { + "start_time": "2024-06-24T01:00:00.000Z", + "end_time": "2024-06-26T01:00:00.000Z" + }, + "qos_metrics": { + "min_bw": { + "value": 12, + "strict": true + }, + "max_delay": { + "value": 4, + "strict": false + }, + "max_number_oxps": { + "value": 7, + "strict": true + } + }, + "notifications": [ + { + "email": "muhaziz@fiu.edu" + } + ] +} diff --git a/src/sdx_datamodel/data/requests/v1.0/test-request-vlan-translation.json b/src/sdx_datamodel/data/requests/v1.0/test-request-vlan-translation.json new file mode 100644 index 0000000..c0c6ba3 --- /dev/null +++ b/src/sdx_datamodel/data/requests/v1.0/test-request-vlan-translation.json @@ -0,0 +1,13 @@ +{ + "name": "VLAN between AMPATH/300 and TENET/150", + "endpoints": [ + { + "port_id": "urn:sdx:port:tenet.ac.za:Tenet03:50", + "vlan": "150" + }, + { + "port_id": "urn:sdx:port:ampath.net:Ampath3:50", + "vlan": "300" + } + ] +} diff --git a/src/sdx_datamodel/models/__init__.py b/src/sdx_datamodel/models/__init__.py index bcbee5c..84093e7 100644 --- a/src/sdx_datamodel/models/__init__.py +++ b/src/sdx_datamodel/models/__init__.py @@ -4,7 +4,6 @@ from __future__ import absolute_import # import models into model package -from sdx_datamodel.models.connection import Connection from sdx_datamodel.models.link import Link from sdx_datamodel.models.link_measurement_period import LinkMeasurementPeriod from sdx_datamodel.models.location import Location diff --git a/src/sdx_datamodel/models/connection.py b/src/sdx_datamodel/models/connection.py deleted file mode 100644 index e213c19..0000000 --- a/src/sdx_datamodel/models/connection.py +++ /dev/null @@ -1,768 +0,0 @@ -# coding: utf-8 - -from __future__ import absolute_import - -from datetime import date, datetime # noqa: F401 -from typing import Dict, List # noqa: F401 - -from sdx_datamodel import util -from sdx_datamodel.models.base_model_ import Model -from sdx_datamodel.models.link import Link # noqa: F401,E501 -from sdx_datamodel.models.port import Port # noqa: F401,E501 - - -class Connection(Model): - """NOTE: This class is auto generated by the swagger code generator program. - - Do not edit the class manually. - """ - - def __init__( - self, - id=None, - name=None, - ingress_port=None, - egress_port=None, - quantity=None, - start_time=None, - end_time=None, - multi_path=None, - preempt=None, - backup_path_type=None, - exclusive_links=None, - inclusive_links=None, - bandwidth_required=None, - bandwidth_measured=None, - latency_required=None, - latency_measured=None, - packetloss_required=None, - packetloss_measured=None, - availability_required=None, - availability_measured=None, - max_number_oxps=None, - paths=None, - status=None, - complete=False, - ): # noqa: E501 - """Connection - a model defined in Swagger - - :param id: The id of this Connection. # noqa: E501 - :type id: str - :param name: The name of this Connection. # noqa: E501 - :type name: str - :param ingress_port: The ingress_port of this Connection. # noqa: E501 - :type ingress_port: Port - :param egress_port: The egress_port of this Connection. # noqa: E501 - :type egress_port: Port - :param quantity: The quantity of this Connection. # noqa: E501 - :type quantity: int - :param start_time: The start_time of this Connection. # noqa: E501 - :type start_time: datetime - :param end_time: The end_time of this Connection. # noqa: E501 - :type end_time: datetime - :param multi_path: The multi_path of this Connection. # noqa: E501 - :type multi_path: bool - :param preempt: The preempt of this Connection. # noqa: E501 - :type preempt: bool - :param backup_path_type: The backup_path_type of this Connection. # noqa: E501 - :type backup_path_type: str - :param exclusive_links: The exclusive_links of this Connection. # noqa: E501 - :type exclusive_links: List[Link] - :param inclusive_links: The inclusive_links of this Connection. # noqa: E501 - :type inclusive_links: List[Link] - :param bandwidth_required: The bandwidth_required of this Connection. # noqa: E501 - :type bandwidth_required: float - :param bandwidth_measured: The bandwidth_measured of this Connection. # noqa: E501 - :type bandwidth_measured: float - :param latency_required: The latency_required of this Connection. # noqa: E501 - :type latency_required: float - :param latency_measured: The latency_measured of this Connection. # noqa: E501 - :type latency_measured: float - :param packetloss_required: The packetloss_required of this Connection. # noqa: E501 - :type packetloss_required: float - :param packetloss_measured: The packetloss_measured of this Connection. # noqa: E501 - :type packetloss_measured: float - :param availability_required: The availability_required of this Connection. # noqa: E501 - :type availability_required: float - :param availability_measured: The availability_measured of this Connection. # noqa: E501 - :type availability_measured: float - :param paths: The paths of this Connection. # noqa: E501 - :type paths: List[str] - :param status: The status of this Connection. # noqa: E501 - :type status: str - :param complete: The complete of this Connection. # noqa: E501 - :type complete: bool - """ - self.swagger_types = { - "id": str, - "name": str, - "ingress_port": Port, - "egress_port": Port, - "quantity": int, - "start_time": datetime, - "end_time": datetime, - "multi_path": bool, - "preempt": bool, - "backup_path_type": str, - "exclusive_links": List[Link], - "inclusive_links": List[Link], - "bandwidth_required": float, - "bandwidth_measured": float, - "latency_required": float, - "latency_measured": float, - "packetloss_required": float, - "packetloss_measured": float, - "availability_required": float, - "availability_measured": float, - "max_number_oxps": int, - "paths": List[str], - "status": str, - "complete": bool, - } - - self.attribute_map = { - "id": "id", - "name": "name", - "ingress_port": "ingress_port", - "egress_port": "egress_port", - "quantity": "quantity", - "start_time": "start_time", - "end_time": "end_time", - "multi_path": "multi_path", - "preempt": "preempt", - "backup_path_type": "backup_path_type", - "exclusive_links": "exclusive_links", - "inclusive_links": "inclusive_links", - "bandwidth_required": "bandwidth_required", - "bandwidth_measured": "bandwidth_measured", - "latency_required": "latency_required", - "latency_measured": "latency_measured", - "packetloss_required": "packetloss_required", - "packetloss_measured": "packetloss_measured", - "availability_required": "availability_required", - "availability_measured": "availability_measured", - "max_number_oxps": "max_number_oxps", - "paths": "paths", - "status": "status", - "complete": "complete", - } - self._id = id - self._name = name - self._ingress_port = ingress_port - self._egress_port = egress_port - self._quantity = quantity - self._start_time = start_time - self._end_time = end_time - self._multi_path = multi_path - self._preempt = preempt - self._backup_path_type = backup_path_type - self._exclusive_links = exclusive_links - self._inclusive_links = inclusive_links - self._bandwidth_required = bandwidth_required - self._bandwidth_measured = bandwidth_measured - self._latency_required = latency_required - self._latency_measured = latency_measured - self._packetloss_required = packetloss_required - self._packetloss_measured = packetloss_measured - self._availability_required = availability_required - self._availability_measured = availability_measured - self._max_number_oxps = max_number_oxps - self._paths = paths - self._status = status - self._complete = complete - - @classmethod - def from_dict(cls, dikt): - """Returns the dict as a model - - :param dikt: A dict. - :type: dict - :return: The connection of this Connection. # noqa: E501 - :rtype: Connection - """ - return util.deserialize_model(dikt, cls) - - @property - def id(self): - """Gets the id of this Connection. - - - :return: The id of this Connection. - :rtype: str - """ - return self._id - - @id.setter - def id(self, id): - """Sets the id of this Connection. - - - :param id: The id of this Connection. - :type id: str - """ - if id is None: - raise ValueError( - "Invalid value for `id`, must not be `None`" - ) # noqa: E501 - - self._id = id - - @property - def name(self): - """Gets the name of this Connection. - - - :return: The name of this Connection. - :rtype: str - """ - return self._name - - @name.setter - def name(self, name): - """Sets the name of this Connection. - - - :param name: The name of this Connection. - :type name: str - """ - if name is None: - raise ValueError( - "Invalid value for `name`, must not be `None`" - ) # noqa: E501 - - self._name = name - - @property - def ingress_port(self): - """Gets the ingress_port of this Connection. - - - :return: The ingress_port of this Connection. - :rtype: Port - """ - return self._ingress_port - - @ingress_port.setter - def ingress_port(self, ingress_port): - """Sets the ingress_port of this Connection. - - - :param ingress_port: The ingress_port of this Connection. - :type ingress_port: Port - """ - if ingress_port is None: - raise ValueError( - "Invalid value for `ingress_port`, must not be `None`" - ) # noqa: E501 - - self._ingress_port = ingress_port - - # port setter - def set_ingress_port(self, ingress_port: Port) -> Port: - """ - Set ingress_port for this Connection. - - :param ingress_port: The ingress_port of this Connection. - :type: Port - """ - if ingress_port is None: - raise ValueError( - "Invalid value for `ingress_port`: must not be `None`" - ) - - # TODO: use tooling (mypy, pytypes, etc) to enforce types. - if not isinstance(ingress_port, Port): - raise TypeError( - "Invalid type for `ingress_port`: must be of type `Port`" - ) - - self._ingress_port = ingress_port - return self._ingress_port - - @property - def egress_port(self): - """Gets the egress_port of this Connection. - - - :return: The egress_port of this Connection. - :rtype: Port - """ - return self._egress_port - - @egress_port.setter - def egress_port(self, egress_port): - """Sets the egress_port of this Connection. - - - :param egress_port: The egress_port of this Connection. - :type egress_port: Port - """ - if egress_port is None: - raise ValueError( - "Invalid value for `egress_port`, must not be `None`" - ) # noqa: E501 - - self._egress_port = egress_port - - # port setter - def set_egress_port(self, egress_port: Port) -> Port: - """ - Set egress_port for this Connection. - - :param egress_port: The egress_port of this Connection. - :type: Port - """ - if egress_port is None: - raise ValueError( - "Invalid value for `egress_port`: must not be `None`" - ) - - # TODO: use tooling (mypy, pytypes, etc) to enforce types. - if not isinstance(egress_port, Port): - raise TypeError( - "Invalid type for `egress_port`, must be of type `Port`" - ) - - self._egress_port = egress_port - return self._egress_port - - @property - def quantity(self): - """Gets the quantity of this Connection. - - - :return: The quantity of this Connection. - :rtype: int - """ - return self._quantity - - @quantity.setter - def quantity(self, quantity): - """Sets the quantity of this Connection. - - - :param quantity: The quantity of this Connection. - :type quantity: int - """ - - self._quantity = quantity - - @property - def start_time(self): - """Gets the start_time of this Connection. - - - :return: The start_time of this Connection. - :rtype: datetime - """ - return self._start_time - - @start_time.setter - def start_time(self, start_time): - """Sets the start_time of this Connection. - - - :param start_time: The start_time of this Connection. - :type start_time: datetime - """ - - self._start_time = start_time - - @property - def end_time(self): - """Gets the end_time of this Connection. - - - :return: The end_time of this Connection. - :rtype: datetime - """ - return self._end_time - - @end_time.setter - def end_time(self, end_time): - """Sets the end_time of this Connection. - - - :param end_time: The end_time of this Connection. - :type end_time: datetime - """ - - self._end_time = end_time - - @property - def multi_path(self): - """Gets the multi_path of this Connection. - - - :return: The multi_path of this Connection. - :rtype: bool - """ - return self._multi_path - - @multi_path.setter - def multi_path(self, multi_path): - """Sets the multi_path of this Connection. - - - :param multi_path: The multi_path of this Connection. - :type multi_path: bool - """ - - self._multi_path = multi_path - - @property - def preempt(self): - """Gets the preempt of this Connection. - - - :return: The preempt of this Connection. - :rtype: bool - """ - return self._preempt - - @preempt.setter - def preempt(self, preempt): - """Sets the preempt of this Connection. - - - :param preempt: The preempt of this Connection. - :type preempt: bool - """ - - self._preempt = preempt - - @property - def backup_path_type(self): - """Gets the backup_path_type of this Connection. - - - :return: The backup_path_type of this Connection. - :rtype: str - """ - return self._backup_path_type - - @backup_path_type.setter - def backup_path_type(self, backup_path_type): - """Sets the backup_path_type of this Connection. - - - :param backup_path_type: The backup_path_type of this Connection. - :type backup_path_type: str - """ - allowed_values = ["0", "1", "2", "3"] # noqa: E501 - if backup_path_type not in allowed_values: - raise ValueError( - "Invalid value for `backup_path_type` ({0}), must be one of {1}".format( - backup_path_type, allowed_values - ) - ) - - self._backup_path_type = backup_path_type - - @property - def exclusive_links(self): - """Gets the exclusive_links of this Connection. - - - :return: The exclusive_links of this Connection. - :rtype: List[Link] - """ - return self._exclusive_links - - @exclusive_links.setter - def exclusive_links(self, exclusive_links): - """Sets the exclusive_links of this Connection. - - - :param exclusive_links: The exclusive_links of this Connection. - :type exclusive_links: List[Link] - """ - - self._exclusive_links = exclusive_links - - @property - def inclusive_links(self): - """Gets the inclusive_links of this Connection. - - - :return: The inclusive_links of this Connection. - :rtype: List[Link] - """ - return self._inclusive_links - - @inclusive_links.setter - def inclusive_links(self, inclusive_links): - """Sets the inclusive_links of this Connection. - - - :param inclusive_links: The inclusive_links of this Connection. - :type inclusive_links: List[Link] - """ - - self._inclusive_links = inclusive_links - - @property - def bandwidth_required(self): - """Gets the bandwidth_required of this Connection. - - - :return: The bandwidth_required of this Connection. - :rtype: float - """ - return self._bandwidth_required - - @bandwidth_required.setter - def bandwidth_required(self, bandwidth_required): - """Sets the bandwidth_required of this Connection. - - - :param bandwidth_required: The bandwidth_required of this Connection. - :type bandwidth_required: float - """ - - self._bandwidth_required = bandwidth_required - - @property - def bandwidth_measured(self): - """Gets the bandwidth_measured of this Connection. - - - :return: The bandwidth_measured of this Connection. - :rtype: float - """ - return self._bandwidth_measured - - @bandwidth_measured.setter - def bandwidth_measured(self, bandwidth_measured): - """Sets the bandwidth_measured of this Connection. - - - :param bandwidth_measured: The bandwidth_measured of this Connection. - :type bandwidth_measured: float - """ - - self._bandwidth_measured = bandwidth_measured - - @property - def latency_required(self): - """Gets the latency_required of this Connection. - - - :return: The latency_required of this Connection. - :rtype: float - """ - return self._latency_required - - @latency_required.setter - def latency_required(self, latency_required): - """Sets the latency_required of this Connection. - - - :param latency_required: The latency_required of this Connection. - :type latency_required: float - """ - - self._latency_required = latency_required - - @property - def latency_measured(self): - """Gets the latency_measured of this Connection. - - - :return: The latency_measured of this Connection. - :rtype: float - """ - return self._latency_measured - - @latency_measured.setter - def latency_measured(self, latency_measured): - """Sets the latency_measured of this Connection. - - - :param latency_measured: The latency_measured of this Connection. - :type latency_measured: float - """ - - self._latency_measured = latency_measured - - @property - def packetloss_required(self): - """Gets the packetloss_required of this Connection. - - - :return: The packetloss_required of this Connection. - :rtype: float - """ - return self._packetloss_required - - @packetloss_required.setter - def packetloss_required(self, packetloss_required): - """Sets the packetloss_required of this Connection. - - - :param packetloss_required: The packetloss_required of this Connection. - :type packetloss_required: float - """ - - self._packetloss_required = packetloss_required - - @property - def packetloss_measured(self): - """Gets the packetloss_measured of this Connection. - - - :return: The packetloss_measured of this Connection. - :rtype: float - """ - return self._packetloss_measured - - @packetloss_measured.setter - def packetloss_measured(self, packetloss_measured): - """Sets the packetloss_measured of this Connection. - - - :param packetloss_measured: The packetloss_measured of this Connection. - :type packetloss_measured: float - """ - - self._packetloss_measured = packetloss_measured - - @property - def availability_required(self): - """Gets the availability_required of this Connection. - - - :return: The availability_required of this Connection. - :rtype: float - """ - return self._availability_required - - @availability_required.setter - def availability_required(self, availability_required): - """Sets the availability_required of this Connection. - - - :param availability_required: The availability_required of this Connection. - :type availability_required: float - """ - - self._availability_required = availability_required - - @property - def availability_measured(self): - """Gets the availability_measured of this Connection. - - - :return: The availability_measured of this Connection. - :rtype: float - """ - return self._availability_measured - - @availability_measured.setter - def availability_measured(self, availability_measured): - """Sets the availability_measured of this Connection. - - - :param availability_measured: The availability_measured of this Connection. - :type availability_measured: float - """ - - self._availability_measured = availability_measured - - @property - def max_number_oxps(self): - """Gets the max_number_oxps of this Connection. - - - :return: The max_number_oxps of this Connection. - :rtype: int - """ - return self._max_number_oxps - - @max_number_oxps.setter - def max_number_oxps(self, max_number_oxps): - """Sets the max_number_oxps of this Connection. - - - :param max_number_oxps: The max_number_oxps of this Connection. - :type max_number_oxps: int - """ - - self._max_number_oxps = max_number_oxps - - @property - def paths(self): - """Gets the paths of this Connection. - - - :return: The paths of this Connection. - :rtype: List[str] - """ - return self._paths - - @paths.setter - def paths(self, paths): - """Sets the paths of this Connection. - - - :param paths: The paths of this Connection. - :type paths: List[str] - """ - - self._paths = paths - - @property - def status(self): - """Gets the status of this Connection. - - Connection Status # noqa: E501 - - :return: The status of this Connection. - :rtype: str - """ - return self._status - - @status.setter - def status(self, status): - """Sets the status of this Connection. - - Connection Status # noqa: E501 - - :param status: The status of this Connection. - :type status: str - """ - allowed_values = [ - "success", - "fail", - "scheduled", - "provisioining", - ] # noqa: E501 - if status not in allowed_values: - raise ValueError( - "Invalid value for `status` ({0}), must be one of {1}".format( - status, allowed_values - ) - ) - - self._status = status - - @property - def complete(self): - """Gets the complete of this Connection. - - - :return: The complete of this Connection. - :rtype: bool - """ - return self._complete - - @complete.setter - def complete(self, complete): - """Sets the complete of this Connection. - - - :param complete: The complete of this Connection. - :type complete: bool - """ - - self._complete = complete diff --git a/src/sdx_datamodel/models/connection_request.py b/src/sdx_datamodel/models/connection_request.py new file mode 100644 index 0000000..d93e1ab --- /dev/null +++ b/src/sdx_datamodel/models/connection_request.py @@ -0,0 +1,332 @@ +# This module represents AW-SDX connection requests using Pydantic. +# Here we implement Service Provisioning Data Model Specification 1.0, +# available at https://sdx-docs.readthedocs.io. + +import math +import re +from datetime import datetime + +import pytz +from pydantic import ( + BaseModel, + EmailStr, + Field, + PositiveFloat, + PositiveInt, + RootModel, + computed_field, + field_validator, +) + +__all__ = [ + "Service", + "Port", + "ConnectionRequest", + "ConnectionRequestV1", + "ConnectionRequestV0", +] + +# Regular expression used for matching VLAN ranges like "100:200". +VLAN_RANGE_PATTERN = r"(\d+):(\d+)" + + +class EndPoint(BaseModel): + port_id: str = Field(frozen=True) + vlan: str = Field(frozen=True) + + @field_validator("vlan") + @classmethod + def validate_vlan(cls, value: str) -> str: + # an integer like "100" is valid, and it has to be in [1,4095] + # range. + if EndPoint.is_integer(value): + if int(value) not in range(1, 4095): + raise ValueError(f"vlan {value} is not in [1,4095] range") + return value + + # a range like "1:100" is valid. + match = re.match(VLAN_RANGE_PATTERN, value) + if match: + x = int(match.group(1)) + y = int(match.group(2)) + if x not in range(1, 4095): + raise ValueError(f"vlan {x} is invalid: not in [1,4095] range") + if y not in range(1, 4095): + raise ValueError(f"vlan {y} is invalid: not in [1,4095] range") + if x > y: + raise ValueError(f"vlan {value} is invalid: {x} > {y}") + # this range is probably okay. + return value + + # "any", "all", and "untagged" also are valid. + if value not in ("any", "all", "untagged"): + raise ValueError(f"VLAN {value} is not valid") + + # By now we should have have exhausted all possible checks; + # just return the value. + return value + + @classmethod + def is_integer(cls, value) -> bool: + try: + int(value) + return True + except ValueError: + return False + + +class NotificationEmail(BaseModel): + email: EmailStr = Field(frozen=True) + + +class Scheduling(BaseModel): + # TODO: use timestamp validation + start_time: datetime | None = Field(frozen=True, default=datetime.now()) + end_time: datetime | None = Field(frozen=True, default=None) + + @field_validator("start_time", "end_time", mode="before") + def parse_datetime(cls, value): + """Convert ISO8601 string to datetime if it's a string.""" + if isinstance(value, str): + if value == "": + # FIXME: this is a workaround. One of the test files have + # start_time = "" and end_time = "", which does not seem + # to conform to provisioniong spec. Both start_time and + # end_time have to be either a timestamp, or absent. + return None + try: + # Parse ISO8601 string and ensure it's timezone-aware + dt = datetime.fromisoformat(value.replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = pytz.UTC.localize(dt) + return dt + except ValueError: + raise ValueError("Invalid ISO8601 datetime format") + return value + + @field_validator("end_time") + def check_time_relationship(cls, end_time, info): + """Validate that end_time is greater than start_time when both are present.""" + start_time = info.data.get("start_time") + if end_time is not None and start_time is not None: + if end_time <= start_time: + raise ValueError("end_time must be greater than start_time") + return end_time + + +class MinimumBandwidth(BaseModel): + value: PositiveInt = Field(frozen=True, default=0) + strict: bool = Field(frozen=True, default=False) + + +class MaximumDelay(BaseModel): + value: PositiveInt = Field(frozen=True, default=math.inf) + strict: bool = Field(frozen=True, default=False) + + +class MaximumOXP(BaseModel): + value: PositiveInt = Field(frozen=True, default=math.inf) + strict: bool = Field(frozen=True, default=False) + + +class QoSMetrics(BaseModel): + min_bw: MinimumBandwidth | None = Field(frozen=True, default=None) + max_delay: MaximumDelay | None = Field(frozen=True, default=None) + max_number_oxps: MaximumOXP | None = Field(frozen=True, default=None) + + +class Service(BaseModel): + # The l2vpn_ptp of this Service. + l2vpn_ptp: object = Field(frozen=True, default=None) + + # The l2vpn_ptmp of this Service. + l2vpn_ptmp: object = Field(frozen=True, default=None) + + # The monitoring_capability of this Service. + monitoring_capability: str = Field(frozen=True, default=None) + + # The owner of this Service. + owner: str = Field(frozen=True, default=None) + + # The private_attributes of this Service. + private_attributes: list[str] = Field(frozen=True, default=None) + + # The provisioning_system of this Service. + provisioning_system: str = Field(frozen=True, default=None) + + # The provisioning_url of this Service. + provisioning_url: str = Field(frozen=True, default=None) + + # The vendor of this Service. + vendor: list[str] = Field(frozen=True, default=None) + + +class Port(BaseModel): + # The id of this Port. + id: str = Field(frozen=True) + + # The name of this Port. + name: str = Field(frozen=True, default="unknown") + + # The entities of this Port. + entities: list[str] = Field(frozen=True, default=[]) + + # The short_name of this Port. + short_name: str = Field(frozen=True, default="") + + # The node of this Port. + node: str = Field(frozen=True, default=None) + + # The vlan range of this Port. + vlan_range: str | None = Field(frozen=True, default=None) + + # `label` and `label_range` are V0 artifacts. Should be safe to + # remove them when we remove v0 connection requests. See + # data/requests/v0/test_request_p2p.json for a usage example. + label: str = Field(frozen=True, default=None) + label_range: str = Field(frozen=True, default=None) + + # The status of this Port. + status: str = Field(frozen=True, default=None) + + # The state of this Port. + state: str = Field(frozen=True, default=None) + + # The nni of this Port. + nni: str = Field(frozen=True, default=None) + + # The technology/bandwidth of this Port. + type: str = Field(frozen=True, default=None) + + # The services of this Port. + services: Service = Field(frozen=True, default=None) + + # The private_attributes of this Port. + private_attributes: list[str] = Field(frozen=True, default=[]) + + +class ConnectionRequestV1(BaseModel): + # The `id` field is not in the SDX Provisioning Spec 1.0, but PCE + # needs it, and SDX-Controller generates it upon the arrival of a + # request. We need `id` to be an assignable field. + id: str | None = Field(frozen=False, default=None) + + name: str = Field(frozen=True) + endpoints: list[EndPoint] = Field(frozen=True) + + description: str | None = Field(frozen=True, default=None) + notifications: list[NotificationEmail] | None = Field( + frozen=True, default=None + ) + scheduling: Scheduling | None = Field(frozen=True, default=None) + qos_metrics: QoSMetrics | None = Field(frozen=True, default=None) + + # Add the properties that PCE needs. + @computed_field # type: ignore[prop-decorator] + @property + def ingress_port(self) -> Port: + ep = self.endpoints[0] + return Port(id=ep.port_id, vlan_range=ep.vlan) + + @computed_field # type: ignore[prop-decorator] + @property + def egress_port(self) -> Port: + ep = self.endpoints[-1] + return Port(id=ep.port_id, vlan_range=ep.vlan) + + @computed_field # type: ignore[prop-decorator] + @property + def bandwidth_required(self) -> PositiveInt: + if self.qos_metrics and self.qos_metrics.min_bw: + return self.qos_metrics.min_bw.value + return 0 + + @computed_field # type: ignore[prop-decorator] + @property + def latency_required(self) -> PositiveFloat: + if self.qos_metrics and self.qos_metrics.max_delay: + return self.qos_metrics.max_delay.value + return math.inf + + @computed_field # type: ignore[prop-decorator] + @property + def start_time(self) -> datetime | None: + if self.scheduling: + return self.scheduling.start_time + return None + + @computed_field # type: ignore[prop-decorator] + @property + def end_time(self) -> datetime | None: + if self.scheduling: + return self.scheduling.end_time + return None + + @computed_field # type: ignore[prop-decorator] + @property + def max_number_oxps(self) -> PositiveInt | None: + if self.qos_metrics and self.qos_metrics.max_number_oxps: + return self.qos_metrics.max_number_oxps.value + return None + + @field_validator("endpoints") + @classmethod + def validate_endpoints(cls, endpoints): + if len(endpoints) < 2: + raise ValueError(f"not enough endpoints in {endpoints}") + + vlans = [endpoint.vlan for endpoint in endpoints] + + # If one endpoint has the VLAN range or option “all”, all + # endpoints must have the same value. + if "all" in vlans and not all(map(lambda x: x == "all", vlans)): + raise ValueError( + f"all vlans requested, but not consistently: {vlans}" + ) + + def is_range(value): + """ + match a pattern like "100:200" + """ + return bool(re.match(VLAN_RANGE_PATTERN, value)) + + # When one endpoint has the VLAN range option in use, all + # other endpoint(s) must have the same VLAN range. + if any(map(lambda x: is_range(x), vlans)) and len(set(vlans)) != 1: + raise ValueError( + f"range of vlans requested, but not consistently: {vlans}" + ) + + return endpoints + + +class ConnectionRequestV0(BaseModel): + """ + Backward compatibility for original request format. + """ + + id: str = Field(frozen=True) + name: str = Field(frozen=True, default=None) + + ingress_port: Port = Field(frozen=True) + egress_port: Port = Field(frozen=True) + + start_time: datetime | None = Field(frozen=True, default=None) + end_time: datetime | None = Field(frozen=True, default=None) + + bandwidth_required: PositiveInt | None = Field(frozen=True, default=0) + latency_required: PositiveInt | None = Field(frozen=True, default=0) + + +class ConnectionRequest(RootModel): + """ + A convenience class to validate either version of the request. + """ + + root: ConnectionRequestV1 | ConnectionRequestV0 + + def __getattr__(self, name): + """ + Convenience method to access the root value directly. + """ + return getattr(self.root, name) diff --git a/src/sdx_datamodel/parsing/connectionhandler.py b/src/sdx_datamodel/parsing/connectionhandler.py index d227813..f35e201 100644 --- a/src/sdx_datamodel/parsing/connectionhandler.py +++ b/src/sdx_datamodel/parsing/connectionhandler.py @@ -1,6 +1,6 @@ import json -from sdx_datamodel.models.connection import Connection +from sdx_datamodel.models.connection_request import ConnectionRequest from sdx_datamodel.models.port import Port from sdx_datamodel.parsing.porthandler import PortHandler @@ -19,7 +19,7 @@ class ConnectionHandler: Python dicts. """ - def import_connection_data(self, data: dict) -> Connection: + def import_connection_data(self, data: dict) -> ConnectionRequest: """ Create a Connection from connection data encoded in a dict. @@ -80,7 +80,7 @@ def import_connection_data(self, data: dict) -> Connection: except ServiceNotSupportedException as e: raise e - return Connection( + return ConnectionRequest( id=id, name=name, start_time=start_time, @@ -92,7 +92,7 @@ def import_connection_data(self, data: dict) -> Connection: egress_port=egress_port, ) - def import_connection(self, path) -> Connection: + def import_connection(self, path) -> ConnectionRequest: """ Import connection descritpion from a file. diff --git a/src/sdx_datamodel/validation/connectionvalidator.py b/src/sdx_datamodel/validation/connectionvalidator.py deleted file mode 100644 index 7eac7a1..0000000 --- a/src/sdx_datamodel/validation/connectionvalidator.py +++ /dev/null @@ -1,321 +0,0 @@ -""""" -Checks for Connection objects to be in the expected format. -""" - -import logging -from datetime import datetime -from re import match - -import pytz - -from sdx_datamodel.models.connection import Connection -from sdx_datamodel.models.port import Port -from sdx_datamodel.parsing.exceptions import ( - AttributeNotSupportedException, - InvalidVlanRangeException, - MissingAttributeException, - ServiceNotSupportedException, -) - - -class ConnectionValidator: - """ - The validation class made to validate a Connection request - """ - - def __init__(self, connection): - if not isinstance(connection, Connection): - raise ValueError( - "ConnectionValidator must be passed a Connection object" - ) - - self._connection = connection - - self._logger = logging.getLogger(__name__) - - def is_valid(self) -> bool: - errors = self.validate(raise_error=True) - for error in errors: - self._logger.error(error) - return not bool(errors) - - def validate(self, raise_error=True) -> [str]: - errors = self._validate_connection(self._connection) - if errors and raise_error: - raise ValueError("\n".join(errors)) - return errors - - def _validate_connection(self, conn: Connection): - """ - Validate that the connection provided meets the JSON schema. - - A connection must have the following: - - - It must meet object standard - - - It must have the default fields: id, name, ingress_port, - and egress_port - - :param connection: The connection being evaluated - - :return: A list of any issues in the data. - """ - - errors = [] - errors += self._validate_object_defaults(conn) - errors += self._validate_port(conn.ingress_port, conn) - errors += self._validate_port(conn.egress_port, conn) - - if len(errors) > 0: - return errors - - errors += self._validate_connection_vlan( - conn.ingress_port.vlan_range, conn.egress_port.vlan_range - ) - - if conn.start_time or conn.end_time: - errors += self._validate_time(conn.start_time, conn.end_time, conn) - - if conn.latency_required: - errors += self._validate_qos_metrics_value( - "max_delay", conn.latency_required, 1000 - ) - - if conn.bandwidth_required: - errors += self._validate_qos_metrics_value( - "min_bw", conn.bandwidth_required, 100 - ) - - if conn.max_number_oxps: - errors += self._validate_qos_metrics_value( - "max_number_oxps", conn.max_number_oxps, 100 - ) - return errors - - def _validate_qos_metrics_value(self, metric, value, max_value): - """ - Validate that the QoS Metrics provided meets the XSD standards. - - A connection must have the following: - - - It must meet object default standards. - - - The max_delay must be a number - - - The max_number_oxps must be a number between 0 and 100 - - :param qos_metrics: The QoS Metrics being evaluated. - - :return: A list of any issues in the data. - """ - errors = [] - - if not isinstance(value, int): - errors.append( - f"Strict QoS requirements: {value} {metric} must be a number" - ) - if not (0 <= value <= max_value): - errors.append( - f"Strict QoS requirements: {value} {metric} must be between 0 and 1000" - ) - - return errors - - def _validate_port(self, port: Port, conn: Connection): - """ - Validate that the port provided meets the XSD standards. - A port must have the following: - - It must meet object default standards. - - A port must belong to a topology - - The node is optional - :param port: The port being evaluated. - :param topology: The Topology. - :return: A list of any issues in the data. - """ - - errors = [] - if not port: - errors.append(f"{port.__class__.__name__} must exist") - - errors += self._validate_object_defaults(port) - - if port.vlan_range is None: - errors.append( - f"{port.__class__.__name__} {port._name} must have a vlan" - ) - - """ - node = find_node(port,topology) - if topology and node not in topology.nodes: - errors.append( - 'listed node id {} does not exist in parent Topology {}'.format( - node.id, node, topology.id - ) - ) - """ - return errors - - def _validate_connection_vlan(self, ingress_vlan: str, egress_vlan: str): - """ - validate VLAN in connection request. - - VLAN is of the following: 1-4095, "100:200", "any", "all" or "untagged" - """ - errors = [] - if not isinstance(ingress_vlan, str): - errors.append( - f"VLAN ({ingress_vlan}) must be a str, but is {type(ingress_vlan)}" - ) - return errors - - if not isinstance(egress_vlan, str): - errors.append( - f"VLAN ({egress_vlan}) must be a str, but is {type(egress_vlan)}" - ) - return errors - - if ingress_vlan == "all" or egress_vlan == "all": - if ingress_vlan != "all" or egress_vlan != "all": - errors.append( - "Invalid VLAN: If one VLAN is 'all', the other must also be 'all'" - ) - return errors - - if ":" in ingress_vlan or ":" in egress_vlan: - if ingress_vlan != egress_vlan: - errors.append( - f"VLAN ranges must be equal: {ingress_vlan} != {egress_vlan}" - ) - return errors - - error = self._validate_vlan(ingress_vlan) - if error: - errors.append(error) - error = self._validate_vlan(egress_vlan) - if error: - errors.append(error) - - return errors - - def _validate_vlan(self, vlan: str): - if vlan == "any" or vlan == "untagged": - return None - - if ":" in vlan: - v1 = vlan.split(":")[0] - v2 = vlan.split(":")[1] - else: - v2 = v1 = vlan - - if not v1.isdigit() or not v2.isdigit(): - if v1 == v2: - error = f"VLAN range {vlan} is invalid: {v1} is not a number" - else: - error = f"VLAN range {vlan} is invalid: {v1} or {v2} is not a number" - return error - - v1 = int(v1) - v2 = int(v2) - - if v1 < 1 or v2 < 1 or v1 > 4095 or v2 > 4095: - if v1 == v2: - error = f"VLAN range {vlan} is invalid: {v1} is out of range (1-4095)" - else: - error = f"VLAN range {vlan} is invalid: {v1} or {v2} is out of range (1-4095)" - return error - - if v1 > v2: - error = f"VLAN range {vlan} is invalid: {v1} > {v2}" - return error - - def _validate_time(self, start_time: str, end_time: str, conn: Connection): - """ - Validate that the time provided meets the XSD standards. - - :param start_time, end_time: time being validated - - :return: A list of any issues in the data. - """ - utc = pytz.UTC - errors = [] - now = datetime.now().replace(tzinfo=utc) - if not start_time: - start_time = str(datetime.now()) - try: - start_time_obj = datetime.fromisoformat(start_time) - start_time = start_time_obj.replace(tzinfo=utc) - if start_time < now: - errors.append( - f"Scheduling not possible: {start_time} start_time cannot be before the current time" - ) - - if (start_time - now).total_seconds() > 300: - raise AttributeNotSupportedException( - f"Scheduling advanced reservation is not supported: start_time {start_time} " - ) - except ValueError: - errors.append( - f"Scheduling not possible: {start_time} start_time is not in a valid ISO format" - ) - if end_time: - try: - end_time_obj = datetime.fromisoformat(end_time) - end_time = end_time_obj.replace(tzinfo=utc) - if end_time < now or end_time < start_time: - errors.append( - f"Scheduling not possible: {end_time} end_time cannot be before the current or start time" - ) - except ValueError: - errors.append( - f"Scheduling not possible: {end_time} end_time is not in a valid ISO format" - ) - raise AttributeNotSupportedException( - f"Scheduling advanced reservation is not supported: end_time: {end_time} " - ) - - return errors - - def _validate_object_defaults(self, sdx_object): - """ - Validate that default fields meets the XSD standards. - - The object must have the following: - - - The object must have an ID - - - The object ID must be a string - - - The object must have a name - - - The object name must be a string - - - If the object has a short name, it must be a string - - - If the object has a version, it must be a string in ISO - format - - - All the additional properties on the object are proper - - :param sdx_object: The sdx Model Object being evaluated. - - :return: A list of any issues in the data. - """ - errors = [] - if not sdx_object._id: - errors.append(f"{sdx_object.__class__.__name__} must have an ID") - if not isinstance(sdx_object._id, str): - errors.append( - f"{sdx_object.__class__.__name__} ID must be a string" - ) - if not sdx_object._name: - errors.append( - f"{sdx_object.__class__.__name__} {sdx_object._name} " - f"must have a name" - ) - if not isinstance(sdx_object._name, str): - errors.append( - f"{sdx_object.__class__.__name__} {sdx_object._name} " - f"name must be a string" - ) - - return errors diff --git a/tests/__init__.py b/tests/__init__.py index d9b5af8..3aa9869 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,12 +1,6 @@ +from importlib.resources import files from pathlib import Path -try: - # Use stdlib modules with Python > 3.8. - from importlib.resources import files -except: - # Use compatibility library with Python 3.8. - from importlib_resources import files - class TestData: # Some data files are in src/sdx_datamodel/data. @@ -22,11 +16,24 @@ class TestData: TOPOLOGY_FILE_ZAOXI = TOPOLOGY_DIR / "zaoxi.json" REQUESTS_DIR = PACKAGE_DATA_DIR / "requests" - CONNECTION_FILE_REQ = REQUESTS_DIR / "test_request.json" - CONNECTION_FILE_REQ_NO_NODE = REQUESTS_DIR / "test_request_no_node.json" - CONNECTION_FILE_P2P = REQUESTS_DIR / "test_request_p2p.json" - CONNECTION_FILE_L2VPN_P2P_v2 = ( - REQUESTS_DIR / "test-request-amlight_sax-p2p-v2.json" + + CONNECTION_FILE_REQ_v0 = REQUESTS_DIR / "v0" / "test_request.json" + CONNECTION_FILE_REQ_NO_NODE_v0 = ( + REQUESTS_DIR / "v0" / "test_request_no_node.json" + ) + CONNECTION_FILE_P2P_v0 = REQUESTS_DIR / "v0" / "test_request_p2p.json" + + CONNECTION_FILE_L2VPN_P2P_v1 = ( + REQUESTS_DIR / "v1.0" / "test-request-l2vpn-p2p.json" + ) + CONNECTION_FILE_L2VPN_VLAN_TRANS_v1 = ( + REQUESTS_DIR / "v1.0" / "test-request-vlan-translation.json" + ) + CONNECTION_FILE_L2VPN_AMLIGHT_ZAOXI_v1 = ( + REQUESTS_DIR / "v1.0" / "test-request-amlight-zaoxi-p2p.json" + ) + CONNECTION_FILE_L2VPN_P2P_AMLIGHT_SAX_v1 = ( + REQUESTS_DIR / "v1.0" / "test-request-amlight_sax-p2p-v2.json" ) # The Remaining test data files are in tests/data. diff --git a/tests/test_connection_handler.py b/tests/test_connection_handler.py index 9273dba..45b68e2 100644 --- a/tests/test_connection_handler.py +++ b/tests/test_connection_handler.py @@ -1,128 +1,204 @@ import json import unittest +from pathlib import Path -from sdx_datamodel.models.connection import Connection -from sdx_datamodel.parsing.connectionhandler import ConnectionHandler -from sdx_datamodel.parsing.exceptions import MissingAttributeException +from pydantic import ValidationError + +from sdx_datamodel.models.connection_request import ( + ConnectionRequest, + ConnectionRequestV0, + ConnectionRequestV1, +) from . import TestData +def make_connection_request_from_json(path: Path) -> ConnectionRequest: + # data = json.loads(path.read_text()) + data = path.read_text() + return ConnectionRequest.model_validate_json(data) + + class ConnectionHandlerTests(unittest.TestCase): """Test ConnectionHandler class.""" def test_import_connection_p2p(self): - connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_P2P - ) - self.assertIsInstance(connection, Connection) + c = make_connection_request_from_json(TestData.CONNECTION_FILE_P2P_v0) + self.assertIsInstance(c, ConnectionRequest) + self.assertIsInstance(c.root, ConnectionRequestV0) def test_import_connection_req(self): - connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ - ) - self.assertIsInstance(connection, Connection) + c = make_connection_request_from_json(TestData.CONNECTION_FILE_REQ_v0) + self.assertIsInstance(c, ConnectionRequest) + self.assertIsInstance(c.root, ConnectionRequestV0) def test_import_connection_req_no_node(self): - connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ_NO_NODE + c = make_connection_request_from_json( + TestData.CONNECTION_FILE_REQ_NO_NODE_v0 ) - self.assertIsInstance(connection, Connection) + self.assertIsInstance(c, ConnectionRequest) + self.assertIsInstance(c.root, ConnectionRequestV0) def test_connection_setters(self): - connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_P2P - ) + c = make_connection_request_from_json(TestData.CONNECTION_FILE_P2P_v0) + self.assertIsInstance(c, ConnectionRequest) + self.assertIsInstance(c.root, ConnectionRequestV0) - self.assertIsInstance(connection, Connection) + # `name` property exists, but we can't assign to it. + self.assertIsNotNone(c.name) with self.assertRaises(ValueError) as ex: - connection.name = None + c.name = None self.assertIn( - "Invalid value for `name`, must not be `None`", ex.exception.args + '"ConnectionRequest" object has no field "name"', ex.exception.args ) + # `id` property exists, but we can't assign to it. + self.assertIsNotNone(c.id) + with self.assertRaises(ValueError) as ex: - connection.id = None + c.id = None self.assertIn( - "Invalid value for `id`, must not be `None`", ex.exception.args + '"ConnectionRequest" object has no field "id"', ex.exception.args ) def test_import_connection_missing_required_attributes(self): """Exception expected when required attributes are missing.""" - data = {} + + # All three required fields (id, ingress_port, and + # egress_port) are missing in input data. self.assertRaisesRegex( - MissingAttributeException, - f"Missing required attribute 'id' while parsing <{data}>", - ConnectionHandler().import_connection_data, - data, + ValidationError, + "3 validation errors for ConnectionRequest", + ConnectionRequestV0.model_validate, + {}, ) - data = {"id": "id"} + # data = {"id": "id"} + # self.assertRaisesRegex( + # MissingAttributeException, + # f"Missing required attribute 'name' while parsing <{data}>", + # ConnectionHandler().import_connection_data, + # data, + # ) + + # missing ingress_port and egress_port. self.assertRaisesRegex( - MissingAttributeException, - f"Missing required attribute 'name' while parsing <{data}>", - ConnectionHandler().import_connection_data, - data, + ValidationError, + "2 validation errors for ConnectionRequest", + ConnectionRequestV0.model_validate, + {"id": "id"}, + ) + + # data = {"id": "id", "name": "name"} + # self.assertRaisesRegex( + # MissingAttributeException, + # f"Missing required attribute 'ingress_port' while parsing <{data}>", + # ConnectionHandler().import_connection_data, + # data, + # ) + + # missing ingress_port and egress_port. + self.assertRaisesRegex( + ValidationError, + "2 validation errors for ConnectionRequest", + ConnectionRequestV0.model_validate, + {"id": "id", "name": "name"}, ) - data = {"id": "id", "name": "name"} + # missing ingress_port and egress_port. self.assertRaisesRegex( - MissingAttributeException, - f"Missing required attribute 'ingress_port' while parsing <{data}>", - ConnectionHandler().import_connection_data, - data, + ValidationError, + "2 validation errors for ConnectionRequestV0", + ConnectionRequestV0.model_validate, + {"id": "id", "name": "name", "endpoints": None}, ) - data = {"id": "id", "name": "name", "endpoints": None} + # `endpoints` should be a valid list self.assertRaisesRegex( - MissingAttributeException, - f"Missing required attribute 'ingress_port' while parsing <{data}>", - ConnectionHandler().import_connection_data, - data, + ValidationError, + "1 validation error for ConnectionRequest", + ConnectionRequestV1.model_validate, + {"id": "id", "name": "name", "endpoints": None}, ) - data = { - "id": "id", - "name": "name", - "ingress_port": {"id": "urn:ingress", "name": "test"}, - "egress_port": None, - } + # data = { + # "id": "id", + # "name": "name", + # "ingress_port": {"id": "urn:ingress", "name": "test"}, + # "egress_port": None, + # } + # self.assertRaisesRegex( + # MissingAttributeException, + # f"Missing required attribute 'egress_port' while parsing <{data}>", + # ConnectionHandler().import_connection_data, + # data, + # ) + + # invalid `egress_port`. self.assertRaisesRegex( - MissingAttributeException, - f"Missing required attribute 'egress_port' while parsing <{data}>", - ConnectionHandler().import_connection_data, - data, + ValidationError, + "1 validation error for ConnectionRequestV0", + ConnectionRequestV0.model_validate, + { + "id": "id", + "name": "name", + "ingress_port": {"id": "urn:ingress", "name": "test"}, + "egress_port": None, + }, ) - data = {"id": "id", "egress_port": None, "ingress_port": None} + # data = {"id": "id", "egress_port": None, "ingress_port": None} + # self.assertRaisesRegex( + # MissingAttributeException, + # f"Missing required attribute 'name' while parsing <{data}>", + # ConnectionHandler().import_connection_data, + # data, + # ) + + # Both `egress_port` and `ingress_port` are invalid. self.assertRaisesRegex( - MissingAttributeException, - f"Missing required attribute 'name' while parsing <{data}>", - ConnectionHandler().import_connection_data, - data, + ValidationError, + "2 validation errors for ConnectionRequestV0", + ConnectionRequestV0.model_validate, + {"id": "id", "egress_port": None, "ingress_port": None}, ) def test_import_connection_missing_optional_attributes(self): # All required attributes are set, so no error expected when # optional attributes are missing. - ingress_port = {"id": "ingress_port_id", "name": "ingress_port_name"} - egress_port = {"id": "egress_port_id", "name": "egress_port_name"} - connection = ConnectionHandler().import_connection_data( + # ingress_port = {"id": "ingress_port_id", "name": "ingress_port_name"} + # egress_port = {"id": "egress_port_id", "name": "egress_port_name"} + # connection = ConnectionHandler().import_connection_data( + # { + # "id": "id", + # "name": "name", + # "ingress_port": ingress_port, + # "egress_port": egress_port, + # } + # ) + + connection = ConnectionRequestV0.model_validate( { "id": "id", "name": "name", - "ingress_port": ingress_port, - "egress_port": egress_port, + "ingress_port": { + "id": "ingress_port_id", + "name": "ingress_port_name", + }, + "egress_port": { + "id": "egress_port_id", + "name": "egress_port_name", + }, } ) - self.assertIsInstance(connection, Connection) + self.assertIsInstance(connection, ConnectionRequestV0) - self.assertIsInstance(connection.swagger_types, dict) - self.assertIsInstance(connection.attribute_map, dict) + # self.assertIsInstance(connection.swagger_types, dict) + # self.assertIsInstance(connection.attribute_map, dict) self.assertEqual(connection.id, "id") self.assertEqual(connection.name, "name") @@ -130,78 +206,93 @@ def test_import_connection_missing_optional_attributes(self): self.assertIsNotNone(connection.egress_port) self.assertIsNotNone(connection.ingress_port) - self.assertIsNone(connection.status) - self.assertIsNone(connection.start_time) - self.assertIsNone(connection.end_time) - - self.assertIsInstance(connection.to_dict(), dict) - self.assertIsInstance(connection.to_str(), str) - - self.assertEqual(connection.latency_required, None) - connection.latency_required = 10 - self.assertEqual(connection.latency_required, 10) - - self.assertEqual(connection.bandwidth_required, None) - connection.bandwidth_required = 10 - self.assertEqual(connection.bandwidth_required, 10) - - self.assertRaisesRegex( - ValueError, - "Invalid value for `ingress_port`: must not be `None`", - connection.set_ingress_port, - None, - ) - - self.assertRaisesRegex( - TypeError, - "Invalid type for `ingress_port`: must be of type `Port`", - connection.set_ingress_port, - {}, - ) - - self.assertRaisesRegex( - ValueError, - "Invalid value for `egress_port`: must not be `None`", - connection.set_egress_port, - None, - ) - - self.assertRaisesRegex( - TypeError, - "Invalid type for `egress_port`, must be of type `Port`", - connection.set_egress_port, - {}, - ) + # self.assertIsNone(connection.status) + # self.assertIsNone(connection.start_time) + # self.assertIsNone(connection.end_time) + + # self.assertIsInstance(connection.to_dict(), dict) + # self.assertIsInstance(connection.to_str(), str) + + self.assertEqual(connection.latency_required, 0) + # connection.latency_required = 10 + # self.assertEqual(connection.latency_required, 10) + + self.assertEqual(connection.bandwidth_required, 0) + # connection.bandwidth_required = 10 + # self.assertEqual(connection.bandwidth_required, 10) + + # self.assertRaisesRegex( + # ValueError, + # "Invalid value for `ingress_port`: must not be `None`", + # connection.set_ingress_port, + # None, + # ) + + # self.assertRaisesRegex( + # TypeError, + # "Invalid type for `ingress_port`: must be of type `Port`", + # connection.set_ingress_port, + # {}, + # ) + + # self.assertRaisesRegex( + # ValueError, + # "Invalid value for `egress_port`: must not be `None`", + # connection.set_egress_port, + # None, + # ) + + # self.assertRaisesRegex( + # TypeError, + # "Invalid type for `egress_port`, must be of type `Port`", + # connection.set_egress_port, + # {}, + # ) def test_connection_handler_no_ingress_port(self): - connection_data = json.loads(TestData.CONNECTION_FILE_P2P.read_text()) + connection_data = json.loads( + TestData.CONNECTION_FILE_P2P_v0.read_text() + ) connection_data["ingress_port"] = None self.assertRaisesRegex( - MissingAttributeException, - f"Missing required attribute 'ingress_port' while parsing <{connection_data}>", - ConnectionHandler().import_connection_data, + ValidationError, + "1 validation error for ConnectionRequestV0", + ConnectionRequestV0.model_validate, connection_data, ) def test_connection_handler_no_egress_port(self): - connection_data = json.loads(TestData.CONNECTION_FILE_P2P.read_text()) + connection_data = json.loads( + TestData.CONNECTION_FILE_P2P_v0.read_text() + ) connection_data["egress_port"] = None + # self.assertRaisesRegex( + # MissingAttributeException, + # f"Missing required attribute 'egress_port' while parsing <{connection_data}>", + # ConnectionHandler().import_connection_data, + # connection_data, + # ) + self.assertRaisesRegex( - MissingAttributeException, - f"Missing required attribute 'egress_port' while parsing <{connection_data}>", - ConnectionHandler().import_connection_data, + ValidationError, + "1 validation error for ConnectionRequestV0", + ConnectionRequestV0.model_validate, connection_data, ) - """Test ConnectionHandler class for Port P2P Connection Request Spec v2.""" - def test_import_connection_port_p2p_v2(self): - connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_L2VPN_P2P_v2 + """ + Test ConnectionHandler class for Port P2P Connection Request + Spec v1. + """ + connection_data = json.loads( + TestData.CONNECTION_FILE_L2VPN_P2P_v1.read_text() ) - self.assertIsInstance(connection, Connection) + + c = ConnectionRequestV1.model_validate(connection_data) + self.assertIsInstance(c, ConnectionRequestV1) data = { "name": "new-connection", @@ -210,8 +301,13 @@ def test_import_connection_port_p2p_v2(self): {"port_id": "id2"}, ], } - with self.assertRaises(MissingAttributeException): - connection = ConnectionHandler().import_connection_data(data) + + self.assertRaisesRegex( + ValidationError, + "1 validation error for ConnectionRequestV1", + ConnectionRequestV1.model_validate, + data, + ) if __name__ == "__main__": diff --git a/tests/test_connection_request.py b/tests/test_connection_request.py new file mode 100644 index 0000000..2674a6f --- /dev/null +++ b/tests/test_connection_request.py @@ -0,0 +1,816 @@ +import datetime +import json +import unittest + +from pydantic import ValidationError + +from sdx_datamodel.models.connection_request import ( + ConnectionRequest, + ConnectionRequestV0, + ConnectionRequestV1, + Scheduling, +) + +from . import TestData + + +class TestConnectionRequest(unittest.TestCase): + + def test_request_v1(self): + testdata = json.loads( + TestData.CONNECTION_FILE_L2VPN_P2P_v1.read_text() + ) + request = ConnectionRequest(**testdata) + + self.assertIsInstance(request, ConnectionRequest) + self.assertIsInstance(request.root, ConnectionRequestV1) + + def test_request_v0(self): + testdata = json.loads(TestData.CONNECTION_FILE_P2P_v0.read_text()) + request = ConnectionRequest(**testdata) + + self.assertIsInstance(request, ConnectionRequest) + self.assertIsInstance(request.root, ConnectionRequestV0) + + +class TestConnectionRequestV1(unittest.TestCase): + """ + Tests for ConnectionRequestV1. + """ + + def test_connection_request_basic(self): + """ + Test a basic connection request. + """ + testdata = json.loads( + TestData.CONNECTION_FILE_L2VPN_VLAN_TRANS_v1.read_text() + ) + request = ConnectionRequestV1(**testdata) + + self.assertIsInstance(request, ConnectionRequestV1) + + self.assertEqual(request.name, "VLAN between AMPATH/300 and TENET/150") + + self.assertEqual(len(request.endpoints), 2) + + self.assertEqual( + request.endpoints[0].port_id, "urn:sdx:port:tenet.ac.za:Tenet03:50" + ) + self.assertEqual(request.endpoints[0].vlan, "150") + + self.assertEqual( + request.endpoints[1].port_id, "urn:sdx:port:ampath.net:Ampath3:50" + ) + self.assertEqual(request.endpoints[1].vlan, "300") + + def test_connection_request_amlight_zaoxi(self): + """ + Validate the example request. + """ + testdata = json.loads( + TestData.CONNECTION_FILE_L2VPN_AMLIGHT_ZAOXI_v1.read_text() + ) + request = ConnectionRequestV1(**testdata) + + self.assertIsInstance(request, ConnectionRequestV1) + + self.assertEqual(request.name, "new-connection") + self.assertEqual(request.description, "a test circuit") + + self.assertEqual(len(request.endpoints), 2) + + self.assertEqual( + request.endpoints[0].port_id, + "urn:sdx:port:amlight.net:A1:3", + ) + self.assertEqual(request.endpoints[0].vlan, "101") + + self.assertEqual( + request.endpoints[1].port_id, "urn:sdx:port:zaoxi:B2:1" + ) + self.assertEqual(request.endpoints[1].vlan, "101") + + def test_connection_request_amlight_sax(self): + testdata = json.loads( + TestData.CONNECTION_FILE_L2VPN_P2P_AMLIGHT_SAX_v1.read_text() + ) + request = ConnectionRequestV1(**testdata) + + self.assertEqual(request.name, "new-connection") + self.assertEqual(request.description, "a test circuit") + + self.assertEqual(len(request.endpoints), 2) + + self.assertEqual( + request.endpoints[0].port_id, + "urn:sdx:port:ampath.net:Ampath3:50", + ) + self.assertEqual(request.endpoints[0].vlan, "302") + + self.assertEqual( + request.endpoints[1].port_id, "urn:sdx:port:sax.net:Sax01:50" + ) + self.assertEqual(request.endpoints[1].vlan, "302") + + self.assertIsNone(request.scheduling.start_time) + self.assertIsNone(request.scheduling.end_time) + + self.assertEqual(request.qos_metrics.min_bw.value, 2) + self.assertEqual(request.qos_metrics.min_bw.strict, True) + + self.assertEqual(request.qos_metrics.max_delay.value, 50) + self.assertEqual(request.qos_metrics.max_delay.strict, False) + + self.assertEqual(request.qos_metrics.max_number_oxps.value, 7) + self.assertEqual(request.qos_metrics.max_number_oxps.strict, True) + + self.assertIsNotNone(request.notifications) + self.assertEqual(request.notifications[0].email, "muhaziz@fiu.edu") + + def test_connection_request_empty(self): + """ + Empty requests are invalid. + """ + testdata = {} + + self.assertRaisesRegex( + ValidationError, + "2 validation errors for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_no_endpoints(self): + """ + Requests that contain no endpoints are invalid. + """ + testdata = { + "name": "no-endpoints", + } + + self.assertRaisesRegex( + ValidationError, + "1 validation error for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_empty_endpoints(self): + """ + Requests that have empty endpoints are invalid. + """ + testdata = { + "name": "no-endpoints", + "endpoints": [], + } + + self.assertRaisesRegex( + ValidationError, + "1 validation error for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_one_endpoint(self): + """ + Requests that contain only one endpoint are invalid. + """ + testdata = { + "name": "no-endpoints", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "100", + }, + ], + } + + self.assertRaisesRegex( + ValidationError, + "1 validation error for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_vlan_not_string(self): + """ + VLANs are required to be strings. + """ + testdata = { + "name": "Bad connection request: vlan must be string", + "endpoints": [ + { + "port_id": "urn:sdx:port:tenet.ac.za:Tenet03:50", + "vlan": 100, + }, + { + "port_id": "urn:sdx:port:ampath.net:Ampath3:50", + "vlan": 100.0, + }, + ], + } + + # Both VLANs are not strings; expect two validation errors. + self.assertRaisesRegex( + ValidationError, + "2 validation errors for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_vlan_invalid_number(self): + """ + VLANs are required to be in the [1,4095] range. + """ + testdata = { + "name": "Bad connection request: vlan must be in [1,4095] range", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "-1", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "5000", + }, + ], + } + + # Both VLANs are not in the [1,4095] range; expect two + # validation errors. + self.assertRaisesRegex( + ValidationError, + "2 validation errors for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_vlan_invalid_range(self): + """ + Ranges are required to be valid numbers. + """ + testdata = { + "name": "Bad connection request: vlan must be in [1,4095] range", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "100:not-a-number", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "200:100", + }, + ], + } + + # Both VLANs are not in the [1,4095] range; expect two + # validation errors. + self.assertRaisesRegex( + ValidationError, + "2 validation errors for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_vlan_invalid_integers_in_range(self): + """ + Numbers in ranges are required to be within [1,4095] range. + """ + testdata = { + "name": "Bad connection request: vlan must be in [1,4095] range", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "1:10000", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "10000:1", + }, + ], + } + + # Both VLANs are not in the [1,4095] range; expect two + # validation errors. + self.assertRaisesRegex( + ValidationError, + "2 validation errors for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_vlan_invalid_strings_in_range(self): + """ + Ranges should be in numbers, and numbers in ranges are + required to be within [1,4095] range. + """ + testdata = { + "name": "Bad connection request: vlan must be in [1,4095] range", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "1:10000", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "10000:any", + }, + ], + } + + # Both VLANs are not in the [1,4095] range; expect two + # validation errors. + self.assertRaisesRegex( + ValidationError, + "2 validation errors for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_vlan_all_all(self): + """ + VLANs in both endpoints can be `all`. + """ + testdata = { + "name": "Good connection request: both VLANs are `all`", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "all", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "all", + }, + ], + } + + request = ConnectionRequestV1(**testdata) + + self.assertEqual(request.endpoints[0].vlan, "all") + self.assertEqual(request.endpoints[1].vlan, "all") + + def test_connection_request_vlan_any_any(self): + """ + VLANs in both endpoints can be `any`. + """ + testdata = { + "name": "Good connection request: both VLANs are `any`", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "any", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "any", + }, + ], + } + + request = ConnectionRequestV1(**testdata) + + self.assertEqual(request.endpoints[0].vlan, "any") + self.assertEqual(request.endpoints[1].vlan, "any") + + def test_connection_request_vlan_invalid_strings(self): + """ + VLANs are required to be numbers, ranges, or certain strings + such as "all", "any", or "untagged". + """ + testdata = { + "name": "Bad connection request: vlan must be in [1,4095] range", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "unknown", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "random", + }, + ], + } + + # Both VLANs are not any of the expected strings; expect two + # validation errors. + self.assertRaisesRegex( + ValidationError, + "2 validation errors for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_vlan_valid_range(self): + """ + When one endpoint has the VLAN range option in use, all other + endpoint(s) must have the same VLAN range. + """ + request_name = "Connection request with valid vlan ranges" + port0_id = "urn:sdx:port:example.net:p:1" + vlan_range = "100:200" + port1_id = "urn:sdx:port:example.net:p:2" + testdata = { + "name": request_name, + "endpoints": [ + { + "port_id": port0_id, + "vlan": vlan_range, + }, + { + "port_id": port1_id, + "vlan": vlan_range, + }, + ], + } + + request = ConnectionRequestV1(**testdata) + self.assertEqual(request.name, request_name) + self.assertEqual(request.endpoints[0].port_id, port0_id) + self.assertEqual(request.endpoints[0].vlan, vlan_range) + self.assertEqual(request.endpoints[1].port_id, port1_id) + self.assertEqual(request.endpoints[1].vlan, vlan_range) + + def test_connection_request_vlan_invalid_unmatched_range(self): + """ + When one endpoint has the VLAN range option in use, all other + endpoint(s) must have the same VLAN range. + """ + request_name = "Connection request with valid vlan ranges" + port0_id = "urn:sdx:port:example.net:p:1" + port0_vlan = "100:200" + port1_id = "urn:sdx:port:example.net:p:2" + port1_vlan = "200:300" + testdata = { + "name": request_name, + "endpoints": [ + { + "port_id": port0_id, + "vlan": port0_vlan, + }, + { + "port_id": port1_id, + "vlan": port1_vlan, + }, + ], + } + + self.assertRaisesRegex( + ValidationError, + "1 validation error for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_vlan_inverse_invalid_range(self): + """ + Ranges like n1:n2 where n2 < n1 are invalid. + """ + request_name = "Connection request with valid vlan ranges" + port0_id = "urn:sdx:port:example.net:p:1" + vlan_range = "200:100" + port1_id = "urn:sdx:port:example.net:p:2" + testdata = { + "name": request_name, + "endpoints": [ + { + "port_id": port0_id, + "vlan": vlan_range, + }, + { + "port_id": port1_id, + "vlan": vlan_range, + }, + ], + } + + self.assertRaisesRegex( + ValidationError, + "2 validation errors for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_with_optional_fields(self): + """ + Test a connection request that has optional fields. + """ + testdata = json.loads( + TestData.CONNECTION_FILE_L2VPN_P2P_v1.read_text() + ) + request = ConnectionRequestV1(**testdata) + + self.assertIsInstance(request, ConnectionRequestV1) + + self.assertEqual(request.name, "new-connection") + self.assertEqual(request.description, "a test circuit") + + self.assertEqual(len(request.endpoints), 2) + + self.assertIsInstance( + request.scheduling.start_time, + datetime.datetime, + ) + self.assertIsInstance( + request.scheduling.start_time, + datetime.datetime, + ) + + # TODO: `TzInfo` is a type internal to Pydantic. I don't know + # how to compare a thing containing `TzInfo`. + + # self.assertEqual( + # request.scheduling.start_time, + # datetime.datetime(2024, 6, 24, 1, 0, tzinfo=TzInfo(UTC)), + # ) + # self.assertEqual( + # request.scheduling.start_time, + # datetime.datetime(2024, 6, 26, 1, 0, tzinfo=TzInfo(UTC)), + # ) + + self.assertEqual(request.qos_metrics.min_bw.value, 12) + self.assertEqual(request.qos_metrics.min_bw.strict, True) + + self.assertEqual(request.qos_metrics.max_delay.value, 4) + self.assertEqual(request.qos_metrics.max_delay.strict, False) + + self.assertEqual(request.qos_metrics.max_number_oxps.value, 7) + self.assertEqual(request.qos_metrics.max_number_oxps.strict, True) + + def test_connection_request_immutable(self): + """ + Test that we are unable to accidentally mutate fields. + """ + + testdata = json.loads( + TestData.CONNECTION_FILE_L2VPN_P2P_v1.read_text() + ) + request = ConnectionRequestV1(**testdata) + + self.assertIsInstance(request, ConnectionRequestV1) + + with self.assertRaises(ValidationError): + request.name = "another-name" + + with self.assertRaises(ValidationError): + request.description = "another-description" + + with self.assertRaises(ValidationError): + request.endpoints = [] + + # TODO: mutating endpoints do not raise validation error. Why? + + # with self.assertRaises(ValidationError): + # request.endpoints[0] = {} + + # with self.assertRaises(ValidationError): + # request.endpoints[1] = {} + + with self.assertRaises(ValidationError): + request.notifications = None + + with self.assertRaises(ValidationError): + request.notifications[0].email = None + + with self.assertRaises(ValidationError): + request.scheduling = None + + with self.assertRaises(ValidationError): + request.scheduling.start_time = None + + with self.assertRaises(ValidationError): + request.scheduling.end_time = None + + with self.assertRaises(ValidationError): + request.qos_metrics = None + + with self.assertRaises(ValidationError): + request.qos_metrics.min_bw = None + + with self.assertRaises(ValidationError): + request.qos_metrics.min_bw.value = None + + with self.assertRaises(ValidationError): + request.qos_metrics.min_bw.strict = True + + with self.assertRaises(ValidationError): + request.qos_metrics.max_delay.value = None + + with self.assertRaises(ValidationError): + request.qos_metrics.max_delay.strict = None + + with self.assertRaises(ValidationError): + request.qos_metrics.max_number_oxps = None + + with self.assertRaises(ValidationError): + request.qos_metrics.max_number_oxps.value = None + + with self.assertRaises(ValidationError): + request.qos_metrics.max_number_oxps.strict = None + + def test_connection_request_all_vlans_valid(self): + """ + If one VLAN in a request is "all", all VLANs should be "all". + """ + testdata = { + "name": "Bad connection request with 'all' in vlans", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "1", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "all", + }, + ], + } + + self.assertRaisesRegex( + ValidationError, + "1 validation error for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + def test_connection_request_valid_email(self): + """ + Requests containing valid notification emails. + """ + testdata = { + "name": "Connection request with valid email in notifications", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "any", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "any", + }, + ], + "notifications": [ + {"email": "alice@example.net"}, + {"email": "bob@example.org"}, + ], + } + + request = ConnectionRequestV1(**testdata) + self.assertEqual(request.notifications[0].email, "alice@example.net") + self.assertEqual(request.notifications[1].email, "bob@example.org") + + def test_connection_request_invalid_email(self): + """ + Requests containing invalid notification emails. + """ + testdata = { + "name": "Connection request with invalid email in notifications", + "endpoints": [ + { + "port_id": "urn:sdx:port:example.net:p:1", + "vlan": "any", + }, + { + "port_id": "urn:sdx:port:example.net:p:2", + "vlan": "any", + }, + ], + "notifications": [{"email": "@alice"}], + } + + self.assertRaisesRegex( + ValidationError, + "1 validation error for ConnectionRequestV1", + ConnectionRequestV1, + **testdata, + ) + + +class TestConnectionScheduling(unittest.TestCase): + """ + Some test cases for Scheduling in particular. + """ + + def test_scheduling_empty(self): + s = Scheduling().dict(exclude_unset=True) + self.assertIsInstance(s, dict) + + def test_scheduling_only_start_time(self): + s = Scheduling(start_time="2025-03-13T10:00:00Z") + self.assertIsInstance(s.dict(), dict) + + def test_scheduling_both_timestamps_valid(self): + # Both times valid + s = Scheduling( + start_time="2025-03-13T10:00:00Z", end_time="2025-03-13T12:00:00Z" + ) + self.assertIsInstance(s.dict(), dict) + + def test_scheduling_end_time_before_start_time(self): + # Invalid case: end_time before start_time + self.assertRaisesRegex( + ValidationError, + "1 validation error for Scheduling", + Scheduling, + start_time="2025-03-13T12:00:00Z", + end_time="2025-03-13T10:00:00Z", + ) + + def test_scheduling_bad_iso8601(self): + # Invalid case: bad ISO8601 format + self.assertRaisesRegex( + ValidationError, + "1 validation error for Scheduling", + Scheduling, + start_time="not-a-date", + ) + + +class TestConnectionRequestV0(unittest.TestCase): + """ + Tests for ConnectionRequestV0. + """ + + def test_connection_request_v0_basic(self): + """ + Basic checks for a basic V0 connection request. + """ + + testdata = json.loads(TestData.CONNECTION_FILE_REQ_v0.read_text()) + request = ConnectionRequestV0(**testdata) + + self.assertIsInstance(request, ConnectionRequestV0) + + self.assertEqual(request.id, "285eea4b-1e86-4d54-bd75-f14b8cb4a63a") + self.assertEqual(request.name, "Test connection request") + + self.assertEqual(request.bandwidth_required, 10) + self.assertEqual(request.latency_required, 300) + + # No start/end times in this request. + self.assertIsNone(request.start_time) + self.assertIsNone(request.end_time) + + self.assertEqual(request.ingress_port.id, "urn:sdx:port:zaoxi:A1:2") + self.assertEqual(request.ingress_port.name, "Novi100:2") + self.assertEqual( + request.ingress_port.node, "urn:ogf:network:sdx:node:zaoxi:A1" + ) + self.assertEqual(request.ingress_port.status, "up") + + self.assertEqual( + request.egress_port.id, "urn:sdx:port:amlight.net:A1:1" + ) + self.assertEqual(request.egress_port.name, "Novi100:1") + self.assertEqual( + request.egress_port.node, "urn:sdx:node:amlight.net:A1" + ) + self.assertEqual(request.egress_port.status, "up") + + def test_connection_request_v0_no_node(self): + """ + Check a V0 connection request that does not carry the optional + fields. + """ + + testdata = json.loads( + TestData.CONNECTION_FILE_REQ_NO_NODE_v0.read_text() + ) + request = ConnectionRequestV0(**testdata) + + self.assertIsInstance(request, ConnectionRequestV0) + + self.assertIsNone(request.ingress_port.node) + self.assertIsNone(request.ingress_port.status) + + self.assertIsNone(request.egress_port.node) + self.assertIsNone(request.egress_port.status) + + def test_connection_request_v0_p2p(self): + """ + Check the P2P request in v0 format. + """ + testdata = json.loads(TestData.CONNECTION_FILE_P2P_v0.read_text()) + request = ConnectionRequestV0(**testdata) + + self.assertIsInstance(request, ConnectionRequestV0) + + self.assertIsNotNone(request.ingress_port.id) + self.assertIsNotNone(request.ingress_port.name) + self.assertIsNotNone(request.ingress_port.short_name) + self.assertIsNotNone(request.ingress_port.label) + self.assertIsNotNone(request.ingress_port.label_range) + self.assertIsNotNone(request.ingress_port.node) + self.assertIsNone(request.ingress_port.status) + + self.assertIsNotNone(request.egress_port.id) + self.assertIsNotNone(request.egress_port.name) + self.assertIsNotNone(request.egress_port.short_name) + self.assertIsNotNone(request.egress_port.label) + self.assertIsNotNone(request.egress_port.label_range) + self.assertIsNotNone(request.egress_port.node) + self.assertIsNone(request.egress_port.status) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_connection_v2.py b/tests/test_connection_v2.py index 463758e..15a7e70 100644 --- a/tests/test_connection_v2.py +++ b/tests/test_connection_v2.py @@ -2,23 +2,22 @@ import unittest from sdx_datamodel.models.connection_qos_metrics import ConnectionQosMetrics +from sdx_datamodel.models.connection_request import ConnectionRequest from sdx_datamodel.models.connection_scheduling import ConnectionScheduling -from sdx_datamodel.models.connection_v2 import Connection from sdx_datamodel.models.link import Link from sdx_datamodel.models.port import Port from sdx_datamodel.parsing.connectionhandler import ConnectionHandler -from sdx_datamodel.validation.connectionvalidator import ConnectionValidator class TestConnection(unittest.TestCase): - def _get_validator(self, data): - """ - Return a validator for the given file. - """ - handler = ConnectionHandler() - connection = handler.import_connection_data(data) - return ConnectionValidator(connection) + # def _get_validator(self, data): + # """ + # Return a validator for the given file. + # """ + # handler = ConnectionHandler() + # connection = handler.import_connection_data(data) + # return ConnectionValidator(connection) def test_connection(self): # Create test data @@ -34,7 +33,7 @@ def test_connection(self): exclusive_links = [Link(id="link1"), Link(id="link2")] # Create a Connection instance - connection = Connection( + connection = ConnectionRequest( id="connection1", name="Test Connection", endpoints=endpoints, @@ -82,12 +81,14 @@ def test_connection_invalid_qos_metrics(self): validator = self._get_validator(connection_request) - with self.assertRaises(ValueError) as ex: - validator.is_valid() + c = ConnectionRequest.model_validate(connection_request) + + # with self.assertRaises(ValueError) as ex: + # validator.is_valid() - errors = ex.exception.args[0].splitlines() - print(f"{errors}") - self.assertEqual(len(errors), 3) + # errors = ex.exception.args[0].splitlines() + # print(f"{errors}") + # self.assertEqual(len(errors), 3) if __name__ == "__main__": diff --git a/tests/test_connection_validator.py b/tests/test_connection_validator.py index a91c9b6..722e522 100644 --- a/tests/test_connection_validator.py +++ b/tests/test_connection_validator.py @@ -1,14 +1,17 @@ import datetime import unittest -from sdx_datamodel.models.connection import Connection +import json + +from sdx_datamodel.models.connection_request import ConnectionRequest from sdx_datamodel.models.port import Port from sdx_datamodel.parsing.connectionhandler import ConnectionHandler from sdx_datamodel.parsing.exceptions import ( AttributeNotSupportedException, ServiceNotSupportedException, ) -from sdx_datamodel.validation.connectionvalidator import ConnectionValidator +# from sdx_datamodel.validation.connectionvalidator import ConnectionValidator +from sdx_datamodel.models.connection_request import ConnectionRequest from . import TestData @@ -22,29 +25,32 @@ def _get_validator(self, path): """ Return a validator for the given file. """ - handler = ConnectionHandler() - connection = handler.import_connection(path) - return ConnectionValidator(connection) + # handler = ConnectionHandler() + # connection = handler.import_connection(path) + data = json.loads(path.read_text()) + return ConnectionRequest(data) def test_connection_json_p2p(self): """ Validate a JSON document descibing a connection. """ - validator = self._get_validator(TestData.CONNECTION_FILE_P2P) + validator = self._get_validator(TestData.CONNECTION_FILE_P2P_v0) self.assertTrue(validator.is_valid()) def test_connection_json_req(self): """ Validate a JSON document descibing a connection. """ - validator = self._get_validator(TestData.CONNECTION_FILE_REQ) + validator = self._get_validator(TestData.CONNECTION_FILE_REQ_v0) self.assertTrue(validator.is_valid()) def test_connection_json_req_no_node(self): """ Validate a JSON document descibing a "node-less" connection. """ - validator = self._get_validator(TestData.CONNECTION_FILE_REQ_NO_NODE) + validator = self._get_validator( + TestData.CONNECTION_FILE_REQ_NO_NODE_v0 + ) self.assertTrue(validator.is_valid()) def test_connection_json_req_bad_name(self): @@ -52,7 +58,7 @@ def test_connection_json_req_bad_name(self): Connection name must be a string. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.name = 42 @@ -66,7 +72,7 @@ def test_connection_json_req_bad_id(self): Connection ID must be a string. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.id = 42 @@ -80,7 +86,7 @@ def test_connection_json_req_bad_ports(self): Ingress and egress ports must have valid names and IDs. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.ingress_port.name = 32 @@ -128,7 +134,7 @@ def test_connection_object(self): status="unknown", ) - connection = Connection( + connection = ConnectionRequest( id="test_place_connection_id", name="test_place_connection_name", ingress_port=ingress_port, @@ -141,7 +147,7 @@ def test_connection_object(self): complete=False, ) - self.assertIsInstance(connection, Connection) + self.assertIsInstance(connection, ConnectionRequest) def test_connection_object_invalid(self): """ @@ -163,7 +169,7 @@ def test_connection_object_invalid(self): status="unknown", ) - connection = Connection( + connection = ConnectionRequest( id="test_place_connection_id", name="test_place_connection_name", ingress_port=ingress_port, @@ -173,7 +179,7 @@ def test_connection_object_invalid(self): complete=False, ) - self.assertIsInstance(connection, Connection) + self.assertIsInstance(connection, ConnectionRequest) with self.assertRaises(ValueError) as ex: validator = ConnectionValidator(connection).is_valid() @@ -188,7 +194,7 @@ def test_validate_vlan_valid(self): Test _validate_vlan with a valid VLAN range. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.egress_port.vlan_range = "100-200" validator = ConnectionValidator(connection) @@ -201,7 +207,7 @@ def test_validate_vlan_invalid(self): Test _validate_vlan with an invalid VLAN range. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.egress_port.vlan_range = "5000" validator = ConnectionValidator(connection) @@ -216,7 +222,7 @@ def test_validate_vlan_invalid_range(self): Test _validate_vlan with an invalid VLAN range format. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.egress_port.vlan_range = "200:100" validator = ConnectionValidator(connection) @@ -249,7 +255,7 @@ def test_validate_time_valid(self): Test _validate_time with a valid time range. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.start_time = str( datetime.datetime.now() + datetime.timedelta(seconds=100) @@ -265,7 +271,7 @@ def test_validate_time_invalid_format(self): Test _validate_time with an invalid time format. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.start_time = "invalid_time_format" validator = ConnectionValidator(connection) @@ -282,7 +288,7 @@ def test_validate_time_past_time(self): Test _validate_time with a past time. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.start_time = str( datetime.datetime.now() - datetime.timedelta(days=1) @@ -301,7 +307,7 @@ def test_start_time_too_far_in_future(self): Test start_time that is too far in the future. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.start_time = str( datetime.datetime.now() + datetime.timedelta(seconds=600) @@ -321,7 +327,7 @@ def test_attribute_not_supported_exception(self): Test for AttributeNotSupportedException. """ connection = ConnectionHandler().import_connection( - TestData.CONNECTION_FILE_REQ + TestData.CONNECTION_FILE_REQ_v0 ) connection.end_time = str( diff --git a/tests/test_schemas.py b/tests/test_schemas.py index 23b97ae..33bf113 100644 --- a/tests/test_schemas.py +++ b/tests/test_schemas.py @@ -41,13 +41,13 @@ def _read_schema(self, path: Path): def test_connection_request_schema_v1(self): jsonschema.validate( - self._read_json(TestData.CONNECTION_FILE_REQ), + self._read_json(TestData.CONNECTION_FILE_REQ_v0), self._read_schema(self.CONNECTION_SCHEMA_V1_FILE), ) def test_connection_request_schema(self): jsonschema.validate( - self._read_json(TestData.CONNECTION_FILE_L2VPN_P2P_v2), + self._read_json(TestData.CONNECTION_FILE_L2VPN_P2P_v1), self._read_schema(self.CONNECTION_SCHEMA_FILE), ) diff --git a/tox.ini b/tox.ini index 5d3daad..9731756 100644 --- a/tox.ini +++ b/tox.ini @@ -1,7 +1,5 @@ [tox] env_list = - py39 - py310 py311 py312