openstack/python-mistralclient
Code Issues Proposed changes

Modify API tests to check the calls

Browse Source 
Since often mocked response is returned untouched by the tested
function, we need to check not only that it returns the proper
response, but also that it makes a right request too.

Change-Id: I87bf79d9c8648f8377c0c27b1cd4992a6e5af598
Closes-Bug: #1288169
This commit is contained in:
Kirill Izotov
2014-03-07 08:29:48 +07:00
parent 9680b76b99
commit 95b0fa9660
5 changed files with 197 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
mistralclient/tests/base.py
View File

@@ -16,6 +16,7 @@
import unittest2
import mock
import json
from mistralclient.api import client
@@ -23,13 +24,12 @@ from mistralclient.api import client
class FakeResponse(object):
"""Fake response for testing Mistral Client."""
def __init__(self, status_code, json_values={}, content=None):
def __init__(self, status_code, content=None):
self.status_code = status_code
self.json_values = json_values
self.content = content
def json(self):
return self.json_values
return json.loads(self.content)
class BaseClientTest(unittest2.TestCase):
@@ -41,22 +41,39 @@ class BaseClientTest(unittest2.TestCase):
self.tasks = self._client.tasks
self.listeners = self._client.listeners
def mock_http_get(self, json, status_code=200):
def mock_http_get(self, content, status_code=200):
if isinstance(content, dict):
content = json.dumps(content)
self._client.http_client.get = \
mock.MagicMock(return_value=FakeResponse(status_code, json))
mock.MagicMock(return_value=FakeResponse(status_code, content))
return self._client.http_client.get
def mock_http_post(self, content, status_code=201):
if isinstance(content, dict):
content = json.dumps(content)
def mock_http_post(self, json, status_code=201):
self._client.http_client.post = \
mock.MagicMock(return_value=FakeResponse(status_code, json))
mock.MagicMock(return_value=FakeResponse(status_code, content))
return self._client.http_client.post
def mock_http_put(self, content, status_code=200):
if isinstance(content, dict):
content = json.dumps(content)
def mock_http_put(self, json, status_code=200):
self._client.http_client.put = \
mock.MagicMock(return_value=FakeResponse(status_code, json))
mock.MagicMock(return_value=FakeResponse(status_code, content))
return self._client.http_client.put
def mock_http_delete(self, status_code=204):
self._client.http_client.delete = \
mock.MagicMock(return_value=FakeResponse(status_code))
return self._client.http_client.delete
class BaseCommandTest(unittest2.TestCase):
def setUp(self):

88
mistralclient/tests/test_executions.py
View File

