-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathrouter.py
176 lines (139 loc) · 6.04 KB
/
router.py
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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
# Jinja2 Template Engine
from jinja2 import Template, Environment
import napalm
import const
# arranged print
from pprint import pprint, pformat
class Router:
def __init__(self, hostname, os, ipaddress, username, password):
self.hostname = hostname
self.username = username
self.password = password
self.ipaddress = ipaddress
self.os = os
self.driver = napalm.get_network_driver(self.os)
self.device = self.driver(hostname=ipaddress, username=username, password=password)
def open(self):
self.device.open()
def close(self):
self.device.close()
def commit(self):
try:
self.device.commit_config()
commit_result = True
except:
commit_result = False
return commit_result
def discard_config(self):
return self.device.discard_config()
def compare_config(self):
return self.device.compare_config()
def get_config(self):
return self.device.get_config()
def replace(self,config):
self.device.load_replace_candidate(config=config)
return True
def check_hostname(self):
hostname_fetched = self.device.get_facts()['hostname']
if hostname_fetched == self.hostname :
return True, hostname_fetched
else:
return False, hostname_fetched
def call_getters(self,func_name,extra_param=None):
if extra_param == None:
return eval('self.device.'+func_name)()
else:
return eval('self.device.'+func_name)(extra_param)
def load_config(self, operation_name, operation_param=None):
result = False
message = ''
if operation_name == 'set_add_interface_ipv4':
template_filename = './set_templates/' + self.os + '/add_interface_ipv4.j2'
template_param = operation_param
elif operation_name == 'set_add_network_ipv4':
template_filename = './set_templates/' + self.os + '/add_network_ipv4.j2'
template_param = operation_param
elif operation_name == 'set_add_bgp_neighbor_ipv4':
template_filename = './set_templates/' + self.os + '/add_bgp_neighbor_ipv4.j2'
template_param = operation_param
elif operation_name == 'set_add_bgp_policy_ipv4':
template_filename = './set_templates/' + self.os + '/add_bgp_policy_ipv4.j2'
template_param = operation_param
else:
pass
try:
# This is only brawsing for check candidate configuration before setting the router.
config_txt = self.generate_from_jinja2(
template_filename,
template_param)
#print(config_txt) #debug point
self.device.load_merge_candidate(config=config_txt)
#TODO: you can use load_template() instead of load_merge_candidate().
message = config_txt
result = True
except Exception as err:
result = False
print ("Error : "+str(err))
message= str(err)
return result, message
def validate_operation(self, validate_dst):
base_str = ''
for validate_oper in validate_dst:
if isinstance(validate_oper,dict):
validate_filename = list(validate_oper.keys())[0]
else:
validate_filename = validate_oper
rule_path=const.VALIDATE_TEMPLATE_PATH+validate_filename+'.j2'
temp_param = self.allocate_validation_param(validate_filename,validate_oper)
if temp_param:
base_str += self.generate_from_jinja2(rule_path,temp_param)
yml_path = self.save_as_yml(base_str,const.VALIDATE_RULE_PATH)
return self.device.compliance_report(yml_path)
def allocate_validation_param(self,oper_name ,oper_dict):
if oper_name in 'interfaces':
ifvalidate_list = {'interfaces':{'interfaces_name':[]}}
facts_result = self.call_getters('get_facts')
for if_name in facts_result['interface_list']:
for if_prefix in const.IF_PLEFIX_LIST[self.os]:
if if_name.startswith(if_prefix):
ifvalidate_list['interfaces']['interfaces_name'].append(if_name)
return ifvalidate_list
elif oper_name in 'environment':
envvalidate_param = {'cards':[]}
env_result = self.call_getters('get_environment')
for env_order, env_param in env_result.items():
if env_order in 'cpu':
for card_name in env_param.keys():
envvalidate_param['cards'].append({
'card_name' : card_name,
'cpu_maxrate': oper_dict['environment']['cpu_maxrate']
})
elif env_order in 'memory':
envvalidate_param['memory_maxrate'] = oper_dict['environment']['memory_maxrate']
elif env_order in 'fans':
#TBD
pass
elif env_order in 'temperature':
#TBD
pass
return envvalidate_param
elif oper_name in 'bgp_summary':
bgp_result = self.call_getters('get_bgp_neighbors')
if len(bgp_result) != 0:
return{'bgp_neighbor' : list(bgp_result['global']['peers'].keys())}
else:
return False
else:
return oper_dict
def save_as_yml(self,yml_data,save_dir):
save_filepath = save_dir+const.VALIDATE_RULE_FILENAME
with open(save_filepath, 'w') as f:
f.write(yml_data)
f.close
return save_filepath
def generate_from_jinja2(self, template_filename, template_param):
# read template file (jinja2 format)
with open(template_filename, 'r') as f:
template_jinja2 = f.read()
# generate nwtest file from template file
return Environment().from_string(template_jinja2).render(template_param)