Newer
Older
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
import copy
from typing import Dict, List, Tuple
from common.DeviceTypes import DeviceTypeEnum
from common.tools.object_factory.ConfigRule import json_config_rule_set
from context.proto.context_pb2 import DeviceDriverEnum, DeviceOperationalStatusEnum
DEVICE_DISABLED = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_DISABLED
DEVICE_EMUPR_TYPE = DeviceTypeEnum.EMULATED_PACKET_ROUTER.value
DEVICE_EMUOLS_TYPE = DeviceTypeEnum.EMULATED_OPTICAL_LINE_SYSTEM.value
DEVICE_EMU_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_UNDEFINED]
DEVICE_EMU_ADDRESS = '127.0.0.1'
DEVICE_EMU_PORT = '0'
DEVICE_PR_TYPE = DeviceTypeEnum.PACKET_ROUTER.value
DEVICE_PR_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_OPENCONFIG]
DEVICE_TAPI_TYPE = DeviceTypeEnum.OPTICAL_LINE_SYSTEM.value
DEVICE_TAPI_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_TRANSPORT_API]
DEVICE_P4_TYPE = DeviceTypeEnum.P4_SWITCH.value
DEVICE_P4_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_P4]
def json_device_id(device_uuid : str):
return {'device_uuid': {'uuid': device_uuid}}
def json_device(
device_uuid : str, device_type : str, status : DeviceOperationalStatusEnum, endpoints : List[Dict] = [],
config_rules : List[Dict] = [], drivers : List[Dict] = []
):
return {
'device_id' : json_device_id(device_uuid),
'device_type' : device_type,
'device_config' : {'config_rules': copy.deepcopy(config_rules)},
'device_operational_status': status,
'device_drivers' : copy.deepcopy(drivers),
'device_endpoints' : copy.deepcopy(endpoints),
}
def json_device_emulated_packet_router_disabled(
device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
drivers : List[Dict] = DEVICE_EMU_DRIVERS
):
return json_device(
device_uuid, DEVICE_EMUPR_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules,
drivers=drivers)
def json_device_emulated_tapi_disabled(
device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
drivers : List[Dict] = DEVICE_EMU_DRIVERS
):
return json_device(
device_uuid, DEVICE_EMUOLS_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules,
drivers=drivers)
def json_device_packetrouter_disabled(
device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
drivers : List[Dict] = DEVICE_PR_DRIVERS
):
return json_device(
device_uuid, DEVICE_PR_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers)
def json_device_tapi_disabled(
device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
drivers : List[Dict] = DEVICE_TAPI_DRIVERS
):
return json_device(
device_uuid, DEVICE_TAPI_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers)
def json_device_p4_disabled(
device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
drivers : List[Dict] = DEVICE_P4_DRIVERS
):
return json_device(
device_uuid, DEVICE_P4_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers)
def json_device_connect_rules(address : str, port : int, settings : Dict = {}):
return [
json_config_rule_set('_connect/address', address),
json_config_rule_set('_connect/port', port),
json_config_rule_set('_connect/settings', settings),
]
def json_device_emulated_connect_rules(
endpoint_descriptors : List[Tuple[str, str, List[int]]], address : str = DEVICE_EMU_ADDRESS,
port : int = DEVICE_EMU_PORT
):
settings = {'endpoints': [
{'uuid': endpoint_uuid, 'type': endpoint_type, 'sample_types': sample_types}
for endpoint_uuid,endpoint_type,sample_types in endpoint_descriptors
]}
return json_device_connect_rules(address, port, settings=settings)