ext/action_helpers/python/romke/test.py

   1 #!/bin/env python
   2 # -*- coding: utf-8 -*- vim: set ts=4 et sw=4 fdm=indent :
   3 import unittest, tempfile, simplejson, os, random
   4
   5 import mcollectiveah
   6
   7 class TestFunctions(unittest.TestCase):
   8     def test_raise_environ(self):
   9         try:
  10             del os.environ['MCOLLECTIVE_REQUEST_FILE']
  11             del os.environ['MCOLLECTIVE_REPLY_FILE']
  12         except: pass
  13         self.assertRaises(mcollectiveah.MCollectiveActionNoEnv, mcollectiveah.MCollectiveAction)
  14
  15     def test_raise_file_error(self):
  16         os.environ['MCOLLECTIVE_REQUEST_FILE'] = '/tmp/mcollectiveah-test-request.%d' % random.randrange(100000)
  17         os.environ['MCOLLECTIVE_REPLY_FILE'] = '/tmp/mcollectiveah-test-reply.%d' % random.randrange(100000)
  18
  19         self.assertRaises(mcollectiveah.MCollectiveActionFileError, mcollectiveah.MCollectiveAction)
  20
  21         os.unlink(os.environ['MCOLLECTIVE_REPLY_FILE'])
  22
  23     def test_echo(self):
  24         tin = tempfile.NamedTemporaryFile(mode='w', delete=False)
  25         self.data = {'message': 'test'}
  26
  27         simplejson.dump(self.data, tin)
  28         os.environ['MCOLLECTIVE_REQUEST_FILE'] = tin.name
  29         tin.close()
  30
  31         tout = tempfile.NamedTemporaryFile(mode='w')
  32         os.environ['MCOLLECTIVE_REPLY_FILE'] = tout.name
  33         tout.close()
  34
  35         mc = mcollectiveah.MCollectiveAction()
  36         mc.reply['message'] = mc.request['message']
  37         del mc
  38
  39         tout = open(os.environ['MCOLLECTIVE_REPLY_FILE'], 'r')
  40         data = simplejson.load(tout)
  41         tout.close()
  42
  43         self.assertEqual(data, self.data)
  44
  45
  46         os.unlink(os.environ['MCOLLECTIVE_REQUEST_FILE'])
  47         os.unlink(os.environ['MCOLLECTIVE_REPLY_FILE'])
  48
  49 if __name__ == '__main__':
  50     unittest.main()