@@ -15,8 +15,10 @@
# limitations under the License.
import unittest2
import json
from mistralclient.tests import base
from mistralclient.api.executions import Execution
# TODO: Later we need additional tests verifying all the errors etc.
@@ -37,85 +39,103 @@ EXECS = [
}
]
URL_TEMPLATE = '/workbooks/%s/executions'
URL_TEMPLATE_ID = '/workbooks/%s/executions/%s'
class TestExecutions(base.BaseClientTest):
def test_create(self):
self.mock_http_post(json=EXECS[0])
mock = self.mock_http_post(content=EXECS[0])
body = {
'task': EXECS[0]['target_task'],
'context': EXECS[0]['context'],
'workbook_name': EXECS[0]['workbook_name']
}
ex = self.executions.create(EXECS[0]['workbook_name'],
EXECS[0]['target_task'],
EXECS[0]['context'])
self.assertIsNotNone(ex)
self.assertEqual(EXECS[0]['id'], ex.id)
self.assertEqual(EXECS[0]['workbook_name'], ex.workbook_name)
self.assertEqual(EXECS[0]['target_task'], ex.target_task)
self.assertEqual(EXECS[0]['state'], ex.state)
self.assertEqual(EXECS[0]['context'], ex.context)
self.assertEquals(Execution(self.executions, EXECS[0]).__dict__,
ex.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE % EXECS[0]['workbook_name'],
json.dumps(body))
def test_create_with_empty_context(self):
execs = EXECS[0].copy()
execs.pop('context')
self.mock_http_post(json=execs)
ex = self.executions.create(execs['workbook_name'],
execs['target_task'])
with self.assertRaises(AttributeError):
ex.context
mock = self.mock_http_post(content=EXECS[0])
body = {
'task': EXECS[0]['target_task'],
'workbook_name': EXECS[0]['workbook_name']
}
self.executions.create(EXECS[0]['workbook_name'],
EXECS[0]['target_task'])
mock.assert_called_once_with(
URL_TEMPLATE % EXECS[0]['workbook_name'],
json.dumps(body))
@unittest2.expectedFailure
def test_create_failure1(self):
self.mock_http_post(content=EXECS[0])
self.executions.create(EXECS[0]['workbook_name'],
EXECS[0]['target_task'],
"sdfsdf")
@unittest2.expectedFailure
def test_create_failure2(self):
self.mock_http_post(content=EXECS[0])
self.executions.create(EXECS[0]['workbook_name'],
EXECS[0]['target_task'],
list('343', 'sdfsd'))
def test_update(self):
self.mock_http_put(json=EXECS[0])
mock = self.mock_http_put(content=EXECS[0])
body = {
'workbook_name': EXECS[0]['workbook_name'],
'id': EXECS[0]['id'],
'state': EXECS[0]['state']
}
ex = self.executions.update(EXECS[0]['workbook_name'],
EXECS[0]['id'],
EXECS[0]['state'])
self.assertIsNotNone(ex)
self.assertEqual(EXECS[0]['id'], ex.id)
self.assertEqual(EXECS[0]['workbook_name'], ex.workbook_name)
self.assertEqual(EXECS[0]['target_task'], ex.target_task)
self.assertEqual(EXECS[0]['state'], ex.state)
self.assertEqual(EXECS[0]['context'], ex.context)
self.assertEquals(Execution(self.executions, EXECS[0]).__dict__,
ex.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE_ID % (EXECS[0]['workbook_name'], EXECS[0]['id']),
json.dumps(body))
def test_list(self):
self.mock_http_get(json={'executions': EXECS})
mock = self.mock_http_get(content={'executions': EXECS})
executions = self.executions.list(EXECS[0]['workbook_name'])
self.assertEqual(1, len(executions))
ex = executions[0]
self.assertEqual(EXECS[0]['id'], ex.id)
self.assertEqual(EXECS[0]['workbook_name'], ex.workbook_name)
self.assertEqual(EXECS[0]['target_task'], ex.target_task)
self.assertEqual(EXECS[0]['state'], ex.state)
self.assertEqual(EXECS[0]['context'], ex.context)
self.assertEquals(Execution(self.executions, EXECS[0]).__dict__,
ex.__dict__)
mock.assert_called_once_with(URL_TEMPLATE % EXECS[0]['workbook_name'])
def test_get(self):
self.mock_http_get(json=EXECS[0])
mock = self.mock_http_get(content=EXECS[0])
ex = self.executions.get(EXECS[0]['workbook_name'], EXECS[0]['id'])
self.assertEqual(EXECS[0]['id'], ex.id)
self.assertEqual(EXECS[0]['workbook_name'], ex.workbook_name)
self.assertEqual(EXECS[0]['target_task'], ex.target_task)
self.assertEqual(EXECS[0]['state'], ex.state)
self.assertEqual(EXECS[0]['context'], ex.context)
self.assertEquals(Execution(self.executions, EXECS[0]).__dict__,
ex.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE_ID % (EXECS[0]['workbook_name'], EXECS[0]['id']))
def test_delete(self):
self.mock_http_delete(status_code=204)
mock = self.mock_http_delete(status_code=204)
# Just make sure it doesn't throw any exceptions.
self.executions.delete(EXECS[0]['workbook_name'], EXECS[0]['id'])
mock.assert_called_once_with(
URL_TEMPLATE_ID % (EXECS[0]['workbook_name'], EXECS[0]['id']))

72
mistralclient/tests/test_listeners.py
View File

