openstack/python-mistralclient
Code Issues Proposed changes

Removing v1 stuff (API methods, commands, tests)

Browse Source 
Change-Id: I561a75aaa0a303d528159831a3a9bc8363a2877d
This commit is contained in:
Renat Akhmerov 2015-04-08 14:24:09 +06:00
parent 649dfbc28e
commit f1bbe75134
32 changed files with 31 additions and 2300 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
AUTHORS
View File

@ -6,9 +6,12 @@ Dmitri Zimine <dz@stackstorm.com>
Jeremy Stanley <fungi@yuggoth.org>
Kirill Izotov <enykeev@stackstorm.com>
Leandro I. Costantino <leandro.i.costantino@intel.com>
Lingxian Kong <konglingxian@huawei.com>
Nikolay Mahotkin <nmakhotkin@mirantis.com>
Pierre-Arthur MATHIEU <pierre-arthur.mathieu@hp.com>
Renat Akhmerov <rakhmerov@mirantis.com>
Tetiana Lashchova <tlashchova@mirantis.com>
Thomas Goirand <zigo@debian.org>
Timur Nurlygayanov <tnurlygayanov@mirantis.com>
Winson Chan <wcchan@stackstorm.com>

13
functionaltests/resources/v1/wb_v1.yaml
View File

@ -1,13 +0,0 @@
---
Namespaces:
Greetings:
actions:
hello:
class: std.echo
base-parameters:
output: Hello!
Workflow:
tasks:
hello:
action: Greetings.hello

31
mistralclient/api/client.py
View File

@ -14,7 +14,6 @@
import six
from mistralclient.api.v1 import client as client_v1
from mistralclient.api.v2 import client as client_v2
@ -24,30 +23,28 @@ def client(mistral_url=None, username=None, api_key=None,
auth_token=None, user_id=None, cacert=None):
if mistral_url and not isinstance(mistral_url, six.string_types):
raise RuntimeError('Mistral url should be string')
raise RuntimeError('Mistral url should be a string.')
if not mistral_url:
mistral_url = "http://localhost:8989/v2"
version = determine_client_version(mistral_url)
if version == 1:
client_cls = client_v1.Client
else:
client_cls = client_v2.Client
return client_cls(mistral_url=mistral_url, username=username,
api_key=api_key, project_name=project_name,
auth_url=auth_url, project_id=project_id,
endpoint_type=endpoint_type,
service_type=service_type, auth_token=auth_token,
user_id=user_id, cacert=cacert)
return client_v2.Client(
mistral_url=mistral_url,
username=username,
api_key=api_key,
project_name=project_name,
auth_url=auth_url,
project_id=project_id,
endpoint_type=endpoint_type,
service_type=service_type,
auth_token=auth_token,
user_id=user_id,
cacert=cacert
)
def determine_client_version(mistral_url):
if mistral_url.find("v2") != -1:
return 2
elif mistral_url.find("v1") != -1:
return 1
raise RuntimeError("Can not determine mistral API version")

0
mistralclient/api/v1/__init__.py
View File

112
mistralclient/api/v1/client.py
View File

@ -1,112 +0,0 @@
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import six
from mistralclient.api import httpclient
from mistralclient.api.v1 import executions
from mistralclient.api.v1 import listeners
from mistralclient.api.v1 import tasks
from mistralclient.api.v1 import workbooks
class Client(object):
def __init__(self, mistral_url=None, username=None, api_key=None,
project_name=None, auth_url=None, project_id=None,
endpoint_type='publicURL', service_type='workflow',
auth_token=None, user_id=None, cacert=None):
if mistral_url and not isinstance(mistral_url, six.string_types):
raise RuntimeError('Mistral url should be string')
if auth_url:
(mistral_url, auth_token, project_id, user_id) = (
self.authenticate(mistral_url, username, api_key,
project_name, auth_url, project_id,
endpoint_type, service_type, auth_token,
user_id, cacert))
if not mistral_url:
mistral_url = "http://localhost:8989/v1"
self.http_client = httpclient.HTTPClient(mistral_url,
auth_token,
project_id,
user_id)
# Create all resource managers.
self.workbooks = workbooks.WorkbookManager(self)
self.executions = executions.ExecutionManager(self)
self.tasks = tasks.TaskManager(self)
self.listeners = listeners.ListenerManager(self)
def authenticate(self, mistral_url=None, username=None, api_key=None,
project_name=None, auth_url=None, project_id=None,
endpoint_type='publicURL', service_type='workflow',
auth_token=None, user_id=None, cacert=None):
if (not (project_name or project_id) or
not (isinstance(project_name, six.string_types) or
isinstance(project_id, six.string_types))):
raise RuntimeError('Either project name or project id should'
' be non-empty string')
if project_name and project_id:
raise RuntimeError('Only project name or '
'project id should be set')
if (not (username or user_id) or
not (isinstance(username, six.string_types) or
isinstance(user_id, six.string_types))):
raise RuntimeError('Either user name or user id should'
' be non-empty string')
if username and user_id:
raise RuntimeError('Only user name or user id'
' should be set')
keystone_client = _get_keystone_client(auth_url)
keystone = keystone_client.Client(
username=username,
user_id=user_id,
password=api_key,
token=auth_token,
tenant_id=project_id,
tenant_name=project_name,
auth_url=auth_url,
endpoint=auth_url,
cacert=cacert)
keystone.authenticate()
token = keystone.auth_token
user_id = keystone.user_id
project_id = keystone.project_id
if not mistral_url:
catalog = keystone.service_catalog.get_endpoints(
service_type=service_type,
endpoint_type=endpoint_type
)
if service_type in catalog:
service = catalog.get(service_type)
mistral_url = service[0].get('url') if service else None
return mistral_url, token, project_id, user_id
def _get_keystone_client(auth_url):
if "v2.0" in auth_url:
from keystoneclient.v2_0 import client
else:
from keystoneclient.v3 import client
return client

94
mistralclient/api/v1/executions.py
View File

@ -1,94 +0,0 @@
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
from mistralclient.api import base
from mistralclient import exceptions as ex
class Execution(base.Resource):
resource_name = 'Execution'
class ExecutionManager(base.ResourceManager):
resource_class = Execution
def _get_context_as_str(self, context):
msg = 'Context must be a dictionary or json compatible string:'
try:
if isinstance(context, dict):
context = json.dumps(context)
else:
json.loads(context)
except Exception as e:
raise ex.IllegalArgumentException(' '.join((msg, str(e))))
return context
def create(self, workbook_name, task, context=None):
self._ensure_not_empty(workbook_name=workbook_name, task=task)
data = {
'workbook_name': workbook_name,
'task': task
}
if context:
data['context'] = self._get_context_as_str(context)
return self._create('/workbooks/%s/executions' % workbook_name, data)
def update(self, workbook_name, id, state):
self._ensure_not_empty(id=id, state=state)
data = {
'state': state
}
if workbook_name:
uri = '/workbooks/%s/executions/%s' % (workbook_name, id)
else:
uri = '/executions/%s' % id
return self._update(uri, data)
def list(self, workbook_name):
if workbook_name:
uri = '/workbooks/%s/executions' % workbook_name
else:
uri = '/executions'
return self._list(uri, 'executions')
def get(self, workbook_name, id):
self._ensure_not_empty(id=id)
if workbook_name:
uri = '/workbooks/%s/executions/%s' % (workbook_name, id)
else:
uri = '/executions/%s' % id
return self._get(uri)
def delete(self, workbook_name, id):
self._ensure_not_empty(id=id)
if workbook_name:
uri = '/workbooks/%s/executions/%s' % (workbook_name, id)
else:
uri = '/executions/%s' % id
self._delete(uri)

