Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_db_version_good(mocker, mock_info_log):
mocker.patch.object(app.puppetdb, 'current_version', return_value='4.2.0')
err = 'PuppetDB Version %d.%d.%d' % (4, 2, 0)
result = utils.get_db_version(app.puppetdb)
mock_info_log.assert_called_with(err)
assert (4, 0, 0) < result
assert (4, 2, 0) == result
assert (3, 2, 0) < result
assert (4, 3, 0) > result
assert (5, 0, 0) > result
assert (4, 2, 1) > result
def test_db_invalid_version(mocker, mock_err_log):
mocker.patch.object(app.puppetdb, 'current_version', return_value='4')
err = u"Unable to determine version from string: '%s'" % (4)
result = utils.get_db_version(app.puppetdb)
mock_err_log.assert_called_with(err)
assert (4, 0, 0) < result
assert (4, 2, 0) == result
def test_db_http_error(mocker, mock_err_log):
err = "NotFound"
def raise_http_error():
x = Response()
x.status_code = 404
x.reason = err
raise HTTPError(err, response=x)
mocker.patch.object(app.puppetdb, 'current_version',
side_effect=raise_http_error)
result = utils.get_db_version(app.puppetdb)
mock_err_log.assert_called_with(err)
assert result == ()
def test_db_empty_response(mocker, mock_err_log):
err = "Empty Response"
def connection_error():
raise EmptyResponseError(err)
mocker.patch.object(app.puppetdb, 'current_version',
side_effect=connection_error)
result = utils.get_db_version(app.puppetdb)
mock_err_log.assert_called_with(err)
assert result == ()
def test_db_connection_error(mocker, mock_err_log):
err = "ConnectionError"
def connection_error():
x = Response()
x.status_code = 500
x.reason = err
raise ConnectionError(err, response=x)
mocker.patch.object(app.puppetdb, 'current_version',
side_effect=connection_error)
result = utils.get_db_version(app.puppetdb)
mock_err_log.assert_called_with(err)
assert result == ()
envs = environments()
metrics = {
'num_nodes': 0,
'num_resources': 0,
'avg_resources_node': 0}
check_env(env, envs)
if env == '*':
query = app.config['OVERVIEW_FILTER']
prefix = 'puppetlabs.puppetdb.population'
query_type = ''
# Puppet DB version changed the query format from 3.2.0
# to 4.0 when querying mbeans
if get_db_version(puppetdb) < (4, 0, 0):
query_type = 'type=default,'
num_nodes = get_or_abort(
puppetdb.metric,
"{0}{1}".format(prefix, ':%sname=num-nodes' % query_type))
num_resources = get_or_abort(
puppetdb.metric,
"{0}{1}".format(prefix, ':%sname=num-resources' % query_type))
avg_resources_node = get_or_abort(
puppetdb.metric,
"{0}{1}".format(prefix,
':%sname=avg-resources-per-node' % query_type))
metrics['num_nodes'] = num_nodes['Value']
metrics['num_resources'] = num_resources['Value']
metrics['avg_resources_node'] = "{0:10.0f}".format(
avg_resources_node['Value'])
def radiator(env):
"""This view generates a simplified monitoring page
akin to the radiator view in puppet dashboard
"""
envs = environments()
check_env(env, envs)
if env == '*':
query_type = ''
if get_db_version(puppetdb) < (4, 0, 0):
query_type = 'type=default,'
query = None
metrics = get_or_abort(
puppetdb.metric,
'puppetlabs.puppetdb.population:%sname=num-nodes' % query_type)
num_nodes = metrics['Value']
else:
query = AndOperator()
metric_query = ExtractOperator()
query.add(EqualsOperator("catalog_environment", env))
metric_query.add_field(FunctionOperator('count'))
metric_query.add_query(query)
metrics = get_or_abort(
puppetdb._query,