12711868108

Committed 10 Jan 2025 02:57PM UTC coverage: 77.873% (+0.2%) from 77.648%

Build # 12711868108

Build Type

Pull #157

github

Committed by

web-flow

Commit Message

Merge 79568aa20 into 11bd074f3

Pull Request Pull Request #157: validating against time and qos_metrics values

Run Details

41 of 52 new or added lines in 4 files covered. (78.85%)

1 existing line in 1 file now uncovered.

1369 of 1758 relevant lines covered (77.87%)

1.56 hits per line

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

89.29

/src/sdx_datamodel/validation/connectionvalidator.py

"""""
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

ISO_FORMAT = r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[-+]\d{2}:\d{2}"

ISO_TIME_FORMAT = r"(^(?:[1-9]\d{3}-(?:(?:0[1-9]|1[0-2])-(?:0[1-9]|1\d|2[0-8])|(?:0[13-9]|1[0-2])-(?:29|30)|(?:0[13578]|1[02])-31)|(?:[1-9]\d(?:0[48]|[2468][048]|[13579][26])|(?:[2468][048]|[13579][26])00)-02-29)T(?:[01]\d|2[0-3]):[0-5]\d:[0-5]\d(?:Z|[+-][01]\d:[0-5]\d)$)"  # noqa: E501


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 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.bandwidth_required, 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

        :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"{value} {metric} must be a number")
        if not (0 <= value <= max_value):
            errors.append(f"{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)

        """
        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_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 = []
        # if not match(ISO_TIME_FORMAT, time):
        #    errors.append(f"{time} time needs to be in full ISO format")
        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 < datetime.now().replace(tzinfo=utc):
                errors.append(
                    f"{start_time} start_time cannot be before the current time"
                )
        except ValueError:
            errors.append(
                f"{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 < datetime.now().replace(tzinfo=utc)
                    or end_time < start_time
                ):
                    errors.append(
                        f"{end_time} end_time cannot be before the current or start time"
                    )
            except ValueError:
                errors.append(
                    f"{end_time} end_time is not in a valid ISO format"
                )

        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

