dprevot/netflow
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Merge pull request #19 from grafolean/fix/tests

Browse source 
Fix failing tests (wrong index when accessing netflow records)
This commit is contained in:
cookie 2020-03-19 15:38:27 +00:00 committed by GitHub
parent 565f829945 096c7d6f4f
commit 647f4b3748
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 3 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
tests.py
View file

@ -95,12 +95,12 @@ class TestSoftFlowExport(unittest.TestCase):
# check number of "things" in the packets (flows + templates)
# template packet = 10 things
# other packets = 12 things
self.assertEqual(sum(p[1].header.count for p in pkts), (num-1)*12 + 10)
self.assertEqual(sum(p[2].header.count for p in pkts), (num-1)*12 + 10)
# check number of flows in the packets
# template packet = 8 flows (2 templates)
# other packets = 12 flows
self.assertEqual(sum(len(p[1].flows) for p in pkts), (num-1)*12 + 8)
self.assertEqual(sum(len(p[2].flows) for p in pkts), (num-1)*12 + 8)
def test_recv_all_packets_template_first(self):
"""Test all packets are received when the template is sent first"""
@ -130,7 +130,7 @@ class TestSoftFlowExport(unittest.TestCase):
def test_analyzer(self):
"""Test thar the analyzer doesn't break and outputs the correct number of lines"""
pkts, _, _ = send_recv_packets([TEMPLATE_PACKET, *PACKETS])
data = {p[0]: [f.data for f in p[1].flows] for p in pkts}
data = {p[0]: [f.data for f in p[2].flows] for p in pkts}
analyzer = subprocess.run(
[sys.executable, 'analyzer.py'],
input=json.dumps(data),