69
mistralclient/api/v1/listeners.py
View File

@ -1,69 +0,0 @@
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from mistralclient.api import base
class Listener(base.Resource):
resource_name = 'Listener'
class ListenerManager(base.ResourceManager):
resource_class = Listener
def create(self, workbook_name, webhook, description=None, events=None):
# TODO(rakhmerov): need to describe what events is (data type)
self._ensure_not_empty(workbook_name=workbook_name,
webhook=webhook)
data = {
'workbook_name': workbook_name,
'description': description,
'webhook': webhook,
'events': events
}
return self._create('/workbooks/%s/listeners' % workbook_name, data)
def update(self, workbook_name, id, webhook=None, description=None,
events=None):
# TODO(everyone): need to describe what events is
self._ensure_not_empty(workbook_name=workbook_name, id=id)
data = {
'id': id,
'workbook_name': workbook_name,
'description': description,
'webhook': webhook,
'events': events
}
return self._update('/workbooks/%s/listeners/%s' %
(workbook_name, id), data)
def list(self, workbook_name):
self._ensure_not_empty(workbook_name=workbook_name)
return self._list('/workbooks/%s/listeners' % workbook_name,
'listeners')
def get(self, workbook_name, id):
self._ensure_not_empty(workbook_name=workbook_name, id=id)
return self._get('/workbooks/%s/listeners/%s' % (workbook_name, id))
def delete(self, workbook_name, id):
self._ensure_not_empty(workbook_name=workbook_name, id=id)
self._delete('/workbooks/%s/listeners/%s' % (workbook_name, id))

67
mistralclient/api/v1/tasks.py
View File

@ -1,67 +0,0 @@
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from mistralclient.api import base
class Task(base.Resource):
resource_name = 'Task'
class TaskManager(base.ResourceManager):
resource_class = Task
def update(self, workbook_name, execution_id, id, state):
self._ensure_not_empty(id=id, state=state)
data = {
'state': state
}
if execution_id:
if workbook_name:
uri = ('/workbooks/%s/executions/%s/tasks/%s' %
(workbook_name, execution_id, id))
else:
uri = '/executions/%s/tasks/%s' % (execution_id, id)
else:
uri = '/tasks/%s' % id
return self._update(uri, data)
def list(self, workbook_name, execution_id):
if execution_id:
if workbook_name:
uri = ('/workbooks/%s/executions/%s/tasks' %
(workbook_name, execution_id))
else:
uri = '/executions/%s/tasks' % execution_id
else:
uri = '/tasks'
return self._list(uri, 'tasks')
def get(self, workbook_name, execution_id, id):
self._ensure_not_empty(id=id)
if execution_id:
if workbook_name:
uri = ('/workbooks/%s/executions/%s/tasks/%s' %
(workbook_name, execution_id, id))
else:
uri = '/executions/%s/tasks/%s' % (execution_id, id)
else:
uri = '/tasks/%s' % id
return self._get(uri)

78
mistralclient/api/v1/workbooks.py
View File

@ -1,78 +0,0 @@
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from mistralclient.api import base
class Workbook(base.Resource):
resource_name = 'Workbook'
class WorkbookManager(base.ResourceManager):
resource_class = Workbook
def create(self, name, description=None, tags=None):
self._ensure_not_empty(name=name)
data = {
'name': name,
'description': description,
'tags': tags,
}
return self._create('/workbooks', data)
def update(self, name, description=None, tags=None):
self._ensure_not_empty(name=name)
data = {
'name': name,
'description': description,
'tags': tags,
}
return self._update('/workbooks/%s' % name, data)
def list(self):
return self._list('/workbooks', 'workbooks')
def get(self, name):
self._ensure_not_empty(name=name)
return self._get('/workbooks/%s' % name)
def delete(self, name):
self._ensure_not_empty(name=name)
self._delete('/workbooks/%s' % name)
def upload_definition(self, name, text):
self._ensure_not_empty(name=name)
resp = self.client.http_client.put('/workbooks/%s/definition' % name,
text,
headers={'content-type':
'text/plain'})
if resp.status_code != 200:
self._raise_api_exception(resp)
def get_definition(self, name):
self._ensure_not_empty(name=name)
resp = self.client.http_client.get('/workbooks/%s/definition'
% name)
if resp.status_code != 200:
self._raise_api_exception(resp)
return resp.content

0
mistralclient/commands/v1/__init__.py
View File

168
mistralclient/commands/v1/executions.py
View File

@ -1,168 +0,0 @@
# Copyright 2014 StackStorm, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import json
import logging
from cliff import command
from cliff import lister
from cliff import show
from mistralclient.api.v1 import executions as e
LOG = logging.getLogger(__name__)
def format(execution=None):
columns = (
'ID',
'Workbook',
'Task',
'State'
)
if execution:
data = (
execution.id,
execution.workbook_name,
execution.task,
execution.state
)
else:
data = (tuple('<none>' for _ in range(len(columns))),)
return (columns, data)
class List(lister.Lister):
"List all executions"
def get_parser(self, prog_name):
parser = super(List, self).get_parser(prog_name)
parser.add_argument(
'workbook',
help='Execution workbook')
return parser
def take_action(self, parsed_args):
data = [format(execution)[1] for execution
in e.ExecutionManager(self.app.client)
.list(parsed_args.workbook)]
if data:
return (format()[0], data)
else:
return format()
class Get(show.ShowOne):
"Show specific execution"
def get_parser(self, prog_name):
parser = super(Get, self).get_parser(prog_name)
parser.add_argument(
'workbook',
help='Execution workbook')
parser.add_argument(
'id',
help='Execution identifier')
return parser
def take_action(self, parsed_args):
execution = e.ExecutionManager(self.app.client).get(
parsed_args.workbook,
parsed_args.id)
return format(execution)
class Create(show.ShowOne):
"Create new execution"
def get_parser(self, prog_name):
parser = super(Create, self).get_parser(prog_name)
parser.add_argument(
'workbook',
help='Execution workbook')
parser.add_argument(
'task',
help='Execution task')
parser.add_argument(
'context',
help='Execution context')
return parser
def take_action(self, parsed_args):
try:
ctx = json.loads(parsed_args.context)
except:
ctx = open(parsed_args.context).read()
execution = e.ExecutionManager(self.app.client).create(
parsed_args.workbook,
parsed_args.task,
ctx)
return format(execution)
class Delete(command.Command):
"Delete execution"
def get_parser(self, prog_name):
parser = super(Delete, self).get_parser(prog_name)
parser.add_argument(
'workbook',
help='Execution workbook')
parser.add_argument(
'id',
help='Execution identifier')
return parser
def take_action(self, parsed_args):
e.ExecutionManager(self.app.client).delete(
parsed_args.workbook, parsed_args.id)
class Update(show.ShowOne):
"Update execution"
def get_parser(self, prog_name):
parser = super(Update, self).get_parser(prog_name)
parser.add_argument(
'workbook',
help='Execution workbook')
parser.add_argument(
'id',
help='Execution identifier')
parser.add_argument(
'state',
choices=['RUNNING', 'SUSPENDED', 'STOPPED', 'SUCCESS', 'ERROR'],
help='Execution state')
return parser
def take_action(self, parsed_args):
execution = e.ExecutionManager(self.app.client).update(
parsed_args.workbook,
parsed_args.id,
parsed_args.state)
return format(execution)

