diff --git a/SAS/TMSS/services/websocket/test/t_websocket_service.py b/SAS/TMSS/services/websocket/test/t_websocket_service.py index 3adb826d11dfa0251c434d5a2a2f2eaef52d8c01..92a18cc185114305fa203f06b0809b298961fbca 100755 --- a/SAS/TMSS/services/websocket/test/t_websocket_service.py +++ b/SAS/TMSS/services/websocket/test/t_websocket_service.py @@ -75,7 +75,6 @@ class TestSubtaskSchedulingService(unittest.TestCase): on_close=on_close) self.ws.run_forever() - websocket.enableTrace(True) self.t = Thread(target=thread_ws_starter, daemon=True) self.t.start() @@ -98,7 +97,7 @@ class TestSubtaskSchedulingService(unittest.TestCase): def test_01(self): ''' - This test starts a scheduling service and tmss, creates a chain of subtasks, finishes the first, and checks if the successors are then scheduled. + This test starts a websocket service and tmss. Creates, updates and deletes objects to check if json_blobs from the ws service are properly received. ''' logger.info(' -- test_01_for_expected_behaviour -- ') @@ -120,74 +119,77 @@ class TestSubtaskSchedulingService(unittest.TestCase): json_blob['object'] = json_test self.assertEqual(json_blob, self.msg_queue.popleft()) + # Test creations # Test scheduling_unit_draft create - logger.info('\n\n\nCreate su_draft\n\n') su_draft = self.test_data_creator.post_data_and_get_response_as_json_object( self.test_data_creator.SchedulingUnitDraft(), '/scheduling_unit_draft/') test_object(su_draft, self.ObjTypes.SCHED_UNIT_DRAFT, self.ObjActions.CREATE) # Test task_draft create - logger.info('\n\n\nCreate task_draft\n\n') task_draft = self.test_data_creator.post_data_and_get_response_as_json_object( self.test_data_creator.TaskDraft(scheduling_unit_draft_url=su_draft['url']), '/task_draft/') test_object(task_draft, self.ObjTypes.TASK_DRAFT, self.ObjActions.CREATE) # Test scheduling_unit_blueprint create - logger.info('\n\n\nCreate su_blueprint\n\n') su_blueprint = self.test_data_creator.post_data_and_get_response_as_json_object( self.test_data_creator.SchedulingUnitBlueprint(scheduling_unit_draft_url=su_draft['url']), '/scheduling_unit_blueprint/') test_object(su_blueprint, self.ObjTypes.SCHED_UNIT_BLUEPRINT, self.ObjActions.CREATE) # Test task_blueprint create - logger.info('\n\n\nCreate task_blueprint\n\n') task_blueprint = self.test_data_creator.post_data_and_get_response_as_json_object( self.test_data_creator.TaskBlueprint(scheduling_unit_blueprint_url=su_blueprint['url'], draft_url=task_draft['url']), '/task_blueprint/') test_object(task_blueprint, self.ObjTypes.TASK_BLUEPRINT, self.ObjActions.CREATE) # Test subtask create - logger.info('\n\n\nCreate subtask\n\n') subtask = self.test_data_creator.post_data_and_get_response_as_json_object( self.test_data_creator.Subtask(task_blueprint_url=task_blueprint['url']), '/subtask/') test_object(subtask, self.ObjTypes.SUBTASK, self.ObjActions.CREATE) - # TODO: Add task_draft and su_draft updates # Test updates with self.tmss_test_env.create_tmss_client() as client: # Test subtask update - logger.info('\n\n\nUpdate subtask\n\n') client.set_subtask_status(subtask['id'], 'scheduled') subtask = requests.get(subtask['url'], auth=self.test_data_creator.auth).json() test_object(subtask, self.ObjTypes.SUBTASK, self.ObjActions.UPDATE) + # Test task_blueprint update - logger.info('\n\n\nUpdate task_blueprint\n\n') task_blueprint = requests.get(task_blueprint['url'], auth=self.test_data_creator.auth).json() test_object(task_blueprint, self.ObjTypes.TASK_BLUEPRINT, self.ObjActions.UPDATE) + # Test scheduling_unit_blueprint update - logger.info('\n\n\nUpdate su_blueprint\n\n') su_blueprint = requests.get(su_blueprint['url'], auth=self.test_data_creator.auth).json() test_object(su_blueprint, self.ObjTypes.SCHED_UNIT_BLUEPRINT, self.ObjActions.UPDATE) + # Test scheduling_unit_draft update + su_draft['description'] = 'This is an update test' + su_draft = requests.put(su_draft['url'], json=su_draft, auth=self.test_data_creator.auth).json() + test_object(su_draft, self.ObjTypes.SCHED_UNIT_DRAFT, self.ObjActions.UPDATE) + + # Test task_draft update + task_draft['description'] = 'This is an update test' + task_draft = requests.put(task_draft['url'], json=task_draft, auth=self.test_data_creator.auth).json() + test_object(task_draft, self.ObjTypes.TASK_DRAFT, self.ObjActions.UPDATE) + # Test deletions - # Test substask delete - logger.info('\n\n\nDelete subtask\n\n') + # Test subtask delete requests.delete(subtask['url'], auth=self.test_data_creator.auth) test_object({'id': subtask['id']}, self.ObjTypes.SUBTASK, self.ObjActions.DELETE) + # Test task_blueprint delete - logger.info('\n\n\nDelete task_blueprint\n\n') requests.delete(task_blueprint['url'], auth=self.test_data_creator.auth) test_object({'id': task_blueprint['id']}, self.ObjTypes.TASK_BLUEPRINT, self.ObjActions.DELETE) + # Test scheduling_unit_blueprint delete - logger.info('\n\n\nDelete su_blueprint\n\n') requests.delete(su_blueprint['url'], auth=self.test_data_creator.auth) test_object({'id': su_blueprint['id']}, self.ObjTypes.SCHED_UNIT_BLUEPRINT, self.ObjActions.DELETE) + # Test task_draft delete - logger.info('\n\n\nDelete task_draft\n\n') requests.delete(task_draft['url'], auth=self.test_data_creator.auth) test_object({'id': task_draft['id']}, self.ObjTypes.TASK_DRAFT, self.ObjActions.DELETE) + # Test scheduling_unit_draft delete - logger.info('\n\n\nDelete su_draft\n\n') requests.delete(su_draft['url'], auth=self.test_data_creator.auth) test_object({'id': su_draft['id']}, self.ObjTypes.SCHED_UNIT_DRAFT, self.ObjActions.DELETE)