Note: The binary representation is not extracted from the real packets
and based on the implemetation, so might be required to be fixup with
the real packets.
Signed-off-by: IWASE Yusuke <iwase.yusu...@gmail.com>
---
ryu/tests/unit/packet/test_zebra.py | 626 ++++++++++++++++++++++++++++++++++++
1 file changed, 626 insertions(+)
diff --git a/ryu/tests/unit/packet/test_zebra.py
b/ryu/tests/unit/packet/test_zebra.py
index 8fb7613..b5fb9c5 100644
--- a/ryu/tests/unit/packet/test_zebra.py
+++ b/ryu/tests/unit/packet/test_zebra.py
@@ -16,11 +16,13 @@
from __future__ import print_function
import os
+import socket
import sys
import unittest
from nose.tools import eq_
from nose.tools import ok_
+from nose.tools import raises
import six
from ryu.lib import pcaplib
@@ -65,3 +67,627 @@ class Test_zebra(unittest.TestCase):
pkt.serialize()
eq_(buf, pkt.data,
"b'%s' != b'%s'" % (binary_str(buf), binary_str(pkt.data)))
+
+
+class TestZebraMessage(unittest.TestCase):
+
+ def test_get_header_size(self):
+ eq_(zebra.ZebraMessage.V0_HEADER_SIZE,
+ zebra.ZebraMessage.get_header_size(0))
+ eq_(zebra.ZebraMessage.V1_HEADER_SIZE,
+ zebra.ZebraMessage.get_header_size(2))
+ eq_(zebra.ZebraMessage.V3_HEADER_SIZE,
+ zebra.ZebraMessage.get_header_size(3))
+ eq_(zebra.ZebraMessage.V3_HEADER_SIZE,
+ zebra.ZebraMessage.get_header_size(4))
+
+ @raises(ValueError)
+ def test_get_header_size_invalid_version(self):
+ eq_(zebra.ZebraMessage.V0_HEADER_SIZE,
+ zebra.ZebraMessage.get_header_size(0xff))
+
+
+class TestZebraRedistributeAdd(unittest.TestCase):
+ buf = (
+ b'\x02' # route_type
+ )
+ route_type = zebra.ZEBRA_ROUTE_CONNECT
+
+ def test_parser(self):
+ body = zebra.ZebraRedistributeAdd.parse(self.buf, version=3)
+
+ eq_(self.route_type, body.route_type)
+
+ buf = body.serialize(version=3)
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraIPv4ImportLookup(unittest.TestCase):
+ buf = (
+ b'\x18'
+ b'\xc0\xa8\x01\x01' # prefix
+ )
+ prefix = '192.168.1.1/24'
+ metric = None
+ nexthop_num = 0
+ from_zebra = False
+
+ def test_parser(self):
+ body = zebra.ZebraIPv4ImportLookup.parse(self.buf)
+
+ eq_(self.prefix, body.prefix)
+ eq_(self.metric, body.metric)
+ eq_(self.nexthop_num, len(body.nexthops))
+ eq_(self.from_zebra, body.from_zebra)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraIPv4ImportLookupFromZebra(unittest.TestCase):
+ buf = (
+ b'\xc0\xa8\x01\x01' # prefix
+ b'\x00\x00\x00\x14' # metric
+ b'\x01' # nexthop_num
+ b'\x01' # nexthop_type
+ b'\x00\x00\x00\x02' # ifindex
+ )
+ prefix = '192.168.1.1'
+ metric = 0x14
+ nexthop_num = 1
+ nexthop_type = zebra.ZEBRA_NEXTHOP_IFINDEX
+ ifindex = 2
+ from_zebra = True
+
+ def test_parser(self):
+ body = zebra.ZebraIPv4ImportLookup.parse_from_zebra(self.buf)
+
+ eq_(self.prefix, body.prefix)
+ eq_(self.metric, body.metric)
+ eq_(self.nexthop_num, len(body.nexthops))
+ eq_(self.nexthop_type, body.nexthops[0].type)
+ eq_(self.ifindex, body.nexthops[0].ifindex)
+ eq_(self.from_zebra, body.from_zebra)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraIPv4NexthopLookupMRib(unittest.TestCase):
+ buf = (
+ b'\xc0\xa8\x01\x01' # addr
+ )
+ addr = '192.168.1.1'
+ distance = None
+ metric = None
+ nexthop_num = 0
+
+ def test_parser(self):
+ body = zebra.ZebraIPv4NexthopLookupMRib.parse(self.buf)
+
+ eq_(self.addr, body.addr)
+ eq_(self.distance, body.distance)
+ eq_(self.metric, body.metric)
+ eq_(self.nexthop_num, len(body.nexthops))
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraIPv4NexthopLookupMRibFromZebra(unittest.TestCase):
+ buf = (
+ b'\xc0\xa8\x01\x01' # addr
+ b'\x01' # distance
+ b'\x00\x00\x00\x14' # metric
+ b'\x01' # nexthop_num
+ b'\x01' # nexthop_type
+ b'\x00\x00\x00\x02' # ifindex
+ )
+ addr = '192.168.1.1'
+ distance = 1
+ metric = 0x14
+ nexthop_num = 1
+ nexthop_type = zebra.ZEBRA_NEXTHOP_IFINDEX
+ ifindex = 2
+
+ def test_parser(self):
+ body = zebra.ZebraIPv4NexthopLookupMRib.parse(self.buf)
+
+ eq_(self.addr, body.addr)
+ eq_(self.distance, body.distance)
+ eq_(self.metric, body.metric)
+ eq_(self.nexthop_num, len(body.nexthops))
+ eq_(self.nexthop_type, body.nexthops[0].type)
+ eq_(self.ifindex, body.nexthops[0].ifindex)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraNexthopUpdateIPv6(unittest.TestCase):
+ buf = (
+ b'\x00\x0a' # family
+ b'\x40' # prefix_len
+ b'\x20\x01\x0d\xb8' # prefix
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x14' # metric
+ b'\x01' # nexthop_num
+ b'\x01' # nexthop_type
+ b'\x00\x00\x00\x02' # ifindex
+ )
+ family = socket.AF_INET6
+ prefix = '2001:db8::/64'
+ metric = 0x14
+ nexthop_num = 1
+ nexthop_type = zebra.ZEBRA_NEXTHOP_IFINDEX
+ ifindex = 2
+
+ def test_parser(self):
+ body = zebra.ZebraNexthopUpdate.parse(self.buf)
+
+ eq_(self.family, body.family)
+ eq_(self.prefix, body.prefix)
+ eq_(self.metric, body.metric)
+ eq_(self.nexthop_num, len(body.nexthops))
+ eq_(self.nexthop_type, body.nexthops[0].type)
+ eq_(self.ifindex, body.nexthops[0].ifindex)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraInterfaceNbrAddressAdd(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # ifindex
+ b'\x02' # family
+ b'\xc0\xa8\x01\x00' # prefix
+ b'\x18' # prefix_len
+ )
+ ifindex = 1
+ family = socket.AF_INET
+ prefix = '192.168.1.0/24'
+
+ def test_parser(self):
+ body = zebra.ZebraInterfaceNbrAddressAdd.parse(self.buf)
+
+ eq_(self.ifindex, body.ifindex)
+ eq_(self.family, body.family)
+ eq_(self.prefix, body.prefix)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraInterfaceBfdDestinationUpdate(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # ifindex
+ b'\x02' # dst_family
+ b'\xc0\xa8\x01\x01' # dst_prefix
+ b'\x18' # dst_prefix_len
+ b'\x04' # status
+ b'\x02' # src_family
+ b'\xc0\xa8\x01\x02' # src_prefix
+ b'\x18' # src_prefix_len
+ )
+ ifindex = 1
+ dst_family = socket.AF_INET
+ dst_prefix = '192.168.1.1/24'
+ status = zebra.BFD_STATUS_UP
+ src_family = socket.AF_INET
+ src_prefix = '192.168.1.2/24'
+
+ def test_parser(self):
+ body = zebra.ZebraInterfaceBfdDestinationUpdate.parse(self.buf)
+
+ eq_(self.ifindex, body.ifindex)
+ eq_(self.dst_family, body.dst_family)
+ eq_(self.dst_prefix, body.dst_prefix)
+ eq_(self.status, body.status)
+ eq_(self.src_family, body.src_family)
+ eq_(self.src_prefix, body.src_prefix)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraBfdDestinationRegisterMultiHopEnabled(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # pid
+ b'\x00\x02' # dst_family
+ b'\xc0\xa8\x01\x01' # dst_prefix
+ b'\x00\x00\x00\x10' # min_rx_timer
+ b'\x00\x00\x00\x20' # min_tx_timer
+ b'\x01' # detect_mult
+ b'\x01' # multi_hop
+ b'\x00\x02' # src_family
+ b'\xc0\xa8\x01\x02' # src_prefix
+ b'\x05' # multi_hop_count
+ )
+ pid = 1
+ dst_family = socket.AF_INET
+ dst_prefix = '192.168.1.1'
+ min_rx_timer = 0x10
+ min_tx_timer = 0x20
+ detect_mult = 1
+ multi_hop = 1
+ src_family = socket.AF_INET
+ src_prefix = '192.168.1.2'
+ multi_hop_count = 5
+ ifname = None
+
+ def test_parser(self):
+ body = zebra.ZebraBfdDestinationRegister.parse(self.buf)
+
+ eq_(self.pid, body.pid)
+ eq_(self.dst_family, body.dst_family)
+ eq_(self.dst_prefix, body.dst_prefix)
+ eq_(self.min_rx_timer, body.min_rx_timer)
+ eq_(self.min_tx_timer, body.min_tx_timer)
+ eq_(self.detect_mult, body.detect_mult)
+ eq_(self.multi_hop, body.multi_hop)
+ eq_(self.src_family, body.src_family)
+ eq_(self.src_prefix, body.src_prefix)
+ eq_(self.multi_hop_count, body.multi_hop_count)
+ eq_(self.ifname, body.ifname)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraBfdDestinationRegisterMultiHopDisabled(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # pid
+ b'\x00\x02' # dst_family
+ b'\xc0\xa8\x01\x01' # dst_prefix
+ b'\x00\x00\x00\x10' # min_rx_timer
+ b'\x00\x00\x00\x20' # min_tx_timer
+ b'\x01' # detect_mult
+ b'\x00' # multi_hop
+ b'\x00\x02' # src_family
+ b'\xc0\xa8\x01\x02' # src_prefix
+ b'\x04' # ifname_len
+ b'eth0' # ifname
+ )
+ pid = 1
+ dst_family = socket.AF_INET
+ dst_prefix = '192.168.1.1'
+ min_rx_timer = 0x10
+ min_tx_timer = 0x20
+ detect_mult = 1
+ multi_hop = 0
+ src_family = socket.AF_INET
+ src_prefix = '192.168.1.2'
+ multi_hop_count = None
+ ifname = 'eth0'
+
+ def test_parser(self):
+ body = zebra.ZebraBfdDestinationRegister.parse(self.buf)
+
+ eq_(self.pid, body.pid)
+ eq_(self.dst_family, body.dst_family)
+ eq_(self.dst_prefix, body.dst_prefix)
+ eq_(self.min_rx_timer, body.min_rx_timer)
+ eq_(self.min_tx_timer, body.min_tx_timer)
+ eq_(self.detect_mult, body.detect_mult)
+ eq_(self.multi_hop, body.multi_hop)
+ eq_(self.src_family, body.src_family)
+ eq_(self.src_prefix, body.src_prefix)
+ eq_(self.multi_hop_count, body.multi_hop_count)
+ eq_(self.ifname, body.ifname)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraBfdDestinationRegisterMultiHopEnabledIPv6(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # pid
+ b'\x00\x0a' # dst_family
+ b'\x20\x01\x0d\xb8' # dst_prefix
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x01'
+ b'\x00\x00\x00\x10' # min_rx_timer
+ b'\x00\x00\x00\x20' # min_tx_timer
+ b'\x01' # detect_mult
+ b'\x01' # multi_hop
+ b'\x00\x0a' # src_family
+ b'\x20\x01\x0d\xb8' # src_prefix
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x02'
+ b'\x05' # multi_hop_count
+ )
+ pid = 1
+ dst_family = socket.AF_INET6
+ dst_prefix = '2001:db8::1'
+ min_rx_timer = 0x10
+ min_tx_timer = 0x20
+ detect_mult = 1
+ multi_hop = 1
+ src_family = socket.AF_INET6
+ src_prefix = '2001:db8::2'
+ multi_hop_count = 5
+ ifname = None
+
+ def test_parser(self):
+ body = zebra.ZebraBfdDestinationRegister.parse(self.buf)
+
+ eq_(self.pid, body.pid)
+ eq_(self.dst_family, body.dst_family)
+ eq_(self.dst_prefix, body.dst_prefix)
+ eq_(self.min_rx_timer, body.min_rx_timer)
+ eq_(self.min_tx_timer, body.min_tx_timer)
+ eq_(self.detect_mult, body.detect_mult)
+ eq_(self.multi_hop, body.multi_hop)
+ eq_(self.src_family, body.src_family)
+ eq_(self.src_prefix, body.src_prefix)
+ eq_(self.multi_hop_count, body.multi_hop_count)
+ eq_(self.ifname, body.ifname)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraBfdDestinationDeregisterMultiHopEnabled(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # pid
+ b'\x00\x02' # dst_family
+ b'\xc0\xa8\x01\x01' # dst_prefix
+ b'\x01' # multi_hop
+ b'\x00\x02' # src_family
+ b'\xc0\xa8\x01\x02' # src_prefix
+ b'\x05' # multi_hop_count
+ )
+ pid = 1
+ dst_family = socket.AF_INET
+ dst_prefix = '192.168.1.1'
+ multi_hop = 1
+ src_family = socket.AF_INET
+ src_prefix = '192.168.1.2'
+ multi_hop_count = 5
+ ifname = None
+
+ def test_parser(self):
+ body = zebra.ZebraBfdDestinationDeregister.parse(self.buf)
+
+ eq_(self.pid, body.pid)
+ eq_(self.dst_family, body.dst_family)
+ eq_(self.dst_prefix, body.dst_prefix)
+ eq_(self.multi_hop, body.multi_hop)
+ eq_(self.src_family, body.src_family)
+ eq_(self.src_prefix, body.src_prefix)
+ eq_(self.multi_hop_count, body.multi_hop_count)
+ eq_(self.ifname, body.ifname)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraBfdDestinationDeregisterMultiHopDisabled(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # pid
+ b'\x00\x02' # dst_family
+ b'\xc0\xa8\x01\x01' # dst_prefix
+ b'\x00' # multi_hop
+ b'\x00\x02' # src_family
+ b'\xc0\xa8\x01\x02' # src_prefix
+ b'\x04' # ifname_len
+ b'eth0' # ifname
+ )
+ pid = 1
+ dst_family = socket.AF_INET
+ dst_prefix = '192.168.1.1'
+ multi_hop = 0
+ src_family = socket.AF_INET
+ src_prefix = '192.168.1.2'
+ multi_hop_count = None
+ ifname = 'eth0'
+
+ def test_parser(self):
+ body = zebra.ZebraBfdDestinationDeregister.parse(self.buf)
+
+ eq_(self.pid, body.pid)
+ eq_(self.dst_family, body.dst_family)
+ eq_(self.dst_prefix, body.dst_prefix)
+ eq_(self.multi_hop, body.multi_hop)
+ eq_(self.src_family, body.src_family)
+ eq_(self.src_prefix, body.src_prefix)
+ eq_(self.multi_hop_count, body.multi_hop_count)
+ eq_(self.ifname, body.ifname)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraBfdDestinationDeregisterMultiHopEnabledIPv6(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # pid
+ b'\x00\x0a' # dst_family
+ b'\x20\x01\x0d\xb8' # dst_prefix
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x01'
+ b'\x01' # multi_hop
+ b'\x00\x0a' # src_family
+ b'\x20\x01\x0d\xb8' # src_prefix
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x02'
+ b'\x05' # multi_hop_count
+ )
+ pid = 1
+ dst_family = socket.AF_INET6
+ dst_prefix = '2001:db8::1'
+ multi_hop = 1
+ src_family = socket.AF_INET6
+ src_prefix = '2001:db8::2'
+ multi_hop_count = 5
+ ifname = None
+
+ def test_parser(self):
+ body = zebra.ZebraBfdDestinationDeregister.parse(self.buf)
+
+ eq_(self.pid, body.pid)
+ eq_(self.dst_family, body.dst_family)
+ eq_(self.dst_prefix, body.dst_prefix)
+ eq_(self.multi_hop, body.multi_hop)
+ eq_(self.src_family, body.src_family)
+ eq_(self.src_prefix, body.src_prefix)
+ eq_(self.multi_hop_count, body.multi_hop_count)
+ eq_(self.ifname, body.ifname)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraVrfAdd(unittest.TestCase):
+ buf = (
+ b'VRF1' # vrf_name
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ )
+ vrf_name = 'VRF1'
+
+ def test_parser(self):
+ body = zebra.ZebraVrfAdd.parse(self.buf)
+
+ eq_(self.vrf_name, body.vrf_name)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraInterfaceVrfUpdate(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # ifindex
+ b'\x00\x02' # vrf_id
+ )
+ ifindex = 1
+ vrf_id = 2
+
+ def test_parser(self):
+ body = zebra.ZebraInterfaceVrfUpdate.parse(self.buf)
+
+ eq_(self.ifindex, body.ifindex)
+ eq_(self.vrf_id, body.vrf_id)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraInterfaceEnableRadv(unittest.TestCase):
+ buf = (
+ b'\x00\x00\x00\x01' # ifindex
+ b'\x00\x00\x01\x00' # interval
+ )
+ ifindex = 1
+ interval = 0x100
+
+ def test_parser(self):
+ body = zebra.ZebraInterfaceEnableRadv.parse(self.buf)
+
+ eq_(self.ifindex, body.ifindex)
+ eq_(self.interval, body.interval)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraMplsLabelsAddIPv4(unittest.TestCase):
+ buf = (
+ b'\x09' # route_type
+ b'\x00\x00\x00\x02' # family
+ b'\xc0\xa8\x01\x00' # prefix
+ b'\x18' # prefix_len
+ b'\xc0\xa8\x01\x01' # gate_addr
+ b'\x10' # distance
+ b'\x00\x00\x00\x64' # in_label
+ b'\x00\x00\x00\x03' # out_label
+ )
+ route_type = zebra.ZEBRA_ROUTE_BGP
+ family = socket.AF_INET
+ prefix = '192.168.1.0/24'
+ gate_addr = '192.168.1.1'
+ distance = 0x10
+ in_label = 100
+ out_label = zebra.MPLS_IMP_NULL_LABEL
+
+ def test_parser(self):
+ body = zebra.ZebraMplsLabelsAdd.parse(self.buf)
+
+ eq_(self.route_type, body.route_type)
+ eq_(self.family, body.family)
+ eq_(self.prefix, body.prefix)
+ eq_(self.gate_addr, body.gate_addr)
+ eq_(self.distance, body.distance)
+ eq_(self.in_label, body.in_label)
+ eq_(self.out_label, body.out_label)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
+
+
+class TestZebraMplsLabelsAddIPv6(unittest.TestCase):
+ buf = (
+ b'\x09' # route_type
+ b'\x00\x00\x00\x0a' # family
+ b'\x20\x01\x0d\xb8' # prefix
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x40' # prefix_len
+ b'\x20\x01\x0d\xb8' # gate_addr
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x01'
+ b'\x10' # distance
+ b'\x00\x00\x00\x64' # in_label
+ b'\x00\x00\x00\x03' # out_label
+ )
+ route_type = zebra.ZEBRA_ROUTE_BGP
+ family = socket.AF_INET6
+ prefix = '2001:db8::/64'
+ gate_addr = '2001:db8::1'
+ distance = 0x10
+ in_label = 100
+ out_label = zebra.MPLS_IMP_NULL_LABEL
+
+ def test_parser(self):
+ body = zebra.ZebraMplsLabelsAdd.parse(self.buf)
+
+ eq_(self.route_type, body.route_type)
+ eq_(self.family, body.family)
+ eq_(self.prefix, body.prefix)
+ eq_(self.gate_addr, body.gate_addr)
+ eq_(self.distance, body.distance)
+ eq_(self.in_label, body.in_label)
+ eq_(self.out_label, body.out_label)
+
+ buf = body.serialize()
+
+ eq_(binary_str(self.buf), binary_str(buf))
--
2.7.4
------------------------------------------------------------------------------
Check out the vibrant tech community on one of the world's most
engaging tech sites, Slashdot.org! http://sdm.link/slashdot
_______________________________________________
Ryu-devel mailing list
Ryu-devel@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/ryu-devel
