test/resources/linux/application.py

   1 #!/usr/bin/env python
   2 from neco.execution.ec import ExperimentController
   3 from neco.execution.resource import ResourceState
   4 from neco.execution.trace import TraceAttr
   5 from neco.resources.linux.node import LinuxNode
   6 from neco.resources.linux.application import LinuxApplication
   7
   8 from test_utils import skipIfNotAlive
   9
  10 import os
  11 import time
  12 import tempfile
  13 import unittest
  14
  15 class LinuxApplicationTestCase(unittest.TestCase):
  16     def setUp(self):
  17         self.fedora_host = 'nepi2.pl.sophia.inria.fr'
  18         self.fedora_user = 'inria_nepi'
  19
  20         self.ubuntu_host = 'roseval.pl.sophia.inria.fr'
  21         self.ubuntu_user = 'alina'
  22
  23         self.target = 'nepi5.pl.sophia.inria.fr'
  24
  25     @skipIfNotAlive
  26     def t_ping(self, host, user):
  27         from neco.execution.resource import ResourceFactory
  28
  29         ResourceFactory.register_type(LinuxNode)
  30         ResourceFactory.register_type(LinuxApplication)
  31
  32         ec = ExperimentController()
  33
  34         node = ec.register_resource("LinuxNode")
  35         ec.set(node, "hostname", host)
  36         ec.set(node, "username", user)
  37         ec.set(node, "cleanHome", True)
  38         ec.set(node, "cleanProcesses", True)
  39
  40         app = ec.register_resource("LinuxApplication")
  41         cmd = "ping -c5 %s" % self.target
  42         ec.set(app, "command", cmd)
  43
  44         ec.register_connection(app, node)
  45
  46         try:
  47             ec.deploy()
  48
  49             while not ec.state(app) == ResourceState.FINISHED:
  50                 time.sleep(0.5)
  51
  52             self.assertTrue(ec.state(node) == ResourceState.STARTED)
  53             self.assertTrue(ec.state(app) == ResourceState.FINISHED)
  54
  55             stdout = ec.trace(app, 'stdout')
  56             size = ec.trace(app, 'stdout', attr = TraceAttr.SIZE)
  57             self.assertEquals(len(stdout), size)
  58
  59             block = ec.trace(app, 'stdout', attr = TraceAttr.STREAM, block = 5, offset = 1)
  60             self.assertEquals(block, stdout[5:10])
  61
  62             path = ec.trace(app, 'stdout', attr = TraceAttr.PATH)
  63             rm = ec.get_resource(app)
  64             p = os.path.join(rm.home, 'stdout')
  65             self.assertEquals(path, p)
  66
  67         finally:
  68             ec.shutdown()
  69
  70     def test_deploy_fedora(self):
  71         self.t_ping(self.fedora_host, self.fedora_user)
  72
  73     def test_deploy_ubuntu(self):
  74         self.t_ping(self.ubuntu_host, self.ubuntu_user)
  75
  76
  77 if __name__ == '__main__':
  78     unittest.main()
  79