Add Cinder block pool metrics (#25)

The PR adds the cinder block pool metrics so that we can now retrieve
insights into the environment on the capacity all active storage pools.
This PR also creates tests for the new plugins and reduces duplication
across plugins by centralizing the runner method into the init of the
tests module.

This should now bring total code coverage to 100%

Signed-off-by: Kevin Carter <kevin.carter@rackspace.com>
This commit is contained in:
Kevin Carter 2017-03-19 15:31:46 -05:00 committed by Major Hayden
parent b2b2f8e7b5
commit e808ba1f7d
13 changed files with 538 additions and 79 deletions

View File

@ -1,3 +1,7 @@
[run]
branch = True
source = monitorstack
[report]
exclude_lines =
pragma: no cover

View File

@ -129,7 +129,7 @@ def process_result(results, output_format, **kwargs):
sys.exit(exit_code)
if __name__ == '__main__':
if __name__ == '__main__': # pragma: no cover
topdir = os.path.normpath(
os.path.join(
os.path.abspath(

View File

@ -0,0 +1,72 @@
# Copyright 2017, 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.
"""Get block storage usage from the available pools."""
import click
from monitorstack import utils
from monitorstack.cli import pass_context
from monitorstack.utils import os_utils as ost
DOC = """Get block storage totals from the available pools."""
COMMAND_NAME = 'os_block_pools_totals'
@click.command(COMMAND_NAME, short_help=DOC)
@click.option('--config-file',
help='OpenStack configuration file',
default='openstack.ini')
@pass_context
def cli(ctx, config_file):
"""Get nova cores quotas."""
setattr(cli, '__doc__', DOC)
output = {
'measurement_name': COMMAND_NAME,
'meta': {
'block_pools': 'totals'
},
'variables': {}
}
config = utils.read_config(config_file=config_file)['cinder']
interface = config.pop('interface', 'internal')
_ost = ost.OpenStack(os_auth_args=config)
try:
variables = output['variables']
total_capacity_gb = 0
free_capacity_gb = 0
for item in _ost.get_volume_pool_stats(interface=interface):
cap = item['capabilities']
output['meta'][cap.get('pool_name')] = True
free_capacity_gb += float(cap.get('free_capacity_gb', 0))
total_capacity_gb += float(cap.get('total_capacity_gb', 0))
else:
used_capacity = total_capacity_gb - free_capacity_gb
total_percent = 100 * (free_capacity_gb / total_capacity_gb)
variables['cinder_total_percent_used'] = total_percent
variables['cinder_total_free_capacity'] = free_capacity_gb
variables['cinder_total_used_capacity'] = used_capacity
variables['cinder_total_capacity'] = total_capacity_gb
except Exception as exp:
output['exit_code'] = 1
output['message'] = '{} failed -- {}'.format(
COMMAND_NAME,
utils.log_exception(exp=exp)
)
else:
output['exit_code'] = 0
output['message'] = '{} is ok'.format(COMMAND_NAME)
finally:
return output

View File

@ -0,0 +1,75 @@
# Copyright 2017, 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.
"""Get block storage usage from the available pools."""
import click
from monitorstack import utils
from monitorstack.cli import pass_context
from monitorstack.utils import os_utils as ost
DOC = """Get block storage usage from the available pools."""
COMMAND_NAME = 'os_block_pools_usage'
@click.command(COMMAND_NAME, short_help=DOC)
@click.option('--config-file',
help='OpenStack configuration file',
default='openstack.ini')
@pass_context
def cli(ctx, config_file):
"""Get nova cores quotas."""
setattr(cli, '__doc__', DOC)
output = {
'measurement_name': COMMAND_NAME,
'meta': {
'block_pools': 'usage'
},
'variables': {}
}
config = utils.read_config(config_file=config_file)['cinder']
interface = config.pop('interface', 'internal')
_ost = ost.OpenStack(os_auth_args=config)
try:
variables = output['variables']
for item in _ost.get_volume_pool_stats(interface=interface):
cap = item['capabilities']
total_capacity_gb = float(cap.get('total_capacity_gb', 0))
free_capacity_gb = float(cap.get('free_capacity_gb', 0))
percent_used = 100 * (free_capacity_gb / total_capacity_gb)
pool_name = cap.get('pool_name')
output['meta'][pool_name] = True
free_metric = '{}_free_capacity_gb'.format(pool_name)
variables[free_metric] = free_capacity_gb
total_metric = '{}_total_capacity_gb'.format(pool_name)
variables[total_metric] = total_capacity_gb
percent_metric = '{}_percent_used'.format(pool_name)
variables[percent_metric] = percent_used
except Exception as exp:
output['exit_code'] = 1
output['message'] = '{} failed -- {}'.format(
COMMAND_NAME,
utils.log_exception(exp=exp)
)
else:
output['exit_code'] = 0
output['message'] = '{} is ok'.format(COMMAND_NAME)
finally:
return output

View File

@ -20,11 +20,11 @@ import time
import traceback
try:
if sys.version_info > (3, 2, 0):
if sys.version_info > (3, 2, 0): # pragma: no cover
import configparser as ConfigParser
else:
else: # pragma: no cover
import ConfigParser
except ImportError:
except ImportError: # pragma: no cover
raise SystemExit('No configparser module was found.')
import diskcache

View File

@ -13,9 +13,19 @@
# limitations under the License.
"""OpenStack-related utilities."""
import sys
try:
from openstack import connection as os_conn
except ImportError as e:
if sys.version_info > (3, 2, 0): # pragma: no cover
import urllib.parse as urlparse
else: # pragma: no cover
import urlparse
except ImportError: # pragma: no cover
raise SystemExit('No urlparse module was found.')
try:
from openstack import connection as os_conn # pragma: no cover
except ImportError as e: # pragma: no cover
raise SystemExit('OpenStack plugins require access to the OpenStackSDK.'
' Please install "python-openstacksdk".'
' ERROR: %s' % str(e))
@ -42,6 +52,22 @@ class OpenStack(object):
"""
return os_conn.Connection(**self.os_auth_args)
def _session_req(self, path, service_type, interface='internal'):
"""Return compute resource limits for a project.
:param path: URL path to make a request against.
:type path: str
:param interface: Interface name, normally [internal, public, admin].
:type interface: str
:returns: dict
"""
endpoint_url = self.conn.session.get_endpoint(
interface=interface,
service_type=service_type
)
sess_url = urlparse.urljoin(endpoint_url, path)
return self.conn.session.get(sess_url).json()
def get_consumer_usage(self, servers=None, marker=None, limit=512):
"""Retrieve current usage by an OpenStack cloud consumer.
@ -81,14 +107,12 @@ class OpenStack(object):
:type interface: str
:returns: dict
"""
url = self.conn.session.get_endpoint(
interface=interface,
service_type='compute'
path = '/os-quota-sets/' + project_id
return self._session_req(
path=path,
service_type='compute',
interface=interface
)
quota_data = self.conn.session.get(
url + '/os-quota-sets/' + project_id
)
return quota_data.json()
def get_projects(self):
"""Retrieve a list of projects.
@ -182,3 +206,17 @@ class OpenStack(object):
:returns: str
"""
return self.get_flavor(flavor_id=flavor_id)['name']
def get_volume_pool_stats(self, interface='internal'):
"""Return volume pool usages.
:param interface: Interface name, normally [internal, public, admin].
:type interface: str
:returns: dict
"""
path = '/scheduler-stats/get_pools?detail=True'
return self._session_req(
path=path,
service_type='volume',
interface=interface
)

View File

@ -14,16 +14,51 @@
# limitations under the License.
"""This an __init__.py."""
import json
import os
from click.testing import CliRunner
from monitorstack import utils
from monitorstack.cli import cli
def read_config():
"""Load the test config file."""
"""Load the test config file.
:returns: dict
"""
os_config_file = os.path.expanduser(
os.path.abspath(
os.path.dirname(__file__) + '/files/test-openstack.ini'
)
)
return utils.read_config(os_config_file)
def runner(module, extra_args=None):
"""Run click CLI tests.
:param module: Name of module to run.
:type module: str
:param extra_args: List of extra arguments to pass to the CLI.
:type extra_args: list
:returns: dict || object
"""
_runner = CliRunner()
args = [
'-f', 'json',
module
]
if extra_args:
args.extend(extra_args)
result = _runner.invoke(cli, args)
try:
return json.loads(result.output)
except ValueError:
return result
def fake_version_info(major, minor, serial):
"""Return tuple for fake python version info."""
return major, minor, serial

View File

@ -159,6 +159,18 @@ class TestOsUtils(unittest.TestCase):
"""Tear down the test."""
pass
def test__session_req(self):
"""Test retrieving block pool stats."""
with mock.patch('openstack.connection.Connection') as MockClass:
MockClass.return_value = MockedOpenStackConn()
limits = self.osu._session_req(
path='test/path',
service_type='test-service',
interface='test-interface'
)
u = 'https://127.0.1.1/test-interface/test/path'
self.assertEquals(limits, {'url': u})
def test_get_consumer_usage(self):
"""Test retrieving consumer usage."""
with mock.patch('openstack.connection.Connection') as MockClass:
@ -197,8 +209,19 @@ class TestOsUtils(unittest.TestCase):
"""Test retrieving consumer limits."""
with mock.patch('openstack.connection.Connection') as MockClass:
MockClass.return_value = MockedOpenStackConn()
limits = self.osu.get_compute_limits(project_id='not-a-uuid')
u = 'https://127.0.1.1/internal/compute/os-quota-sets/not-a-uuid'
limits = self.osu.get_compute_limits(project_id='not-a-uuid1')
u = 'https://127.0.1.1/os-quota-sets/not-a-uuid1'
self.assertEquals(limits, {'url': u})
def test_get_compute_limits_interface_set(self):
"""Test retrieving consumer limits."""
with mock.patch('openstack.connection.Connection') as MockClass:
MockClass.return_value = MockedOpenStackConn()
limits = self.osu.get_compute_limits(
interface='test',
project_id='not-a-uuid2'
)
u = 'https://127.0.1.1/os-quota-sets/not-a-uuid2'
self.assertEquals(limits, {'url': u})
def test_get_projects(self):
@ -244,3 +267,19 @@ class TestOsUtils(unittest.TestCase):
MockClass.return_value = MockedOpenStackConn()
flavor_name = self.osu.get_flavor_name(flavor_id=12345)
self.assertEquals(flavor_name, 'test_12345')
def test_get_volume_pool_stats(self):
"""Test retrieving block pool stats."""
with mock.patch('openstack.connection.Connection') as MockClass:
MockClass.return_value = MockedOpenStackConn()
limits = self.osu.get_volume_pool_stats()
u = 'https://127.0.1.1/scheduler-stats/get_pools?detail=True'
self.assertEquals(limits, {'url': u})
def test_get_volume_pool_stats_interface_set(self):
"""Test retrieving block pool stats."""
with mock.patch('openstack.connection.Connection') as MockClass:
MockClass.return_value = MockedOpenStackConn()
limits = self.osu.get_volume_pool_stats(interface='test')
u = 'https://127.0.1.1/scheduler-stats/get_pools?detail=True'
self.assertEquals(limits, {'url': u})

View File

@ -21,6 +21,8 @@ from click.testing import CliRunner
from monitorstack.cli import cli
import tests # Import the test base module
def _runner(module):
runner = CliRunner()
@ -99,12 +101,12 @@ class TestKvm(unittest.TestCase):
def test_run_failure_no_libvirt(self):
"""Ensure the run() method works."""
sys.modules.pop('libvirt', None)
result = _runner('kvm')
self.assertTrue(isinstance(result, SystemExit))
result = tests.runner('kvm')
self.assertTrue(isinstance(result.exception, SystemExit))
def test_run_failure(self):
"""Ensure the run() method works."""
sys.modules['libvirt'] = LibvirtStubFailed()
result = _runner('kvm')
result = tests.runner('kvm')
assert result['measurement_name'] == 'kvm'
assert result['exit_code'] == 1

View File

@ -0,0 +1,120 @@
# Copyright 2017, 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 KVM plugin."""
from monitorstack.utils.os_utils import OpenStack as Ost
import tests # Import the test base module
def get_volume_pool_stats(*args, **kwargs):
"""Mocked get_consumer_usage()."""
return [
{
'name': 'name1',
'capabilities': {
'pool_name': 'pool_name1',
'total_capacity_gb': 100,
'free_capacity_gb': 50
}
},
{
'name': 'name2',
'capabilities': {
'pool_name': 'pool_name2',
'total_capacity_gb': 100,
'free_capacity_gb': 50
}
}
]
class TestOsBlock(object):
"""Tests for the os_vm.* monitors."""
def test_os_block_pools_totals_success(self, monkeypatch):
"""Ensure os_block_pools_totals method works with success."""
monkeypatch.setattr(
Ost,
'get_volume_pool_stats',
get_volume_pool_stats
)
result = tests.runner(
'os_block_pools_totals',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
variables = result['variables']
meta = result['meta']
assert variables['cinder_total_free_capacity'] == 100
assert variables['cinder_total_percent_used'] == 50
assert variables['cinder_total_used_capacity'] == 100
assert variables['cinder_total_capacity'] == 200
assert meta['block_pools'] == 'totals'
assert meta['pool_name1'] is True
assert meta['pool_name2'] is True
assert result['measurement_name'] == 'os_block_pools_totals'
def test_os_block_pools_totals_failure(self):
"""Ensure os_block_pools_totals method works with success."""
result = tests.runner(
'os_block_pools_totals',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_block_pools_totals'
assert result['exit_code'] == 1
def test_os_block_pools_usage_success(self, monkeypatch):
"""Ensure os_block_pools_totals method works with success."""
monkeypatch.setattr(
Ost,
'get_volume_pool_stats',
get_volume_pool_stats
)
result = tests.runner(
'os_block_pools_usage',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
variables = result['variables']
meta = result['meta']
assert variables['pool_name1_free_capacity_gb'] == 50
assert variables['pool_name2_total_capacity_gb'] == 100
assert variables['pool_name1_percent_used'] == 50
assert variables['pool_name1_total_capacity_gb'] == 100
assert variables['pool_name2_free_capacity_gb'] == 50
assert variables['pool_name2_percent_used'] == 50
assert meta['block_pools'] == 'usage'
assert meta['pool_name1'] is True
assert meta['pool_name2'] is True
assert result['measurement_name'] == 'os_block_pools_usage'
def test_os_block_pools_usage_failure(self):
"""Ensure os_block_pools_totals method works with success."""
result = tests.runner(
'os_block_pools_usage',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_block_pools_usage'
assert result['exit_code'] == 1

View File

@ -13,22 +13,9 @@
# limitations under the License.
"""Tests for the KVM plugin."""
import json
from click.testing import CliRunner
from monitorstack.cli import cli
from monitorstack.utils.os_utils import OpenStack as Ost
def _runner(module):
runner = CliRunner()
result = runner.invoke(cli, [
'-f', 'json',
module,
'--config-file', 'tests/files/test-openstack.ini',
])
return json.loads(result.output)
import tests # Import the test base module
class MockProject(object):
@ -96,21 +83,32 @@ def mock_get_compute_limits(*args, **kwargs):
}
class TestOs(object):
class TestOsVm(object):
"""Tests for the os_vm.* monitors."""
def test_os_vm_quota_cores_success(self, monkeypatch):
"""Ensure os_vm_quota_cores method works with success."""
monkeypatch.setattr(Ost, 'get_projects', mock_get_projects)
monkeypatch.setattr(Ost, 'get_compute_limits', mock_get_compute_limits)
result = _runner('os_vm_quota_cores')
result = tests.runner(
'os_vm_quota_cores',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_quota_cores'
assert result['meta'] == {'quotas': 'cores'}
def test_os_vm_quota_cores_failure(self):
"""Ensure os_vm_quota_cores method works with failure."""
result = _runner('os_vm_quota_cores')
result = tests.runner(
'os_vm_quota_cores',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_quota_cores'
assert result['meta'] == {'quotas': 'cores'}
@ -119,13 +117,25 @@ class TestOs(object):
monkeypatch.setattr(Ost, 'get_projects', mock_get_projects)
monkeypatch.setattr(Ost, 'get_compute_limits', mock_get_compute_limits)
result = _runner('os_vm_quota_instance')
result = tests.runner(
'os_vm_quota_instance',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_quota_instance'
assert result['meta'] == {'quotas': 'instances'}
def test_os_vm_quota_instance_failure(self):
"""Ensure os_vm_quota_cores method works with failure."""
result = _runner('os_vm_quota_instance')
result = tests.runner(
'os_vm_quota_instance',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_quota_instance'
assert result['meta'] == {'quotas': 'instances'}
@ -134,13 +144,25 @@ class TestOs(object):
monkeypatch.setattr(Ost, 'get_projects', mock_get_projects)
monkeypatch.setattr(Ost, 'get_compute_limits', mock_get_compute_limits)
result = _runner('os_vm_quota_ram')
result = tests.runner(
'os_vm_quota_ram',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_quota_ram'
assert result['meta'] == {'quotas': 'ram'}
def test_os_vm_quota_ram_failure(self):
"""Ensure os_vm_quota_ram method works with failure."""
result = _runner('os_vm_quota_ram')
result = tests.runner(
'os_vm_quota_ram',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_quota_ram'
assert result['meta'] == {'quotas': 'ram'}
@ -151,14 +173,26 @@ class TestOs(object):
monkeypatch.setattr(Ost, 'get_project_name', mock_get_project_name)
monkeypatch.setattr(Ost, 'get_consumer_usage', mock_get_consumer_usage)
result = _runner('os_vm_used_cores')
result = tests.runner(
'os_vm_used_cores',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_used_cores'
assert result['meta']['used'] == 'cores'
assert result['meta']['flavor_one']
def test_os_vm_used_cores_failure(self):
"""Ensure os_vm_used_cores method works with failure."""
result = _runner('os_vm_used_cores')
result = tests.runner(
'os_vm_used_cores',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_used_cores'
assert result['meta'] == {'used': 'cores'}
@ -169,14 +203,26 @@ class TestOs(object):
monkeypatch.setattr(Ost, 'get_project_name', mock_get_project_name)
monkeypatch.setattr(Ost, 'get_consumer_usage', mock_get_consumer_usage)
result = _runner('os_vm_used_disk')
result = tests.runner(
'os_vm_used_disk',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_used_disk'
assert result['meta']['used'] == 'disk'
assert result['meta']['flavor_one']
def test_os_vm_used_disk_failure(self):
"""Ensure os_vm_used_disk method works with failure."""
result = _runner('os_vm_used_disk')
result = tests.runner(
'os_vm_used_disk',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_used_disk'
assert result['meta'] == {'used': 'disk'}
@ -187,14 +233,26 @@ class TestOs(object):
monkeypatch.setattr(Ost, 'get_project_name', mock_get_project_name)
monkeypatch.setattr(Ost, 'get_consumer_usage', mock_get_consumer_usage)
result = _runner('os_vm_used_instance')
result = tests.runner(
'os_vm_used_instance',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_used_instance'
assert result['meta']['used'] == 'instances'
assert result['variables'] == {'test_name': 1}
def test_os_vm_used_instance_failure(self):
"""Ensure os_vm_used_instance method works with failure."""
result = _runner('os_vm_used_instance')
result = tests.runner(
'os_vm_used_instance',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_used_instance'
assert result['meta'] == {'used': 'instances'}
@ -205,7 +263,13 @@ class TestOs(object):
monkeypatch.setattr(Ost, 'get_project_name', mock_get_project_name)
monkeypatch.setattr(Ost, 'get_consumer_usage', mock_get_consumer_usage)
result = _runner('os_vm_used_ram')
result = tests.runner(
'os_vm_used_ram',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_used_ram'
assert result['meta']['used'] == 'ram'
assert result['meta']['flavor_one']
@ -213,6 +277,12 @@ class TestOs(object):
def test_os_vm_used_ram_failure(self):
"""Ensure os_vm_used_ram method works with failure."""
result = _runner('os_vm_used_ram')
result = tests.runner(
'os_vm_used_ram',
extra_args=[
'--config-file',
'tests/files/test-openstack.ini'
]
)
assert result['measurement_name'] == 'os_vm_used_ram'
assert result['meta'] == {'used': 'ram'}

View File

@ -13,38 +13,29 @@
# limitations under the License.
"""Tests for the process plugin."""
import json
import mock
from click.testing import CliRunner
from monitorstack.cli import cli
from monitorstack.plugins import process
import tests # Import the test base module
class TestUptime(object):
"""Tests for the uptime monitor class."""
def test_run_failure(self):
"""Ensure the run() method works."""
runner = CliRunner()
process_name = 'dont-go-chasing-waterfalls'
result = runner.invoke(cli, [
'-f', 'json',
'process', process_name])
result_json = json.loads(result.output)
assert result_json['variables'] == {process_name: 0}
assert result.exit_code == 1
result = tests.runner('process', extra_args=[process_name])
assert result['variables'] == {process_name: 0}
assert result['exit_code'] == 1
def test_run_success(self):
"""Ensure the run() method works."""
runner = CliRunner()
process_name = '/'
result = runner.invoke(cli, [
'-f', 'json',
'process', process_name])
result_json = json.loads(result.output)
assert result_json['variables'] == {process_name: 1}
assert result.exit_code == 0
result = tests.runner('process', extra_args=[process_name])
assert result['variables'] == {process_name: 1}
assert result['exit_code'] == 0
def test_check_process_success(self, monkeypatch):
"""Ensure the check_process() method works."""
@ -57,5 +48,23 @@ class TestUptime(object):
def test_get_cmdlines(self):
"""Ensure the get_cmdlines() method works."""
cmdlines = process.get_cmdlines()
assert isinstance(cmdlines, list)
assert isinstance(process.get_cmdlines(), list)
def test_get_cmdlines_exception(self, monkeypatch):
"""Ensure the get_cmdlines() method works."""
class _RaisePid(object):
pid = 'not-a-pid'
@staticmethod
def cmdline():
raise process.psutil.NoSuchProcess('not-a-pid')
def _mock_process_iter():
return [_RaisePid, _RaisePid, _RaisePid]
with mock.patch('psutil.process_iter') as MockClass:
MockClass.return_value = _mock_process_iter()
process_name = 'dont-go-chasing-waterfalls'
result = tests.runner('process', extra_args=[process_name])
assert result['variables'] == {process_name: 0}
assert result['exit_code'] == 1

View File

@ -13,24 +13,19 @@
# limitations under the License.
"""Tests for the uptime plugin."""
import json
from click.testing import CliRunner
from monitorstack.cli import cli
from monitorstack.plugins.uptime import get_uptime
import tests # Import the test base module
class TestUptime(object):
"""Tests for the uptime monitor class."""
def test_run(self):
"""Ensure the run() method works."""
runner = CliRunner()
result = runner.invoke(cli, ['-f', 'json', 'uptime'])
result_json = json.loads(result.output)
assert 'uptime' in result_json['variables']
assert result.exit_code == 0
result = tests.runner('uptime')
assert 'uptime' in result['variables']
assert result['exit_code'] == 0
def test_get_uptime(self):
"""Ensure the cli() method works."""