Adding vg_check from the maas plugins

This change adds the volume group check from the
maas plugins, and also adds a cli helper to run bash command.

Change-Id: I303a843b0abaea721758e182a9c8f3e2db33e85d
Signed-off-by: Michael Rice <michael.rice@rackspace.com>
Signed-off-by: Kevin Carter <kevin.carter@rackspace.com>
This commit is contained in:
Michael Rice 2017-05-11 14:35:32 -05:00 committed by Michael Rice
parent cee20d7740
commit b676952bbe
7 changed files with 242 additions and 16 deletions

View File

@ -0,0 +1,71 @@
# Copyright 2017, Michael Rice <michael@michaelrice.org>
#
# 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 platform
import click
from monitorstack import utils
from monitorstack.cli import pass_context
from monitorstack.utils.cli import run_command
DOC = """Check a given volume group"""
COMMAND_NAME = 'vg_check'
@click.command(COMMAND_NAME, short_help=DOC)
@click.option('--volume_group', nargs=1, type=str, required=True)
@pass_context
def cli(ctx, volume_group):
"""
Given volume group name get the total size and free space
:param ctx: Click context
:param volume_group: Name of volume group
:type volume_group: str
:return:
"""
exit_code, total_size, free = check_volgrp(volume_group)
output = {
'exit_code': exit_code,
'measurement_name': COMMAND_NAME,
'meta': {
'platform': platform.platform(),
}
}
if exit_code == 0:
output['message'] = '{} check for volume group {} is ok'.format(
COMMAND_NAME, volume_group)
output['variables'] = {
'vg_{}_total_size_M'.format(volume_group): total_size,
'vg_{}_free_M'.format(volume_group): free,
'vg_{}_used_M'.format(volume_group): total_size - free
}
if exit_code != 0:
# if exit_code is not 0 then 'free' actually has our error output.
# and with py3 it is bytes so we convert to str first.
output['message'] = '{} for {} failed -- {}'.format(
COMMAND_NAME, volume_group, utils.log_exception(str(free))
)
return output
def check_volgrp(name):
command = ('vgs {} --noheadings --units M '
'--nosuffix -o vg_size,vg_free'.format(name))
retcode, output, err = run_command(command)
if retcode != 0:
return retcode, output, err
totalsize, free = [int(float(x)) for x in output.split()]
return retcode, totalsize, free

26
monitorstack/utils/cli.py Normal file
View File

@ -0,0 +1,26 @@
# Copyright 2017, Michael Rice <michael@michaelrice.org>
#
# 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 shlex
import subprocess
def run_command(arg):
proc = subprocess.Popen(shlex.split(arg),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
shell=False)
out, err = proc.communicate()
ret = proc.returncode
return ret, out, err

View File

@ -31,7 +31,8 @@ def runner(module, extra_args=None):
""" """
_runner = CliRunner() _runner = CliRunner()
args = [ args = [
'-f', 'json', '-f',
'json',
module module
] ]
if extra_args: if extra_args:

View File

@ -34,3 +34,13 @@ def read_config():
def fake_version_info(major, minor, serial): def fake_version_info(major, minor, serial):
"""Return tuple for fake python version info.""" """Return tuple for fake python version info."""
return major, minor, serial return major, minor, serial
class FakePopen(object):
"""Fake Shell Commands."""
def __init__(self, return_code=0, *args, **kwargs):
self.returncode = return_code
@staticmethod
def communicate():
return 'stdout', 'stderr'

View File

@ -0,0 +1,56 @@
# Copyright 2018, Kevin Carter <kevin@cloudnull.com>
#
# 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.
"""Tests for the cli utils plugin."""
import unittest
import mock
from monitorstack.utils import cli
import tests # Import the test base module
class TestCliUtils(unittest.TestCase):
"""Tests for the utilities."""
def setUp(self):
"""Setup the test."""
# load the base class for these tests.
self.communicate_patched = mock.patch(
'monitorstack.utils.cli.subprocess.Popen'
)
self.communicate = self.communicate_patched.start()
def tearDown(self):
"""Tear down the test."""
self.communicate_patched.stop()
def test_run_command_success(self):
self.communicate.return_value = tests.unit.FakePopen()
ret, out, err = cli.run_command(
arg='test_command'
)
self.assertEqual(out, 'stdout')
self.assertEqual(ret, 0)
def test_run_command_fail(self):
self.communicate.return_value = tests.unit.FakePopen(
return_code=1
)
ret, out, err = cli.run_command(
arg='test_command'
)
self.assertEqual(err, 'stderr')
self.assertEqual(ret, 1)

View File

@ -13,26 +13,12 @@
# limitations under the License. # limitations under the License.
"""Tests for the KVM plugin.""" """Tests for the KVM plugin."""
import json
import sys import sys
import unittest import unittest
from click.testing import CliRunner
from monitorstack.cli import cli
import tests.unit # Import the test base module import tests.unit # Import the test base module
def _runner(module):
runner = CliRunner()
result = runner.invoke(cli, ['-f', 'json', module])
try:
return json.loads(result.output)
except Exception:
return result.exception
class LibvirtStub(object): class LibvirtStub(object):
"""Stubbed libvirt class.""" """Stubbed libvirt class."""
@ -91,7 +77,7 @@ class TestKvm(unittest.TestCase):
def test_run_success(self): def test_run_success(self):
"""Ensure the run() method works.""" """Ensure the run() method works."""
result = _runner('kvm') result = tests.runner('kvm')
variables = result['variables'] variables = result['variables']
meta = result['meta'] meta = result['meta']
assert 'kvm_vms' in variables assert 'kvm_vms' in variables

View File

@ -0,0 +1,76 @@
# Copyright 2017, Michael Rice <michael@michaelrice.org>
#
# 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 unittest
import mock
from monitorstack.plugins import vg_check
import tests.unit # Import the test base module
class VolumeGroupTestCases(unittest.TestCase):
def setUp(self):
"""Setup the test."""
# load the base class for these tests.
self.communicate_patched = mock.patch(
'monitorstack.utils.cli.subprocess.Popen'
)
self.communicate = self.communicate_patched.start()
def tearDown(self):
"""Tear down the test."""
self.communicate_patched.stop()
@mock.patch("monitorstack.utils.cli.subprocess")
def test_check_volgrp_returns_with_vg_not_found(self, mock_path):
"""
When the volume group is not found an exit status code of 5 is
returned by the system. When a non 0 is returned the expected
result from this call is is an error message with a blank total
"""
mock_path.Popen.return_value.returncode = 5
mock_path.Popen.return_value.communicate.return_value = \
("", "Volume group foo not found")
ret_code, total, free = vg_check.check_volgrp("foo")
assert ret_code == 5
assert total == ""
assert free == "Volume group foo not found"
class TestVolumeGroup(object):
def test_cli_would_exec_command(self, monkeypatch):
def mock_get_vgs(name):
"""Mock the check_volgrp() method."""
return 0, 100, 99
monkeypatch.setattr(
vg_check,
'check_volgrp',
mock_get_vgs
)
result = tests.runner(
'vg_check',
extra_args=['--volume_group', 'test']
)
variables = result['variables']
assert 'vg_test_used_M' in variables
assert variables['vg_test_used_M'] == 1
assert 'vg_test_free_M' in variables
assert variables['vg_test_free_M'] == 99
assert 'vg_test_total_size_M' in variables
assert variables['vg_test_total_size_M'] == 100