1	"""""
2	Checks for Connection objects to be in the expected format.
3	"""
4
5	import logging	2✔
6	from datetime import datetime	2✔
7	from re import match	2✔
8
9	import pytz	2✔
10
11	from sdx_datamodel.models.connection import Connection	2✔
12	from sdx_datamodel.models.port import Port	2✔
13
14	ISO_FORMAT = r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[-+]\d{2}:\d{2}"	2✔
15
16	ISO_TIME_FORMAT = r"(^(?:[1-9]\d{3}-(?:(?:0[1-9]\|1[0-2])-(?:0[1-9]\|1\d\|2[0-8])\|(?:0[13-9]\|1[0-2])-(?:29\|30)\|(?:0[13578]\|1[02])-31)\|(?:[1-9]\d(?:0[48]\|[2468][048]\|[13579][26])\|(?:[2468][048]\|[13579][26])00)-02-29)T(?:[01]\d\|2[0-3]):[0-5]\d:[0-5]\d(?:Z\|[+-][01]\d:[0-5]\d)$)" # noqa: E501	2✔
17
18
19	class ConnectionValidator:	2✔
20	"""
21	The validation class made to validate a Connection request
22	"""
23
24	def __init__(self, connection):	2✔
25	if not isinstance(connection, Connection):	2✔
26	raise ValueError(	2✔
27	"ConnectionValidator must be passed a Connection object"
28	)
29
30	self._connection = connection	2✔
31
32	self._logger = logging.getLogger(__name__)	2✔
33
34	def is_valid(self) -> bool:	2✔
35	errors = self.validate(raise_error=True)	2✔
36	for error in errors:	2✔
37	self._logger.error(error)	×
38	return not bool(errors)	2✔
39
40	def validate(self, raise_error=True) -> [str]:	2✔
41	errors = self._validate_connection(self._connection)	2✔
42	if errors and raise_error:	2✔
43	raise ValueError("\n".join(errors))	2✔
44	return errors	2✔
45
46	def _validate_connection(self, conn: Connection):	2✔
47	"""
48	Validate that the connection provided meets the JSON schema.
49
50	A connection must have the following:
51
52	- It must meet object standard
53
54	- It must have the default fields: id, name, ingress_port,
55	and egress_port
56
57	:param connection: The connection being evaluated
58
59	:return: A list of any issues in the data.
60	"""
61
62	errors = []	2✔
63	errors += self._validate_object_defaults(conn)	2✔
64	errors += self._validate_port(conn.ingress_port, conn)	2✔
65	errors += self._validate_port(conn.egress_port, conn)	2✔
66
67	if conn.start_time or conn.end_time:	2✔
68	errors += self._validate_time(conn.start_time, conn.end_time, conn)	2✔
69
70	if conn.latency_required:	2✔
71	errors += self._validate_qos_metrics_value(	2✔
72	"max_delay", conn.latency_required, 1000
73	)
74
75	if conn.bandwidth_required:	2✔
76	errors += self._validate_qos_metrics_value(	2✔
77	"min_bw", conn.bandwidth_required, 100
78	)
79
80	if conn.max_number_oxps:	2✔
81	errors += self._validate_qos_metrics_value(	2✔
82	"max_number_oxps", conn.bandwidth_required, 100
83	)
84	return errors	2✔
85
86	def _validate_qos_metrics_value(self, metric, value, max_value):	2✔
87	"""
88	Validate that the QoS Metrics provided meets the XSD standards.
89
90	A connection must have the following:
91
92	- It must meet object default standards.
93
94	- The max_delay must be a number
95
96	- The max_number_oxps must be a number
97
98	:param qos_metrics: The QoS Metrics being evaluated.
99
100	:return: A list of any issues in the data.
101	"""
102	errors = []	2✔
103
104	if not isinstance(value, int):	2✔
NEW 105	errors.append(f"{value} {metric} must be a number")	×
106	if not (0 <= value <= max_value):	2✔
107	errors.append(f"{value} {metric} must be between 0 and 1000")	2✔
108
109	return errors	2✔
110
111	def _validate_port(self, port: Port, conn: Connection):	2✔
112	"""
113	Validate that the port provided meets the XSD standards.
114	A port must have the following:
115	- It must meet object default standards.
116	- A port must belong to a topology
117	- The node is optional
118	:param port: The port being evaluated.
119	:param topology: The Topology.
120	:return: A list of any issues in the data.
121	"""
122
123	errors = []	2✔
124	if not port:	2✔
125	errors.append(f"{port.__class__.__name__} must exist")	×
126
127	errors += self._validate_object_defaults(port)	2✔
128
129	"""	2✔
130	node = find_node(port,topology)
131	if topology and node not in topology.nodes:
132	errors.append(
133	'listed node id {} does not exist in parent Topology {}'.format(
134	node.id, node, topology.id
135	)
136	)
137	"""
138	return errors	2✔
139
140	def _validate_time(self, start_time: str, end_time: str, conn: Connection):	2✔
141	"""
142	Validate that the time provided meets the XSD standards.
143
144	:param start_time, end_time: time being validated
145
146	:return: A list of any issues in the data.
147	"""
148	utc = pytz.UTC	2✔
149	errors = []	2✔
150	# if not match(ISO_TIME_FORMAT, time):
151	# errors.append(f"{time} time needs to be in full ISO format")
152	if not start_time:	2✔
153	start_time = str(datetime.now())	2✔
154	try:	2✔
155	start_time_obj = datetime.fromisoformat(start_time)	2✔
156	start_time = start_time_obj.replace(tzinfo=utc)	2✔
157	if start_time < datetime.now().replace(tzinfo=utc):	2✔
158	errors.append(	2✔
159	f"{start_time} start_time cannot be before the current time"
160	)
NEW 161	except ValueError:	×
NEW 162	errors.append(	×
163	f"{start_time} start_time is not in a valid ISO format"
164	)
165	if end_time:	2✔
166	try:	2✔
167	end_time_obj = datetime.fromisoformat(end_time)	2✔
168	end_time = end_time_obj.replace(tzinfo=utc)	2✔
169	if (	2✔
170	end_time < datetime.now().replace(tzinfo=utc)
171	or end_time < start_time
172	):
173	errors.append(	2✔
174	f"{end_time} end_time cannot be before the current or start time"
175	)
NEW 176	except ValueError:	×
NEW 177	errors.append(	×
178	f"{end_time} end_time is not in a valid ISO format"
179	)
180
181	return errors	2✔
182
183	def _validate_object_defaults(self, sdx_object):	2✔
184	"""
185	Validate that default fields meets the XSD standards.
186
187	The object must have the following:
188
189	- The object must have an ID
190
191	- The object ID must be a string
192
193	- The object must have a name
194
195	- The object name must be a string
196
197	- If the object has a short name, it must be a string
198
199	- If the object has a version, it must be a string in ISO
200	format
201
202	- All the additional properties on the object are proper
203
204	:param sdx_object: The sdx Model Object being evaluated.
205
206	:return: A list of any issues in the data.
207	"""
208	errors = []	2✔
209	if not sdx_object._id:	2✔
210	errors.append(f"{sdx_object.__class__.__name__} must have an ID")	×
211	if not isinstance(sdx_object._id, str):	2✔
212	errors.append(	2✔
213	f"{sdx_object.__class__.__name__} ID must be a string"
214	)
215	if not sdx_object._name:	2✔
216	errors.append(	×
217	f"{sdx_object.__class__.__name__} {sdx_object._name} "
218	f"must have a name"
219	)
220	if not isinstance(sdx_object._name, str):	2✔
221	errors.append(	2✔
222	f"{sdx_object.__class__.__name__} {sdx_object._name} "
223	f"name must be a string"
224	)
225
226	return errors	2✔

atlanticwave-sdx / datamodel / 12711868108

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