test_zebra: Enhance test cases

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.yusuke0@gmail.com>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>
This commit is contained in:
IWASE Yusuke 2017-09-05 10:13:45 +09:00 committed by FUJITA Tomonori
parent aedd3920b3
commit bf9b8ac588

View File

@ -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))