task_queue=LCM_TASK_QUEUE,
)
- assert [
- call.kwargs["workflow"] == WORKFLOW_VNF_INSTANTIATE
- and call.kwargs["arg"].vnfr_uuid
- for call in mock_execute_child_workflow.call_args_list
- ] == vnfr_ids
+ self.assertEqual(
+ [
+ call.kwargs["workflow"] == WORKFLOW_VNF_INSTANTIATE
+ and call.kwargs["arg"].vnfr_uuid
+ for call in mock_execute_child_workflow.call_args_list
+ ],
+ vnfr_ids,
+ )
- assert_lcm_op_states(
+ self.assert_lcm_op_states(
self.mock_update_lcm_operation_state_tracker.call_args_list,
[LcmOperationState.PROCESSING, LcmOperationState.COMPLETED],
)
- assert_ns_states(
+ self.assert_ns_states(
self.mock_update_ns_state_tracker.call_args_list, [NsState.INSTANTIATED]
)
],
debug_mode=True,
):
- # TODO: Check this WorkflowFailureError is caused by a TestException
- with self.assertRaises(WorkflowFailureError):
+ with self.assertRaises(WorkflowFailureError) as e_info:
await env.client.execute_workflow(
NsInstantiateWorkflow,
arg=self.input,
id=self.input.nslcmop["nsInstanceId"],
task_queue=LCM_TASK_QUEUE,
)
+ self.assertEqual(e_info.exception.cause.cause.type, "TestException")
- assert_lcm_op_states(
+ self.assert_lcm_op_states(
self.mock_update_lcm_operation_state_tracker.call_args_list,
[LcmOperationState.PROCESSING, LcmOperationState.FAILED],
)
],
debug_mode=True,
):
- # TODO: Check this WorkflowFailureError is caused by a TestException
- with self.assertRaises(WorkflowFailureError):
+ with self.assertRaises(WorkflowFailureError) as e_info:
await env.client.execute_workflow(
NsInstantiateWorkflow,
arg=self.input,
id=self.input.nslcmop["nsInstanceId"],
task_queue=LCM_TASK_QUEUE,
)
- assert_ns_states(
+ self.assertEqual(e_info.exception.cause.cause.type, "TestException")
+ self.assert_ns_states(
self.mock_update_ns_state_tracker.call_args_list, [NsState.INSTANTIATED]
)
+ def assert_ns_states(self, call_args_list, expected_states):
+ """Asserts that the NS state was set to each of the expected states (in order)."""
+ self.assertEqual(
+ [call.args[0].state for call in call_args_list], expected_states
+ )
-def assert_ns_states(call_args_list, expected_states):
- """Asserts that the NS state was set to each of the expected states (in order)."""
- assert [call.args[0].state for call in call_args_list] == expected_states
-
-
-def assert_lcm_op_states(call_args_list, expected_states):
- """Asserts that the LCM operation was set to each of the exepcted states (in order)."""
- assert [call.args[0].op_state for call in call_args_list] == expected_states
+ def assert_lcm_op_states(self, call_args_list, expected_states):
+ """Asserts that the LCM operation was set to each of the exepcted states (in order)."""
+ self.assertEqual(
+ [call.args[0].op_state for call in call_args_list], expected_states
+ )