@@ -14,7 +14,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import json
from mistralclient.tests import base
from mistralclient.api.listeners import Listener
# TODO: later we need additional tests verifying all the errors etc.
@@ -27,62 +30,85 @@ LISTENERS = [
}
]
URL_TEMPLATE = '/workbooks/%s/listeners'
URL_TEMPLATE_ID = '/workbooks/%s/listeners/%s'
class TestListeners(base.BaseClientTest):
def test_create(self):
self.mock_http_post(json=LISTENERS[0])
mock = self.mock_http_post(content=LISTENERS[0])
body = {
'workbook_name': LISTENERS[0]['workbook_name'],
'description': LISTENERS[0]['description'],
'webhook': LISTENERS[0]['webhook'],
'events': None
}
lsnr = self.listeners.create(LISTENERS[0]['workbook_name'],
LISTENERS[0]['webhook'],
LISTENERS[0]['description'])
self.assertIsNotNone(lsnr)
self.assertEqual(LISTENERS[0]['id'], lsnr.id)
self.assertEqual(LISTENERS[0]['workbook_name'], lsnr.workbook_name)
self.assertEqual(LISTENERS[0]['webhook'], lsnr.webhook)
self.assertEqual(LISTENERS[0]['description'], lsnr.description)
self.assertEquals(Listener(self.listeners, LISTENERS[0]).__dict__,
lsnr.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE % (LISTENERS[0]['workbook_name']),
json.dumps(body))
def test_update(self):
self.mock_http_put(json=LISTENERS[0])
mock = self.mock_http_put(content=LISTENERS[0])
body = {
'id': LISTENERS[0]['id'],
'workbook_name': LISTENERS[0]['workbook_name'],
'description': LISTENERS[0]['description'],
'webhook': LISTENERS[0]['webhook'],
'events': None
}
lsnr = self.listeners.update(LISTENERS[0]['workbook_name'],
LISTENERS[0]['id'],
LISTENERS[0]['webhook'],
LISTENERS[0]['description'])
self.assertIsNotNone(lsnr)
self.assertEqual(LISTENERS[0]['id'], lsnr.id)
self.assertEqual(LISTENERS[0]['workbook_name'], lsnr.workbook_name)
self.assertEqual(LISTENERS[0]['webhook'], lsnr.webhook)
self.assertEqual(LISTENERS[0]['description'], lsnr.description)
self.assertEquals(Listener(self.listeners, LISTENERS[0]).__dict__,
lsnr.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE_ID % (LISTENERS[0]['workbook_name'],
LISTENERS[0]['id']),
json.dumps(body))
def test_list(self):
self.mock_http_get(json={'listeners': LISTENERS})
mock = self.mock_http_get(content={'listeners': LISTENERS})
listeners = self.listeners.list(LISTENERS[0]['workbook_name'])
self.assertEqual(1, len(listeners))
lsnr = listeners[0]
self.assertEqual(LISTENERS[0]['id'], lsnr.id)
self.assertEqual(LISTENERS[0]['workbook_name'], lsnr.workbook_name)
self.assertEqual(LISTENERS[0]['webhook'], lsnr.webhook)
self.assertEqual(LISTENERS[0]['description'], lsnr.description)
self.assertEquals(Listener(self.listeners, LISTENERS[0]).__dict__,
lsnr.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE % (LISTENERS[0]['workbook_name']))
def test_get(self):
self.mock_http_get(json=LISTENERS[0])
mock = self.mock_http_get(content=LISTENERS[0])
lsnr = self.listeners.get(LISTENERS[0]['workbook_name'],
LISTENERS[0]['id'])
self.assertEqual(LISTENERS[0]['id'], lsnr.id)
self.assertEqual(LISTENERS[0]['workbook_name'], lsnr.workbook_name)
self.assertEqual(LISTENERS[0]['webhook'], lsnr.webhook)
self.assertEqual(LISTENERS[0]['description'], lsnr.description)
self.assertEquals(Listener(self.listeners, LISTENERS[0]).__dict__,
lsnr.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE_ID % (LISTENERS[0]['workbook_name'],
LISTENERS[0]['id']))
def test_delete(self):
self.mock_http_delete(status_code=204)
mock = self.mock_http_delete(status_code=204)
# Just make sure it doesn't throw any exceptions.
self.listeners.delete(LISTENERS[0]['workbook_name'],
LISTENERS[0]['id'])
mock.assert_called_once_with(
URL_TEMPLATE_ID % (LISTENERS[0]['workbook_name'],
LISTENERS[0]['id']))

