4565137249

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #62 from atlanticwave-sdx/61.connection-validation

Run Details

199 of 400 branches covered (49.75%)

Branch coverage included in aggregate %.

116 of 125 new or added lines in 11 files covered. (92.8%)

25 existing lines in 4 files now uncovered.

945 of 1194 relevant lines covered (79.15%)

2.37 hits per line

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

68.75

/src/sdx/datamodel/validation/topologyvalidator.py

"""
Checks for Topology objects to be in the expected format.
"""
from re import match

from sdx.datamodel.models.link import Link
from sdx.datamodel.models.location import Location
from sdx.datamodel.models.node import Node
from sdx.datamodel.models.service import Service
from sdx.datamodel.models.topology import SDX_INSTITUTION_ID, Topology

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


class TopologyValidator:
    """
    The validation class made to validate a Topology
    """

    def __init__(self, topology: Topology):
        if not isinstance(topology, Topology):
            raise ValueError("TopologyValidator expects a Topology object")
        self._topology = topology

    def is_valid(self) -> bool:
        errors = self.validate(self._topology, raise_error=False)
        for error in errors:
            print(error)
        return not bool(errors)

    def validate(self, topology=None, raise_error=True) -> [str]:
        if not topology and self._topology:
            topology = self._topology
        errors = self._validate_topology(topology)
        if errors and raise_error:
            raise ValueError("\n".join(errors))
        return errors

    def _validate_topology(self, topology: Topology):
        """
        Validate that the topology provided meets the JSON schema.

        A topology must have the following:

            - It must meet object standard

            - It must have the default fields: id, name, version,
              time_stamp, nodes, and links

            - It must have a Primary owner assigned

            - It must have its primary owner in its institution list

            - It must have the global institution in the institution
              list

        :param topology: The topology being evaluated

        :return: A list of any issues in the data.
        """
        errors = []
        errors += self._validate_object_defaults(topology)

        if SDX_INSTITUTION_ID not in topology.id:
            errors.append(
                f"Global Institution must be in topology {topology.id}"
            )

        service = topology.get_domain_service()
        if service is not None:
            errors += self._validate_service(service, topology)
        for node in topology.get_nodes():
            errors += self._validate_node(node, topology)
        for link in topology.get_links():
            errors += self._validate_link(link, topology)

        return errors

    def _validate_service(self, service: Service, topology: Topology):
        """
        Check that institution descritpion meets the XSD standards.

        A institution must have the following:

            - It must meet object default standards.

            - It's Location values are valid

            - The Institution Type must be in the list of valid
              Institution types

        :param institution: The Institution being evaluated.
        :param topology: The Parent Topology.

        :return: A list of any issues in the data.
        """
        errors = []
        errors += self._validate_object_defaults(service)

        return errors

    def _validate_version(self, version, time_stamp, topology: Topology):
        """
        Validate that version and time_stamp meets ISO standards.

            - It must meet object default standards.

            - It's Location values are valid

            - The Institution Type must be in the list of valid
              Institution types

        :param version: The topology version.
        :param time_stamp: The topology time stamp.
        :param topology: The Parent Topology.

        :return: A list of any issues in the data.
        """
        errors = []
        if version:
            if not isinstance(version, str):
                errors.append(f"{topology.id} version must be a string")
            elif not match(ISO_FORMAT, version):
                errors.append(
                    f"{topology.id} version must be datetime ISO format"
                )

        if not match(ISO_FORMAT, time_stamp):
            errors.append(
                f"time_stamp {time_stamp} needs to be in full ISO format"
            )

        return errors

    def _validate_node(self, node: Node, topology: Topology):
        """
        Validate that the node provided meets the XSD standards.

        A node must have the following:

            - It must meet object default standards.

            - It's Location values are valid

        :param node: The Node being evaluated.
        :param topology: The Parent Topology.

        :return: A list of any issues in the data.
        """
        errors = []

        errors += self._validate_object_defaults(node)
        errors += self._validate_location(node.get_location())

        return errors

    def _validate_link(self, link: Link, topology: Topology):
        """
        Validate that the link provided meets the XSD standards.

        A link 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 link: The Link being evaluated.
        :param topology: The Parent Topology.

        :return: A list of any issues in the data.
        """
        errors = []
        errors += self._validate_object_defaults(link)

        if len(link._ports) != 2:
            errors.append(
                f"Link {link.id} must connect between 2 ports. "
                f"Currently {link._ports}"
            )
        for port in link._ports:
            if not isinstance(port, (dict, str)):
                errors.append(
                    f"Link {link.id} Port {port} should be "
                    f"a string or a dict. Not {port.__class__.__name__}"
                )
        # TODO: Check ports are in the current topology

        return errors

    def _validate_object_defaults(self, sdx_object):
        """
        Validate that object fields meets 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} must have a name"
            )
        if not isinstance(sdx_object._name, str):
            errors.append(
                f"{sdx_object.__class__.__name__} {sdx_object._name} name must be a string"
            )

        return errors

    def _validate_location(self, location: Location, enforce_coordinates=True):
        """
        Validate that the object location fields meets the XSD
        standards.  The location must have the following:

            - A location must have a longitude

            - A location's longitude muse be a floating point value

            - A location's longitude must be between -180 and -180

            - A location must have a latitude

            - A location must be a floating point value

            - A location's latitude must be between -90 and 90

            - A location's altitude must be a floating point value

            - A location's UN/LOCODE must be a string value

            - A location's address must be a string or a list of
              strings

        :param location: The Location Object being evaluated.
        :param enforce_coordinates: A boolean determining if longitude
            and latitude should be enforced

        :return: A list of any issues in the data.
        """
        errors = []
        if location.longitude is None and enforce_coordinates:
            errors.append(
                f"{location.__class__.__name__} {location.id} "
                f"Longitude must be set to a value"
            )
        try:
            if location.longitude is not None:
                if not -180 <= float(location.longitude) <= 180:
                    errors.append(
                        f"{location.__class__.__name__} {location.id} "
                        f"Longitude must be a value that is between -180 and 180"
                    )
        except ValueError:
            errors.append(
                f"{location.__class__.__name__} {location.id} "
                f"Longitude must be a floating point value"
            )

        if location.latitude is None and enforce_coordinates:
            errors.append(
                f"{location.__class__.__name__} {location.id} "
                f"Latitude must be set to a value"
            )
        try:
            if location.latitude is not None:
                if not -90 <= float(location.latitude) <= 90:
                    errors.append(
                        f"{location.__class__.__name__} {location.id} "
                        f"Latitude must be a value that is between -90 and 90"
                    )
        except ValueError:
            errors.append(
                f"{location.__class__.__name__} {location.id} "
                f"Latitude must be a floating point value"
            )

        try:
            if location.latitude:
                float(location.latitude)
        except ValueError:
            errors.append(
                f"{location.__class__.__name__} {location.id} "
                f"Latitude must be a floating point value"
            )

        if not location.address:
            errors.append(
                f"{location.__class__.__name__} {location._id} "
                f"Address must exist"
            )
        if not type(location.address) == str:
            errors.append(
                f"{location.__class__.__name__} {location._id} "
                f"Address {location.address} must be a string"
            )

        return errors

1	"""
2	Checks for Topology objects to be in the expected format.
3	"""
4	from re import match	3✔
5
6	from sdx.datamodel.models.link import Link	3✔
7	from sdx.datamodel.models.location import Location	3✔
8	from sdx.datamodel.models.node import Node	3✔
9	from sdx.datamodel.models.service import Service	3✔
10	from sdx.datamodel.models.topology import SDX_INSTITUTION_ID, Topology	3✔
11
12	ISO_FORMAT = r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[-+]\d{2}:\d{2}"	3✔
13
14
15	class TopologyValidator:	3✔
16	"""
17	The validation class made to validate a Topology
18	"""
19
20	def __init__(self, topology: Topology):	3✔
21	if not isinstance(topology, Topology):	3!
NEW 22	raise ValueError("TopologyValidator expects a Topology object")	×
23	self._topology = topology	3✔
24
25	def is_valid(self) -> bool:	3✔
26	errors = self.validate(self._topology, raise_error=False)	3✔
27	for error in errors:	3✔
28	print(error)	3✔
29	return not bool(errors)	3✔
30
31	def validate(self, topology=None, raise_error=True) -> [str]:	3✔
32	if not topology and self._topology:	3✔
33	topology = self._topology	3✔
34	errors = self._validate_topology(topology)	3✔
35	if errors and raise_error:	3✔
36	raise ValueError("\n".join(errors))	3✔
37	return errors	3✔
38
39	def _validate_topology(self, topology: Topology):	3✔
40	"""
41	Validate that the topology provided meets the JSON schema.
42
43	A topology must have the following:
44
45	- It must meet object standard
46
47	- It must have the default fields: id, name, version,
48	time_stamp, nodes, and links
49
50	- It must have a Primary owner assigned
51
52	- It must have its primary owner in its institution list
53
54	- It must have the global institution in the institution
55	list
56
57	:param topology: The topology being evaluated
58
59	:return: A list of any issues in the data.
60	"""
61	errors = []	3✔
62	errors += self._validate_object_defaults(topology)	3✔
63
64	if SDX_INSTITUTION_ID not in topology.id:	3✔
65	errors.append(	3✔
66	f"Global Institution must be in topology {topology.id}"
67	)
68
69	service = topology.get_domain_service()	3✔
70	if service is not None:	3✔
71	errors += self._validate_service(service, topology)	3✔
72	for node in topology.get_nodes():	3✔
73	errors += self._validate_node(node, topology)	3✔
74	for link in topology.get_links():	3✔
75	errors += self._validate_link(link, topology)	3✔
76
77	return errors	3✔
78
79	def _validate_service(self, service: Service, topology: Topology):	3✔
80	"""
81	Check that institution descritpion meets the XSD standards.
82
83	A institution must have the following:
84
85	- It must meet object default standards.
86
87	- It's Location values are valid
88
89	- The Institution Type must be in the list of valid
90	Institution types
91
92	:param institution: The Institution being evaluated.
93	:param topology: The Parent Topology.
94
95	:return: A list of any issues in the data.
96	"""
97	errors = []	3✔
98	errors += self._validate_object_defaults(service)	3✔
99
100	return errors	3✔
101
102	def _validate_version(self, version, time_stamp, topology: Topology):	3✔
103	"""
104	Validate that version and time_stamp meets ISO standards.
105
106	- It must meet object default standards.
107
108	- It's Location values are valid
109
110	- The Institution Type must be in the list of valid
111	Institution types
112
113	:param version: The topology version.
114	:param time_stamp: The topology time stamp.
115	:param topology: The Parent Topology.
116
117	:return: A list of any issues in the data.
118	"""
UNCOV 119	errors = []	×
120	if version:	×
121	if not isinstance(version, str):	×
NEW 122	errors.append(f"{topology.id} version must be a string")	×
123	elif not match(ISO_FORMAT, version):	×
124	errors.append(	×
125	f"{topology.id} version must be datetime ISO format"
126	)
127
UNCOV 128	if not match(ISO_FORMAT, time_stamp):	×
129	errors.append(	×
130	f"time_stamp {time_stamp} needs to be in full ISO format"
131	)
132
UNCOV 133	return errors	×
134
135	def _validate_node(self, node: Node, topology: Topology):	3✔
136	"""
137	Validate that the node provided meets the XSD standards.
138
139	A node must have the following:
140
141	- It must meet object default standards.
142
143	- It's Location values are valid
144
145	:param node: The Node being evaluated.
146	:param topology: The Parent Topology.
147
148	:return: A list of any issues in the data.
149	"""
150	errors = []	3✔
151
152	errors += self._validate_object_defaults(node)	3✔
153	errors += self._validate_location(node.get_location())	3✔
154
155	return errors	3✔
156
157	def _validate_link(self, link: Link, topology: Topology):	3✔
158	"""
159	Validate that the link provided meets the XSD standards.
160
161	A link must have the following:
162
163	- It must meet object default standards.
164
165	- A link can only connect to 2 nodes
166
167	- The nodes that a link is connected to must be in the
168	parent Topology's nodes list
169
170	:param link: The Link being evaluated.
171	:param topology: The Parent Topology.
172
173	:return: A list of any issues in the data.
174	"""
175	errors = []	3✔
176	errors += self._validate_object_defaults(link)	3✔
177
178	if len(link._ports) != 2:	3!
UNCOV 179	errors.append(	×
180	f"Link {link.id} must connect between 2 ports. "
181	f"Currently {link._ports}"
182	)
183	for port in link._ports:	3✔
184	if not isinstance(port, (dict, str)):	3!
UNCOV 185	errors.append(	×
186	f"Link {link.id} Port {port} should be "
187	f"a string or a dict. Not {port.__class__.__name__}"
188	)
189	# TODO: Check ports are in the current topology
190
191	return errors	3✔
192
193	def _validate_object_defaults(self, sdx_object):	3✔
194	"""
195	Validate that object fields meets XSD standards.
196
197	The object must have the following:
198
199	- The object must have an ID
200
201	- The object ID must be a string
202
203	- The object must have a name
204
205	- The object name must be a string
206
207	- If the object has a short name, it must be a string
208
209	- If the object has a version, it must be a string in ISO
210	format
211
212	- All the additional properties on the object are proper
213
214	:param sdx_object: The sdx Model Object being evaluated.
215
216	:return: A list of any issues in the data.
217	"""
218	errors = []	3✔
219	if not sdx_object._id:	3!
NEW 220	errors.append(f"{sdx_object.__class__.__name__} must have an ID")	×
221	if not isinstance(sdx_object._id, str):	3!
UNCOV 222	errors.append(	×
223	f"{sdx_object.__class__.__name__} ID must be a string"
224	)
225	if not sdx_object._name:	3!
UNCOV 226	errors.append(	×
227	f"{sdx_object.__class__.__name__} {sdx_object._name} must have a name"
228	)
229	if not isinstance(sdx_object._name, str):	3!
UNCOV 230	errors.append(	×
231	f"{sdx_object.__class__.__name__} {sdx_object._name} name must be a string"
232	)
233
234	return errors	3✔
235
236	def _validate_location(self, location: Location, enforce_coordinates=True):	3✔
237	"""
238	Validate that the object location fields meets the XSD
239	standards. The location must have the following:
240
241	- A location must have a longitude
242
243	- A location's longitude muse be a floating point value
244
245	- A location's longitude must be between -180 and -180
246
247	- A location must have a latitude
248
249	- A location must be a floating point value
250
251	- A location's latitude must be between -90 and 90
252
253	- A location's altitude must be a floating point value
254
255	- A location's UN/LOCODE must be a string value
256
257	- A location's address must be a string or a list of
258	strings
259
260	:param location: The Location Object being evaluated.
261	:param enforce_coordinates: A boolean determining if longitude
262	and latitude should be enforced
263
264	:return: A list of any issues in the data.
265	"""
266	errors = []	3✔
267	if location.longitude is None and enforce_coordinates:	3!
UNCOV 268	errors.append(	×
269	f"{location.__class__.__name__} {location.id} "
270	f"Longitude must be set to a value"
271	)
272	try:	3✔
273	if location.longitude is not None:	3!
274	if not -180 <= float(location.longitude) <= 180:	3!
UNCOV 275	errors.append(	×
276	f"{location.__class__.__name__} {location.id} "
277	f"Longitude must be a value that is between -180 and 180"
278	)
UNCOV 279	except ValueError:	×
280	errors.append(	×
281	f"{location.__class__.__name__} {location.id} "
282	f"Longitude must be a floating point value"
283	)
284
285	if location.latitude is None and enforce_coordinates:	3!
UNCOV 286	errors.append(	×
287	f"{location.__class__.__name__} {location.id} "
288	f"Latitude must be set to a value"
289	)
290	try:	3✔
291	if location.latitude is not None:	3!
292	if not -90 <= float(location.latitude) <= 90:	3!
UNCOV 293	errors.append(	×
294	f"{location.__class__.__name__} {location.id} "
295	f"Latitude must be a value that is between -90 and 90"
296	)
UNCOV 297	except ValueError:	×
298	errors.append(	×
299	f"{location.__class__.__name__} {location.id} "
300	f"Latitude must be a floating point value"
301	)
302
303	try:	3✔
304	if location.latitude:	3!
305	float(location.latitude)	3✔
UNCOV 306	except ValueError:	×
307	errors.append(	×
308	f"{location.__class__.__name__} {location.id} "
309	f"Latitude must be a floating point value"
310	)
311
312	if not location.address:	3✔
313	errors.append(	3✔
314	f"{location.__class__.__name__} {location._id} "
315	f"Address must exist"
316	)
317	if not type(location.address) == str:	3!
UNCOV 318	errors.append(	×
319	f"{location.__class__.__name__} {location._id} "
320	f"Address {location.address} must be a string"
321	)
322
323	return errors	3✔

atlanticwave-sdx / datamodel / 4565137249

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