tests/topotests/pim_basic/mcast-tx.py

   1 #!/usr/bin/env python3
   2 #
   3 # mcast-tx.py
   4 #
   5 # Copyright (c) 2018 Cumulus Networks, Inc.
   6 #
   7 # Permission to use, copy, modify, and/or distribute this software
   8 # for any purpose with or without fee is hereby granted, provided
   9 # that the above copyright notice and this permission notice appear
  10 # in all copies.
  11 #
  12 # THE SOFTWARE IS PROVIDED "AS IS" AND Cumulus Networks DISCLAIMS ALL WARRANTIES
  13 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  14 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  15 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  16 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  17 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  18 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  19 # OF THIS SOFTWARE.
  20 #
  21
  22 import argparse
  23 import logging
  24 import socket
  25 import struct
  26 import time
  27 import sys
  28
  29 logging.basicConfig(
  30     level=logging.DEBUG, format="%(asctime)s %(levelname)5s: %(message)s"
  31 )
  32
  33 # Color the errors and warnings in red
  34 logging.addLevelName(
  35     logging.ERROR, "\033[91m  %s\033[0m" % logging.getLevelName(logging.ERROR)
  36 )
  37 logging.addLevelName(
  38     logging.WARNING, "\033[91m%s\033[0m" % logging.getLevelName(logging.WARNING)
  39 )
  40 log = logging.getLogger(__name__)
  41
  42 parser = argparse.ArgumentParser(description="Multicast packet generator")
  43 parser.add_argument("group", help="Multicast IP")
  44 parser.add_argument("ifname", help="Interface name")
  45 parser.add_argument("--port", type=int, help="UDP port number", default=1000)
  46 parser.add_argument("--ttl", type=int, help="time-to-live", default=20)
  47 parser.add_argument("--count", type=int, help="Packets to send", default=1)
  48 parser.add_argument("--interval", type=int, help="ms between packets", default=100)
  49 args = parser.parse_args()
  50
  51 # Create the datagram socket
  52 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  53
  54 # IN.SO_BINDTODEVICE is not defined in some releases of python but it is 25
  55 # https://github.com/sivel/bonding/issues/10
  56 #
  57 # Bind our socket to ifname
  58 #
  59 # Note ugly python version incompatibility
  60 #
  61 if sys.version_info[0] > 2:
  62     sock.setsockopt(
  63         socket.SOL_SOCKET,
  64         25,
  65         struct.pack("%ds" % len(args.ifname), args.ifname.encode("utf-8")),
  66     )
  67 else:
  68     sock.setsockopt(
  69         socket.SOL_SOCKET, 25, struct.pack("%ds" % len(args.ifname), args.ifname)
  70     )
  71
  72 # We need to make sure our sendto() finishes before we close the socket
  73 sock.setblocking(1)
  74
  75 # Set the time-to-live
  76 sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, struct.pack("b", args.ttl))
  77
  78 ms = args.interval / 1000.0
  79
  80 # Send data to the multicast group
  81 for x in range(args.count):
  82     log.info(
  83         "TX multicast UDP packet to %s:%d on %s" % (args.group, args.port, args.ifname)
  84     )
  85
  86     #
  87     # Note ugly python version incompatibility
  88     #
  89     if sys.version_info[0] > 2:
  90         sent = sock.sendto(b"foobar %d" % x, (args.group, args.port))
  91     else:
  92         sent = sock.sendto("foobar %d" % x, (args.group, args.port))
  93
  94     if args.count > 1 and ms:
  95         time.sleep(ms)
  96
  97 sock.close()