55
mistralclient/tests/test_tasks.py
View File

@@ -14,7 +14,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import json
from mistralclient.tests import base
from mistralclient.api.tasks import Task
# TODO: later we need additional tests verifying all the errors etc.
@@ -31,10 +34,19 @@ TASKS = [
}
]
URL_TEMPLATE = '/workbooks/%s/executions/%s/tasks'
URL_TEMPLATE_ID = '/workbooks/%s/executions/%s/tasks/%s'
class TestTasks(base.BaseClientTest):
def test_update(self):
self.mock_http_put(json=TASKS[0])
mock = self.mock_http_put(content=TASKS[0])
body = {
'workbook_name': TASKS[0]['workbook_name'],
'execution_id': TASKS[0]['execution_id'],
'id': TASKS[0]['id'],
'state': TASKS[0]['state']
}
task = self.tasks.update(TASKS[0]['workbook_name'],
TASKS[0]['execution_id'],
@@ -42,43 +54,36 @@ class TestTasks(base.BaseClientTest):
TASKS[0]['state'])
self.assertIsNotNone(task)
self.assertEqual(TASKS[0]['id'], task.id)
self.assertEqual(TASKS[0]['workbook_name'], task.workbook_name)
self.assertEqual(TASKS[0]['execution_id'], task.execution_id)
self.assertEqual(TASKS[0]['description'], task.description)
self.assertEqual(TASKS[0]['action'], task.action)
self.assertEqual(TASKS[0]['state'], task.state)
self.assertEqual(TASKS[0]['tags'], task.tags)
self.assertEquals(Task(self.tasks, TASKS[0]).__dict__, task.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE_ID % (TASKS[0]['workbook_name'],
TASKS[0]['execution_id'],
TASKS[0]['id']),
json.dumps(body))
def test_list(self):
self.mock_http_get(json={'tasks': TASKS})
mock = self.mock_http_get(content={'tasks': TASKS})
tasks = self.tasks.list(TASKS[0]['workbook_name'],
TASKS[0]['execution_id'])
self.assertEqual(1, len(tasks))
task = tasks[0]
self.assertEqual(TASKS[0]['id'], task.id)
self.assertEqual(TASKS[0]['workbook_name'], task.workbook_name)
self.assertEqual(TASKS[0]['execution_id'], task.execution_id)
self.assertEqual(TASKS[0]['description'], task.description)
self.assertEqual(TASKS[0]['action'], task.action)
self.assertEqual(TASKS[0]['state'], task.state)
self.assertEqual(TASKS[0]['tags'], task.tags)
self.assertEquals(Task(self.tasks, TASKS[0]).__dict__, task.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE % (TASKS[0]['workbook_name'],
TASKS[0]['execution_id']))
def test_get(self):
self.mock_http_get(json=TASKS[0])
mock = self.mock_http_get(content=TASKS[0])
task = self.tasks.get(TASKS[0]['workbook_name'],
TASKS[0]['execution_id'],
TASKS[0]['id'])
self.assertEqual(TASKS[0]['id'], task.id)
self.assertEqual(TASKS[0]['workbook_name'], task.workbook_name)
self.assertEqual(TASKS[0]['execution_id'], task.execution_id)
self.assertEqual(TASKS[0]['description'], task.description)
self.assertEqual(TASKS[0]['action'], task.action)
self.assertEqual(TASKS[0]['state'], task.state)
self.assertEqual(TASKS[0]['tags'], task.tags)
self.assertEquals(Task(self.tasks, TASKS[0]).__dict__, task.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE_ID % (TASKS[0]['workbook_name'],
TASKS[0]['execution_id'],
TASKS[0]['id']))

63
mistralclient/tests/test_workbooks.py
View File

