-
Notifications
You must be signed in to change notification settings - Fork 1
/
test_pim6_basic.py
133 lines (112 loc) · 3.72 KB
/
test_pim6_basic.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-or-later
# Copyright (C) 2018-2022 David Lamparter for NetDEF, Inc.
"""
Basic IPv6 PIM test.
"""
from topotato.v1 import *
from topotato.multicast import *
from topotato.scapy import ScapySend
from scapy.all import (
IPv6,
ICMPv6MLReport2,
ICMPv6MLDMultAddrRec,
IPv6ExtHdrHopByHop,
RouterAlert,
UDP,
)
@topology_fixture()
def topology(topo):
"""
[ h1 ]--{ lan }--[ r1 ]---[ r2 ]---[ h2 ]
"""
class Configs(FRRConfigs):
routers = ["r1", "r2"]
zebra = """
#% extends "boilerplate.conf"
#% block main
debug zebra events
debug zebra packet
debug zebra rib detailed
debug zebra nht detailed
#% endblock
"""
staticd = """
#% extends "boilerplate.conf"
#% block main
#% if router.name == 'r2'
ipv6 route {{ routers['r1'].lo_ip6[0].ip }}/128 {{ routers['r1'].iface_to('r2').ll6 }} {{ router.iface_to('r1').ifname }}
ipv6 route {{ topo.lans['lan'].ip6[0] }} {{ routers['r1'].iface_to('r2').ll6 }} {{ router.iface_to('r1').ifname }}
#% elif router.name == 'r1'
ipv6 route {{ routers['r2'].lo_ip6[0].ip }}/128 {{ routers['r2'].iface_to('r1').ll6 }} {{ router.iface_to('r2').ifname }}
#% endif
#% endblock
"""
pim6d = """
#% extends "boilerplate.conf"
#% block main
#% if frr.has_defun("debug_mld_cmd")
debug mld
#% endif
debug pimv6 trace
#% if router.name in ['r1', 'r2']
interface lo
ipv6 pim
##
#% for iface in router.ifaces
!
interface {{ iface.ifname }}
ipv6 pim
ipv6 pim hello 1 5
ipv6 mld
#% endfor
!
ipv6 pim rp {{ routers['r1'].lo_ip6[0].ip }} ff00::/8
#% endif
#% endblock
"""
class PIM6Basic(TestBase, AutoFixture, topo=topology, configs=Configs):
@topotatofunc
def prepare(self, topo, h1, h2, r1, r2):
self.receiver = MulticastReceiver(h2, h2.iface_to('r2'))
for rt in [r1, r2]:
yield from AssertVtysh.make(rt, "zebra", "show ipv6 route")
self.receiver = MulticastReceiver(h2, h2.iface_to('r2'))
yield from AssertVtysh.make(r1, "pim6d", "show ipv6 pim rp-info")
yield from AssertVtysh.make(r2, "pim6d", "show ipv6 pim rp-info", f"""
RP address group/prefix-list OIF I am RP Source Group-Type
{ r1.lo_ip6[0].ip } ff00::/8 r2-r1 no Static ASM
""", maxwait=5.0)
@topotatofunc
def test_ssm(self, topo, h1, h2, r1, r2):
"""
Join a (S,G) on MLD and try forwarding a packet on it.
"""
srcaddr = h1.iface_to('lan').ip6[0].ip
yield from self.receiver.join('ff05::2345', srcaddr)
logchecks = yield from AssertLog.make(r2, 'pim6d', '[MLD default:r2-h2 (%s,ff05::2345)] NOINFO => JOIN' % srcaddr, maxwait=3.0)
@logchecks.skip_on_exception
def need_debug_mld(testitem):
testitem.instance.r2.require_defun("debug_mld_cmd")
yield from AssertLog.make(r1, 'pim6d', 'pim_forward_start: (S,G)=(%s,ff05::2345) oif=r1-r2' % srcaddr, maxwait=3.0)
ip = IPv6(hlim=255, src=srcaddr, dst="ff05::2345")
udp = UDP(sport=9999, dport=9999)
yield from ScapySend.make(
h1,
"h1-lan",
pkt = ip/udp,
)
yield from ScapySend.make(
h1,
"h1-lan",
pkt = ip/udp,
)
yield from ScapySend.make(
h1,
"h1-lan",
pkt = ip/udp,
)
def expect_pkt(ipv6: IPv6, udp: UDP):
return ipv6.src == str(srcaddr) and ipv6.dst == 'ff05::2345' \
and udp.dport == 9999
yield from AssertPacket.make("r2_h2", maxwait=4.0, pkt=expect_pkt)