test/vpp_igmp.py

   1
   2 from vpp_object import VppObject
   3 import socket
   4
   5
   6 class IGMP_MODE:
   7     ROUTER = 0
   8     HOST = 1
   9
  10
  11 class IGMP_FILTER:
  12     INCLUDE = 1
  13     EXCLUDE = 0
  14
  15
  16 def find_igmp_state(states, itf, gaddr, saddr):
  17     for s in states:
  18         if s.sw_if_index == itf.sw_if_index and \
  19            s.gaddr.address == socket.inet_pton(socket.AF_INET, gaddr) and \
  20            s.saddr.address == socket.inet_pton(socket.AF_INET, saddr):
  21             return True
  22     return False
  23
  24
  25 def wait_for_igmp_event(test, timeout, itf, gaddr, saddr, ff):
  26     ev = test.vapi.wait_for_event(timeout, "igmp_event")
  27     if ev.sw_if_index == itf.sw_if_index and \
  28        ev.gaddr.address == socket.inet_pton(socket.AF_INET, gaddr) and \
  29        ev.saddr.address == socket.inet_pton(socket.AF_INET, saddr) and \
  30        ev.filter == ff:
  31         return True
  32     return False
  33
  34
  35 class IgmpSG():
  36     def __init__(self, gaddr, saddrs):
  37         self.gaddr = gaddr
  38         self.gaddr_p = socket.inet_pton(socket.AF_INET, gaddr)
  39         self.saddrs = saddrs
  40         self.saddrs_p = []
  41         self.saddrs_encoded = []
  42         for s in saddrs:
  43             ss = socket.inet_pton(socket.AF_INET, s)
  44             self.saddrs_p.append(ss)
  45             self.saddrs_encoded.append({'address': ss})
  46
  47
  48 class IgmpRecord():
  49     def __init__(self, sg, type):
  50         self.sg = sg
  51         self.type = type
  52
  53
  54 class VppHostState(VppObject):
  55     def __init__(self, test, filter, sw_if_index, sg):
  56         self._test = test
  57         self.sw_if_index = sw_if_index
  58         self.filter = filter
  59         self.sg = sg
  60
  61     def add_vpp_config(self):
  62         self._test.vapi.igmp_listen(
  63             self.filter, self.sw_if_index,
  64             self.sg.saddrs_encoded, self.sg.gaddr_p)
  65
  66     def remove_vpp_config(self):
  67         self._test.vapi.igmp_listen(
  68             self.filter,
  69             self.sw_if_index,
  70             [],
  71             self.sg.gaddr_p)
  72
  73     def __str__(self):
  74         return self.object_id()
  75
  76     def object_id(self):
  77         return "%s:%d" % (self.sg, self.sw_if_index)
  78
  79     def query_vpp_config(self):
  80         return self._test.vapi.igmp_dump()