@@ -14,8 +14,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import mock
import json
from mistralclient.tests import base
from mistralclient.api.workbooks import Workbook
# TODO: later we need additional tests verifying all the errors etc.
@@ -52,71 +54,82 @@ Workflow:
param2: 42
"""
URL_TEMPLATE = '/workbooks'
URL_TEMPLATE_NAME = '/workbooks/%s'
URL_TEMPLATE_DEFINITION = '/workbooks/%s/definition'
class TestWorkbooks(base.BaseClientTest):
def test_create(self):
self.mock_http_post(json=WORKBOOKS[0])
mock = self.mock_http_post(content=WORKBOOKS[0])
wb = self.workbooks.create(WORKBOOKS[0]['name'],
WORKBOOKS[0]['description'],
WORKBOOKS[0]['tags'])
self.assertIsNotNone(wb)
self.assertEqual(WORKBOOKS[0]['name'], wb.name)
self.assertEqual(WORKBOOKS[0]['description'], wb.description)
self.assertEqual(WORKBOOKS[0]['tags'], wb.tags)
self.assertEquals(Workbook(self.workbooks, WORKBOOKS[0]).__dict__,
wb.__dict__)
mock.assert_called_once_with(URL_TEMPLATE, json.dumps(WORKBOOKS[0]))
def test_update(self):
self.mock_http_put(json=WORKBOOKS[0])
mock = self.mock_http_put(content=WORKBOOKS[0])
wb = self.workbooks.update(WORKBOOKS[0]['name'],
WORKBOOKS[0]['description'],
WORKBOOKS[0]['tags'])
self.assertIsNotNone(wb)
self.assertEqual(WORKBOOKS[0]['name'], wb.name)
self.assertEqual(WORKBOOKS[0]['description'], wb.description)
self.assertEqual(WORKBOOKS[0]['tags'], wb.tags)
self.assertEquals(Workbook(self.workbooks, WORKBOOKS[0]).__dict__,
wb.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE_NAME % WORKBOOKS[0]['name'],
json.dumps(WORKBOOKS[0]))
def test_list(self):
self.mock_http_get(json={'workbooks': WORKBOOKS})
mock = self.mock_http_get(content={'workbooks': WORKBOOKS})
workbooks = self.workbooks.list()
self.assertEqual(1, len(workbooks))
wb = workbooks[0]
self.assertEqual(WORKBOOKS[0]['name'], wb.name)
self.assertEqual(WORKBOOKS[0]['description'], wb.description)
self.assertEqual(WORKBOOKS[0]['tags'], wb.tags)
self.assertEquals(Workbook(self.workbooks, WORKBOOKS[0]).__dict__,
wb.__dict__)
mock.assert_called_once_with(URL_TEMPLATE)
def test_get(self):
self.mock_http_get(json=WORKBOOKS[0])
mock = self.mock_http_get(content=WORKBOOKS[0])
wb = self.workbooks.get(WORKBOOKS[0]['name'])
self.assertIsNotNone(wb)
self.assertEqual(WORKBOOKS[0]['name'], wb.name)
self.assertEqual(WORKBOOKS[0]['description'], wb.description)
self.assertEqual(WORKBOOKS[0]['tags'], wb.tags)
self.assertEquals(Workbook(self.workbooks, WORKBOOKS[0]).__dict__,
wb.__dict__)
mock.assert_called_once_with(URL_TEMPLATE_NAME % WORKBOOKS[0]['name'])
def test_delete(self):
self.mock_http_delete(status_code=204)
mock = self.mock_http_delete(status_code=204)
# Just make sure it doesn't throw any exceptions.
self.workbooks.delete(WORKBOOKS[0]['name'])
def test_upload_definition(self):
self.mock_http_put(None, status_code=200)
mock.assert_called_once_with(URL_TEMPLATE_NAME % WORKBOOKS[0]['name'])
def test_upload_definition(self):
mock = self.mock_http_put(None, status_code=200)
# Just make sure it doesn't throw any exceptions.
self.workbooks.upload_definition("my_workbook", WB_DEF)
mock.assert_called_once_with(
URL_TEMPLATE_DEFINITION % WORKBOOKS[0]['name'],
WB_DEF,
headers={'content-type': 'text/plain'})
def test_get_definition(self):
self._client.http_client.get =\
mock.MagicMock(return_value=base.FakeResponse(200, None, WB_DEF))
mock = self.mock_http_get(status_code=200, content=WB_DEF)
text = self.workbooks.get_definition("my_workbook")
self.assertEqual(WB_DEF, text)
mock.assert_called_once_with(URL_TEMPLATE_DEFINITION
% WORKBOOKS[0]['name'])