Skip to content

Commit 0eb664d

Browse files
simingysimingy
authored
Merge pull request #5 from RunSi/master
Refactored CRC code
2 parents 258fb9e + ceedf50 commit 0eb664d

File tree

1 file changed

+83
-73
lines changed

1 file changed

+83
-73
lines changed

‎crc_errors/CRC_Count_check.py

+83-73
Original file line numberDiff line numberDiff line change
@@ -7,117 +7,125 @@
77
from tabulate import tabulate
88

99
# Needed for aetest script
10-
from ats import aetest
11-
from ats.log.utils import banner
10+
from pyats import aetest
11+
from pyats.log.utils import banner
1212

1313
# Genie Imports
14-
from genie.conf import Genie
15-
from genie.abstract import Lookup
14+
from genie.testbed import load
1615

17-
# import the genie libs
18-
from genie.libs import ops # noqa
1916

2017
# Get your logger for your script
2118
log = logging.getLogger(__name__)
2219

23-
2420
###################################################################
2521
# COMMON SETUP SECTION #
2622
###################################################################
2723

2824
class common_setup(aetest.CommonSetup):
2925
""" Common Setup section """
3026

31-
# CommonSetup have subsection.
32-
# You can have 1 to as many subsection as wanted
33-
3427
# Connect to each device in the testbed
3528
@aetest.subsection
3629
def connect(self, testbed):
37-
genie_testbed = Genie.init(testbed)
30+
genie_testbed = load(testbed)
3831
self.parent.parameters['testbed'] = genie_testbed
3932
device_list = []
4033
for device in genie_testbed.devices.values():
4134
log.info(banner(
42-
"Connect to device '{d}'".format(d=device.name)))
35+
f"Connect to device '{device.name}'"))
4336
try:
4437
device.connect()
38+
device_list.append(device)
4539
except Exception as e:
46-
self.failed("Failed to establish connection to '{}'".format(
47-
device.name))
48-
49-
device_list.append(device)
40+
log.info(f"Failed to establish connection to '{device.name}'")
5041

51-
# Pass list of devices the to testcases
5242
self.parent.parameters.update(dev=device_list)
43+
log.debug(self.parent.parameters)
44+
45+
@aetest.subsection
46+
def create_testcases(self):
47+
#below creates a loop over the CRC_count_Check Class using dev as the iterator
48+
aetest.loop.mark(CRC_count_Check, dev=self.parent.parameters['dev'])
49+
5350

5451

5552
###################################################################
5653
# TESTCASES SECTION #
5754
###################################################################
5855

59-
# Testcase name : vxlan_consistency_check
60-
class CRC_count_check(aetest.Testcase):
61-
""" This is user Testcases section """
56+
#Capture Interfaces statistics on the device and tabulate
57+
#Also look for CRC Errors. If CRC Errors fail then change
58+
#variable 'passing' from 0 to 1
59+
class CRC_count_Check(aetest.Testcase):
60+
@aetest.setup
61+
def setup(self, dev):
62+
#Setup has been marked for looping in Common Setup(create_testcases) with the argument dev
63+
#dev is the list of devices in the current testbed
64+
self.dev = dev
65+
log.info(banner(f"Gathering Interface Information from {dev.name}"))
66+
self.interface_info = dev.learn('interface')
67+
68+
list_of_interfaces=self.interface_info.info.keys()
69+
self.mega_dict = {}
70+
self.mega_dict[dev.name] = {}
71+
self.mega_tabular = []
72+
self.passing=0
73+
74+
75+
for ints, props in self.interface_info.info.items():
76+
counters = props.get('counters')
77+
if counters:
78+
smaller_tabular = []
79+
if 'in_crc_errors' in counters:
80+
self.mega_dict[dev.name][ints] = counters['in_crc_errors']
81+
smaller_tabular.append(dev.name)
82+
smaller_tabular.append(ints)
83+
smaller_tabular.append(str(counters['in_crc_errors']))
84+
if counters['in_crc_errors']:
85+
smaller_tabular.append('Failed')
86+
self.passing=1
87+
else:
88+
smaller_tabular.append('Passed')
6289