133
mistralclient/commands/v1/tasks.py
View File

@ -1,133 +0,0 @@
# Copyright 2014 StackStorm, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import logging
from cliff import lister
from cliff import show
from mistralclient.api.v1 import tasks as t
LOG = logging.getLogger(__name__)
def format(task=None):
columns = (
'ID',
'Workbook',
'Execution',
'Name',
'Description',
'State',
'Tags'
)
if task:
data = (
task.id,
task.workbook_name,
task.execution_id,
task.name,
task.description or '<none>',
task.state,
task.tags and ', '.join(task.tags) or '<none>'
)
else:
data = (tuple('<none>' for _ in range(len(columns))),)
return (columns, data)
class List(lister.Lister):
"List all tasks"
def get_parser(self, prog_name):
parser = super(List, self).get_parser(prog_name)
parser.add_argument(
'workbook',
help='Task workbook')
parser.add_argument(
'execution',
help='Task execution')
return parser
def take_action(self, parsed_args):
data = [format(task)[1] for task
in t.TaskManager(self.app.client)
.list(parsed_args.workbook,
parsed_args.execution)]
if data:
return (format()[0], data)
else:
return format()
class Get(show.ShowOne):
"Show specific task"
def get_parser(self, prog_name):
parser = super(Get, self).get_parser(prog_name)
parser.add_argument(
'workbook',
help='Task workbook')
parser.add_argument(
'execution',
help='Task execution')
parser.add_argument(
'id',
help='Task identifier')
return parser
def take_action(self, parsed_args):
execution = t.TaskManager(self.app.client).get(
parsed_args.workbook,
parsed_args.execution,
parsed_args.id)
return format(execution)
class Update(show.ShowOne):
"Update task"
def get_parser(self, prog_name):
parser = super(Update, self).get_parser(prog_name)
parser.add_argument(
'workbook',
help='Task workbook')
parser.add_argument(
'execution',
help='Task execution')
parser.add_argument(
'id',
help='Task identifier')
parser.add_argument(
'state',
choices=['IDLE', 'RUNNING', 'SUCCESS', 'ERROR'],
help='Task state')
return parser
def take_action(self, parsed_args):
execution = t.TaskManager(self.app.client).update(
parsed_args.workbook,
parsed_args.execution,
parsed_args.id,
parsed_args.state)
return format(execution)

195
mistralclient/commands/v1/workbooks.py
View File

@ -1,195 +0,0 @@
# Copyright 2014 StackStorm, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import argparse
import logging
from cliff import command
from cliff import lister
from cliff import show
from mistralclient.api.v1 import workbooks as w
LOG = logging.getLogger(__name__)
def format(workbook=None):
columns = (
'Name',
'Description',
'Tags'
)
if workbook:
data = (
workbook.name,
workbook.description or '<none>',
', '.join(workbook.tags) or '<none>'
)
else:
data = (tuple('<none>' for _ in range(len(columns))),)
return (columns, data)
class List(lister.Lister):
"List all workbooks"
def take_action(self, parsed_args):
data = [format(workbook)[1] for workbook
in w.WorkbookManager(self.app.client).list()]
if data:
return (format()[0], data)
else:
return format()
class Get(show.ShowOne):
"Show specific workbook"
def get_parser(self, prog_name):
parser = super(Get, self).get_parser(prog_name)
parser.add_argument(
'name',
help='Workbook name')
return parser
def take_action(self, parsed_args):
workbook = w.WorkbookManager(self.app.client).get(parsed_args.name)
return format(workbook)
class Create(show.ShowOne):
"Create new workbook"
def get_parser(self, prog_name):
parser = super(Create, self).get_parser(prog_name)
parser.add_argument(
'name',
help='Workbook name')
parser.add_argument(
'description',
nargs='?',
help='Workbook description')
parser.add_argument(
'tags',
nargs='?',
help='Workbook tags separated by ","')
parser.add_argument(
'definition',
nargs='?',
type=argparse.FileType('r'),
help='Workbook definition file'
)
return parser
def take_action(self, parsed_args):
workbook = w.WorkbookManager(self.app.client).create(
parsed_args.name,
parsed_args.description,
str(parsed_args.tags).split(','))
if parsed_args.definition:
w.WorkbookManager(self.app.client).upload_definition(
parsed_args.name,
parsed_args.definition.read())
return format(workbook)
class Delete(command.Command):
"Delete workbook"
def get_parser(self, prog_name):
parser = super(Delete, self).get_parser(prog_name)
parser.add_argument(
'name',
help='Workbook name')
return parser
def take_action(self, parsed_args):
w.WorkbookManager(self.app.client).delete(parsed_args.name)
class Update(show.ShowOne):
"Update workbook"
def get_parser(self, prog_name):
parser = super(Update, self).get_parser(prog_name)
parser.add_argument(
'name',
help='Workbook name')
parser.add_argument(
'description',
nargs='?',
help='Workbook description')
parser.add_argument(
'tags',
nargs='*',
help='Workbook tags')
return parser
def take_action(self, parsed_args):
workbook = w.WorkbookManager(self.app.client).update(
parsed_args.name,
parsed_args.description,
parsed_args.tags)
return format(workbook)
class UploadDefinition(command.Command):
"Upload workbook definition"
def get_parser(self, prog_name):
parser = super(UploadDefinition, self).get_parser(prog_name)
parser.add_argument(
'name',
help='Workbook name')
parser.add_argument(
'path',
type=argparse.FileType('r'),
help='Workbook definition file')
return parser
def take_action(self, parsed_args):
w.WorkbookManager(self.app.client).upload_definition(
parsed_args.name,
parsed_args.path.read())
class GetDefinition(command.Command):
"Show workbook definition"
def get_parser(self, prog_name):
parser = super(GetDefinition, self).get_parser(prog_name)
parser.add_argument(
'name',
help='Workbook name')
return parser
def take_action(self, parsed_args):
definition = w.WorkbookManager(
self.app.client).get_definition(parsed_args.name)
self.app.stdout.write(definition)

