src/emuvim/test/base.py

   1 """
   2 Helper module that implements helpers for test implementations.
   3 """
   4
   5 import unittest
   6 import os
   7 import subprocess
   8 import docker
   9 from emuvim.dcemulator.net import DCNetwork
  10 from mininet.clean import cleanup
  11 from mininet.node import Controller
  12
  13 class SimpleTestTopology(unittest.TestCase):
  14     """
  15         Helper class to do basic test setups.
  16         s1 -- s2 -- s3 -- ... -- sN
  17     """
  18
  19     def __init__(self, *args, **kwargs):
  20         self.net = None
  21         self.s = []   # list of switches
  22         self.h = []   # list of hosts
  23         self.d = []   # list of docker containers
  24         self.dc = []  # list of data centers
  25         self.docker_cli = None
  26         super(SimpleTestTopology, self).__init__(*args, **kwargs)
  27
  28     def createNet(
  29             self,
  30             nswitches=0, ndatacenter=0, nhosts=0, ndockers=0,
  31             autolinkswitches=False, controller=Controller):
  32         """
  33         Creates a Mininet instance and automatically adds some
  34         nodes to it.
  35
  36         Attention, we should always use Mininet's default controller
  37         for our tests. Only use other controllers if you want to test
  38         specific controller functionality.
  39         """
  40         self.net = DCNetwork(controller=controller)
  41
  42         # add some switches
  43         for i in range(0, nswitches):
  44             self.s.append(self.net.addSwitch('s%d' % i))
  45         # if specified, chain all switches
  46         if autolinkswitches:
  47             for i in range(0, len(self.s) - 1):
  48                 self.net.addLink(self.s[i], self.s[i + 1])
  49         # add some data centers
  50         for i in range(0, ndatacenter):
  51             self.dc.append(
  52                 self.net.addDatacenter(
  53                     'datacenter%d' % i,
  54                     metadata={"unittest_dc": i}))
  55         # add some hosts
  56         for i in range(0, nhosts):
  57             self.h.append(self.net.addHost('h%d' % i))
  58         # add some dockers
  59         for i in range(0, ndockers):
  60             self.d.append(self.net.addDocker('d%d' % i, dimage="ubuntu:trusty"))
  61
  62     def startNet(self):
  63         self.net.start()
  64
  65     def stopNet(self):
  66         self.net.stop()
  67
  68     def getDockerCli(self):
  69         """
  70         Helper to interact with local docker instance.
  71         """
  72         if self.docker_cli is None:
  73             self.docker_cli = docker.Client(
  74                 base_url='unix://var/run/docker.sock')
  75         return self.docker_cli
  76
  77     def getContainernetContainers(self):
  78         """
  79         List the containers managed by containernet
  80         """
  81         return self.getDockerCli().containers(filters={"label": "com.containernet"})
  82
  83     @staticmethod
  84     def setUp():
  85         pass
  86
  87     @staticmethod
  88     def tearDown():
  89         cleanup()
  90         # make sure that all pending docker containers are killed
  91         with open(os.devnull, 'w') as devnull:
  92             subprocess.call(
  93                 "sudo docker rm -f $(sudo docker ps --filter 'label=com.containernet' -a -q)",
  94                 stdout=devnull,
  95                 stderr=devnull,
  96                 shell=True)