File: openflow.py

package info (click to toggle)
openvswitch 3.6.0-6
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 99,632 kB
  • sloc: sh: 1,683,183; ansic: 313,349; python: 28,192; xml: 21,442; makefile: 548; javascript: 191
file content (66 lines) | stat: -rwxr-xr-x 1,920 bytes parent folder | download | duplicates (3)
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
import unittest
import logging
from mininet.net import Mininet
import mininet.log as log
from mininet.node import OVSController, OVSKernelSwitch

Switch = OVSKernelSwitch
Controller = OVSController
logging.basicConfig(level=logging.DEBUG)
log.setLogLevel('info')


class BasicOpenflowTest(unittest.TestCase):

    def addHost(self, N):
        logging.debug("Creating host h%s and add to net.", N)
        name = 'h%d' % N
        ip = '10.0.0.%d' % N
        return self.net.addHost(name, ip=ip)

    def setUp(self):
        self.net = Mininet(controller=Controller, switch=Switch)

        logging.info("Creating controllers")
        self.net.addController('c1', command='ovs-testcontroller')

        logging.info("Creating switches")
        s1 = self.net.addSwitch('s1', protocols="OpenFlow10")
        s2 = self.net.addSwitch('s2', protocols="OpenFlow10")

        logging.info("Creating hosts (7 on each switch)")
        hosts1 = [self.addHost(n) for n in (1, 2, 3, 4, 5, 6, 7)]
        hosts2 = [self.addHost(n) for n in (8, 9, 10, 11, 12, 13, 14)]

        logging.info("Creating links")
        for h in hosts1:
            self.net.addLink(s1, h)
        for h in hosts2:
            self.net.addLink(s2, h)
        self.net.addLink(s1, s2)

        logging.info("Starting network")
        self.net.start()

    def testPingAll(self):
        logging.info("Testing network")
        packetLoss = self.net.pingAll()
        self.assertTrue(
            packetLoss == 0,
            "Packet loss during ping test %s" %
            packetLoss)

    def testIPerfTCP(self):
        logging.info("Running TCP performance test")
        self.net.iperf()

    def testIPerfUDP(self):
        logging.info("Running UDP performance test")
        self.net.iperf(l4Type='UDP')

    def tearDown(self):
        logging.info("Stopping network")
        self.net.stop()

if __name__ == '__main__':
    unittest.main()