35
mistralclient/shell.py
View File

@ -20,9 +20,6 @@ import logging
import sys
from mistralclient.api import client
import mistralclient.commands.v1.executions
import mistralclient.commands.v1.tasks
import mistralclient.commands.v1.workbooks
import mistralclient.commands.v2.action_executions
import mistralclient.commands.v2.actions
import mistralclient.commands.v2.cron_triggers
@ -181,8 +178,10 @@ class MistralShell(app.App):
def initialize_app(self, argv):
self._clear_shell_commands()
self._set_shell_commands(self._get_commands(
client.determine_client_version(self.options.mistral_url)))
ver = client.determine_client_version(self.options.mistral_url)
self._set_shell_commands(self._get_commands(ver))
self.client = client.client(mistral_url=self.options.mistral_url,
username=self.options.username,
@ -208,32 +207,10 @@ class MistralShell(app.App):
self.command_manager.commands.pop(k)
def _get_commands(self, version):
if version == 1:
return self._get_commands_v1()
else:
if version == 2:
return self._get_commands_v2()
@staticmethod
def _get_commands_v1():
return {
'workbook-list': mistralclient.commands.v1.workbooks.List,
'workbook-get': mistralclient.commands.v1.workbooks.Get,
'workbook-create': mistralclient.commands.v1.workbooks.Create,
'workbook-delete': mistralclient.commands.v1.workbooks.Delete,
'workbook-update': mistralclient.commands.v1.workbooks.Update,
'workbook-upload-definition':
mistralclient.commands.v1.workbooks.UploadDefinition,
'workbook-get-definition':
mistralclient.commands.v1.workbooks.GetDefinition,
'execution-list': mistralclient.commands.v1.executions.List,
'execution-get': mistralclient.commands.v1.executions.Get,
'execution-create': mistralclient.commands.v1.executions.Create,
'execution-delete': mistralclient.commands.v1.executions.Delete,
'execution-update': mistralclient.commands.v1.executions.Update,
'task-list': mistralclient.commands.v1.tasks.List,
'task-get': mistralclient.commands.v1.tasks.Get,
'task-update': mistralclient.commands.v1.tasks.Update,
}
return {}
@staticmethod
def _get_commands_v2():

0
mistralclient/tests/functional/cli/v1/__init__.py
View File

348
mistralclient/tests/functional/cli/v1/cli_tests.py
View File

@ -1,348 +0,0 @@
# Copyright (c) 2014 Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
from tempest_lib import exceptions
from mistralclient.tests.functional.cli import base
MISTRAL_URL = "http://localhost:8989/v1"
class SimpleMistralCLITests(base.MistralCLIAuth):
"""Basic tests, check '-list', '-help' commands."""
_mistral_url = MISTRAL_URL
@classmethod
def setUpClass(cls):
super(SimpleMistralCLITests, cls).setUpClass()
def test_workbooks_list(self):
workbooks = self.parser.listing(self.mistral('workbook-list'))
self.assertTableStruct(workbooks,
['Name', 'Description', 'Tags'])
def test_executions_list(self):
executions = self.parser.listing(
self.mistral('execution-list', params='""'))
self.assertTableStruct(executions,
['ID', 'Workbook', 'Task', 'State'])
def test_tasks_list(self):
tasks = self.parser.listing(
self.mistral('task-list', params='"", ""'))
self.assertTableStruct(tasks,
['ID', 'Workbook', 'Execution', 'Name',
'Description', 'State', 'Tags'])
class ClientTestBase(base.MistralCLIAuth):
_mistral_url = MISTRAL_URL
@classmethod
def setUpClass(cls):
super(ClientTestBase, cls).setUpClass()
cls.wb_def = os.path.relpath(
'functionaltests/resources/v1/wb_v1.yaml', os.getcwd())
def tearDown(self):
super(ClientTestBase, self).tearDown()
for wb in self.parser.listing(self.mistral('workbook-list')):
if wb['Name'] != "<none>":
execs = self.parser.listing(self.mistral(
'execution-list', params='{0}'.format(wb['Name'])))
ids = [ex['ID'] for ex in execs]
for id in ids:
if id != "<none>":
self.parser.listing(self.mistral(
'execution-delete',
params='"{0}" "{1}"'.format(wb['Name'], id)))
self.parser.listing(self.mistral(
'workbook-delete', params=wb['Name']))
def get_value_of_field(self, obj, field):
return [o['Value'] for o in obj
if o['Field'] == "{0}".format(field)][0]
class WorkbookCLITests(ClientTestBase):
"""Test suite checks commands to work with workbooks."""
@classmethod
def setUpClass(cls):
super(WorkbookCLITests, cls).setUpClass()
def test_workbook_create_delete(self):
wb = self.parser.listing(self.mistral('workbook-create', params='wb'))
self.assertTableStruct(wb, ['Field', 'Value'])
name = self.get_value_of_field(wb, "Name")
self.assertEqual('wb', name)
wbs = self.parser.listing(self.mistral('workbook-list'))
self.assertIn('wb', [workbook['Name'] for workbook in wbs])
self.parser.listing(self.mistral('workbook-delete', params='wb'))
wbs = self.parser.listing(self.mistral('workbook-list'))
self.assertNotIn('wb', [workbook['Name'] for workbook in wbs])
def test_workbook_update(self):
self.parser.listing(self.mistral('workbook-create', params='wb'))
wb = self.parser.listing(self.mistral(
'workbook-update', params='"wb" "Test Description" "tag"'))
self.assertTableStruct(wb, ['Field', 'Value'])
name = self.get_value_of_field(wb, "Name")
description = self.get_value_of_field(wb, "Description")
tags = self.get_value_of_field(wb, "Tags")
self.assertEqual('wb', name)
self.assertIn('Test Description', description)
self.assertIn('tag', tags)
def test_workbook_upload_get_definition(self):
self.parser.listing(self.mistral('workbook-create', params='wb'))
self.parser.listing(self.mistral(
'workbook-upload-definition',
params='"wb" "{0}"'.format(self.wb_def)))
definition = self.mistral('workbook-get-definition', params='wb')
self.assertNotIn('404 Not Found', definition)
class ExecutionCLITests(ClientTestBase):
"""Test suite checks commands to work with executions."""
def setUp(self):
super(ExecutionCLITests, self).setUp()
self.mistral('workbook-create', params='wb')
self.mistral('workbook-upload-definition',
params='"wb" "{0}"'.format(self.wb_def))
def tearDown(self):
super(ExecutionCLITests, self).tearDown()
def test_execution_create_delete(self):
execution = self.parser.listing(self.mistral(
'execution-create', params='"wb" "hello" "{}"'))
self.assertTableStruct(execution, ['Field', 'Value'])
exec_id = self.get_value_of_field(execution, 'ID')
wb = self.get_value_of_field(execution, 'Workbook')
task = self.get_value_of_field(execution, 'Task')
status = self.get_value_of_field(execution, 'State')
self.assertEqual('wb', wb)
self.assertEqual('hello', task)
self.assertEqual('RUNNING', status)
execs = self.parser.listing(
self.mistral('execution-list', params='wb'))
self.assertIn(exec_id, [ex['ID'] for ex in execs])
self.assertIn(wb, [ex['Workbook'] for ex in execs])
self.assertIn(task, [ex['Task'] for ex in execs])
self.assertIn('SUCCESS', [ex['State'] for ex in execs])
self.parser.listing(self.mistral(
'execution-delete', params='"wb" "{0}"'.format(exec_id)))
execs = self.parser.listing(
self.mistral('execution-list', params='wb'))
self.assertNotIn(exec_id, [ex['ID'] for ex in execs])
def test_update_execution(self):
execution = self.parser.listing(self.mistral(
'execution-create', params='"wb" "hello" "{}"'))
self.assertTableStruct(execution, ['Field', 'Value'])
exec_id = self.get_value_of_field(execution, 'ID')
status = self.get_value_of_field(execution, 'State')
self.assertEqual('RUNNING', status)
execution = self.parser.listing(self.mistral(
'execution-update', params='"wb" "{0}" "STOPPED"'.format(exec_id)))
self.assertTableStruct(execution, ['Field', 'Value'])
updated_exec_id = self.get_value_of_field(execution, 'ID')
status = self.get_value_of_field(execution, 'State')
self.assertEqual(exec_id, updated_exec_id)
self.assertEqual('STOPPED', status)
def test_get_execution(self):
execution = self.parser.listing(self.mistral(
'execution-create', params='"wb" "hello" "{}"'))
exec_id = self.get_value_of_field(execution, 'ID')
execution = self.parser.listing(self.mistral(
'execution-get', params='"wb" "{0}"'.format(exec_id)))
gotten_id = self.get_value_of_field(execution, 'ID')
wb = self.get_value_of_field(execution, 'Workbook')
task = self.get_value_of_field(execution, 'Task')
self.assertEqual(exec_id, gotten_id)
self.assertEqual('wb', wb)
self.assertEqual('hello', task)
class TaskCLITests(ClientTestBase):
"""Test suite checks commands to work with tasks."""
def test_get_task(self):
self.mistral('workbook-create', params='wb')
self.mistral('workbook-upload-definition',
params='"wb" "{0}"'.format(self.wb_def))
execution = self.parser.listing(self.mistral(
'execution-create', params='"wb" "hello" "{}"'))
exec_id = self.get_value_of_field(execution, 'ID')
tasks = self.parser.listing(
self.mistral('task-list', params='"wb" "{0}"'.format(exec_id)))
task_id = [task['ID'] for task in tasks][0]
task = self.parser.listing(self.mistral(
'task-get', params='"wb" "hello" "{0}"'.format(task_id)))
gotten_id = self.get_value_of_field(task, 'ID')
wb = self.get_value_of_field(task, 'Workbook')
self.assertEqual(task_id, gotten_id)
self.assertEqual('wb', wb)
class NegativeCLITests(ClientTestBase):
"""This class contains negative tests."""
def test_wb_list_extra_param(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'workbook-list', params='param')
def test_wb_get_unexist_wb(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'workbook-get', params='wb')
def test_wb_get_without_param(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'workbook-get')
def test_wb_create_same_name(self):
self.mistral('workbook-create', params='wb')
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'workbook-create', params='wb')
def test_wb_create_with_wrong_path_to_definition(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral,
'workbook-create', params='wb pam pam pam')
def test_wb_delete_unexist_wb(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'workbook-delete', params='wb')
def test_wb_update_unexist_wb(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'workbook-update', params='wb pam pam')
def test_wb_upload_definition_unexist_wb(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral,
'workbook-upload-definition', params='wb')
def test_wb_upload_definition_using_wrong_path(self):
self.mistral('workbook-create', params='wb')
self.assertRaises(exceptions.CommandFailed,
self.mistral,
'workbook-upload-definition', params='wb param')
def test_wb_get_definition_wb_without_definition(self):
self.mistral('workbook-create', params='wb')
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'workbook-get-definition', params='wb')
def test_ex_list_unexist_wb(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'execution-list', params='wb')
def test_ex_create_unexist_wb(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'execution-create', params='wb')
def test_ex_create_unexist_task(self):
self.mistral('workbook-create', params='wb')
self.assertRaises(exceptions.CommandFailed,
self.mistral,
'execution-create', params='wb param {}')
def test_ex_create_without_context(self):
self.mistral('workbook-create', params='wb')
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'execution-create', params='wb param')
def test_ex_create_wrong_context_format(self):
self.mistral('workbook-create', params='wb')
self.assertRaises(exceptions.CommandFailed,
self.mistral,
'execution-create', params='wb param pam')
def test_ex_get_from_nonexist_wb(self):
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'execution-get', params='wb id')
def test_ex_get_nonexist_execution(self):
self.mistral('workbook-create', params='wb')
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'execution-get', params='wb id')
def test_ex_update_set_wrong_state(self):
self.mistral('workbook-create', params='wb')
self.mistral('workbook-upload-definition',
params='"wb" "{0}"'.format(self.wb_def))
execution = self.parser.listing(self.mistral(
'execution-create', params='"wb" "hello" "{}"'))
exec_id = self.get_value_of_field(execution, 'ID')
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'execution-update',
params='"wb" "{0}" "OK"'.format(exec_id))
def test_task_get_nonexisting_task(self):
self.mistral('workbook-create', params='wb')
self.mistral('workbook-upload-definition',
params='"wb" "{0}"'.format(self.wb_def))
self.mistral('execution-create', params='"wb" "hello" "{}"')
self.assertRaises(exceptions.CommandFailed,
self.mistral, 'task-get',
params='"wb" "hello" "id"')

