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
import json, logging, requests
from requests.auth import HTTPBasicAuth
from typing import Optional
LOGGER = logging.getLogger(__name__)
HTTP_OK_CODES = {
200, # OK
201, # Created
202, # Accepted
204, # No Content
}
def create_connectivity_service(
root_url, uuid, node_id_src, tp_id_src, node_id_dst, tp_id_dst, vlan_id,
auth : Optional[HTTPBasicAuth] = None, timeout : Optional[int] = None
):
url = '{:s}/nmswebs/restconf/data/ietf-eth-tran-service:etht-svc'.format(root_url)
headers = {'content-type': 'application/json'}
data = {
'etht-svc-instances': [
{
'etht-svc-name': uuid,
'etht-svc-type': 'ietf-eth-tran-types:p2p-svc',
'etht-svc-end-points': [
{
'etht-svc-access-points': [
{'access-node-id': node_id_src, 'access-ltp-id': tp_id_src, 'access-point-id': '1'}
],
'outer-tag': {'vlan-value': vlan_id, 'tag-type': 'ietf-eth-tran-types:classify-c-vlan'},
'etht-svc-end-point-name': '{:s}:{:s}'.format(str(node_id_src), str(tp_id_src)),
'service-classification-type': 'ietf-eth-tran-types:vlan-classification'
},
{
'etht-svc-access-points': [
{'access-node-id': node_id_dst, 'access-ltp-id': tp_id_dst, 'access-point-id': '2'}
],
'outer-tag': {'vlan-value': vlan_id, 'tag-type': 'ietf-eth-tran-types:classify-c-vlan'},
'etht-svc-end-point-name': '{:s}:{:s}'.format(str(node_id_dst), str(tp_id_dst)),
'service-classification-type': 'ietf-eth-tran-types:vlan-classification'
}
]
}
]
}
results = []
try:
LOGGER.info('Connectivity service {:s}: {:s}'.format(str(uuid), str(data)))
response = requests.post(
url=url, data=json.dumps(data), timeout=timeout, headers=headers, verify=False, auth=auth)
LOGGER.info('Microwave Driver response: {:s}'.format(str(response)))
except Exception as e: # pylint: disable=broad-except
LOGGER.exception('Exception creating ConnectivityService(uuid={:s}, data={:s})'.format(str(uuid), str(data)))
results.append(e)
else:
if response.status_code not in HTTP_OK_CODES:
msg = 'Could not create ConnectivityService(uuid={:s}, data={:s}). status_code={:s} reply={:s}'
LOGGER.error(msg.format(str(uuid), str(data), str(response.status_code), str(response)))
results.append(response.status_code in HTTP_OK_CODES)
return results
def delete_connectivity_service(root_url, uuid, auth : Optional[HTTPBasicAuth] = None, timeout : Optional[int] = None):
url = '{:s}/nmswebs/restconf/data/ietf-eth-tran-service:etht-svc/etht-svc-instances={:s}'
url = url.format(root_url, uuid)
results = []
try:
response = requests.delete(url=url, timeout=timeout, verify=False, auth=auth)
except Exception as e: # pylint: disable=broad-except
LOGGER.exception('Exception deleting ConnectivityService(uuid={:s})'.format(str(uuid)))
results.append(e)
else:
if response.status_code not in HTTP_OK_CODES:
msg = 'Could not delete ConnectivityService(uuid={:s}). status_code={:s} reply={:s}'
LOGGER.error(msg.format(str(uuid), str(response.status_code), str(response)))
results.append(response.status_code in HTTP_OK_CODES)
return results
if __name__ == '__main__':
ROOT_URL = 'https://127.0.0.1:8443'
SERVICE_UUID = 'my-service'
create_connectivity_service(ROOT_URL, SERVICE_UUID, '172.18.0.1', '1', '172.18.0.2', '2', 300)
delete_connectivity_service(ROOT_URL, SERVICE_UUID)