tests.py

   1 #!/usr/bin/env python
   2
   3 import unittest2 as unittest
   4 import config
   5 import auto
   6 import sys
   7 import util
   8 from modules.base import CannotAutomateException
   9
  10 import magiclink
  11 import constants
  12
  13 import logging
  14 logging.basicConfig(level=logging.DEBUG)
  15
  16 '''tests for mudpuppy'''
  17
  18 class OrchestraTests(unittest.TestCase):
  19         def test_orchestra_client(self):
  20                 oc = util.OrchestraClient()
  21                 jobid = oc.submit_job('madworld','one',config.default_orchestra_target)
  22                 status = oc.wait_for_completion(jobid)
  23                 self.assertEquals(status['Status'], 'FAIL')
  24                 jobid = oc.submit_job('helloworld','one',config.default_orchestra_target)
  25                 status = oc.wait_for_completion(jobid)
  26                 self.assertEquals(status['Status'], 'OK')
  27
  28 class ModuleTests(unittest.TestCase):
  29         def setUp(self):
  30                 reload(auto)
  31         def tearDown(self):
  32                 reload(auto)
  33         def test_1_load_modules(self):
  34                 config.installed_modules = ('test.TestModule', )
  35                 modules = auto.load_modules()
  36                 self.assertEquals(len(modules),1)
  37
  38                 # Shouldn't allow duplication of loads
  39                 self.assertRaises(auto.ModuleLoadException, auto.load_modules)
  40         def test_allocation_modules(self):
  41                 config.installed_modules = ('test.TestModule', )
  42                 modules = auto.load_modules()
  43
  44                 _itemdata = {'name': 'FakeItem'}
  45                 bots = auto.get_modules_for_item(_itemdata,{})
  46                 self.assertEquals(bots, [])
  47
  48                 _itemdata = {'name': 'TestModuleItem'}
  49                 bots = auto.get_modules_for_item(_itemdata,{})
  50                 for botclass in bots:
  51                         self.assertIsNotNone(botclass)
  52                         bot = botclass(_itemdata,{})
  53                         self.assertEquals(bot.module_name, 'TestModule')
  54         def test_question_modules(self):
  55                 config.installed_modules = ('test.TestFakeIP', )
  56                 modules = auto.load_modules()
  57                 _item_state = {'name': 'TestFakeIPitem'}
  58                 bots = auto.get_modules_for_item(_item_state,{})
  59                 for botclass in bots:
  60                         bot = botclass(_item_state,{})
  61                         self.assertIsNotNone(bot)
  62                         self.assertEquals(bot.module_name, 'TestFakeIP')
  63                         self.assertEquals(bot.do_magic(), {'task_metadata_update': {'ip': '10.0.0.1'}})
  64         def test_module_generated_cannot_automate(self):
  65                 config.installed_modules = ('test.TestCannotAutomate', )
  66                 modules = auto.load_modules()
  67                 _item_state = {'name': 'TestCannotAutomateItem'}
  68                 bots = auto.get_modules_for_item(_item_state,{})
  69                 for botclass in bots:
  70                         self.assertIsNotNone(botclass)
  71                         bot = botclass(_item_state, {})
  72                         self.assertRaises(CannotAutomateException, bot.do_magic)
  73
  74 class MagicAPITests(unittest.TestCase):
  75         def test_000_get_tasks(self):
  76                 api = magiclink.MagicAPI()
  77                 tasks = api.get_tasks()
  78                 self.assertIsInstance(tasks, list)
  79         def test_create_delete_tasks(self):
  80                 api = magiclink.MagicAPI()
  81                 taskdata = api.create_task(dict(requirements=[], description="UNITTEST EPHEMERAL TEST TASK", automate=False))
  82                 uuid = taskdata['metadata']['uuid']
  83                 tasks = api.get_tasks()
  84                 self.assertIn(uuid,tasks)
  85                 api.delete_task(uuid)
  86                 tasks = api.get_tasks()
  87                 self.assertNotIn(uuid,tasks)
  88
  89 class MagicLinkTests(unittest.TestCase):
  90         def test_state_changes(self):
  91                 api = magiclink.MagicLink()
  92                 taskdata = api.create_task(dict(requirements=[], description="UNITTEST EPHEMERAL TEST TASK", automate=False))
  93                 uuid = taskdata['metadata']['uuid']
  94                 tasks = api.get_tasks()
  95                 self.assertIn(uuid,tasks)
  96
  97                 avail = api.get_available_items(uuid)
  98                 itemname = avail[0]['name']
  99
 100                 self.assertGreater(len(avail), 0)
 101                 self.assertEqual(api.get_item_state(uuid,itemname),constants.INCOMPLETE)
 102                 success = api.update_item_state(uuid, itemname, constants.INCOMPLETE, constants.IN_PROGRESS)
 103                 self.assertTrue(success)
 104                 self.assertEqual(api.get_item_state(uuid,itemname),constants.IN_PROGRESS)
 105
 106                 success = api.update_item_state(uuid, itemname, constants.IN_PROGRESS, constants.COMPLETE)
 107                 self.assertTrue(success)
 108                 self.assertEqual(api.get_item_state(uuid,itemname),constants.COMPLETE)
 109                 avail = api.get_available_items(uuid)
 110                 self.assertNotEqual(avail[0]['name'],itemname)
 111
 112                 api.delete_task(uuid)
 113                 tasks = api.get_tasks()
 114                 self.assertNotIn(uuid,tasks)
 115
 116 if __name__ == "__main__":
 117         unittest.main()