63-
# First test section
64-
@ aetest.test
65-
def learn_interfaces(self):
66-
""" Sample test section. Only print """
90+
else:
91+
self.mega_dict[dev.name][ints] = None
92+
smaller_tabular.append(dev.name)
93+
smaller_tabular.append(ints)
94+
smaller_tabular.append('N/A')
95+
smaller_tabular.append('N/A')
96+
self.mega_tabular.append(smaller_tabular)
97+
self.mega_tabular.append(['-' * sum(len(i) for i in smaller_tabular)])
6798

68-
self.all_interfaces = {}
69-
for dev in self.parent.parameters['dev']:
70-
log.info(banner("Gathering Interface Information from {}".format(
71-
dev.name)))
72-
abstract = Lookup.from_device(dev)
73-
intf = abstract.ops.interface.interface.Interface(dev)
74-
intf.learn()
75-
self.all_interfaces[dev.name] = intf.info
76-
77-
# Second test section
78-
@ aetest.test
79-
def check_CRC(self):
80-
81-
mega_dict = {}
82-
mega_tabular = []
83-
for device, ints in self.all_interfaces.items():
84-
mega_dict[device] = {}
85-
for name, props in ints.items():
86-
counters = props.get('counters')
87-
if counters:
88-
smaller_tabular = []
89-
if 'in_crc_errors' in counters:
90-
mega_dict[device][name] = counters['in_crc_errors']
91-
smaller_tabular.append(device)
92-
smaller_tabular.append(name)
93-
smaller_tabular.append(str(counters['in_crc_errors']))
94-
if counters['in_crc_errors']:
95-
smaller_tabular.append('Failed')
96-
else:
97-
smaller_tabular.append('Passed')
98-
else:
99-
mega_dict[device][name] = None
100-
smaller_tabular.append(device)
101-
smaller_tabular.append(name)
102-
smaller_tabular.append('N/A')
103-
smaller_tabular.append('N/A')
104-
mega_tabular.append(smaller_tabular)
10599

106-
mega_tabular.append(['-'*sum(len(i) for i in smaller_tabular)])
100+
aetest.loop.mark(self.interface_check, intf=list_of_interfaces)
107101

108-
log.info(tabulate(mega_tabular,
102+
# create table and display. Test fails if variable 'passing' = 1.
103+
# which means there are some CRC errors
104+
@aetest.test
105+
def table_display(self):
106+
log.info(tabulate(self.mega_tabular,
109107
headers=['Device', 'Interface',
110108
'CRC Errors Counter',
111109
'Passed/Failed'],
112110
tablefmt='orgtbl'))
113111

114-
for dev in mega_dict:
115-
for intf in mega_dict[dev]:
116-
if mega_dict[dev][intf]:
117-
self.failed("{d}: {name} CRC ERRORS: {e}".format(
118-
d=dev, name=intf, e=mega_dict[dev][intf]))
112+
if self.passing==1:
113+
self.failed('Some interfaces have CRC errors')
114+
else:
115+
self.passed
116+
117+
# Test created for each interface. If CRC errors then Interface test will fail
118+
@aetest.test
119+
def interface_check(self, intf):
120+
# This test has been marked for loop. intf is the looping argument (list of interfaces)
121+
# Thus this test is run for each interface in the intf list.
122+
#for int, errors in self.parent.parameters['mega'].items():
123+
for int, errors in self.mega_dict[self.dev.name].items():
124+
if errors:
125+
self.failed(f'Interface {int} has crc errors {errors}')
126+
else:
127+
self.passed(f'No errors on {int}')
119128

120-
self.passed("All devices' interfaces CRC ERRORS Count is: 'Zero'")
121129

122130
# #####################################################################
123131
# #### COMMON CLEANUP SECTION ###
@@ -136,10 +144,12 @@ class common_cleanup(aetest.CommonCleanup):
136144
# here is an example of 1 subsection
137145

138146
@aetest.subsection
139-
def clean_everything(self):
140-
""" Common Cleanup Subsection """
141-
log.info("Aetest Common Cleanup ")
147+
148+
def disconnect(self):
149+
log.info("Aetest Common Cleanup disconnecting devices")
150+
for dev in self.parent.parameters['dev']:
151+
dev.disconnect()
142152

143153

144154
if __name__ == '__main__': # pragma: no cover
145-
aetest.main()
155+
aetest.main()

0 commit comments

Comments
 (0)