Reputation: 20916
I have couple of tasks in my workflow..Now, I want to mock the output of Task1 returning exception and just test task2.run method is not called.
class Task1():
def run(self):
try:
return 2,2
except Exception as e:
print e
raise Exception('Task1 Exception')
class Task2():
def run(self,x,y):
try:
return 2 * (x*y)
except Exception as e:
print e
raise Exception('Task2 Exception')
class Workflow():
def run(self,task1, task2):
x,y = task1.run()
print task2.run(x,y)
task1 = Task1()
task2 = Task2()
w = Workflow()
w.run(task1,task2)
Here is my unittest -
import unittest
import mock
from test2 import Task1, Task2, Workflow
class TestWorkflow(unittest.TestCase):
def test_workflow(self):
self.task1 = mock.MagicMock(Task1()) # mocking task1
self.task2 = mock.MagicMock(Task2())
self.task1.side_effect = Exception() # setting task1 exception to an exception object
self.workflow = Workflow()
self.workflow.run(self.task1, self.task2)
self.task2.run.assert_not_called() # checking task2 is not called.
if __name__ == '__main__':
unittest.main()
Exception:-
x,y = task1.run()
ValueError: need more than 0 values to unpack
Upvotes: 1
Views: 3061
Reputation: 5090
Given the syntax, you are using Python 2, but I will refer to the documentation in the Python 3 official documentation anyway. I believe the solution applies.
The first example is:
from unittest.mock import MagicMock
thing = ProductionClass()
thing.method = MagicMock(return_value=3)
thing.method(3, 4, 5, key='value')
#prints 3
thing.method.assert_called_with(3, 4, 5, key='value')
Bringing this your problem you would have the code below instead of your current code. See that I'm assigning the Task1.run and Task2.run methods to a MagicMock instance.
import unittest
import mock
from t import Task1, Task2, Workflow
class TestWorkflow(unittest.TestCase):
def test_workflow(self):
self.task1 = Task1()
self.task1.run = mock.MagicMock() # mocking run from task1
self.task2 = Task2()
self.task2.run = mock.MagicMock()
self.task1.run.side_effect = Exception() # setting task1 exception to an exception object
self.workflow = Workflow()
self.workflow.run(self.task1, self.task2)
self.task2.run.assert_not_called() # checking task2 is not called.
if __name__ == '__main__':
unittest.main()
Upvotes: 1