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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
import grpc, logging
from typing import Dict, List, Set, Tuple
from common.Checkers import chk_options, chk_string
from common.database.api.Database import Database
from common.database.api.context.Constants import DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID
from common.database.api.context.topology.device.Endpoint import Endpoint
from common.database.api.context.topology.device.OperationalStatus import OperationalStatus, \
operationalstatus_enum_values, to_operationalstatus_enum
from common.exceptions.ServiceException import ServiceException
from common.tools.service.DeviceCheckers import check_device_endpoint_exists
from common.tools.service.EndpointIdCheckers import check_endpoint_id
from common.tools.service.EnumCheckers import check_enum
from common.tools.service.DeviceCheckers import check_device_exists, check_device_not_exists
from context.proto.context_pb2 import Device, DeviceId
# For each method name, define acceptable device operational statuses. Empty set means accept all.
ACCEPTED_DEVICE_OPERATIONAL_STATUSES : Dict[str, Set[OperationalStatus]] = {
'CreateService': set([OperationalStatus.ENABLED, OperationalStatus.DISABLED]),
'UpdateService': set([OperationalStatus.KEEP_STATE, OperationalStatus.ENABLED, OperationalStatus.DISABLED]),
}
def _check_device_exists(method_name : str, database : Database, device_id : str):
if method_name in ['CreateDevice']:
check_device_not_exists(database, DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID, device_id)
elif method_name in ['UpdateDevice', 'DeleteDevice']:
check_device_exists(database, DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID, device_id)
else: # pragma: no cover (test requires malforming the code)
msg = 'Unexpected condition [_check_device_exists(method_name={}, device_id={})]'
msg = msg.format(str(method_name), str(device_id))
raise ServiceException(grpc.StatusCode.UNIMPLEMENTED, msg)
def check_device_operational_status(method_name : str, value : str) -> OperationalStatus:
return check_enum(
'OperationalStatus', method_name, value, to_operationalstatus_enum, ACCEPTED_DEVICE_OPERATIONAL_STATUSES)
def check_device_request(
method_name : str, request : Device, database : Database, logger : logging.Logger
) -> Tuple[str, str, str, OperationalStatus, List[Tuple[Endpoint, str]]]:
# ----- Parse attributes -------------------------------------------------------------------------------------------
try:
device_id = chk_string ('device.device_id.device_id.uuid',
request.device_id.device_id.uuid,
allow_empty=False)
device_type = chk_string ('device.device_type',
request.device_type,
allow_empty=False)
device_config = chk_string ('device.device_config.device_config',
request.device_config.device_config,
allow_empty=True)
device_opstat = chk_options('device.devOperationalStatus',
request.devOperationalStatus,
operationalstatus_enum_values())
except Exception as e:
logger.exception('Invalid arguments:')
raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))
device_opstat = check_device_operational_status(method_name, device_opstat)
# ----- Check if device exists in database -------------------------------------------------------------------------
_check_device_exists(method_name, database, device_id)
# ----- Parse endpoints and check if they exist in the database as device endpoints --------------------------------
add_topology_devices_endpoints : Dict[str, Dict[str, Set[str]]] = {}
db_endpoints__port_types : List[Tuple[Endpoint, str]] = []
for endpoint_number,endpoint in enumerate(request.endpointList):
parent_name = 'Endpoint(#{}) of Context({})/Topology({})/Device({})'
parent_name = parent_name.format(endpoint_number, DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID, device_id)
_, ep_device_id, ep_port_id = check_endpoint_id(
logger, endpoint_number, parent_name, endpoint.port_id, add_topology_devices_endpoints,
default_device_if_empty=device_id)
try:
ep_port_type = chk_string('endpoint[#{}].port_type'.format(endpoint_number),
endpoint.port_type,
allow_empty=False)
except Exception as e:
logger.exception('Invalid arguments:')
raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))
db_endpoint = check_device_endpoint_exists(
database, parent_name, DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID, ep_device_id, ep_port_id)
db_endpoints__port_types.append((db_endpoint, ep_port_type))
return device_id, device_type, device_config, device_opstat, db_endpoints__port_types
def check_device_id_request(
method_name : str, request : DeviceId, database : Database, logger : logging.Logger) -> str:
# ----- Parse attributes -------------------------------------------------------------------------------------------
try:
device_id = chk_string('device_id.device_id.uuid',
request.device_id.uuid,
allow_empty=False)
except Exception as e:
logger.exception('Invalid arguments:')
raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))
# ----- Check if device exists in database ---------------------------------------------------------------------------
_check_device_exists(method_name, database, device_id)
return device_id