test/resources/linux/interface.py

   1 #!/usr/bin/env python
   2 from neco.execution.ec import ExperimentController
   3 from neco.execution.resource import ResourceState
   4 from neco.resources.linux.node import LinuxNode
   5 from neco.resources.linux.interface import LinuxInterface
   6 from neco.resources.linux.channel import LinuxChannel
   7 from neco.util.sshfuncs import RUNNING, FINISHED
   8
   9 from test_utils import skipIfNotAlive
  10
  11 import os
  12 import time
  13 import tempfile
  14 import unittest
  15
  16 class LinuxInterfaceTestCase(unittest.TestCase):
  17     def setUp(self):
  18         self.fedora_host = 'nepi2.pl.sophia.inria.fr'
  19         self.fedora_user = 'inria_nepi'
  20
  21         self.ubuntu_host = 'roseval.pl.sophia.inria.fr'
  22         self.ubuntu_user = 'alina'
  23
  24     @skipIfNotAlive
  25     def t_deploy(self, host, user):
  26         from neco.execution.resource import ResourceFactory
  27
  28         ResourceFactory.register_type(LinuxNode)
  29         ResourceFactory.register_type(LinuxInterface)
  30         ResourceFactory.register_type(LinuxChannel)
  31
  32         ec = ExperimentController()
  33
  34         node = ec.register_resource("LinuxNode")
  35         ec.set(node, "hostname", host)
  36         ec.set(node, "username", user)
  37
  38         iface = ec.register_resource("LinuxInterface")
  39         chan = ec.register_resource("LinuxChannel")
  40
  41         ec.register_connection(iface, node)
  42         ec.register_connection(iface, chan)
  43
  44         try:
  45             ec.deploy()
  46
  47             while not all([ ec.state(guid) == ResourceState.STARTED \
  48                     for guid in [node, iface]]):
  49                 time.sleep(0.5)
  50
  51             self.assertTrue(ec.state(node) == ResourceState.STARTED)
  52             self.assertTrue(ec.state(iface) == ResourceState.STARTED)
  53             self.assertTrue(ec.get(iface, "deviceName") == "eth0")
  54
  55         finally:
  56             ec.shutdown()
  57
  58     def test_deploy_fedora(self):
  59         self.t_deploy(self.fedora_host, self.fedora_user)
  60
  61     def test_deploy_ubuntu(self):
  62         self.t_deploy(self.ubuntu_host, self.ubuntu_user)
  63
  64
  65 if __name__ == '__main__':
  66     unittest.main()
  67