Newer
Older
Lluis Gifre Renom
committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import grpc, logging
from typing import Dict, Set, Tuple, Union
from common.Checkers import chk_string
from common.exceptions.ServiceException import ServiceException
from common.database.api.context.Constants import DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID
def check_endpoint_id(
logger : logging.Logger, endpoint_number : int, parent_name : str, endpoint_id : 'EndpointId',
add_topology_devices_endpoints : Dict[str, Dict[str, Set[str]]],
default_device_if_empty : Union[str, None] = None,
predefined_context_id : str = DEFAULT_CONTEXT_ID, acceptable_context_ids : Set[str] = set([DEFAULT_CONTEXT_ID]),
predefined_topology_id : str = DEFAULT_TOPOLOGY_ID, acceptable_topology_ids : Set[str] = set([DEFAULT_TOPOLOGY_ID])
) -> Tuple[str, str, str]:
try:
ep_context_id = chk_string('endpoint_id[#{}].topoId.contextId.contextUuid.uuid'.format(endpoint_number),
endpoint_id.topoId.contextId.contextUuid.uuid,
allow_empty=True)
ep_topology_id = chk_string('endpoint_id[#{}].topoId.topoId.uuid'.format(endpoint_number),
endpoint_id.topoId.topoId.uuid,
allow_empty=True)
ep_device_id = chk_string('endpoint_id[#{}].dev_id.device_id.uuid'.format(endpoint_number),
endpoint_id.dev_id.device_id.uuid,
allow_empty=(default_device_if_empty is not None))
ep_port_id = chk_string('endpoint_id[#{}].port_id.uuid'.format(endpoint_number),
endpoint_id.port_id.uuid,
allow_empty=False)
except Exception as e:
logger.exception('Invalid arguments:')
raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))
if len(ep_context_id) == 0:
# Assumption: if no context is specified for an endpoint_id, use parent context
ep_context_id = predefined_context_id
elif ep_context_id not in acceptable_context_ids:
# Assumption: parent and endpoints should belong to the same context
msg = ' '.join([
'Context({}) in {} mismatches acceptable Contexts({}).',
'Optionally, leave field empty to use predefined Context({}).',
])
msg = msg.format(ep_context_id, parent_name, str(acceptable_context_ids), predefined_context_id)
raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)
if len(ep_topology_id) == 0:
# Assumption: if no topology is specified for an endpoint_id, use parent topology
ep_topology_id = predefined_topology_id
elif ep_topology_id not in acceptable_topology_ids:
msg = ' '.join([
'Topology({}) in {} mismatches acceptable Topologies({}).',
'Optionally, leave field empty to use predefined Topology({}).',
])
msg = msg.format(ep_topology_id, parent_name, str(acceptable_topology_ids), predefined_topology_id)
raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)
if (len(ep_device_id) == 0) and (default_device_if_empty is not None):
ep_device_id = default_device_if_empty
add_devices = add_topology_devices_endpoints.setdefault(ep_topology_id, dict())
if ep_device_id in add_devices:
msg = 'Duplicated Context({})/Topology({})/Device({}) in {}.'
msg = msg.format(ep_context_id, ep_topology_id, ep_device_id, parent_name)
raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)
add_device_and_endpoints = add_devices.setdefault(ep_device_id, set())
# Implicit validation: same device cannot appear 2 times in the list of endpoints
if ep_port_id in add_device_and_endpoints: # pragma: no cover
msg = 'Duplicated Context({})/Topology({})/Device({})/Port({}) in {}.'
msg = msg.format(ep_context_id, ep_topology_id, ep_device_id, ep_port_id, parent_name)
raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)
add_device_and_endpoints.add(ep_port_id)
return ep_topology_id, ep_device_id, ep_port_id