2 Helper module that implements helpers for test implementations.
9 from emuvim
.dcemulator
.net
import DCNetwork
10 from emuvim
.api
.rest
.compute
import RestApiEndpoint
11 from mininet
.clean
import cleanup
12 from mininet
.node
import Controller
14 class SimpleTestTopology(unittest
.TestCase
):
16 Helper class to do basic test setups.
17 s1 -- s2 -- s3 -- ... -- sN
20 def __init__(self
, *args
, **kwargs
):
23 self
.s
= [] # list of switches
24 self
.h
= [] # list of hosts
25 self
.d
= [] # list of docker containers
26 self
.dc
= [] # list of data centers
27 self
.docker_cli
= None
28 super(SimpleTestTopology
, self
).__init
__(*args
, **kwargs
)
32 nswitches
=0, ndatacenter
=0, nhosts
=0, ndockers
=0,
33 autolinkswitches
=False, controller
=Controller
, **kwargs
):
35 Creates a Mininet instance and automatically adds some
38 Attention, we should always use Mininet's default controller
39 for our tests. Only use other controllers if you want to test
40 specific controller functionality.
42 self
.net
= DCNetwork(controller
=controller
, **kwargs
)
43 self
.api
= RestApiEndpoint("127.0.0.1",5000)
45 # start from s1 because ovs does not like to have dpid = 0
46 # and switch name-number is being used by mininet to set the dpid
47 for i
in range(1, nswitches
+1):
48 self
.s
.append(self
.net
.addSwitch('s%d' % i
))
49 # if specified, chain all switches
51 for i
in range(0, len(self
.s
) - 1):
52 self
.net
.addLink(self
.s
[i
], self
.s
[i
+ 1])
53 # add some data centers
54 for i
in range(0, ndatacenter
):
56 self
.net
.addDatacenter(
58 metadata
={"unittest_dc": i
}))
59 # connect data centers to the endpoint
60 for i
in range(0, ndatacenter
):
61 self
.api
.connectDatacenter(self
.dc
[i
])
63 for i
in range(0, nhosts
):
64 self
.h
.append(self
.net
.addHost('h%d' % i
))
66 for i
in range(0, ndockers
):
67 self
.d
.append(self
.net
.addDocker('d%d' % i
, dimage
="ubuntu:trusty"))
78 def getDockerCli(self
):
80 Helper to interact with local docker instance.
82 if self
.docker_cli
is None:
83 self
.docker_cli
= docker
.Client(
84 base_url
='unix://var/run/docker.sock')
85 return self
.docker_cli
87 def getContainernetContainers(self
):
89 List the containers managed by containernet
91 return self
.getDockerCli().containers(filters
={"label": "com.containernet"})
100 # make sure that all pending docker containers are killed
101 with
open(os
.devnull
, 'w') as devnull
:
103 "sudo docker rm -f $(sudo docker ps --filter 'label=com.containernet' -a -q)",