0
mistralclient/tests/functional/client/__init__.py
View File

0
mistralclient/tests/functional/client/v1/__init__.py
View File

104
mistralclient/tests/functional/client/v1/base.py
View File

@ -1,104 +0,0 @@
# Copyright (c) 2014 Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
from tempest import clients
from tempest import config
from tempest_lib.common import rest_client
import testtools
from mistralclient.api import base
from mistralclient.api.v1 import client as mclient
CONF = config.CONF
class ClientAuth(rest_client.RestClient):
def __init__(self, auth_provider, url):
super(ClientAuth, self).__init__(
auth_provider=auth_provider,
service='workflow',
region=CONF.identity.region)
self.mistral_client = mclient.Client(
auth_token=self.auth_provider.get_token(),
mistral_url=url)
class MistralBase(testtools.TestCase):
_mistral_url = None
@classmethod
def setUpClass(cls):
super(MistralBase, cls).setUpClass()
if 'WITHOUT_AUTH' in os.environ:
cls.mistral_client = mclient.Client(
mistral_url=cls._mistral_url)
else:
mgr = clients.Manager()
cls.mistral_client = ClientAuth(
mgr.auth_provider, cls._mistral_url).mistral_client
cls.definition = open(os.path.relpath(
'functionaltests/resources/v1/wb_v1.yaml',
os.getcwd()), 'rb').read()
cls.wb = cls.mistral_client.workbooks.create(
"wb", "Description", ["tags"])
@classmethod
def tearDownClass(cls):
super(MistralBase, cls).tearDownClass()
for wb in cls.mistral_client.workbooks.list():
cls.mistral_client.workbooks.delete(wb.name)
def tearDown(self):
super(MistralBase, self).tearDown()
for ex in self.mistral_client.executions.list(None):
# TODO(akuznetsova): remove try/except after
# https://bugs.launchpad.net/mistral/+bug/1353306
# will be fixed
""" try/except construction was added because of problem
with concurrent transactions in sqlite and appropriate
error: APIException: (OperationalError) database is locked.
This isn't a tests problem, so they are considered passed.
"""
try:
self.mistral_client.executions.delete(None, ex.id)
except base.APIException:
pass
def create_execution(self):
self.mistral_client.workbooks.upload_definition("wb", self.definition)
execution = self.mistral_client.executions.create("wb", "hello")
return execution
def assert_item_in_list(self, items, **props):
def _matches(item, **props):
for prop_name, prop_val in props.iteritems():
v = item[prop_name] if isinstance(item, dict) else getattr(
item, prop_name)
if v != prop_val:
return False
return True
filtered_items = filter(lambda item: _matches(item, **props), items)
return len(filtered_items) != 0

