summaryrefslogtreecommitdiffstats
path: root/src/plugins/igmp/igmp_format.h
blob: b7e6c954ac9392291f958be6591dcf0c0d66f1d4 (plain) 
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

/*
 *------------------------------------------------------------------
 * Copyright (c) 2018 Cisco and/or its affiliates.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *------------------------------------------------------------------
 */

#ifndef _IGMP_FORMAT_H_
#define _IGMP_FORMAT_H_

extern u8 *format_igmp_type (u8 * s, va_list * args);

extern u8 *format_igmp_membership_group_type (u8 * s, va_list * args);

extern u8 *format_igmp_header (u8 * s, va_list * args);

extern u8 *format_igmp_report_v3 (u8 * s, va_list * args);

extern u8 *format_igmp_query_v3 (u8 * s, va_list * args);

extern u8 *format_igmp_filter_mode (u8 * s, va_list * args);

extern u8 *format_igmp_mode (u8 * s, va_list * args);

extern u8 *format_igmp_src_addr_list (u8 * s, va_list * args);

extern u8 *format_igmp_key (u8 * s, va_list * args);

extern u8 *format_igmp_proxy_device_id (u8 * s, va_list * args);

#endif /* IGMP_FORMAT_H */

/*
 * fd.io coding-style-patch-verification: ON
 *
 * Local Variables:
 * eval: (c-set-style "gnu")
 * End:
 */
)) @property def frame_reply(self): """ Ethernet frame modeling a generic reply """ return (Ether(src='00:00:00:00:00:02', dst='00:00:00:00:00:01') / IP(src='4.3.2.1', dst='1.2.3.4') / UDP(sport=20000, dport=10000) / Raw('\xa5' * 100)) @abc.abstractmethod def ip_range(self, start, end): """ range of remote ip's """ pass @abc.abstractmethod def encap_mcast(self, pkt, src_ip, src_mac, vni): """ Encapsulate mcast packet """ pass @abc.abstractmethod def encapsulate(self, pkt, vni): """ Encapsulate packet """ pass @abc.abstractmethod def decapsulate(self, pkt): """ Decapsulate packet """ pass @abc.abstractmethod def check_encapsulation(self, pkt, vni, local_only=False): """ Verify the encapsulation """ pass def assert_eq_pkts(self, pkt1, pkt2): """ Verify the Ether, IP, UDP, payload are equal in both packets """ self.assertEqual(pkt1[Ether].src, pkt2[Ether].src) self.assertEqual(pkt1[Ether].dst, pkt2[Ether].dst) self.assertEqual(pkt1[IP].src, pkt2[IP].src) self.assertEqual(pkt1[IP].dst, pkt2[IP].dst) self.assertEqual(pkt1[UDP].sport, pkt2[UDP].sport) self.assertEqual(pkt1[UDP].dport, pkt2[UDP].dport) self.assertEqual(pkt1[Raw], pkt2[Raw]) def test_decap(self): """ Decapsulation test Send encapsulated frames from pg0 Verify receipt of decapsulated frames on pg1 """ encapsulated_pkt = self.encapsulate(self.frame_request, self.single_tunnel_vni) self.pg0.add_stream([encapsulated_pkt, ]) self.pg1.enable_capture() self.pg_start() # Pick first received frame and check if it's the non-encapsulated # frame out = self.pg1.get_capture(1) pkt = out[0] self.assert_eq_pkts(pkt, self.frame_request) def test_encap(self): """ Encapsulation test Send frames from pg1 Verify receipt of encapsulated frames on pg0 """ self.pg1.add_stream([self.frame_reply]) self.pg0.enable_capture() self.pg_start() # Pick first received frame and check if it's correctly encapsulated. out = self.pg0.get_capture(1) pkt = out[0] self.check_encapsulation(pkt, self.single_tunnel_vni) payload = self.decapsulate(pkt) self.assert_eq_pkts(payload, self.frame_reply) def test_ucast_flood(self): """ Unicast flood test Send frames from pg3 Verify receipt of encapsulated frames on pg0 """ self.pg3.add_stream([self.frame_reply]) self.pg0.enable_capture() self.pg_start() # Get packet from each tunnel and assert it's correctly encapsulated. out = self.pg0.get_capture(self.n_ucast_tunnels) for pkt in out: self.check_encapsulation(pkt, self.ucast_flood_bd, True) payload = self.decapsulate(pkt) self.assert_eq_pkts(payload, self.frame_reply) def test_mcast_flood(self): """ Multicast flood test Send frames from pg2 Verify receipt of encapsulated frames on pg0 """ self.pg2.add_stream([self.frame_reply]) self.pg0.enable_capture() self.pg_start() # Pick first received frame and check if it's correctly encapsulated. out = self.pg0.get_capture(1) pkt = out[0] self.check_encapsulation(pkt, self.mcast_flood_bd, local_only=False, mcast_pkt=True) payload = self.decapsulate(pkt) self.assert_eq_pkts(payload, self.frame_reply) def test_mcast_rcv(self): """ Multicast receive test Send 20 encapsulated frames from pg0 only 10 match unicast tunnels Verify receipt of 10 decap frames on pg2 """ mac = self.pg0.remote_mac ip_range_start = 10 ip_range_end = 30 mcast_stream = [ self.encap_mcast(self.frame_request, ip, mac, self.mcast_flood_bd) for ip in self.ip_range(ip_range_start, ip_range_end)] self.pg0.add_stream(mcast_stream) self.pg2.enable_capture() self.pg_start() out = self.pg2.get_capture(10) for pkt in out: self.assert_eq_pkts(pkt, self.frame_request)