Skip to content
Snippets Groups Projects
Commit 2454e9e9 authored by Nikolay Stanchev's avatar Nikolay Stanchev
Browse files

Aggregator configuration API methods

parent 62124b3c
No related branches found
No related tags found
No related merge requests found
......@@ -8,4 +8,6 @@ ubuntu-xenial-16.04-cloudimg-console.log
.idea/
*.egg
*.pyc
.pytest_cache
.tox
*$py.class
from pyramid.config import Configurator
from pyramid.settings import asbool
from CLMCservice.views import AggregatorConfig
def main(global_config, **settings):
""" This function returns a Pyramid WSGI application."""
# a conversion is necessary so that the configuration value of the aggregator is stored as bool and not as string
aggregator_running = asbool(settings.get('aggregator_running', 'false'))
settings['aggregator_running'] = aggregator_running
config = Configurator(settings=settings)
config.add_route('home', '/')
config.add_route('aggregator', '/aggregator')
config.add_view(AggregatorConfig, attr='get', request_method='GET')
config.add_view(AggregatorConfig, attr='post', request_method='POST')
config.scan()
return config.make_wsgi_app()
# Specific tests related to the CLMC service
\ No newline at end of file
import pytest
from pyramid import testing
from pyramid.httpexceptions import HTTPBadRequest
class TestAggregatorConfig(object):
"""
A pytest-implementation test for the aggregator configuration API calls
"""
@pytest.fixture(autouse=True)
def app_config(self):
"""
A fixture to implement setUp/tearDown functionality for all tests by initializing configuration structure for the web service
"""
self.config = testing.setUp()
self.config.add_settings({'aggregator_running': False})
yield
testing.tearDown()
def test_GET(self):
"""
Tests the GET method for the status of the aggregator.
"""
from CLMCservice.views import AggregatorConfig # nested import so that importing the class view is part of the test itself
assert not self.config.get_settings().get('aggregator_running'), "Initially aggregator is not running."
request = testing.DummyRequest()
response = AggregatorConfig(request).get()
assert type(response) == dict, "Response must be a dictionary representing a JSON object."
assert not response.get('aggregator_running'), "The response of the API call must return the aggregator status being set as False"
assert not self.config.get_settings().get('aggregator_running'), "A GET request must not modify the aggregator status."
@pytest.mark.parametrize("input_val, output_val", [
("True", True),
("true", True),
("1", True),
("False", False),
("false", False),
("0", False),
("t", None),
("f", None),
])
def test_POST(self, input_val, output_val):
"""
Tests the POST method for the status of the aggregator
:param input_val: the input form parameter
:param output_val: the expected output value, None for expecting an Exception
"""
from CLMCservice.views import AggregatorConfig # nested import so that importing the class view is part of the test itself
assert not self.config.get_settings().get('aggregator_running'), "Initially aggregator is not running."
request = testing.DummyRequest()
request.params['running'] = input_val
if output_val is not None:
response = AggregatorConfig(request).post()
assert response == {'aggregator_running': output_val}, "Response of POST request must include the new status of the aggregator"
assert self.config.get_settings().get('aggregator_running') == output_val, "Aggregator status must be updated to running."
else:
error_raised = False
try:
AggregatorConfig(request).post()
except HTTPBadRequest:
error_raised = True
assert error_raised, "Error must be raised in case of an invalid argument."
def str_to_bool(value):
"""
A utility function to convert a string to boolean based on simple rules.
:param value: the value to convert
:return: True or False
:raises ValueError: if value cannot be converted to boolean
"""
if type(value) is not str:
raise ValueError("This method only converts string to booolean.")
if value in ('False', 'false', '0'):
return False
elif value in ('True', 'true', '1'):
return True
else:
raise ValueError("Invalid argument for conversion")
from pyramid.view import view_config
from pyramid.response import Response
from pyramid.view import view_defaults
from pyramid.httpexceptions import HTTPBadRequest
from CLMCservice.utilities import str_to_bool
@view_config(route_name='home')
def my_view(request):
return Response("Hello world")
@view_defaults(route_name='aggregator', renderer='json')
class AggregatorConfig(object):
"""
A class-based view for accessing and mutating the status of the aggregator.
"""
def __init__(self, request):
"""
Initialises the instance of the view with the request argument.
:param request: client's call request
"""
self.request = request
def get(self):
"""
A GET API call for the status of the aggregator.
:return: A JSON response with the status of the aggregator.
"""
aggregator_running = self.request.registry.settings.get('aggregator_running')
return {'aggregator_running': aggregator_running}
def post(self):
"""
A POST API call for the status of the aggregator.
:return: A JSON response to the POST call (success or fail).
:raises HTTPBadRequest: if form argument cannot be converted to boolean
"""
new_status = self.request.params.get('running')
try:
new_status = str_to_bool(new_status)
except ValueError:
raise HTTPBadRequest("Bad request parameter - expected a boolean, received {0}".format(self.request.params.get('running')))
self.request.registry.settings['aggregator_running'] = new_status
# TODO start/stop aggregator based on value of new status
return {'aggregator_running': new_status}
......@@ -11,6 +11,7 @@ pyramid.debug_authorization = false
pyramid.debug_notfound = false
pyramid.debug_routematch = false
pyramid.default_locale_name = en
aggregator_running = false
###
# wsgi server configuration
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment