openstack
/
python-tripleoclient
Code Issues Proposed changes

Merge "Add a command to validate instackenv.json"

This commit is contained in:
Dougal Matthews 2015-08-25 17:34:17 +02:00 committed by Gerrit Code Review
parent 2b681b6818 4765c15420
commit 8802345544
4 changed files with 335 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

245
rdomanager_oscplugin/tests/v1/baremetal/test_baremetal.py
View File

@ -26,6 +26,251 @@ from rdomanager_oscplugin.tests.v1.baremetal import fakes
from rdomanager_oscplugin.v1 import baremetal
class TestValidateInstackEnv(fakes.TestBaremetal):
def setUp(self):
super(TestValidateInstackEnv, self).setUp()
self.instack_json = tempfile.NamedTemporaryFile(mode='w', delete=False)
# Get the command object to test
self.cmd = baremetal.ValidateInstackEnv(self.app, None)
def mock_instackenv_json(self, instackenv_data):
json.dump(instackenv_data, self.instack_json)
self.instack_json.close()
def tearDown(self):
super(TestValidateInstackEnv, self).tearDown()
os.unlink(self.instack_json.name)
def test_success(self):
self.mock_instackenv_json({
"nodes": [{
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_password": "SOME SSH KEY",
"pm_type": "pxe_ssh",
"mac": [
"00:0b:d0:69:7e:59"
],
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(0, self.cmd.error_count)
def test_empty_password(self):
self.mock_instackenv_json({
"nodes": [{
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_password": "",
"pm_type": "pxe_ssh",
"mac": [
"00:0b:d0:69:7e:59"
],
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(1, self.cmd.error_count)
def test_no_password(self):
self.mock_instackenv_json({
"nodes": [{
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_type": "pxe_ssh",
"mac": [
"00:0b:d0:69:7e:59"
],
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(1, self.cmd.error_count)
def test_empty_user(self):
self.mock_instackenv_json({
"nodes": [{
"pm_user": "",
"pm_addr": "192.168.122.1",
"pm_password": "SOME SSH KEY",
"pm_type": "pxe_ssh",
"mac": [
"00:0b:d0:69:7e:59"
],
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(1, self.cmd.error_count)
def test_no_user(self):
self.mock_instackenv_json({
"nodes": [{
"pm_addr": "192.168.122.1",
"pm_password": "SOME SSH KEY",
"pm_type": "pxe_ssh",
"mac": [
"00:0b:d0:69:7e:59"
],
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(1, self.cmd.error_count)
def test_empty_mac(self):
self.mock_instackenv_json({
"nodes": [{
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_password": "SOME SSH KEY",
"pm_type": "pxe_ssh",
"mac": [],
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(1, self.cmd.error_count)
def test_no_mac(self):
self.mock_instackenv_json({
"nodes": [{
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_password": "SOME SSH KEY",
"pm_type": "pxe_ssh",
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(1, self.cmd.error_count)
def test_duplicated_mac(self):
self.mock_instackenv_json({
"nodes": [{
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_password": "KEY1",
"pm_type": "pxe_ssh",
"mac": [
"00:0b:d0:69:7e:58"
],
}, {
"arch": "x86_64",
"pm_user": "stack",
"pm_addr": "192.168.122.2",
"pm_password": "KEY2",
"pm_type": "pxe_ssh",
"mac": [
"00:0b:d0:69:7e:58"
]
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(1, self.cmd.error_count)
@mock.patch('rdomanager_oscplugin.utils.run_shell')
def test_ipmitool_success(self, mock_run_shell):
mock_run_shell.return_value = 0
self.mock_instackenv_json({
"nodes": [{
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_password": "KEY1",
"pm_type": "pxe_ipmitool",
"mac": [
"00:0b:d0:69:7e:59"
],
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(0, self.cmd.error_count)
@mock.patch('rdomanager_oscplugin.utils.run_shell')
def test_ipmitool_failure(self, mock_run_shell):
mock_run_shell.return_value = 1
self.mock_instackenv_json({
"nodes": [{
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_password": "KEY1",
"pm_type": "pxe_ipmitool",
"mac": [
"00:0b:d0:69:7e:59"
],
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(1, self.cmd.error_count)
@mock.patch('rdomanager_oscplugin.utils.run_shell')
def test_duplicated_baremetal_ip(self, mock_run_shell):
mock_run_shell.return_value = 0
self.mock_instackenv_json({
"nodes": [{
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_password": "KEY1",
"pm_type": "pxe_ipmitool",
"mac": [
"00:0b:d0:69:7e:59"
],
}, {
"arch": "x86_64",
"pm_user": "stack",
"pm_addr": "192.168.122.1",
"pm_password": "KEY2",
"pm_type": "pxe_ipmitool",
"mac": [
"00:0b:d0:69:7e:58"
]
}]
})
arglist = ['-f', self.instack_json.name]
parsed_args = self.check_parser(self.cmd, arglist, [])
self.cmd.take_action(parsed_args)
self.assertEqual(1, self.cmd.error_count)
class TestImportBaremetal(fakes.TestBaremetal):
def setUp(self):

5
rdomanager_oscplugin/utils.py
View File

@ -376,3 +376,8 @@ def create_cephx_key():
def run_shell(cmd):
return subprocess.call([cmd], shell=True)
def all_unique(x):
"""Return True if the collection has no duplications."""
return len(set(x)) == len(x)

84
rdomanager_oscplugin/v1/baremetal.py
View File

@ -58,6 +58,90 @@ def _csv_to_nodes_dict(nodes_csv):
return data
class ValidateInstackEnv(command.Command):
"""Validate `instackenv.json` which is used in `baremetal import`."""
auth_required = False
log = logging.getLogger(__name__ + ".ValidateInstackEnv")
def get_parser(self, prog_name):
parser = super(ValidateInstackEnv, self).get_parser(prog_name)
parser.add_argument(
'-f', '--file', dest='instackenv',
help="Path to the instackenv.json file.",
default='instackenv.json')
return parser
def take_action(self, parsed_args):
self.log.debug("take_action(%s)" % parsed_args)
self.error_count = 0
with open(parsed_args.instackenv, 'r') as net_file:
env_data = json.load(net_file)
maclist = []
baremetal_ips = []
for node in env_data['nodes']:
self.log.info("Checking node %s" % node['pm_addr'])
try:
if len(node['pm_password']) == 0:
self.log.error('ERROR: Password 0 length.')
self.error_count += 1
except Exception as e:
self.log.error('ERROR: Password does not exist: %s', e)
self.error_count += 1
try:
if len(node['pm_user']) == 0:
self.log.error('ERROR: User 0 length.')
self.error_count += 1
except Exception as e:
self.log.error('ERROR: User does not exist: %s', e)
self.error_count += 1
try:
if len(node['mac']) == 0:
self.log.error('ERROR: MAC address 0 length.')
self.error_count += 1
maclist.extend(node['mac'])
except Exception as e:
self.log.error('ERROR: MAC address does not exist: %s', e)
self.error_count += 1
if node['pm_type'] == "pxe_ssh":
self.log.debug("Identified virtual node")
if node['pm_type'] == "pxe_ipmitool":
self.log.debug("Identified baremetal node")
cmd = ('ipmitool -R 1 -I lanplus -H %s -U %s -P %s chassis '
'status' % (node['pm_addr'], node['pm_user'],
node['pm_password']))
self.log.debug("Executing: %s", cmd)
status = utils.run_shell(cmd)
if status != 0:
self.log.error('ERROR: ipmitool failed')
self.error_count += 1
baremetal_ips.append(node['pm_addr'])
if not utils.all_unique(baremetal_ips):
self.log.error('ERROR: Baremetals IPs are not all unique.')
self.error_count += 1
else:
self.log.debug('Baremetal IPs are all unique.')
if not utils.all_unique(maclist):
self.log.error('ERROR: MAC addresses are not all unique.')
self.error_count += 1
else:
self.log.debug('MAC addresses are all unique.')
if self.error_count == 0:
print('SUCCESS: found 0 errors')
else:
print('FAILURE: found %d errors' % self.error_count)
class ImportBaremetal(command.Command):
"""Import baremetal nodes from a JSON or CSV file"""

1
setup.cfg
View File

@ -56,6 +56,7 @@ openstack.cli.extension =
rdomanager_oscplugin = rdomanager_oscplugin.plugin
openstack.rdomanager_oscplugin.v1 =
baremetal_instackenv_validate = rdomanager_oscplugin.v1.baremetal:ValidateInstackEnv
baremetal_import = rdomanager_oscplugin.v1.baremetal:ImportBaremetal
baremetal_introspection_bulk_start = rdomanager_oscplugin.v1.baremetal:StartBaremetalIntrospectionBulk
baremetal_introspection_bulk_status = rdomanager_oscplugin.v1.baremetal:StatusBaremetalIntrospectionBulk