4533308467

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #82 from atlanticwave-sdx/48.refactor-connection-handler

Run Details

205 of 450 branches covered (45.56%)

Branch coverage included in aggregate %.

4 of 4 new or added lines in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

903 of 1211 relevant lines covered (74.57%)

2.98 hits per line

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

70.13

/src/sdx/datamodel/validation/connectionvalidator.py

"""""
--------------------------------------------------------------------
Synopsis: A validation class to evaluate that the supplied Connection object contains expected data format
"""
from collections.abc import Iterable
from datetime import *
from re import match

from sdx.datamodel.models.port import Port
from sdx.datamodel.models.connection import Connection

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


class ConnectionValidator:
    """
    The validation class made to validate a Connection request
    """

    def __init__(self):
        super().__init__()
        self.connection = None

    def get_connection(self):
        return self.connection

    def set_connection(self, conn):
        if not isinstance(conn, Connection):
            raise ValueError(
                "The Validator must be passed a Connection object"
            )
        self.connection = conn

    def is_valid(self):
        errors = self.validate(self.connection, raise_error=True)
        for error in errors:
            print(error)
        return not bool(errors)

    def validate(self, conn=None, raise_error=True):
        if not conn and self._connection:
            conn = self.connection
        errors = self._validate_connection(conn)
        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)

        # errors += self._validate_time(conn.start_time, conn)

        # errors += self._validate_time(conn.end_time, conn)
        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("{} must exist".format(port.__class__.__name__))

        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, time: str, conn: Connection):
        """
        Validate that the time provided meets the XSD standards.
        A port must have the following:
         - It must meet object default standards.
         - A link can only connect to 2 nodes
         - The nodes that a link is connected to must be in the parent Topology's nodes list
        :param time: time being validated
        :return: A list of any issues in the data.
        """
        errors = []
        if not match(ISO_TIME_FORMAT, time):
            errors.append(
                "{} time needs to be in full ISO format".format(
                    time,
                )
            )
        return errors

    def _validate_object_defaults(self, sdx_object):
        """
        Validate that the object provided 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(
                "{} must have an ID".format(sdx_object.__class__.__name__)
            )
        if not isinstance(sdx_object._id, str):
            errors.append(
                "{} ID must be a string".format(sdx_object.__class__.__name__)
            )
        if not sdx_object._name:
            errors.append(
                "{} {} must have a name".format(
                    sdx_object.__class__.__name__,
                    sdx_object._name,
                )
            )
        if not isinstance(sdx_object._name, str):
            errors.append(
                "{} {} name must be a String".format(
                    sdx_object.__class__.__name__,
                    sdx_object._name,
                )
            )

        return errors

1	"""""
2	--------------------------------------------------------------------
3	Synopsis: A validation class to evaluate that the supplied Connection object contains expected data format
4	"""
5	from collections.abc import Iterable	4✔
6	from datetime import *	4✔
7	from re import match	4✔
8
9	from sdx.datamodel.models.port import Port	4✔
10	from sdx.datamodel.models.connection import Connection	4✔
11
12	ISO_FORMAT = r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[-+]\d{2}:\d{2}"	4✔
13
14	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)$)"	4✔
15
16
17	class ConnectionValidator:	4✔
18	"""
19	The validation class made to validate a Connection request
20	"""
21
22	def __init__(self):	4✔
23	super().__init__()	4✔
24	self.connection = None	4✔
25
26	def get_connection(self):	4✔
UNCOV 27	return self.connection	×
28
29	def set_connection(self, conn):	4✔
30	if not isinstance(conn, Connection):	4✔
31	raise ValueError(	4✔
32	"The Validator must be passed a Connection object"
33	)
34	self.connection = conn	4✔
35
36	def is_valid(self):	4✔
37	errors = self.validate(self.connection, raise_error=True)	4✔
38	for error in errors:	4!
39	print(error)	×
40	return not bool(errors)	4✔
41
42	def validate(self, conn=None, raise_error=True):	4✔
43	if not conn and self._connection:	4!
44	conn = self.connection	×
45	errors = self._validate_connection(conn)	4✔
46	if errors and raise_error:	4!
47	raise ValueError("\n".join(errors))	×
48	return errors	4✔
49
50	def _validate_connection(self, conn: Connection):	4✔
51	"""
52	Validate that the connection provided meets the JSON schema.
53	A connection must have the following:
54	- It must meet object standard
55	- It must have the default fields: id, name, ingress_port, and egress_port
56	:param connection: The connection being evaluated
57	:return: A list of any issues in the data.
58	"""
59	errors = []	4✔
60	errors += self._validate_object_defaults(conn)	4✔
61
62	errors += self._validate_port(conn.ingress_port, conn)	4✔
63
64	errors += self._validate_port(conn.egress_port, conn)	4✔
65
66	# errors += self._validate_time(conn.start_time, conn)
67
68	# errors += self._validate_time(conn.end_time, conn)
69	return errors	4✔
70
71	def _validate_port(self, port: Port, conn: Connection):	4✔
72	"""
73	Validate that the port provided meets the XSD standards.
74	A port must have the following:
75	- It must meet object default standards.
76	- A port must belong to a topology
77	- The node is optional
78	:param port: The port being evaluated.
79	:param topology: The Topology.
80	:return: A list of any issues in the data.
81	"""
82
83	errors = []	4✔
84	if not port:	4!
85	errors.append("{} must exist".format(port.__class__.__name__))	×
86
87	errors += self._validate_object_defaults(port)	4✔
88
89	"""	1✔
90	node = find_node(port,topology)
91	if topology and node not in topology.nodes:
92	errors.append(
93	'listed node id {} does not exist in parent Topology {}'.format(
94	node.id, node, topology.id
95	)
96	)
97	"""
98	return errors	4✔
99
100	def _validate_time(self, time: str, conn: Connection):	4✔
101	"""
102	Validate that the time provided meets the XSD standards.
103	A port must have the following:
104	- It must meet object default standards.
105	- A link can only connect to 2 nodes
106	- The nodes that a link is connected to must be in the parent Topology's nodes list
107	:param time: time being validated
108	:return: A list of any issues in the data.
109	"""
110	errors = []	×
111	if not match(ISO_TIME_FORMAT, time):	×
112	errors.append(	×
113	"{} time needs to be in full ISO format".format(
114	time,
115	)
116	)
117	return errors	×
118
119	def _validate_object_defaults(self, sdx_object):	4✔
120	"""
121	Validate that the object provided default fields meets the XSD standards.
122	The object must have the following:
123	- The object must have an ID
124	- The object ID must be a string
125	- The object must have a name
126	- The object name must be a string
127	- If the object has a short name, it must be a string
128	- If the object has a version, it must be a string in ISO format
129	- All the additional properties on the object are proper
130	:param sdx_object: The sdx Model Object being evaluated.
131	:return: A list of any issues in the data.
132	"""
133	errors = []	4✔
134	if not sdx_object._id:	4!
135	errors.append(	×
136	"{} must have an ID".format(sdx_object.__class__.__name__)
137	)
138	if not isinstance(sdx_object._id, str):	4!
139	errors.append(	×
140	"{} ID must be a string".format(sdx_object.__class__.__name__)
141	)
142	if not sdx_object._name:	4!
143	errors.append(	×
144	"{} {} must have a name".format(
145	sdx_object.__class__.__name__,
146	sdx_object._name,
147	)
148	)
149	if not isinstance(sdx_object._name, str):	4!
150	errors.append(	×
151	"{} {} name must be a String".format(
152	sdx_object.__class__.__name__,
153	sdx_object._name,
154	)
155	)
156
157	return errors	4✔

atlanticwave-sdx / datamodel / 4533308467

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