diff --git a/puppetboard/app.py b/puppetboard/app.py
index b789959..8fddc6c 100644
--- a/puppetboard/app.py
+++ b/puppetboard/app.py
@@ -25,6 +25,8 @@ from puppetboard.utils import (
jsonprint, prettyprint, Pagination
)
+import werkzeug.exceptions as ex
+
DEFAULT_ORDER_BY = '[{"field": "start_time", "order": "desc"}]'
app = Flask(__name__)
@@ -93,6 +95,22 @@ def utility_processor():
return dict(now=now)
+#
+# 204 doesn't have a mapping in werkzeug, we need to define a custom
+# class and then set it to the mappings.
+#
+class NoContent(ex.HTTPException):
+ code = 204
+ description = '
No content
b,
", utils.formatvalue(x),
+ "Should return stringified value")
+
+ def test_format_val_tuple(self):
+ x = ('a', 'b')
+ self.assertEqual(str(x), utils.formatvalue(x))
+
+
+@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')
+
+
+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)
+
+
+if __name__ == '__main__':
+ unittest.main()