-#!/usr/bin/env python
+#!/usr/bin/env python3
from __future__ import print_function
import binascii
import random
from util import ppp
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from vpp_ip_route import VppIpRoute, VppRoutePath
+from vpp_papi.macaddress import mac_ntop
+from socket import inet_ntop
class VppCFLOW(VppObject):
def enable_exporter(self):
self._test.vapi.set_ipfix_exporter(
- collector_address=self._test.pg0.remote_ip4n,
- src_address=self._test.pg0.local_ip4n,
+ collector_address=self._test.pg0.remote_ip4,
+ src_address=self._test.pg0.local_ip4,
path_mtu=self._mtu,
template_interval=self._timeout)
# packets
self.assertEqual(int(binascii.hexlify(record[2]), 16), 1)
# src mac
- self.assertEqual(':'.join(re.findall('..', record[56].encode(
- 'hex'))), self.pg8.local_mac)
+ self.assertEqual(mac_ntop(record[56]), self.pg8.local_mac)
# dst mac
- self.assertEqual(':'.join(re.findall('..', record[80].encode(
- 'hex'))), self.pg8.remote_mac)
+ self.assertEqual(mac_ntop(record[80]), self.pg8.remote_mac)
flowTimestamp = int(binascii.hexlify(record[156]), 16) >> 32
# flow start timestamp
self.assertAlmostEqual(flowTimestamp, nowUNIX, delta=1)
# ethernet type
self.assertEqual(int(binascii.hexlify(record[256]), 16), 8)
# src ip
- self.assertEqual('.'.join(re.findall('..', record[8].encode(
- 'hex'))),
- '.'.join('{:02x}'.format(int(n)) for n in
- self.pg7.remote_ip4.split('.')))
+ self.assertEqual(inet_ntop(socket.AF_INET, record[8]),
+ self.pg7.remote_ip4)
# dst ip
- self.assertEqual('.'.join(re.findall('..', record[12].encode(
- 'hex'))),
- '.'.join('{:02x}'.format(int(n)) for n in
- "9.0.0.100".split('.')))
+ self.assertEqual(inet_ntop(socket.AF_INET, record[12]),
+ "9.0.0.100")
# protocol (TCP)
self.assertEqual(int(binascii.hexlify(record[4]), 16), 6)
# src port