3 import unittest2 as unittest
8 from modules.base import CannotAutomateException
14 logging.basicConfig(level=logging.DEBUG)
16 '''tests for mudpuppy'''
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')
28 class ModuleTests(unittest.TestCase):
33 def test_1_load_modules(self):
34 config.installed_modules = ('test.TestModule', )
35 modules = auto.load_modules()
36 self.assertEquals(len(modules),1)
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()
44 _itemdata = {'name': 'FakeItem'}
45 bots = auto.get_modules_for_item(_itemdata,{})
46 self.assertEquals(bots, [])
48 _itemdata = {'name': 'TestModuleItem'}
49 bots = auto.get_modules_for_item(_itemdata,{})
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,{})
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,{})
70 self.assertIsNotNone(botclass)
71 bot = botclass(_item_state, {})
72 self.assertRaises(CannotAutomateException, bot.do_magic)
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)
86 tasks = api.get_tasks()
87 self.assertNotIn(uuid,tasks)
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)
97 avail = api.get_available_items(uuid)
98 itemname = avail[0]['name']
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)
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)
112 api.delete_task(uuid)
113 tasks = api.get_tasks()
114 self.assertNotIn(uuid,tasks)
116 if __name__ == "__main__":