Merge branch 'master' of https://github.cyanoptics.com/cord-lab/cord-tester into test
diff --git a/src/test/utils/Channels.py b/src/test/utils/Channels.py
new file mode 100644
index 0000000..90d4f3c
--- /dev/null
+++ b/src/test/utils/Channels.py
@@ -0,0 +1,221 @@
+import threading
+import sys
+import os
+import time
+import monotonic
+import random
+from scapy.all import *
+from McastTraffic import *
+from IGMP import *
+from OnosCtrl import OnosCtrl
+from nose.tools import *
+log.setLevel('DEBUG')
+
+conf.verb = 0
+
+class IgmpChannel:
+
+ IGMP_DST_MAC = "01:00:5e:00:01:01"
+ IGMP_SRC_MAC = "5a:e1:ac:ec:4d:a1"
+ IP_SRC = '1.2.3.4'
+ IP_DST = '224.0.1.1'
+ igmp_eth = Ether(dst = IGMP_DST_MAC, src = IGMP_SRC_MAC, type = ETH_P_IP)
+ igmp_ip = IP(dst = IP_DST, src = IP_SRC)
+
+ def __init__(self, iface = 'veth0', src_list = ['1.2.3.4'], delay = 2):
+ self.iface = iface
+ self.src_list = src_list
+ self.delay = delay
+ self.onos_ctrl = OnosCtrl('org.onosproject.igmp')
+ self.onos_ctrl.activate()
+
+ def igmp_join(self, groups):
+ self.ssm_table_load(groups)
+ igmp = IGMPv3(type = IGMP_TYPE_V3_MEMBERSHIP_REPORT, max_resp_code=30,
+ gaddr='224.0.1.1')
+ for g in groups:
+ gr = IGMPv3gr(rtype=IGMP_V3_GR_TYPE_EXCLUDE, mcaddr=g)
+ gr.sources = self.src_list
+ igmp.grps.append(gr)
+
+ pkt = self.igmp_eth/self.igmp_ip/igmp
+ IGMPv3.fixup(pkt)
+ sendp(pkt, iface=self.iface)
+ if self.delay != 0:
+ time.sleep(self.delay)
+
+ def igmp_leave(self, groups):
+ igmp = IGMPv3(type = IGMP_TYPE_V3_MEMBERSHIP_REPORT, max_resp_code=30,
+ gaddr='224.0.1.1')
+ for g in groups:
+ gr = IGMPv3gr(rtype=IGMP_V3_GR_TYPE_INCLUDE, mcaddr=g)
+ gr.sources = self.src_list
+ igmp.grps.append(gr)
+
+ pkt = self.igmp_eth/self.igmp_ip/igmp
+ IGMPv3.fixup(pkt)
+ sendp(pkt, iface = self.iface)
+ if self.delay != 0:
+ time.sleep(self.delay)
+
+ def onos_load_config(self, config):
+ status, code = self.onos_ctrl.config(config)
+ if status is False:
+ log.info('JSON config request for app %s returned status %d' %code)
+ time.sleep(2)
+
+ def ssm_table_load(self, groups):
+ ssm_dict = {'apps' : { 'org.onosproject.igmp' : { 'ssmTranslate' : [] } } }
+ ssm_xlate_list = ssm_dict['apps']['org.onosproject.igmp']['ssmTranslate']
+ for g in groups:
+ for s in self.src_list:
+ d = {}
+ d['source'] = s
+ d['group'] = g
+ ssm_xlate_list.append(d)
+ self.onos_load_config(ssm_dict)
+
+class Channels(IgmpChannel):
+ Stopped = 0
+ Started = 1
+ Idle = 0
+ Joined = 1
+ def __init__(self, num, iface = 'veth0', iface_mcast = 'veth2', mcast_cb = None):
+ self.num = num
+ self.channels = self.generate(self.num)
+ self.state = self.Stopped
+ self.streams = None
+ self.channel_states = {}
+ self.last_chan = None
+ self.recv_sock = L2Socket(iface = iface, type = ETH_P_IP)
+ self.iface_mcast = iface_mcast
+ self.mcast_cb = mcast_cb
+ for c in range(self.num):
+ self.channel_states[c] = [self.Idle]
+
+ IgmpChannel.__init__(self, iface=iface)
+
+ def generate(self, num):
+ start = (224 << 24) | 1
+ end = start + num + num/256
+ group_addrs = []
+ for i in range(start, end):
+ if i&255:
+ g = '%s.%s.%s.%s' %((i>>24) &0xff, (i>>16)&0xff, (i>>8)&0xff, i&0xff)
+ log.debug('Adding group %s' %g)
+ group_addrs.append(g)
+ return group_addrs
+
+ def start(self):
+ if self.state == self.Stopped:
+ if self.streams:
+ self.streams.stop()
+ self.streams = McastTraffic(self.channels, iface=self.iface_mcast, cb = self.mcast_cb)
+ self.streams.start()
+ self.state = self.Started
+
+ def join(self, chan = None):
+ if chan is None:
+ chan = random.randint(0, self.num)
+ else:
+ if chan >= self.num:
+ chan = 0
+
+ if self.get_state(chan) == self.Joined:
+ return chan
+
+ groups = [self.channels[chan]]
+ self.igmp_join(groups)
+ self.set_state(chan, self.Joined)
+ self.last_chan = chan
+ return chan
+
+ def leave(self, chan):
+ if chan is None:
+ chan = self.last_chan
+ if chan is None or chan >= self.num:
+ return False
+ if self.get_state(chan) != self.Joined:
+ return False
+ groups = [self.channels[chan]]
+ self.igmp_leave(groups)
+ self.set_state(chan, self.Idle)
+ if chan == self.last_chan:
+ self.last_chan = None
+ return True
+
+ def join_next(self, chan = None):
+ if chan is None:
+ chan = self.last_chan
+ if chan is None:
+ return None
+ leave = chan
+ join = chan+1
+ else:
+ leave = chan - 1
+ join = chan
+
+ if join >= self.num:
+ join = 0
+
+ if leave >= 0 and leave != join:
+ self.leave(leave)
+
+ return self.join(join)
+
+ def jump(self):
+ chan = self.last_chan
+ if chan is not None:
+ self.leave(chan)
+ s_next = chan
+ else:
+ s_next = 0
+ chan = random.randint(s_next, self.num)
+ return self.join(chan)
+
+ def gaddr(self, chan):
+ if chan >= self.num:
+ return None
+ return self.channels[chan]
+
+ def recv_cb(self, pkt):
+ '''Default channel receive callback'''
+ log.debug('Received packet from source %s, destination %s' %(pkt[IP].src, pkt[IP].dst))
+ send_time = float(pkt[IP].payload.load)
+ recv_time = monotonic.monotonic()
+ log.debug('Packet received in %.3f usecs' %(recv_time - send_time))
+
+ def recv(self, chan, cb = None, count = 1):
+ if chan is None:
+ return None
+ if type(chan) == type([]) or type(chan) == type(()):
+ channel_list=filter(lambda c: c < self.num, chan)
+ groups = map(lambda c: self.gaddr(c), channel_list)
+ else:
+ groups = (self.gaddr(chan),)
+ if cb is None:
+ cb = self.recv_cb
+ sniff(prn = cb, count=count, lfilter = lambda p: p[IP].dst in groups, opened_socket = self.recv_sock)
+
+ def stop(self):
+ if self.streams:
+ self.streams.stop()
+ self.state = self.Stopped
+
+ def get_state(self, chan):
+ return self.channel_states[chan][0]
+
+ def set_state(self, chan, state):
+ self.channel_states[chan][0] = state
+
+if __name__ == '__main__':
+ num = 2
+ channels = Channels(num)
+ channels.start()
+ for i in range(num):
+ channels.join(i)
+ for i in range(num):
+ channels.recv(i)
+ for i in range(num):
+ channels.leave(i)
+ channels.stop()