airship
/
shipyard
Code Issues Proposed changes

Updates cli to include workflow commands 

Also adds better test coverage for input checks
Changes some input checks to be less coupled to
invoking action/command.

Change-Id: I704b90bf39f4589f78d0253d99d28cf294c40d93
This commit is contained in:
Bryan Strassner 2017-11-09 17:37:40 -06:00
parent 717ea6c9f4
commit 48f1b09268
12 changed files with 612 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
docs/CLI.md
View File

@ -183,7 +183,7 @@ shipyard create configdocs
[--append | --replace]
--filename=<filename> (repeatable)
|
--directory=<directory>
--directory=<directory>
Example:
shipyard create configdocs design --append --filename=site_design.yaml
@ -233,7 +233,7 @@ shipyard describe
Example:
shipyard describe action/01BTG32JW87G0YKA1K29TKNAFX
Equivalent to:
shipyard describe action/01BTG32JW87G0YKA1K29TKNAFX
shipyard describe action 01BTG32JW87G0YKA1K29TKNAFX
shipyard describe step/01BTG32JW87G0YKA1K29TKNAFX/preflight
Equivalent to:
@ -243,6 +243,10 @@ Example:
Equivalent to:
shipyard describe validation 01BTG3PKBS15KCKFZ56XXXBGF2 \
--action=01BTG32JW87G0YKA1K29TKNAFX
shipyard describe workflow/deploy_site__2017-01-01T12:34:56.123456
Equivalent to:
shipyard describe workflow deploy_site__2017-01-01T12:34:56.123456
```
## describe action
@ -277,7 +281,7 @@ Example:
</dl>
## describe validation
Retrieves the validation details assocaited with an action and validation id
Retrieves the validation details associated with an action and validation id
```
shipyard describe validation
<validation id>
@ -288,7 +292,7 @@ Example:
--action=01BTG32JW87G0YKA1K29TKNAFX
```
<dl>
<dt>&lt;step id&gt;</dt>
<dt>&lt;validation id&gt;</dt>
<dd>
The id of the validation found in the describe action response.
</dd>
@ -298,6 +302,23 @@ Example:
</dd>
</dl>
## describe workflow
Retrieves the details for a workflow that is running or has run in the workflow
engine.
```
shipyard describe workflow
<workflow id>
Example:
shipyard describe workflow deploy_site__2017-01-01T12:34:56.123456
```
<dl>
<dt>&lt;workflow id&gt;</dt>
<dd>
The id of the workflow found in the get workflows response.
</dd>
</dl>
# Get Commands
## get actions
Lists the actions that have been invoked.
@ -358,6 +379,26 @@ Example:
</dd>
</dl>
## get workflows
Retrieve workflows that are running or have run in the workflow engine. This
includes processses that may not have been started as an action
(e.g. scheduled tasks).
```
shipyard get workflows
[--since=<date>]
Example:
shipyard get workflows
shipyard get workflows --since=2017-01-01T12:34:56.123456
```
<dl>
<dt>--since=&lt;date&gt;</dt>
<dd>
The historical cutoff date to limit the results of of this response.
</dd>
</dl>
# help commands
Provides topical help for shipyard. Note that --help will provide more
specific command help.
@ -369,7 +410,7 @@ Example:
shipyard help configdocs
```
<dl>
<dt>[&lt;topic&gt;]</dt>
<dt>&lt;topic&gt;</dt>
<dd>
The topic of the help to be displayed. If this parameter is not specified
the list of avaialable topics will be displayed.

2
shipyard_client/cli/action.py
View File

@ -37,7 +37,7 @@ class CliAction(object):
context_marker = self.api_parameters['context_marker']
debug = self.api_parameters['debug']
validate_auth_vars(self, ctx)
validate_auth_vars(ctx, self.api_parameters.get('auth_vars'))
self.logger.debug("Passing environment varibles to the API client")
try:

