2 # -*- coding: utf-8 -*- vim: set ts=4 et sw=4 fdm=indent :
3 import unittest, tempfile, simplejson, os, random
7 class TestFunctions(unittest.TestCase):
8 def test_raise_environ(self):
10 del os.environ['MCOLLECTIVE_REQUEST_FILE']
11 del os.environ['MCOLLECTIVE_REPLY_FILE']
13 self.assertRaises(mcollectiveah.MCollectiveActionNoEnv, mcollectiveah.MCollectiveAction)
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)
19 self.assertRaises(mcollectiveah.MCollectiveActionFileError, mcollectiveah.MCollectiveAction)
21 os.unlink(os.environ['MCOLLECTIVE_REPLY_FILE'])
24 tin = tempfile.NamedTemporaryFile(mode='w', delete=False)
25 self.data = {'message': 'test'}
27 simplejson.dump(self.data, tin)
28 os.environ['MCOLLECTIVE_REQUEST_FILE'] = tin.name
31 tout = tempfile.NamedTemporaryFile(mode='w')
32 os.environ['MCOLLECTIVE_REPLY_FILE'] = tout.name
35 mc = mcollectiveah.MCollectiveAction()
36 mc.reply['message'] = mc.request['message']
39 tout = open(os.environ['MCOLLECTIVE_REPLY_FILE'], 'r')
40 data = simplejson.load(tout)
43 self.assertEqual(data, self.data)
46 os.unlink(os.environ['MCOLLECTIVE_REQUEST_FILE'])
47 os.unlink(os.environ['MCOLLECTIVE_REPLY_FILE'])
49 if __name__ == '__main__':