diff --git a/puppetboard/app.py b/puppetboard/app.py index a709ed2..f236306 100644 --- a/puppetboard/app.py +++ b/puppetboard/app.py @@ -121,7 +121,7 @@ def bad_request(e): @app.errorhandler(403) def forbidden(e): envs = environments() - return render_template('403.html', envs=envs), 400 + return render_template('403.html', envs=envs), 403 @app.errorhandler(404) diff --git a/puppetboard/templates/412.html b/puppetboard/templates/412.html index caa0867..5b829d4 100644 --- a/puppetboard/templates/412.html +++ b/puppetboard/templates/412.html @@ -1,11 +1,5 @@ {% extends 'layout.html' %} -{% block row_fluid %} -
-
-
-

Feature unavailable

-

You've configured Puppetboard with an API version that does not support this feature.

-
-
-
+{% block content %} +

Feature unavailable

+

You've configured Puppetboard with an API version that does not support this feature.

{% endblock %} diff --git a/puppetboard/templates/layout.html b/puppetboard/templates/layout.html index efe7c6f..62f42ba 100644 --- a/puppetboard/templates/layout.html +++ b/puppetboard/templates/layout.html @@ -13,10 +13,10 @@ } {% else %} - + {% endif %} - - + + {% block head %} {% endblock head %} diff --git a/puppetboard/templates/node.html b/puppetboard/templates/node.html index 5d92b8f..75a00a1 100644 --- a/puppetboard/templates/node.html +++ b/puppetboard/templates/node.html @@ -2,7 +2,7 @@ {% import '_macros.html' as macros %} {% block head %} {% if config.DAILY_REPORTS_CHART_ENABLED %} - + {% endif %} {% endblock head %} {% block content %} diff --git a/puppetboard/templates/radiator.html b/puppetboard/templates/radiator.html index 3ef2f67..3a49c12 100644 --- a/puppetboard/templates/radiator.html +++ b/puppetboard/templates/radiator.html @@ -15,7 +15,7 @@ - + @@ -30,7 +30,7 @@ - + @@ -45,7 +45,7 @@ - + @@ -60,7 +60,7 @@ - + @@ -75,7 +75,7 @@ - + @@ -90,7 +90,7 @@ - + diff --git a/requirements-test.txt b/requirements-test.txt index beeb8e9..021a97c 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -4,6 +4,8 @@ mock==1.3.0 pytest==3.0.1 pytest-pep8==1.0.5 pytest-cov==2.2.1 +pytest-mock==1.5.0 cov-core==1.15.0 unittest2==1.1.0; python_version < '2.7' bandit +beautifulsoup4==4.5.3 diff --git a/test/__init__.py b/test/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/test_app.py b/test/test_app.py index c25399e..d5bc9a2 100644 --- a/test/test_app.py +++ b/test/test_app.py @@ -1,19 +1,328 @@ -import os +import pytest +import json from puppetboard import app -import unittest -import tempfile +from pypuppetdb.types import Node +from puppetboard import default_settings + +from bs4 import BeautifulSoup -class AppTestCase(unittest.TestCase): - def setUp(self): - pass +class MockDbQuery(object): + def __init__(self, responses): + self.responses = responses - def tearDown(self): - pass + def get(self, method, **kws): + resp = None + if method in self.responses: + resp = self.responses[method].pop(0) - def test_first_test(self): - self.assertTrue(True) + if 'validate' in resp: + checks = resp['validate']['checks'] + resp = resp['validate']['data'] + for check in checks: + assert check in kws + expected_value = checks[check] + assert expected_value == kws[check] + return resp -if __name__ == '__main__': - unittest.main() +@pytest.fixture +def mock_puppetdb_environments(mocker): + environemnts = [ + {'name': 'production'}, + {'name': 'staging'} + ] + return mocker.patch.object(app.puppetdb, 'environments', + return_value=environemnts) + + +@pytest.fixture +def mock_puppetdb_default_nodes(mocker): + node_list = [ + Node('_', 'node-unreported', + report_timestamp='2013-08-01T09:57:00.000Z', + latest_report_hash='1234567', + catalog_timestamp='2013-08-01T09:57:00.000Z', + facts_timestamp='2013-08-01T09:57:00.000Z', + status='unreported'), + Node('_', 'node-changed', + report_timestamp='2013-08-01T09:57:00.000Z', + latest_report_hash='1234567', + catalog_timestamp='2013-08-01T09:57:00.000Z', + facts_timestamp='2013-08-01T09:57:00.000Z', + status='changed'), + Node('_', 'node-failed', + report_timestamp='2013-08-01T09:57:00.000Z', + latest_report_hash='1234567', + catalog_timestamp='2013-08-01T09:57:00.000Z', + facts_timestamp='2013-08-01T09:57:00.000Z', + status='failed'), + Node('_', 'node-noop', + report_timestamp='2013-08-01T09:57:00.000Z', + latest_report_hash='1234567', + catalog_timestamp='2013-08-01T09:57:00.000Z', + facts_timestamp='2013-08-01T09:57:00.000Z', + status='noop'), + Node('_', 'node-unchanged', + report_timestamp='2013-08-01T09:57:00.000Z', + latest_report_hash='1234567', + catalog_timestamp='2013-08-01T09:57:00.000Z', + facts_timestamp='2013-08-01T09:57:00.000Z', + status='unchanged'), + Node('_', 'node-skipped', + report_timestamp='2013-08-01T09:57:00.000Z', + latest_report_hash='1234567', + catalog_timestamp='2013-08-01T09:57:00.000Z', + facts_timestamp='2013-08-01T09:57:00.000Z', + status='skipped') + + ] + return mocker.patch.object(app.puppetdb, 'nodes', + return_value=iter(node_list)) + + +@pytest.fixture +def client(): + client = app.app.test_client() + return client + + +def test_first_test(): + assert app is not None, ("%s" % reg.app) + + +def test_no_env(client, mock_puppetdb_environments): + rv = client.get('/nonexsistenv/') + + assert rv.status_code == 404 + + +def test_get_index(client, mocker, + mock_puppetdb_environments, + mock_puppetdb_default_nodes): + query_data = { + 'nodes': [[{'count': 10}]], + 'resources': [[{'count': 40}]], + } + + dbquery = MockDbQuery(query_data) + + mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get) + rv = client.get('/') + soup = BeautifulSoup(rv.data, 'html.parser') + assert soup.title.contents[0] == 'Puppetboard' + assert rv.status_code == 200 + + +def test_index_all(client, mocker, + mock_puppetdb_environments, + mock_puppetdb_default_nodes): + + base_str = 'puppetlabs.puppetdb.population:' + query_data = { + 'mbean': [ + { + 'validate': { + 'data': {'Value': '50'}, + 'checks': { + 'path': '%sname=num-nodes' % base_str + } + } + }, + { + 'validate': { + 'data': {'Value': '60'}, + 'checks': { + 'path': '%sname=num-resources' % base_str + } + } + }, + { + 'validate': { + 'data': {'Value': 60.3}, + 'checks': { + 'path': '%sname=avg-resources-per-node' % base_str + } + } + } + ] + } + dbquery = MockDbQuery(query_data) + mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get) + rv = client.get('/%2A/') + + soup = BeautifulSoup(rv.data, 'html.parser') + assert soup.title.contents[0] == 'Puppetboard' + vals = soup.find_all('h1', + {"class": "ui header darkblue no-margin-bottom"}) + + assert len(vals) == 3 + assert vals[0].string == '50' + assert vals[1].string == '60' + assert vals[2].string == ' 60' + + assert rv.status_code == 200 + + +def test_index_division_by_zero(client, mocker): + mock_puppetdb_environments(mocker) + mock_puppetdb_default_nodes(mocker) + + query_data = { + 'nodes': [[{'count': 0}]], + 'resources': [[{'count': 40}]], + } + + dbquery = MockDbQuery(query_data) + + mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get) + + rv = client.get('/') + + assert rv.status_code == 200 + + soup = BeautifulSoup(rv.data, 'html.parser') + assert soup.title.contents[0] == 'Puppetboard' + + vals = soup.find_all('h1', + {"class": "ui header darkblue no-margin-bottom"}) + assert len(vals) == 3 + assert vals[2].string == '0' + + +def test_offline_mode(client, mocker): + app.app.config['OFFLINE_MODE'] = True + + mock_puppetdb_environments(mocker) + mock_puppetdb_default_nodes(mocker) + + query_data = { + 'nodes': [[{'count': 10}]], + 'resources': [[{'count': 40}]], + } + + dbquery = MockDbQuery(query_data) + + mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get) + rv = client.get('/') + soup = BeautifulSoup(rv.data, 'html.parser') + assert soup.title.contents[0] == 'Puppetboard' + for link in soup.find_all('link'): + assert "//" not in link['href'] + + for script in soup.find_all('script'): + if "src" in script.attrs: + assert "//" not in script['src'] + + assert rv.status_code == 200 + + +def test_default_node_view(client, mocker, + mock_puppetdb_environments, + mock_puppetdb_default_nodes): + + rv = client.get('/nodes') + soup = BeautifulSoup(rv.data, 'html.parser') + assert soup.title.contents[0] == 'Puppetboard' + + for label in ['failed', 'changed', 'unreported', 'noop']: + vals = soup.find_all('a', + {"class": "ui %s label status" % label}) + assert len(vals) == 1 + assert 'node-%s' % (label) in vals[0].attrs['href'] + + assert rv.status_code == 200 + + +def test_radiator_view(client, mocker, + mock_puppetdb_environments, + mock_puppetdb_default_nodes): + query_data = { + 'nodes': [[{'count': 10}]], + 'resources': [[{'count': 40}]], + } + + dbquery = MockDbQuery(query_data) + + mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get) + + rv = client.get('/radiator') + + assert rv.status_code == 200 + + soup = BeautifulSoup(rv.data, 'html.parser') + assert soup.title.contents[0] == 'Puppetboard' + assert soup.h1 != 'Not Found' + total = soup.find(class_='total') + + assert '10' in total.text + + +def test_radiator_view_json(client, mocker, + mock_puppetdb_environments, + mock_puppetdb_default_nodes): + query_data = { + 'nodes': [[{'count': 10}]], + 'resources': [[{'count': 40}]], + } + + dbquery = MockDbQuery(query_data) + + mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get) + + rv = client.get('/radiator', headers={'Accept': 'application/json'}) + + assert rv.status_code == 200 + json_data = json.loads(rv.data.decode('utf-8')) + + assert json_data['unreported'] == 1 + assert json_data['noop'] == 1 + assert json_data['failed'] == 1 + assert json_data['changed'] == 1 + assert json_data['skipped'] == 1 + assert json_data['unchanged'] == 1 + + +def test_radiator_view_bad_env(client, mocker): + mock_puppetdb_environments(mocker) + mock_puppetdb_default_nodes(mocker) + + query_data = { + 'nodes': [[{'count': 10}]], + 'resources': [[{'count': 40}]], + } + + dbquery = MockDbQuery(query_data) + + mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get) + + rv = client.get('/nothere/radiator') + + assert rv.status_code == 404 + soup = BeautifulSoup(rv.data, 'html.parser') + assert soup.title.contents[0] == 'Puppetboard' + assert soup.h1.text == 'Not Found' + + +def test_radiator_view_division_by_zero(client, mocker): + mock_puppetdb_environments(mocker) + mock_puppetdb_default_nodes(mocker) + + query_data = { + 'nodes': [[{'count': 0}]], + 'resources': [[{'count': 40}]], + } + + dbquery = MockDbQuery(query_data) + + mocker.patch.object(app.puppetdb, '_query', side_effect=dbquery.get) + + rv = client.get('/radiator') + + assert rv.status_code == 200 + + soup = BeautifulSoup(rv.data, 'html.parser') + assert soup.title.contents[0] == 'Puppetboard' + + total = soup.find(class_='total') + assert '0' in total.text diff --git a/test/test_app_error.py b/test/test_app_error.py new file mode 100644 index 0000000..7b20d84 --- /dev/null +++ b/test/test_app_error.py @@ -0,0 +1,73 @@ +import pytest +from flask import Flask, current_app +from puppetboard import app + +from bs4 import BeautifulSoup + + +@pytest.fixture +def mock_puppetdb_environments(mocker): + environemnts = [ + {'name': 'production'}, + {'name': 'staging'} + ] + + return mocker.patch.object(app.puppetdb, 'environments', + return_value=environemnts) + + +def test_error_no_content(): + result = app.no_content(None) + assert result[0] == '' + assert result[1] == 204 + + +def test_error_bad_request(mock_puppetdb_environments): + with app.app.test_request_context(): + (output, error_code) = app.bad_request(None) + soup = BeautifulSoup(output, 'html.parser') + + assert 'The request sent to PuppetDB was invalid' in soup.p.text + assert error_code == 400 + + +def test_error_forbidden(mock_puppetdb_environments): + with app.app.test_request_context(): + (output, error_code) = app.forbidden(None) + soup = BeautifulSoup(output, 'html.parser') + + long_string = "%s %s" % ('What you were looking for has', + 'been disabled by the administrator') + assert long_string in soup.p.text + assert error_code == 403 + + +def test_error_not_found(mock_puppetdb_environments): + with app.app.test_request_context(): + (output, error_code) = app.not_found(None) + soup = BeautifulSoup(output, 'html.parser') + + long_string = "%s %s" % ('What you were looking for could not', + 'be found in PuppetDB.') + assert long_string in soup.p.text + assert error_code == 404 + + +def test_error_precond(mock_puppetdb_environments): + with app.app.test_request_context(): + (output, error_code) = app.precond_failed(None) + soup = BeautifulSoup(output, 'html.parser') + + long_string = "%s %s" % ('You\'ve configured Puppetboard with an API', + 'version that does not support this feature.') + assert long_string in soup.p.text + assert error_code == 412 + + +def test_error_server(mock_puppetdb_environments): + with app.app.test_request_context(): + (output, error_code) = app.server_error(None) + soup = BeautifulSoup(output, 'html.parser') + + assert 'Internal Server Error' in soup.h2.text + assert error_code == 500 diff --git a/test/test_docker_settings.py b/test/test_docker_settings.py index b9dbb61..a172ae9 100644 --- a/test/test_docker_settings.py +++ b/test/test_docker_settings.py @@ -1,7 +1,8 @@ +import pytest import os from puppetboard import docker_settings -import unittest -import tempfile +from puppetboard import app + try: import future.utils except: @@ -13,80 +14,95 @@ except: pass -class DockerTestCase(unittest.TestCase): - def setUp(self): - for env_var in dir(docker_settings): - if (env_var.startswith('__') or env_var.startswith('_') or - env_var.islower()): - continue +@pytest.fixture(scope='function') +def cleanUpEnv(request): + for env_var in dir(docker_settings): + if (env_var.startswith('__') or env_var.startswith('_') or + env_var.islower()): + continue - if env_var in os.environ: - del os.environ[env_var] - reload(docker_settings) - - def test_default_host_port(self): - self.assertEqual(docker_settings.PUPPETDB_HOST, 'puppetdb') - self.assertEqual(docker_settings.PUPPETDB_PORT, 8080) - - def test_set_host_port(self): - os.environ['PUPPETDB_HOST'] = 'puppetdb' - os.environ['PUPPETDB_PORT'] = '9081' - reload(docker_settings) - self.assertEqual(docker_settings.PUPPETDB_HOST, 'puppetdb') - self.assertEqual(docker_settings.PUPPETDB_PORT, 9081) - - def test_cert_true_test(self): - os.environ['PUPPETDB_SSL_VERIFY'] = 'True' - reload(docker_settings) - self.assertTrue(docker_settings.PUPPETDB_SSL_VERIFY) - os.environ['PUPPETDB_SSL_VERIFY'] = 'true' - reload(docker_settings) - self.assertTrue(docker_settings.PUPPETDB_SSL_VERIFY) - - def test_cert_false_test(self): - os.environ['PUPPETDB_SSL_VERIFY'] = 'False' - reload(docker_settings) - self.assertFalse(docker_settings.PUPPETDB_SSL_VERIFY) - os.environ['PUPPETDB_SSL_VERIFY'] = 'false' - reload(docker_settings) - self.assertFalse(docker_settings.PUPPETDB_SSL_VERIFY) - - def test_cert_path(self): - ca_file = '/usr/ssl/path/ca.pem' - os.environ['PUPPETDB_SSL_VERIFY'] = ca_file - reload(docker_settings) - self.assertEqual(docker_settings.PUPPETDB_SSL_VERIFY, ca_file) - - def validate_facts(self, facts): - self.assertEqual(type(facts), type([])) - self.assertTrue(len(facts) > 0) - for map in facts: - self.assertEqual(type(map), type(())) - self.assertTrue(len(map) == 2) - - def test_inventory_facts_default(self): - self.validate_facts(docker_settings.INVENTORY_FACTS) - - def test_invtory_facts_custom(self): - os.environ['INVENTORY_FACTS'] = "A, B, C, D" - reload(docker_settings) - self.validate_facts(docker_settings.INVENTORY_FACTS) - - def test_graph_facts_defautl(self): - facts = docker_settings.GRAPH_FACTS - self.assertEqual(type(facts), type([])) - self.assertTrue('puppetversion' in facts) - - def test_graph_facts_custom(self): - os.environ['GRAPH_FACTS'] = "architecture, puppetversion, extra" - reload(docker_settings) - facts = docker_settings.GRAPH_FACTS - self.assertEqual(type(facts), type([])) - self.assertEqual(len(facts), 3) - self.assertTrue('puppetversion' in facts) - self.assertTrue('architecture' in facts) - self.assertTrue('extra' in facts) + if env_var in os.environ: + del os.environ[env_var] + reload(docker_settings) + return -if __name__ == '__main__': - unittest.main() +def test_default_host_port(cleanUpEnv): + assert docker_settings.PUPPETDB_HOST == 'puppetdb' + assert docker_settings.PUPPETDB_PORT == 8080 + + +def test_set_host_port(cleanUpEnv): + os.environ['PUPPETDB_HOST'] = 'puppetdb2' + os.environ['PUPPETDB_PORT'] = '9081' + reload(docker_settings) + assert docker_settings.PUPPETDB_HOST == 'puppetdb2' + assert docker_settings.PUPPETDB_PORT == 9081 + + +def test_cert_true_test(cleanUpEnv): + os.environ['PUPPETDB_SSL_VERIFY'] = 'True' + reload(docker_settings) + assert docker_settings.PUPPETDB_SSL_VERIFY is True + os.environ['PUPPETDB_SSL_VERIFY'] = 'true' + reload(docker_settings) + assert docker_settings.PUPPETDB_SSL_VERIFY is True + + +def test_cert_false_test(cleanUpEnv): + os.environ['PUPPETDB_SSL_VERIFY'] = 'False' + reload(docker_settings) + assert docker_settings.PUPPETDB_SSL_VERIFY is False + os.environ['PUPPETDB_SSL_VERIFY'] = 'false' + reload(docker_settings) + assert docker_settings.PUPPETDB_SSL_VERIFY is False + + +def test_cert_path(cleanUpEnv): + ca_file = '/usr/ssl/path/ca.pem' + os.environ['PUPPETDB_SSL_VERIFY'] = ca_file + reload(docker_settings) + assert docker_settings.PUPPETDB_SSL_VERIFY == ca_file + + +def validate_facts(facts): + assert isinstance(facts, list) + assert len(facts) > 0 + for map in facts: + assert isinstance(map, tuple) + assert len(map) == 2 + + +def test_inventory_facts_default(cleanUpEnv): + validate_facts(docker_settings.INVENTORY_FACTS) + + +def test_invtory_facts_custom(cleanUpEnv): + os.environ['INVENTORY_FACTS'] = "A, B, C, D" + reload(docker_settings) + validate_facts(docker_settings.INVENTORY_FACTS) + + +def test_graph_facts_defautl(cleanUpEnv): + facts = docker_settings.GRAPH_FACTS + assert isinstance(facts, list) + assert 'puppetversion' in facts + + +def test_graph_facts_custom(cleanUpEnv): + os.environ['GRAPH_FACTS'] = "architecture, puppetversion, extra" + reload(docker_settings) + facts = docker_settings.GRAPH_FACTS + assert isinstance(facts, list) + assert len(facts) == 3 + assert 'puppetversion' in facts + assert 'architecture' in facts + assert 'extra' in facts + + +def test_bad_log_value(cleanUpEnv): + os.environ['LOGLEVEL'] = 'g' + os.environ['PUPPETBOARD_SETTINGS'] = '../puppetboard/docker_settings.py' + reload(docker_settings) + with pytest.raises(ValueError) as error: + reload(app) diff --git a/test/test_utils.py b/test/test_utils.py index 7a82f48..5974537 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -1,8 +1,4 @@ -try: - import unittest2 as unittest -except ImportError: - import unittest - +import pytest import sys import json import mock @@ -18,141 +14,143 @@ from puppetboard import utils from puppetboard import app from puppetboard.app import NoContent - +from bs4 import BeautifulSoup import logging -class UtilsTestCase(unittest.TestCase): - def setUp(self): - pass +def test_json_format(): + demo = [{'foo': 'bar'}, {'bar': 'foo'}] + sample = json.dumps(demo, indent=2, separators=(',', ': ')) - def teadDown(self): - pass - - def test_json_format(self): - demo = [{'foo': 'bar'}, {'bar': 'foo'}] - sample = json.dumps(demo, indent=2, separators=(',', ': ')) - - self.assertEqual(sample, utils.jsonprint(demo), - "Json formatting has changed") - - def test_format_val_str(self): - x = "some string" - self.assertEqual(x, utils.formatvalue(x), - "Should return same value") - - def test_format_val_array(self): - x = ['a', 'b', 'c'] - self.assertEqual("a, b, c", utils.formatvalue(x), - "Should return comma seperated string") - - def test_format_val_dict_one_layer(self): - x = {'a': 'b'} - self.assertEqual("a => b,
", utils.formatvalue(x), - "Should return stringified value") - - def test_format_val_tuple(self): - x = ('a', 'b') - self.assertEqual(str(x), utils.formatvalue(x)) + assert sample == utils.jsonprint(demo), "Json formatting has changed" -@mock.patch('logging.log') -class GetOrAbortTesting(unittest.TestCase): - - def test_get(self, mock_log): - x = "hello world" - - def test_get_or_abort(): - return x - - self.assertEqual(x, utils.get_or_abort(test_get_or_abort)) - - def test_http_error(self, mock_log): - err = "NotFound" - - def raise_http_error(): - x = Response() - x.status_code = 404 - x.reason = err - raise HTTPError(err, response=x) - - with self.assertRaises(NotFound) as error: - utils.get_or_abort(raise_http_error) - mock_log.error.assert_called_with(err) - - def test_http_connection_error(self, mock_log): - err = "ConnectionError" - - def connection_error(): - x = Response() - x.status_code = 500 - x.reason = err - raise ConnectionError(err, response=x) - - with self.assertRaises(InternalServerError) as error: - utils.get_or_abort(connection_error) - mock_log.error.assert_called_with(err) - - @mock.patch('flask.abort') - def test_http_empty(self, mock_log, flask_abort): - err = "Empty Response" - - def connection_error(): - raise EmptyResponseError(err) - - with self.assertRaises(NoContent) as error: - utils.get_or_abort(connection_error) - mock_log.error.assert_called_with(err) - flask_abort.assert_called_with('204') +def test_format_val_str(): + x = "some string" + assert x == utils.formatvalue(x), "Should return same value" -class yieldOrStop(unittest.TestCase): - - def test_iter(self): - test_list = (0, 1, 2, 3) - - def my_generator(): - for i in test_list: - yield i - - gen = utils.yield_or_stop(my_generator()) - self.assertIsInstance(gen, GeneratorType) - - i = 0 - for val in gen: - self.assertEqual(i, val) - i = i + 1 - - def test_stop_empty(self): - def my_generator(): - yield 1 - raise EmptyResponseError - yield 2 - - gen = utils.yield_or_stop(my_generator()) - for val in gen: - self.assertEqual(1, val) - - def test_stop_conn_error(self): - def my_generator(): - yield 1 - raise ConnectionError - yield 2 - - gen = utils.yield_or_stop(my_generator()) - for val in gen: - self.assertEqual(1, val) - - def test_stop_http_error(self): - def my_generator(): - yield 1 - raise HTTPError - yield 2 - - gen = utils.yield_or_stop(my_generator()) - for val in gen: - self.assertEqual(1, val) +def test_format_val_array(): + x = ['a', 'b', 'c'] + assert "a, b, c" == utils.formatvalue(x) -if __name__ == '__main__': - unittest.main() +def test_format_val_dict_one_layer(): + x = {'a': 'b'} + assert "a => b,
" == utils.formatvalue(x) + + +def test_format_val_tuple(): + x = ('a', 'b') + assert str(x) == utils.formatvalue(x) + + +def test_get(): + x = "hello world" + + def test_get_or_abort(): + return x + + assert x == utils.get_or_abort(test_get_or_abort) + + +def test_pretty_print(): + test_data = [{'hello': 'world'}] + + html = utils.prettyprint(test_data) + soup = BeautifulSoup(html, 'html.parser') + + assert soup.th.text == 'hello' + + +@pytest.fixture +def mock_log(mocker): + return mocker.patch('logging.log') + + +def test_http_error(mock_log): + err = "NotFound" + + def raise_http_error(): + x = Response() + x.status_code = 404 + x.reason = err + raise HTTPError(err, response=x) + + with pytest.raises(NotFound): + utils.get_or_abort(raise_http_error) + mock_log.error.assert_called_once_with(err) + + +def test_http_connection_error(mock_log): + err = "ConnectionError" + + def connection_error(): + x = Response() + x.status_code = 500 + x.reason = err + raise ConnectionError(err, response=x) + + with pytest.raises(InternalServerError): + utils.get_or_abort(connection_error) + mock_log.error.assert_called_with(err) + + +def test_http_empty(mock_log, mocker): + err = "Empty Response" + + def connection_error(): + raise EmptyResponseError(err) + + flask_abort = mocker.patch('flask.abort') + with pytest.raises(NoContent): + utils.get_or_abort(connection_error) + mock_log.error.assert_called_with(err) + flask_abort.assert_called_with('204') + + +def test_iter(): + test_list = (0, 1, 2, 3) + + def my_generator(): + for i in test_list: + yield i + + gen = utils.yield_or_stop(my_generator()) + assert isinstance(gen, GeneratorType) + + i = 0 + for val in gen: + assert i == val + i = i + 1 + + +def test_stop_empty(): + def my_generator(): + yield 1 + raise EmptyResponseError + yield 2 + + gen = utils.yield_or_stop(my_generator()) + for val in gen: + assert 1 == val + + +def test_stop_conn_error(): + def my_generator(): + yield 1 + raise ConnectionError + yield 2 + gen = utils.yield_or_stop(my_generator()) + for val in gen: + assert 1 == val + + +def test_stop_http_error(): + def my_generator(): + yield 1 + raise HTTPError + yield 2 + gen = utils.yield_or_stop(my_generator()) + for val in gen: + assert 1 == val
{{stats['failed']}}
{{stats['unreported']}}
{{stats['noop']}}
{{stats['changed']}}
{{stats['unchanged']}}
{{total}}