20
shipyard_client/cli/describe/actions.py
View File

@ -75,3 +75,23 @@ class DescribeValidation(CliAction):
self.output_format,
self.api_client.get_validation_detail(
action_id=self.action_id, validation_id=self.validation_id))
class DescribeWorkflow(CliAction):
"""Action to describe a workflow"""
def __init__(self, ctx, workflow_id):
"""Initializes api_client, sets parameters, and sets output_format"""
super().__init__(ctx)
self.logger.debug(
"DescribeWorkflow action initialized with workflow_id=%s",
workflow_id)
self.workflow_id = workflow_id
self.output_format = ctx.obj['FORMAT']
def invoke(self):
"""Calls API Client and formats response from API Client"""
self.logger.debug("Calling API Client get_action_detail.")
self.resp_txt = output_formatting(
self.output_format,
self.api_client.get_dag_detail(workflow_id=self.workflow_id))

60
shipyard_client/cli/describe/commands.py
View File

@ -21,7 +21,8 @@ from click_default_group import DefaultGroup
from shipyard_client.cli.describe.actions import DescribeAction
from shipyard_client.cli.describe.actions import DescribeStep
from shipyard_client.cli.describe.actions import DescribeValidation
from shipyard_client.cli.input_checks import check_id
from shipyard_client.cli.describe.actions import DescribeWorkflow
from shipyard_client.cli.input_checks import check_id, check_workflow_id
@click.group(cls=DefaultGroup, default='describe_default_command')
@ -48,27 +49,32 @@ def describe(ctx):
@click.argument('namespace_item')
@click.pass_context
def describe_default_command(ctx, namespace_item):
try:
namespace = namespace_item.split("/")
if (namespace[0] == 'action'):
if namespace[0] == 'action':
ctx.invoke(describe_action, action_id=namespace[1])
elif (namespace[0] == 'step'):
elif namespace[0] == 'step':
ctx.invoke(
describe_step, step_id=namespace[2], action=namespace[1])
elif (namespace[0] == 'validation'):
elif namespace[0] == 'validation':
ctx.invoke(
describe_validation,
validation_id=namespace[1],
action=namespace[2])
elif namespace[0] == 'workflow':
ctx.invoke(
describe_workflow,
workflow_id=namespace[1]
)
else:
raise
except:
ctx.fail("Invalid namespace item. Please utilize one of the following"
" formats for the namespace item. \n"
"action: action/action id \n"
"step: step/action id/step id \n"
"validation: validation/validation id/action id")
raise Exception('Invalid namespaced describe action')
except Exception:
ctx.fail("Invalid namespace item. Please utilize one of the following "
"formats for the namespace item.\n"
"action: action/action id\n"
"step: step/action id/step id\n"
"validation: validation/validation id/action id\n"
"workflow: workflow/workflow id")
DESC_ACTION = """
@ -89,7 +95,7 @@ SHORT_DESC_ACTION = ("Retrieves the detailed information about the supplied"
def describe_action(ctx, action_id):
if not action_id:
click.fail("An action id argument must be passed.")
ctx.fail("An action id argument must be passed.")
check_id(ctx, action_id)
@ -151,3 +157,31 @@ def describe_validation(ctx, validation_id, action):
click.echo(
DescribeValidation(ctx, validation_id, action)
.invoke_and_return_resp())
DESC_WORKFLOW = """
COMMAND: describe workflow \n
DESCRIPTION: Retrieves the detailed information about the supplied workflow
id. \n
FORMAT: shipyard describe workflow <workflow id> \n
EXAMPLE: shipyard describe workflow deploy_site__2017-10-09T21:19:03.000000
"""
SHORT_DESC_WORKFLOW = ("Retrieves the detailed information about the supplied"
" workflow id.")
@describe.command(
'workflow',
help=DESC_WORKFLOW,
short_help=SHORT_DESC_WORKFLOW)
@click.argument('workflow_id')
@click.pass_context
def describe_workflow(ctx, workflow_id):
if not workflow_id:
ctx.fail("An action id argument must be passed.")
check_workflow_id(ctx, workflow_id)
click.echo(DescribeWorkflow(ctx, workflow_id).invoke_and_return_resp())

18
shipyard_client/cli/get/actions.py
View File

@ -68,3 +68,21 @@ class GetRenderedConfigdocs(CliAction):
self.resp_txt = output_formatting(
self.output_format,
self.api_client.get_rendereddocs(version=self.version))
class GetWorkflows(CliAction):
"""Action to get workflows"""
def __init__(self, ctx, since=None):
"""Initializes api_client, sets parameters, and sets output_format"""
super().__init__(ctx)
self.logger.debug("GetWorkflows action initialized.")
self.since = since
self.output_format = ctx.obj['FORMAT']
def invoke(self):
"""Calls API Client and formats response from API Client"""
self.logger.debug("Calling API Client get_actions.")
self.resp_txt = output_formatting(
self.output_format,
self.api_client.get_workflows(self.since))

27
shipyard_client/cli/get/commands.py
View File

@ -19,6 +19,7 @@ import click
from shipyard_client.cli.get.actions import GetActions
from shipyard_client.cli.get.actions import GetConfigdocs
from shipyard_client.cli.get.actions import GetRenderedConfigdocs
from shipyard_client.cli.get.actions import GetWorkflows
@click.group()
@ -141,3 +142,29 @@ def get_renderedconfigdocs(ctx, buffer, committed):
version = 'buffer'
click.echo(GetRenderedConfigdocs(ctx, version).invoke_and_return_resp())
DESC_WORKFLOWS = """
COMMAND: workflows \n
DESCRIPTION: Lists the workflows from airflow. \n
FORMAT: shipyard get workflows [since]\n
EXAMPLE: \n
shipyard get workflows \n
shipyard get workflows --since=2017-11-09T15:02:18Z
"""
SHORT_DESC_WORKFLOWS = "Lists the workflows from airflow."
@get.command(
name='workflows',
help=DESC_WORKFLOWS,
short_help=SHORT_DESC_WORKFLOWS)
@click.option(
'--since',
help=('A boundary in the past within which to retrieve results.'
'Default is 30 days in the past.'))
@click.pass_context
def get_workflows(ctx, since):
click.echo(GetWorkflows(ctx, since).invoke_and_return_resp())

67
shipyard_client/cli/input_checks.py
View File

@ -11,56 +11,81 @@
# 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 arrow
from arrow.parser import ParserError
def check_action_command(ctx, action_command):
"""verifies the action command is valid"""
if (action_command != "deploy_site") and (
action_command != "update_site") and (action_command !=
"redeploy_server"):
"""Verifies the action command is valid"""
if action_command not in ['deploy_site', 'update_site', 'redeploy_server']:
ctx.fail('Invalid action command. The action commands available are '
'deploy_site, update_site, and redeploy_server.')
def check_control_action(ctx, action):
"""verifies the control action is valid"""
if (action != 'pause') and (action != 'unpause') and (action != 'stop'):
"""Verifies the control action is valid"""
if action not in ['pause', 'unpause', 'stop']:
ctx.fail('Invalid action. Please enter pause, unpause, or stop.')
def check_id(ctx, id):
"""verifies the id is valid"""
if (len(id) != 26):
def check_id(ctx, action_id):
"""Verifies a ULID id is in a valid format"""
if action_id is None:
ctx.fail('Invalid ID. None is not a valid action ID.')
if len(action_id) != 26:
ctx.fail('Invalid ID. ID can only be 26 characters.')
if not id.isalnum():
if not action_id.isalnum():
ctx.fail('Invalid ID. ID can only contain letters and numbers.')
def validate_auth_vars(self, ctx):
""" Checks that the required authurization varible have been entered """
def check_workflow_id(ctx, workflow_id):
"""Verifies that a workflow id matches the desired format"""
if workflow_id is None:
ctx.fail('Invalid ID. None is not a valid workflow ID.')
if '__' not in workflow_id:
ctx.fail('Invalid ID. The ID must cotain a double underscore '
'separating the workflow name from the execution date')
input_date_string = workflow_id.split('__')[1]
date_format_ok = True
try:
parsed_dt = arrow.get(input_date_string)
if input_date_string != parsed_dt.format('YYYY-MM-DDTHH:mm:ss.SSSSSS'):
date_format_ok = False
except ParserError:
date_format_ok = False
if not date_format_ok:
ctx.fail('Invalid ID. The date portion of the ID must conform to '
'YYYY-MM-DDTHH:mm:ss.SSSSSS')
def validate_auth_vars(ctx, auth_vars):
"""Checks that the required authurization varible have been entered"""
required_auth_vars = ['auth_url']
auth_vars = self.api_parameters['auth_vars']
err_txt = ""
for var in required_auth_vars:
if auth_vars[var] is None:
self.resp_txt += (
'Missing the required authorization variable: ' + var + '\n')
if self.resp_txt is not "":
self.resp_txt += ('\nMissing the following additional authorization '
'options: ')
err_txt += (
'Missing the required authorization variable: '
'--os_{}\n'.format(var))
if err_txt != "":
err_txt += ('\nMissing the following additional authorization '
'options: ')
for var in auth_vars:
if auth_vars[var] is None and var not in required_auth_vars:
self.resp_txt += '\n--os_' + var
ctx.fail(self.resp_txt)
err_txt += '\n--os_{}'.format(var)
ctx.fail(err_txt)
def check_reformat_parameter(ctx, param):
"""Checks for <name>=<value> format"""
param_dictionary = {}
try:
for p in param:
values = p.split('=')
param_dictionary[values[0]] = values[1]
except:
except Exception:
ctx.fail(
"Invalid parameter or parameter format for " + p +
". Please utilize the format: <parameter name>=<parameter value>")

19
shipyard_client/tests/unit/cli/describe/test_describe_actions.py
View File

@ -15,7 +15,7 @@
import mock
from shipyard_client.cli.describe.actions import \
DescribeAction, DescribeStep, DescribeValidation
DescribeAction, DescribeStep, DescribeValidation, DescribeWorkflow
from shipyard_client.api_client.base_client import BaseClient
from shipyard_client.tests.unit.cli.replace_api_client import \
replace_base_constructor, replace_post_rep, replace_get_resp, \
@ -97,3 +97,20 @@ def test_DescribeValidation(*args):
assert '01BTG32JW87G0YKA1K29TKNAFX' in url
assert 'validationdetails' in url
assert '01BTG3PKBS15KCKFZ56XXXBGF2' in url
@mock.patch.object(BaseClient, '__init__', replace_base_constructor)
@mock.patch.object(BaseClient, 'post_resp', replace_post_rep)
@mock.patch.object(BaseClient, 'get_resp', replace_get_resp)
@mock.patch.object(ShipyardClientContext, '__init__', temporary_context)
@mock.patch(
'shipyard_client.cli.describe.actions.output_formatting',
side_effect=replace_output_formatting)
def test_DescribeWorkflow(*args):
response = DescribeWorkflow(
ctx, 'deploy_site__2017-01-01T12:34:56.123456'
).invoke_and_return_resp()
# test correct function was called
url = response.get('url')
assert 'workflows' in url
assert 'deploy_site__2017-01-01T12:34:56.123456' in url

25
shipyard_client/tests/unit/cli/describe/test_describe_commands.py
View File

@ -18,6 +18,7 @@ from mock import patch, ANY
from shipyard_client.cli.describe.actions import DescribeAction
from shipyard_client.cli.describe.actions import DescribeStep
from shipyard_client.cli.describe.actions import DescribeValidation
from shipyard_client.cli.describe.actions import DescribeWorkflow
from shipyard_client.cli.commands import shipyard
auth_vars = ('--os-project-domain-name=OS_PROJECT_DOMAIN_NAME_test '
@ -116,3 +117,27 @@ def test_describe_validation_negative():
'--action=' + action_id
])
assert 'Error' in results.output
def test_describe_workflow():
"""test describe_workflow"""
workflow_id = 'deploy_site__2017-01-01T12:34:56.123456'
runner = CliRunner()
with patch.object(DescribeWorkflow, '__init__') as mock_method:
runner.invoke(shipyard, [
auth_vars, 'describe', 'workflow', workflow_id])
mock_method.assert_called_once_with(ANY, workflow_id)
def test_describe_workflow_negative():
"""
negative unit test for describe workflow command
verifies invalid workflow_id results in error
"""
workflow_id = 'deploysite20170101T123456123456'
runner = CliRunner()
results = runner.invoke(shipyard, [
auth_vars, 'describe', 'workflow', workflow_id])
assert 'Error' in results.output

31
shipyard_client/tests/unit/cli/get/test_get_actions.py
View File

@ -16,7 +16,7 @@
import mock
from shipyard_client.cli.get.actions import GetActions, GetConfigdocs, \
GetRenderedConfigdocs
GetRenderedConfigdocs, GetWorkflows
from shipyard_client.api_client.base_client import BaseClient
from shipyard_client.tests.unit.cli.replace_api_client import \
replace_base_constructor, replace_post_rep, replace_get_resp, \
@ -35,7 +35,7 @@ auth_vars = {
}
api_parameters = {
'auth_vars': auth_vars,
'context_marker': 'UUID',
'context_marker': '88888888-4444-4444-4444-121212121212',
'debug': False
}
@ -95,3 +95,30 @@ def test_GetRenderedConfigdocs(*args):
assert 'renderedconfigdocs' in url
params = response.get('params')
assert params.get('version') == 'buffer'
@mock.patch.object(BaseClient, '__init__', replace_base_constructor)
@mock.patch.object(BaseClient, 'post_resp', replace_post_rep)
@mock.patch.object(BaseClient, 'get_resp', replace_get_resp)
@mock.patch.object(ShipyardClientContext, '__init__', temporary_context)
@mock.patch(
'shipyard_client.cli.get.actions.output_formatting',
side_effect=replace_output_formatting)
def test_GetWorkflows(*args):
response = GetWorkflows(ctx, since=None).invoke_and_return_resp()
url = response.get('url')
assert 'workflows' in url
assert 'since' not in url
response = GetWorkflows(ctx).invoke_and_return_resp()
url = response.get('url')
assert 'workflows' in url
assert 'since' not in url
since_val = '2017-01-01T12:34:56Z'
response = GetWorkflows(ctx,
since=since_val).invoke_and_return_resp()
url = response.get('url')
assert 'workflows' in url
params = response.get('params')
assert params.get('since') == since_val

51
shipyard_client/tests/unit/cli/get/test_get_commands.py
View File

@ -15,8 +15,12 @@
from click.testing import CliRunner
from mock import patch, ANY
from shipyard_client.cli.get.actions import (GetActions, GetConfigdocs,
GetRenderedConfigdocs)
from shipyard_client.cli.get.actions import (
GetActions,
GetConfigdocs,
GetRenderedConfigdocs,
GetWorkflows
)
from shipyard_client.cli.commands import shipyard
auth_vars = ('--os-project-domain-name=OS_PROJECT_DOMAIN_NAME_test '
@ -37,9 +41,9 @@ def test_get_actions(*args):
def test_get_actions_negative(*args):
"""
negative unit test for get actions command
verifies invalid argument results in error
"""Negative unit test for get actions command.
Verifies invalid argument results in error.
"""
invalid_arg = 'invalid'
@ -60,9 +64,9 @@ def test_get_configdocs(*args):
def test_get_configdocs_negative(*args):
"""
negative unit test for get actions command
verifies invalid argument results in error
"""Negative unit test for get configdocs command.
Verifies invalid argument results in error.
"""
collection = 'design'
@ -83,9 +87,9 @@ def test_get_renderedconfigdocs(*args):
def test_get_renderedconfigdocs_negative(*args):
"""
negative unit test for get actions command
verfies invalid argument results in error
"""Negative unit test for get renderedconfigdocs command.
Verifies invalid argument results in error.
"""
invalid_arg = 'invalid'
@ -93,3 +97,28 @@ def test_get_renderedconfigdocs_negative(*args):
results = runner.invoke(
shipyard, [auth_vars, 'get', 'renderedconfigdocs', invalid_arg])
assert 'Error' in results.output
def test_get_workflows(*args):
"""test get_workflows"""
runner = CliRunner()
with patch.object(GetWorkflows, '__init__') as mock_method:
runner.invoke(shipyard, [auth_vars, 'get', 'workflows'])
mock_method.assert_called_once_with(ANY, None)
since_val = '2017-01-01T12:34:56Z'
since_arg = '--since={}'.format(since_val)
with patch.object(GetWorkflows, '__init__') as mock_method:
runner.invoke(shipyard, [auth_vars, 'get', 'workflows', since_arg])
mock_method.assert_called_once_with(ANY, since_val)
def test_get_workflows_negative(*args):
"""Negative unit test for get workflows command"""
invalid_arg = 'invalid_date'
runner = CliRunner()
results = runner.invoke(
shipyard, [auth_vars, 'get', 'workflows', invalid_arg])
assert 'Error' in results.output