121
mistralclient/tests/functional/client/v1/tests.py
View File

@ -1,121 +0,0 @@
# Copyright (c) 2014 Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from base import MistralBase
class MistralURLDefine(MistralBase):
_mistral_url = "http://localhost:8989/v1"
class Workbooks(MistralURLDefine):
def test_get_workbook_list(self):
wbs = self.mistral_client.workbooks.list()
self.assertIsInstance(wbs, list)
def test_create_workbook(self):
wbs = self.mistral_client.workbooks.list()
self.mistral_client.workbooks.create("new_wb")
wbs_with_new_wb = self.mistral_client.workbooks.list()
self.assertEqual(len(wbs) + 1, len(wbs_with_new_wb))
self.assertTrue(self.assert_item_in_list(
wbs_with_new_wb, name="new_wb"))
def test_get_workbook(self):
received_wb = self.mistral_client.workbooks.get("wb")
self.assertEqual(self.wb.name, received_wb.name)
def test_update_workbook(self):
updated_wb = self.mistral_client.workbooks.update(
"wb", "New Description", ["tags"])
self.assertEqual(self.wb.name, updated_wb.name)
self.assertEqual("New Description", updated_wb.description)
self.assertEqual(["tags"], updated_wb.tags)
def test_upload_get_definition(self):
self.mistral_client.workbooks.upload_definition("wb", self.definition)
received_definition = (self.mistral_client.
workbooks.get_definition("wb"))
self.assertEqual(self.definition, received_definition)
class Executions(MistralURLDefine):
def test_create_execution(self):
execution = self.create_execution()
self.assertEqual("wb", execution.workbook_name)
self.assertNotEqual(execution.id, None)
def test_update_execution(self):
execution = self.create_execution()
updated_exec = self.mistral_client.executions.update(
"wb", execution.id, "ERROR")
self.assertEqual("ERROR", updated_exec.state)
updated_exec = self.mistral_client.executions.update(
None, execution.id, "SUCCESS")
self.assertEqual("SUCCESS", updated_exec.state)
def test_list_executions(self):
execution = self.create_execution()
exec_list = self.mistral_client.executions.list(None)
self.assertTrue(self.assert_item_in_list(
exec_list, id=execution.id))
exec_list = self.mistral_client.executions.list("wb")
self.assertTrue(self.assert_item_in_list(
exec_list, id=execution.id, workbook_name="wb"))
def test_get_execution(self):
execution = self.create_execution()
received_exec = self.mistral_client.executions.get(None, execution.id)
self.assertEqual(execution.id, received_exec.id)
received_exec = self.mistral_client.executions.get("wb", execution.id)
self.assertEqual(execution.id, received_exec.id)
class Tasks(MistralURLDefine):
def test_list_tasks(self):
execution = self.create_execution()
tasks_list = self.mistral_client.tasks.list(None, None)
self.assertIsInstance(tasks_list, list)
tasks_list = self.mistral_client.tasks.list(
execution.workbook_name, None)
self.assertIsInstance(tasks_list, list)
tasks_list = self.mistral_client.tasks.list(None, execution.id)
self.assertIsInstance(tasks_list, list)
tasks_list = self.mistral_client.tasks.list(
execution.workbook_name, execution.id)
self.assertIsInstance(tasks_list, list)
def test_get_task(self):
execution = self.create_execution()
task = self.mistral_client.tasks.list(None, None)[-1]
received_task = self.mistral_client.tasks.get(None, None, task.id)
self.assertIsNotNone(received_task)
task = self.mistral_client.tasks.list("wb", None)[-1]
received_task = self.mistral_client.tasks.get("wb", None, task.id)
self.assertIsNotNone(received_task)
task = self.mistral_client.tasks.list(None, execution.id)[-1]
received_task = self.mistral_client.tasks.get(
None, execution.id, task.id)
self.assertIsNotNone(received_task)
task = self.mistral_client.tasks.list("wb", execution.id)[-1]
received_task = self.mistral_client.tasks.get(
"wb", execution.id, task.id)
self.assertIsNotNone(received_task)

10
mistralclient/tests/unit/resources/ctx.json
View File

