New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Send/Receive #479

Merged
merged 12 commits into from Mar 22, 2018
Copy path View file
@@ -67,7 +67,25 @@ Some examples:
# Filter job list for currently running jobs
$ tower-cli job list --status=running
# Export all objects
$ tower-cli receive --all
# Export all credentials
$ tower-cli receive --credential all
# Export a credential named "My Credential"
$ tower-cli receive --credential "My Credential"
# Import from a JSON file named assets.json
$ tower-cli send assets.json
# Import anything except an organization defined in a JSON file named assets.json
$ tower-cli send --prevent organization assets.json
# Copy all assets from one instance to another
$ tower-cli receive --tower-host tower1.example.com --all | tower-cli send --tower-host tower2.example.com
When in doubt, help is available!
@@ -94,6 +112,7 @@ In specific, ``tower-cli --help`` lists all available resources in the current v
config Read or write tower-cli configuration.
credential Manage credentials within Ansible Tower.
credential_type Manage credential types within Ansible Tower.
empty Empties assets from Tower.
group Manage groups belonging to an inventory.
host Manage hosts belonging to a group within an...
instance Check instances within Ansible Tower.
@@ -108,8 +127,10 @@ In specific, ``tower-cli --help`` lists all available resources in the current v
notification_template Manage notification templates within Ansible...
organization Manage organizations within Ansible Tower.
project Manage projects within Ansible Tower.
receive Export assets from Tower.
role Add and remove users/teams from roles.
schedule Manage schedules within Ansible Tower.
send Import assets into Tower.
setting Manage settings within Ansible Tower.
team Manage teams within Ansible Tower.
user Manage users within Ansible Tower.
Copy path View file
@@ -225,6 +225,8 @@ def test_reading_valid_token(self):
with mock.patch('tower_cli.api.json.load', return_value={'token': 'foobar', 'expires': expires}):
with client.test_mode as t:
t.register('/authtoken/', json.dumps({}), status_code=200, method='OPTIONS')
t.register('/authtoken/', json.dumps({'token': 'foobar', 'expires': expires}), status_code=200,
method='POST')
self.auth(self.req)
self.assertEqual(self.req.headers['Authorization'], 'Token foobar')
@@ -0,0 +1,104 @@
from tower_cli.api import client
from tower_cli.cli.transfer import common
from tower_cli.utils.data_structures import OrderedDict
from tests.compat import unittest
class TransferCommonTests(unittest.TestCase):
"""A set of tests to establish that the Common class works
in the way we expect.
"""
def test_get_api_options(self):
# Assert that an entry without a POST section returns None
with client.test_mode as t:
t.register_json('/inventories/', {'actions': {'PUT': {'FIRST': {'type': 'integer'}}}}, method='OPTIONS')
inventory_options = common.get_api_options('inventory')
self.assertIsNone(inventory_options)
# Assert that an entry with a POST section returns the post section
with client.test_mode as t:
t.register_json('/job_templates/', {'actions': {'POST': {'test': 'string'}}}, method='OPTIONS')
job_template_options = common.get_api_options('job_template')
self.assertEqual(job_template_options, {'test': 'string'}, "Failed to extract POST options")
# Test a cached API options
job_template_options = common.get_api_options('job_template')
self.assertEqual(job_template_options, {'test': 'string'}, "Failed to extract POST options")
def test_map_node_to_post_options(self):
source_node = {
"name": "My Name",
"created_on": "now",
"some_value": "the default value",
"some_other_value": "Not the default",
}
target_node = {}
post_options = {
"name": {"required": True},
"some_value": {"required": False, "default": "the default value"},
# Note, this function does not care if a required value is missing
"some_missing_required_value": {"required": True},
"some_other_value": {"default": "The default"},
}
# First test that nothing happens if post_options is None
common.map_node_to_post_options(None, source_node, target_node)
self.assertEqual(target_node, {}, "Post options of None modified the target node")
common.map_node_to_post_options(post_options, source_node, target_node)
self.assertEqual(target_node, {"name": "My Name", "some_other_value": "Not the default"}, "Failed node mapping")
def test_get_identity(self):
identity = common.get_identity('schedules')
self.assertEqual(identity, 'name', 'Schedules did not get proper identity {}'.format(identity))
identity = common.get_identity('inventory')
self.assertEqual(identity, 'name', 'Inventory did not get proper identity {}'.format(identity))
identity = common.get_identity('user')
self.assertEqual(identity, 'username', 'User did not get proper identity {}'.format(identity))
def test_remove_encrypted_value(self):
test_hash = {
'first': 'ok',
'second': common.ENCRYPTED_VALUE,
'sub': OrderedDict({
'first': common.ENCRYPTED_VALUE,
'second': 'ok',
}),
}
result_hash = {
'first': 'ok',
'second': '',
'sub': {
'first': '',
'second': 'ok',
},
}
common.remove_encrypted_values(test_hash)
self.assertEqual(test_hash, result_hash, "Failed to remove encrypted values from hash")
def test_remove_local_path_from_scm_project(self):
asset = {
'scm_type': 'Manual',
'local_path': 'somewhere',
}
result_asset = {
'scm_type': 'Manual',
'local_path': 'somewhere',
}
# Test a no change for either Manual or '' scm_type
common.remove_local_path_from_scm_project(asset)
self.assertEqual(asset, result_asset, "Incorrectly removed the local path for manual project")
asset['scm_type'] = ''
result_asset['scm_type'] = ''
common.remove_local_path_from_scm_project(asset)
self.assertEqual(asset, result_asset, "Incorrectly removed the local path for blank project")
# Test a change for a git scm_type
asset['scm_type'] = "git"
result_asset['scm_type'] = 'git'
del result_asset['local_path']
common.remove_local_path_from_scm_project(asset)
self.assertEqual(asset, result_asset, "Failed to remove the local path for git project")
@@ -0,0 +1,125 @@
from tower_cli.cli.transfer.logging_command import LoggingCommand
from six.moves import StringIO
import sys
import click
from tests.compat import unittest, mock
class LoggingCommandTests(unittest.TestCase):
"""A set of tests to establish that the LoggingCommand metaclass works
in the way we expect.
"""
def setUp(self):
self.held, sys.stdout = sys.stdout, StringIO()
def clear_string_buffer(self):
sys.stdout.seek(0)
sys.stdout.truncate(0)
def test_print_recap(self):
with mock.patch.object(click, 'get_terminal_size') as mock_method:
mock_method.return_value = [80, 128]
my_logging_command = LoggingCommand()
# Test default play recap (no starting new line)
my_logging_command.print_recap()
recap_msg = "PLAY RECAP *********************************************************************\n" +\
" ok=0 changed=0 warnings=0 failed=0\n\n"
self.assertEquals(sys.stdout.getvalue(), recap_msg, "First recap message failed")
# Test second play recap (has starting new line)
self.clear_string_buffer()
my_logging_command.print_recap()
recap_msg = "\nPLAY RECAP *********************************************************************\n" +\
" ok=0 changed=0 warnings=0 failed=0\n\n"
self.assertEquals(sys.stdout.getvalue(), recap_msg, "Second recap message failed")
# Test play recap with 1 ok
self.clear_string_buffer()
my_logging_command.ok_messages = 1
my_logging_command.print_recap()
recap_msg = "\nPLAY RECAP *********************************************************************\n" +\
" ok=1 changed=0 warnings=0 failed=0\n\n"
self.assertEquals(sys.stdout.getvalue(), recap_msg, "OK recap message failed")
# Test play recap with 1 changed
self.clear_string_buffer()
my_logging_command.changed_messages = 1
my_logging_command.print_recap()
recap_msg = "\nPLAY RECAP *********************************************************************\n" +\
" ok=1 changed=1 warnings=0 failed=0\n\n"
self.assertEquals(sys.stdout.getvalue(), recap_msg, "Changed recap message failed")
# Test play recap with 1 warnings
self.clear_string_buffer()
my_logging_command.warn_messages = 1
my_logging_command.print_recap()
recap_msg = "\nPLAY RECAP *********************************************************************\n" +\
" ok=1 changed=1 warnings=1 failed=0\n\n"
self.assertEquals(sys.stdout.getvalue(), recap_msg, "Warn recap message failed")
# Test play recap with 1 failed
self.clear_string_buffer()
my_logging_command.error_messages = 1
my_logging_command.print_recap()
recap_msg = "\nPLAY RECAP *********************************************************************\n" +\
" ok=1 changed=1 warnings=1 failed=1\n\n"
self.assertEquals(sys.stdout.getvalue(), recap_msg, "Error recap message failed")
def test_my_print_no_color(self):
# Validate that with no_color specified a message comes out straight
my_logging_command = LoggingCommand()
my_logging_command.no_color = True
my_logging_command.my_print("Test Message", fg="red", bold="True", nl="False")
self.assertEquals(sys.stdout.getvalue(), "Test Message\n", "Message with no color did not come out correctly")
def test_print_into(self):
# Validate that the intro just prints a blank line
my_logging_command = LoggingCommand()
my_logging_command.print_intro()
self.assertEquals(sys.stdout.getvalue(), "\n", "Print Intro does not print a new line")
def test_get_terminal_size(self):
with mock.patch.object(click, 'get_terminal_size') as mock_method:
mock_method.return_value = [80, 128]
my_logging_command = LoggingCommand()
my_logging_command.get_rows()
self.assertEquals(my_logging_command.columns, 80, "Did not correctly get back the terminal size")
def test_print_header_row(self):
with mock.patch.object(click, 'get_terminal_size') as mock_method:
mock_method.return_value = [80, 128]
my_logging_command = LoggingCommand()
my_logging_command.print_header_row("inventory", "test inventory")
first_inv_msg = "INVENTORY [test inventory] *****************************************************\n"
self.assertEquals(sys.stdout.getvalue(), first_inv_msg, "First header row incorrect")
self.clear_string_buffer()
my_logging_command.print_header_row("inventory", "test inventory")
second_inv_msg = "\nINVENTORY [test inventory] *****************************************************\n"
self.assertEquals(sys.stdout.getvalue(), second_inv_msg, "Second header row incorrect")
def test_log_error(self):
my_logging_command = LoggingCommand()
my_logging_command.log_error("TestError")
self.assertEquals(sys.stdout.getvalue(), "TestError\n", "Error did not come out correctly")
self.assertEquals(my_logging_command.error_messages, 1)
def test_log_warn(self):
my_logging_command = LoggingCommand()
my_logging_command.log_warn("TestWarn")
self.assertEquals(sys.stdout.getvalue(), " [WARNING]: TestWarn\n", "Warn did not come out correctly")
self.assertEquals(my_logging_command.warn_messages, 1)
def test_log_ok(self):
my_logging_command = LoggingCommand()
my_logging_command.log_ok("TestOK")
self.assertEquals(sys.stdout.getvalue(), "TestOK\n", "OK did not come out correctly")
self.assertEquals(my_logging_command.ok_messages, 1)
def test_log_change(self):
my_logging_command = LoggingCommand()
my_logging_command.log_change("TestChange")
self.assertEquals(sys.stdout.getvalue(), "TestChange\n", "Change did not come out correctly")
self.assertEquals(my_logging_command.changed_messages, 1)
Copy path View file
@@ -54,18 +54,27 @@ def _acquire_token(self):
def _get_auth_token(self):
filename = os.path.expanduser('~/.tower_cli_token.json')
token_json = None
try:
with open(filename) as f:
token_json = json.load(f)
if not isinstance(token_json, dict) or 'token' not in token_json or 'expires' not in token_json or \
dt.utcnow() > dt.strptime(token_json['expires'], TOWER_DATETIME_FMT):
if not isinstance(token_json, dict) or self.cli_client.get_prefix() not in token_json or \
'token' not in token_json[self.cli_client.get_prefix()] or \
'expires' not in token_json[self.cli_client.get_prefix()] or \
dt.utcnow() > dt.strptime(token_json[self.cli_client.get_prefix()]['expires'], TOWER_DATETIME_FMT):
raise Exception("Current token expires.")
return 'Token ' + token_json['token']
return 'Token ' + token_json[self.cli_client.get_prefix()]['token']
except Exception as e:
debug.log('Acquiring and caching auth token due to:\n%s' % str(e), fg='blue', bold=True)
token_json = self._acquire_token()
if not isinstance(token_json, dict) or 'token' not in token_json or 'expires' not in token_json:
raise exc.AuthError('Invalid Tower auth token format: %s' % json.dumps(token_json))
if not isinstance(token_json, dict):
token_json = {}
token_json[self.cli_client.get_prefix()] = self._acquire_token()
if not isinstance(token_json[self.cli_client.get_prefix()], dict) or \
'token' not in token_json[self.cli_client.get_prefix()] or \
'expires' not in token_json[self.cli_client.get_prefix()]:
raise exc.AuthError('Invalid Tower auth token format: %s' % json.dumps(
token_json[self.cli_client.get_prefix()]
))
with open(filename, 'w') as f:
json.dump(token_json, f)
try:
@@ -75,7 +84,7 @@ def _get_auth_token(self):
'Unable to set permissions on {0} - {1} '.format(filename, e),
UserWarning
)
return 'Token ' + token_json['token']
return 'Token ' + token_json[self.cli_client.get_prefix()]['token']
def __call__(self, r):
if 'Authorization' in r.headers:
@@ -188,6 +197,13 @@ def request(self, method, url, *args, **kwargs):
"""Make a request to the Ansible Tower API, and return the
response.
"""
# If the URL has the api/vX at the front strip it off
# This is common to have if you are extracting a URL from an existing object.
# For example, any of the 'related' fields of an object will have this
import re
url = re.sub("^/?api/v[0-9]+/", "", url)
# Piece together the full URL.
use_version = not url.startswith('/o/')
url = '%s%s' % (self.get_prefix(use_version), url.lstrip('/'))
Oops, something went wrong.
ProTip! Use n and p to navigate between commits in a pull request.