295
shipyard_client/tests/unit/cli/test_input_checks.py Normal file
View File

@ -0,0 +1,295 @@
# Copyright 2017 AT&T Intellectual Property. All other 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.
"""Unit tests for input_checks helper module"""
from unittest.mock import Mock
from shipyard_client.cli import input_checks
def test_check_workflow_id_valid():
"""Check that a valid formatted id passes"""
ctx = Mock(side_effect=Exception("failed"))
input_checks.check_workflow_id(
ctx, 'something__2017-01-01T12:34:56.000000')
ctx.fail.assert_not_called()
def test_check_workflow_id_valid_tricky():
"""Check that a valid formatted id passes.
This test provides something that arrow will parse as an invalid
date if the code is not properly set up to separate the date
first.
"""
ctx = Mock(side_effect=Exception("failed"))
input_checks.check_workflow_id(
ctx, '2017-01-01T12:34:99.000__2017-01-01T12:34:56.000000')
ctx.fail.assert_not_called()
def test_check_workflow_id_no_date():
"""Check tha a missing date portion of the string is rejected."""
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_workflow_id(ctx, 'something__')
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'date portion' in str(ctx.mock_calls[0])
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_workflow_id_none():
"""Check that the workflow id check invokes the context.fail on None"""
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_workflow_id(
ctx, None)
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_workflow_id_invalid_separator():
"""Check that the separator check invokes the context.fail"""
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_workflow_id(
ctx, 'something20170101T12:34:56.000000')
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_workflow_id_invalid_date():
"""Check that the date format check invokes the context.fail"""
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_workflow_id(
ctx, 'something__blah0101 12:34:56.000000')
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_workflow_id_invalid_date_format():
"""Check that the date format check invokes the context.fail"""
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_workflow_id(
ctx, 'something__2017-01-01T12:34:56')
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_id_valid():
ctx = Mock(side_effect=Exception("failed"))
input_checks.check_id(ctx, "12345678901234567890123456")
ctx.fail.assert_not_called()
def test_check_id_too_long():
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_id(ctx, "TOOLONGTOOLONGTOOLONGTOOLONGTOOLONG")
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_id_too_short():
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_id(ctx, "TOOSHORT")
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_id_bad_chars():
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_id(ctx, "_ARENOTALLOWED-")
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_id_none():
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_id(ctx, None)
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_control_action_valid():
ctx = Mock(side_effect=Exception("failed"))
input_checks.check_control_action(ctx, 'pause')
input_checks.check_control_action(ctx, 'unpause')
input_checks.check_control_action(ctx, 'stop')
ctx.fail.assert_not_called()
def test_check_control_action_invalid():
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_control_action(ctx, 'completely_bogus')
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_control_action_none():
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_control_action(ctx, None)
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_action_commands():
ctx = Mock(side_effect=Exception("failed"))
input_checks.check_action_command(ctx, 'deploy_site')
input_checks.check_action_command(ctx, 'update_site')
input_checks.check_action_command(ctx, 'redeploy_server')
ctx.fail.assert_not_called()
def test_check_action_commands_invalid():
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_action_command(ctx, "burger_and_fries")
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_action_commands_none():
ctx = Mock(side_effect=Exception("failed"))
try:
input_checks.check_action_command(ctx, None)
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_validate_auth_vars_valid():
ctx = Mock(side_effect=Exception("failed"))
auth_vars = {
'project_domain_name': 'default',
'user_domain_name': 'default',
'project_name': 'service',
'username': 'shipyard',
'password': 'password',
'auth_url': 'abcdefg'
}
input_checks.validate_auth_vars(ctx, auth_vars)
ctx.fail.assert_not_called()
def test_validate_auth_vars_missing_required():
ctx = Mock(side_effect=Exception("failed"))
auth_vars = {
'project_domain_name': 'default',
'user_domain_name': 'default',
'project_name': 'service',
'username': 'shipyard',
'password': 'password',
'auth_url': None
}
try:
input_checks.validate_auth_vars(ctx, auth_vars)
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
assert 'os_auth_url' in str(ctx.mock_calls[0])
assert 'os_username' not in str(ctx.mock_calls[0])
assert 'os_password' not in str(ctx.mock_calls[0])
def test_validate_auth_vars_missing_required_and_others():
ctx = Mock(side_effect=Exception("failed"))
auth_vars = {
'project_domain_name': 'default',
'user_domain_name': 'default',
'project_name': 'service',
'username': None,
'password': 'password',
'auth_url': None
}
try:
input_checks.validate_auth_vars(ctx, auth_vars)
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
assert 'os_auth_url' in str(ctx.mock_calls[0])
assert 'os_username' in str(ctx.mock_calls[0])
assert 'os_password' not in str(ctx.mock_calls[0])
def test_check_reformat_parameter_valid():
ctx = Mock(side_effect=Exception("failed"))
param = ['this=that']
input_checks.check_reformat_parameter(ctx, param)
param = []
input_checks.check_reformat_parameter(ctx, param)
param = ['this=that', 'some=another']
o_params = input_checks.check_reformat_parameter(ctx, param)
assert 'this' in o_params
assert 'some' in o_params
assert 'that' not in o_params
assert 'another' not in o_params
assert o_params['this'] == 'that'
assert o_params['some'] == 'another'
ctx.fail.assert_not_called()
def test_check_reformat_parameter_no_equals_second():
ctx = Mock(side_effect=Exception("failed"))
param = ['this=that', 'someanother']
try:
input_checks.check_reformat_parameter(ctx, param)
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])
def test_check_reformat_parameter_no_equals_first():
ctx = Mock(side_effect=Exception("failed"))
param = ['thisthat', 'some=another']
try:
input_checks.check_reformat_parameter(ctx, param)
except Exception:
pass
# py 3.6: ctx.fail.assert_called()
assert 'call.fail(' in str(ctx.mock_calls[0])