@ -1,5 +1,7 @@
{"context": {
"server": {
"name": "name"
{
"context": {
"server": {
"name": "name"
}
}
}}
}

0
mistralclient/tests/unit/v1/__init__.py
View File

42
mistralclient/tests/unit/v1/base.py
View File

@ -1,42 +0,0 @@
# Copyright 2014 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from mistralclient.api.v1 import client
from mistralclient.tests.unit import base
class BaseClientV1Test(base.BaseClientTest):
def setUp(self):
super(BaseClientV1Test, self).setUp()
self._client = client.Client(project_name="test",
mistral_url="test")
self.workbooks = self._client.workbooks
self.executions = self._client.executions
self.tasks = self._client.tasks
self.listeners = self._client.listeners

81
mistralclient/tests/unit/v1/test_cli_executions.py
View File

@ -1,81 +0,0 @@
# Copyright 2014 StackStorm, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import mock
import pkg_resources as pkg
from mistralclient.api.v1 import executions as e
from mistralclient.commands.v1 import executions
from mistralclient.tests.unit import base
EXECUTION = e.Execution(mock, {
'id': '123',
'workbook_name': 'some',
'task': 'else',
'state': 'RUNNING'
})
class TestCLIExecutions(base.BaseCommandTest):
@mock.patch('mistralclient.api.v1.executions.ExecutionManager.create')
def test_create_ctx_string(self, mock):
mock.return_value = EXECUTION
result = self.call(executions.Create,
app_args=['name', 'id', '{ "context": true }'])
self.assertEqual(('123', 'some', 'else', 'RUNNING'), result[1])
@mock.patch('mistralclient.api.v1.executions.ExecutionManager.create')
def test_create_ctx_file(self, mock):
mock.return_value = EXECUTION
path = pkg.resource_filename('mistralclient',
'tests/unit/resources/ctx.json')
result = self.call(executions.Create,
app_args=['name', 'id', path])
self.assertEqual(('123', 'some', 'else', 'RUNNING'), result[1])
@mock.patch('mistralclient.api.v1.executions.ExecutionManager.update')
def test_update(self, mock):
mock.return_value = EXECUTION
result = self.call(executions.Update,
app_args=['name', 'id', 'SUCCESS'])
self.assertEqual(('123', 'some', 'else', 'RUNNING'), result[1])
@mock.patch('mistralclient.api.v1.executions.ExecutionManager.list')
def test_list(self, mock):
mock.return_value = (EXECUTION,)
result = self.call(executions.List, app_args=['name'])
self.assertEqual([('123', 'some', 'else', 'RUNNING')], result[1])
@mock.patch('mistralclient.api.v1.executions.ExecutionManager.get')
def test_get(self, mock):
mock.return_value = EXECUTION
result = self.call(executions.Get, app_args=['name', 'id'])
self.assertEqual(('123', 'some', 'else', 'RUNNING'), result[1])
@mock.patch('mistralclient.api.v1.executions.ExecutionManager.delete')
def test_delete(self, mock):
result = self.call(executions.Delete, app_args=['name', 'id'])
self.assertIsNone(result)

64
mistralclient/tests/unit/v1/test_cli_tasks.py
View File

@ -1,64 +0,0 @@
# Copyright 2014 StackStorm, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import mock
from mistralclient.api.v1 import tasks as t
from mistralclient.commands.v1 import tasks
from mistralclient.tests.unit import base
TASK = t.Task(mock, {
'id': '123',
'workbook_name': 'some',
'execution_id': 'thing',
'name': 'else',
'description': 'keeps',
'state': 'RUNNING',
'tags': ['a', 'b'],
})
class TestCLIExecutions(base.BaseCommandTest):
@mock.patch('mistralclient.api.v1.tasks.TaskManager.update')
def test_update(self, mock):
mock.return_value = TASK
result = self.call(tasks.Update,
app_args=['workbook', 'executor', 'id', 'ERROR'])
self.assertEqual(
('123', 'some', 'thing', 'else', 'keeps', 'RUNNING', 'a, b'),
result[1])
@mock.patch('mistralclient.api.v1.tasks.TaskManager.list')
def test_list(self, mock):
mock.return_value = (TASK,)
result = self.call(tasks.List, app_args=['workbook', 'executor'])
self.assertEqual(
[('123', 'some', 'thing', 'else', 'keeps', 'RUNNING', 'a, b')],
result[1])
@mock.patch('mistralclient.api.v1.tasks.TaskManager.get')
def test_get(self, mock):
mock.return_value = TASK
result = self.call(tasks.Get, app_args=['workbook', 'executor', 'id'])
self.assertEqual(
('123', 'some', 'thing', 'else', 'keeps', 'RUNNING', 'a, b'),
result[1])

87
mistralclient/tests/unit/v1/test_cli_workbooks.py
View File

@ -1,87 +0,0 @@
# Copyright 2014 StackStorm, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import mock
from mistralclient.api.v1 import workbooks as w
from mistralclient.commands.v1 import workbooks
from mistralclient.tests.unit import base
WORKBOOK = w.Workbook(mock, {
'name': 'a',
'description': 'some',
'tags': ['a', 'b']
})
class TestCLIWorkbooks(base.BaseCommandTest):
@mock.patch('mistralclient.api.v1.workbooks.WorkbookManager.create')
def test_create(self, mock):
mock.return_value = WORKBOOK
result = self.call(workbooks.Create, app_args=['name'])
self.assertEqual(('a', 'some', 'a, b'), result[1])
@mock.patch('mistralclient.api.v1.workbooks.WorkbookManager.update')
def test_update(self, mock):
mock.return_value = WORKBOOK
result = self.call(workbooks.Update, app_args=['name'])
self.assertEqual(('a', 'some', 'a, b'), result[1])
@mock.patch('mistralclient.api.v1.workbooks.WorkbookManager.list')
def test_list(self, mock):
mock.return_value = (WORKBOOK,)
result = self.call(workbooks.List)
self.assertEqual([('a', 'some', 'a, b')], result[1])
@mock.patch('mistralclient.api.v1.workbooks.WorkbookManager.get')
def test_get(self, mock):
mock.return_value = WORKBOOK
result = self.call(workbooks.Get, app_args=['name'])
self.assertEqual(('a', 'some', 'a, b'), result[1])
@mock.patch('mistralclient.api.v1.workbooks.WorkbookManager.delete')
def test_delete(self, mock):
self.assertIsNone(self.call(workbooks.Delete, app_args=['name']))
@mock.patch('argparse.open', create=True)
@mock.patch(
'mistralclient.api.v1.workbooks.WorkbookManager.upload_definition'
)
def test_upload_definition(self, mock, mock_open):
mock.return_value = WORKBOOK
mock_open.return_value = mock.MagicMock(spec=file)
result = self.call(workbooks.UploadDefinition,
app_args=['name', '1.txt'])
self.assertIsNone(result)
@mock.patch('mistralclient.api.v1.workbooks.'
'WorkbookManager.get_definition')
def test_get_definition(self, mock):
mock.return_value = 'sometext'
self.call(workbooks.GetDefinition, app_args=['name'])
self.app.stdout.write.assert_called_with('sometext')

143
mistralclient/tests/unit/v1/test_executions.py
View File

@ -1,143 +0,0 @@
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import unittest2
from mistralclient.api.v1 import executions as e
from mistralclient.tests.unit.v1 import base
# TODO(everyone): Later we need additional tests verifying all the errors etc.
EXECS = [
{
'id': "123",
'workbook_name': "my_workbook",
'task': 'my_task',
'state': 'RUNNING',
'context': """
{
"person": {
"first_name": "John",
"last_name": "Doe"
}
}
"""
}
]
URL_TEMPLATE = '/workbooks/%s/executions'
URL_TEMPLATE_ID = '/workbooks/%s/executions/%s'
class TestExecutions(base.BaseClientV1Test):
def test_create(self):
mock = self.mock_http_post(content=EXECS[0])
body = {
'workbook_name': EXECS[0]['workbook_name'],
'task': EXECS[0]['task'],
'context': EXECS[0]['context']
}
ex = self.executions.create(EXECS[0]['workbook_name'],
EXECS[0]['task'],
EXECS[0]['context'])
self.assertIsNotNone(ex)
self.assertDictEqual(e.Execution(self.executions, EXECS[0]).__dict__,
ex.__dict__)
arg_body = mock.call_args[0][1]
url = mock.call_args[0][0]
self.assertEqual(url, URL_TEMPLATE % EXECS[0]['workbook_name'])
self.assertDictEqual(json.loads(arg_body), body)
def test_create_with_empty_context(self):
mock = self.mock_http_post(content=EXECS[0])
body = {
'workbook_name': EXECS[0]['workbook_name'],
'task': EXECS[0]['task']
}
self.executions.create(EXECS[0]['workbook_name'],
EXECS[0]['task'])
arg_body = mock.call_args[0][1]
url = mock.call_args[0][0]
self.assertEqual(url, URL_TEMPLATE % EXECS[0]['workbook_name'])
self.assertDictEqual(json.loads(arg_body), body)
@unittest2.expectedFailure
def test_create_failure1(self):
self.mock_http_post(content=EXECS[0])
self.executions.create(EXECS[0]['workbook_name'],
EXECS[0]['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]['task'],
list('343', 'sdfsd'))
def test_update(self):
mock = self.mock_http_put(content=EXECS[0])
body = {
'state': EXECS[0]['state']
}
ex = self.executions.update(EXECS[0]['workbook_name'],
EXECS[0]['id'],
EXECS[0]['state'])
self.assertIsNotNone(ex)
self.assertEqual(e.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):
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(e.Execution(self.executions, EXECS[0]).__dict__,
ex.__dict__)
mock.assert_called_once_with(URL_TEMPLATE % EXECS[0]['workbook_name'])
def test_get(self):
mock = self.mock_http_get(content=EXECS[0])
ex = self.executions.get(EXECS[0]['workbook_name'], EXECS[0]['id'])
self.assertEqual(e.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):
mock = self.mock_http_delete(status_code=204)
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']))

112
mistralclient/tests/unit/v1/test_listeners.py
View File

@ -1,112 +0,0 @@
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
from mistralclient.api.v1 import listeners as l
from mistralclient.tests.unit.v1 import base
# TODO(everyone): later we need additional tests verifying all the errors etc.
LISTENERS = [
{
'id': "1",
'workbook_name': "my_workbook",
'description': "My cool Mistral workbook",
'webhook': "http://my.website.org"
}
]
URL_TEMPLATE = '/workbooks/%s/listeners'
URL_TEMPLATE_ID = '/workbooks/%s/listeners/%s'
class TestListeners(base.BaseClientV1Test):
def test_create(self):
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(l.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):
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(l.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):
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(l.Listener(self.listeners, LISTENERS[0]).__dict__,
lsnr.__dict__)
mock.assert_called_once_with(
URL_TEMPLATE % (LISTENERS[0]['workbook_name']))
def test_get(self):
mock = self.mock_http_get(content=LISTENERS[0])
lsnr = self.listeners.get(LISTENERS[0]['workbook_name'],
LISTENERS[0]['id'])
self.assertEqual(l.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):
mock = self.mock_http_delete(status_code=204)
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']))

84
mistralclient/tests/unit/v1/test_tasks.py
View File

@ -1,84 +0,0 @@
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
from mistralclient.api.v1 import tasks as t
from mistralclient.tests.unit.v1 import base
# TODO(everyone): later we need additional tests verifying all the errors etc.
TASKS = [
{
'id': "1",
'workbook_name': "my_workbook",
'execution_id': '123',
'name': 'my_task',
'description': 'My cool task',
'action': 'my_action',
'state': 'RUNNING',
'tags': ['deployment', 'demo']
}
]
URL_TEMPLATE = '/workbooks/%s/executions/%s/tasks'
URL_TEMPLATE_ID = '/workbooks/%s/executions/%s/tasks/%s'
class TestTasks(base.BaseClientV1Test):
def test_update(self):
mock = self.mock_http_put(content=TASKS[0])
body = {
'state': TASKS[0]['state']
}
task = self.tasks.update(TASKS[0]['workbook_name'],
TASKS[0]['execution_id'],
TASKS[0]['id'],
TASKS[0]['state'])
self.assertIsNotNone(task)
self.assertEqual(t.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):
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(t.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):
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(t.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']))

133
mistralclient/tests/unit/v1/test_workbooks.py
View File

@ -1,133 +0,0 @@
# Copyright 2013 - Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
from mistralclient.api.v1 import workbooks as w
from mistralclient.tests.unit.v1 import base
# TODO(everyone): later we need additional tests verifying all the errors etc.
WORKBOOKS = [
{
'name': "my_workbook",
'description': "My cool Mistral workbook",
'tags': ['deployment', 'demo']
}
]
WB_DEF = """
Service:
name: my_service
type: REST
parameters:
baseUrl: http://my.service.org
actions:
action1:
parameters:
url: servers
method: POST
task-parameters:
param1:
optional: false
param2:
optional: false
Workflow:
tasks:
task1:
action: my_service:create-vm
parameters:
param1: 1234
param2: 42
"""
URL_TEMPLATE = '/workbooks'
URL_TEMPLATE_NAME = '/workbooks/%s'
URL_TEMPLATE_DEFINITION = '/workbooks/%s/definition'
class TestWorkbooks(base.BaseClientV1Test):
def test_create(self):
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(w.Workbook(self.workbooks, WORKBOOKS[0]).__dict__,
wb.__dict__)
mock.assert_called_once_with(URL_TEMPLATE, json.dumps(WORKBOOKS[0]))
def test_update(self):
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(w.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):
mock = self.mock_http_get(content={'workbooks': WORKBOOKS})
workbooks = self.workbooks.list()
self.assertEqual(1, len(workbooks))
wb = workbooks[0]
self.assertEqual(w.Workbook(self.workbooks, WORKBOOKS[0]).__dict__,
wb.__dict__)
mock.assert_called_once_with(URL_TEMPLATE)
def test_get(self):
mock = self.mock_http_get(content=WORKBOOKS[0])
wb = self.workbooks.get(WORKBOOKS[0]['name'])
self.assertIsNotNone(wb)
self.assertEqual(w.Workbook(self.workbooks, WORKBOOKS[0]).__dict__,
wb.__dict__)
mock.assert_called_once_with(URL_TEMPLATE_NAME % WORKBOOKS[0]['name'])
def test_delete(self):
mock = self.mock_http_delete(status_code=204)
self.workbooks.delete(WORKBOOKS[0]['name'])
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)
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):
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'])