Loading src/webui/requirements.in +1 −1 Original line number Diff line number Diff line flask flask-wtf flask-healthz flask-unittest grpcio grpcio-health-checking prometheus-client pytest pytest-benchmark lorem-text redis src/webui/service/main/routes.py +6 −4 Original line number Diff line number Diff line Loading @@ -6,14 +6,16 @@ from webui.service.main.forms import ContextForm main = Blueprint('main', __name__) context_client: ContextClient = ContextClient(CONTEXT_SERVICE_ADDRESS, CONTEXT_SERVICE_PORT) @main.route('/', methods=['GET', 'POST']) def home(): # flash('This is an info message', 'info') # flash('This is a danger message', 'danger') client: ContextClient = ContextClient(CONTEXT_SERVICE_ADDRESS, CONTEXT_SERVICE_PORT) client.connect() response = client.ListContextIds(Empty()) client.close() context_client.connect() response = context_client.ListContextIds(Empty()) context_client.close() context_form: ContextForm = ContextForm() context_form.context.choices.append(('', 'Select...')) for context in response.context_ids: Loading src/webui/service/service/routes.py +6 −3 Original line number Diff line number Diff line Loading @@ -6,17 +6,20 @@ from webui.proto.context_pb2 import ContextId, ServiceList, ServiceTypeEnum, Ser service = Blueprint('service', __name__, url_prefix='/service') context_client: ContextClient = ContextClient(CONTEXT_SERVICE_ADDRESS, CONTEXT_SERVICE_PORT) @service.get('/') def home(): # flash('This is an info message', 'info') # flash('This is a danger message', 'danger') client: ContextClient = ContextClient(CONTEXT_SERVICE_ADDRESS, CONTEXT_SERVICE_PORT) request: ContextId = ContextId() request.context_uuid.uuid = session['context_uuid'] service_list: ServiceList = client.ListServices(request) context_client.connect() service_list: ServiceList = context_client.ListServices(request) # print(service_list) client.close() context_client.close() return render_template('service/home.html', services=service_list.services, ste=ServiceTypeEnum, sse=ServiceStatusEnum) Loading src/webui/tests/test_unitary.py +138 −101 Original line number Diff line number Diff line import pytest # import pytest from flask_unittest import ClientTestCase from unittest import mock from flask.testing import FlaskClient from flask.app import Flask from flask.helpers import url_for # from device.client.DeviceClient import DeviceClient from webui.service import create_app from webui.proto.context_pb2 import Empty, DeviceId, DeviceList # @pytest.fixture(scope='session') # def device_client(device_service): # _client = DeviceClient(address='127.0.0.1', port=100) # yield _client # _client.close() @pytest.fixture def app(): class TestWebUI(ClientTestCase): app = create_app(use_config={'TESTING': True, 'SERVER_NAME': 'localhost.localdomain', 'SECRET_KEY': '>s&}24@{]]#k3&^5$f3#?6?h3{W@[}/7z}2pa]>{3&5%RP<)[(', 'WTF_CSRF_ENABLED': False}) yield app def setUp(self, client: FlaskClient) -> None: @pytest.fixture def client(app): with app.test_client() as client: yield client # return app.test_client() self.mocker_delete_device = mock.patch('webui.service.device.routes.device_client.DeleteDevice') self.mocker_delete_device.return_value = Empty() self.mocker_delete_device.start() self.addCleanup(self.mocker_delete_device.stop) def test_routes(app): with app.app_context(): url = url_for('main.home') url = url_for('service.home') url = url_for('device.home') self.mocker_list_devices = mock.patch('webui.service.device.routes.context_client.ListDevices') self.mocker_list_devices.return_value = DeviceList() # returns an empty list self.mocker_list_devices.start() self.addCleanup(self.mocker_list_devices.stop) def test_service_up(client): rw = client.get('/') assert rw.status_code == 200, 'Service is not up!' self.mocker_add_device = mock.patch('webui.service.device.routes.device_client.AddDevice') self.mocker_add_device.return_value = DeviceId() self.mocker_add_device.start() self.addCleanup(self.mocker_add_device.stop) def test_home_page(client): rw = client.get('/') assert rw.status_code == 200, 'Error in the home page!' assert b'Select the working context' in rw.data return super().setUp(client) def test_service_home_page(client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' rw = client.get('/service/') assert rw.status_code == 200 assert b'Services' in rw.data assert b'Add New Service' in rw.data def tearDown(self, client: FlaskClient) -> None: mock.patch.stopall() return super().tearDown(client) def test_device_home_page(client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' rw = client.get('/device/') assert rw.status_code == 200 assert b'Devices' in rw.data assert b'Add New Device' in rw.data @pytest.mark.parametrize('device_id', ( 'DEV1', 'DEV2', 'DEV3', )) def test_device_detail_page(client, device_id): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' rw = client.get(f'/device/detail/{device_id}') assert rw.status_code == 200 assert b'Device' in rw.data assert device_id in rw.data.decode() assert b'Endpoints' in rw.data, 'Missing endpoint information on the device detail page.' # assert b'Add New Device' in rw.data def test_routes(self, client): with self.app.app_context(): url_for('main.home') url_for('service.home') url_for('device.home') def test_device_add_page(client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' rw = client.get('/device/add') assert rw.status_code == 200 assert b'Add New Device' in rw.data assert b'Operational Status' in rw.data, 'Form is not correctly implemented.' assert b'Type' in rw.data, 'Form is not correctly implemented.' assert b'Configurations' in rw.data, 'Form is not correctly implemented.' assert b'Drivers' in rw.data, 'Form is not correctly implemented.' def test_device_add_action(client): def test_device_add_action_success(self, client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' DEVICE_EMU = { Loading @@ -96,18 +54,97 @@ def test_device_add_action(client): 'device_drivers': 0, 'device_endpoints': [], } with mock.patch('webui.service.device.routes.device_client.AddDevice') as mocked_add: mocked_add.return_value = DeviceId() rw = client.post('/device/add', data=DEVICE_EMU, follow_redirects=True) assert b'success' in rw.data rv = client.post('/device/add', data=DEVICE_EMU, follow_redirects=True) self.assertInResponse(b'success', rv) def test_device_delete_action(client): def test_device_delete_action(self, client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' with mock.patch('webui.service.device.routes.device_client.DeleteDevice') as mocked_delete,\ mock.patch('webui.service.device.routes.context_client.ListDevices') as mocked_list: mocked_list.return_value = DeviceList() # returns an empty list rw = client.get('/device/EMULATED/delete', follow_redirects=True) mocked_list.assert_called() mocked_delete.assert_called() assert b'success' in rw.data rv = client.get('/device/EMULATED/delete', follow_redirects=True) # mocked_list.assert_called() # mocked_delete.assert_called() self.assertInResponse(b'success', rv) def test_service_up(self, client): pass # def test_service_up(client): # rw = client.get('/') # assert rw.status_code == 200, 'Service is not up!' # def test_home_page(client): # rw = client.get('/') # assert rw.status_code == 200, 'Error in the home page!' # assert b'Select the working context' in rw.data # def test_service_home_page(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # rw = client.get('/service/') # assert rw.status_code == 200 # assert b'Services' in rw.data # assert b'Add New Service' in rw.data # def test_device_home_page(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # rw = client.get('/device/') # assert rw.status_code == 200 # assert b'Devices' in rw.data # assert b'Add New Device' in rw.data # @pytest.mark.parametrize('device_id', ( # 'DEV1', # 'DEV2', # 'DEV3', # )) # def test_device_detail_page(client, device_id): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # rw = client.get(f'/device/detail/{device_id}') # assert rw.status_code == 200 # assert b'Device' in rw.data # assert device_id in rw.data.decode() # assert b'Endpoints' in rw.data, 'Missing endpoint information on the device detail page.' # # assert b'Add New Device' in rw.data # def test_device_add_page(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # rw = client.get('/device/add') # assert rw.status_code == 200 # assert b'Add New Device' in rw.data # assert b'Operational Status' in rw.data, 'Form is not correctly implemented.' # assert b'Type' in rw.data, 'Form is not correctly implemented.' # assert b'Configurations' in rw.data, 'Form is not correctly implemented.' # assert b'Drivers' in rw.data, 'Form is not correctly implemented.' # def test_device_add_action(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # DEVICE_EMU = { # 'device_id': 'EMULATED', # 'device_type': 'emulated', # 'device_config': '', # 'operational_status': 1, # 'device_drivers': 0, # 'device_endpoints': [], # } # with mock.patch('webui.service.device.routes.device_client.AddDevice') as mocked_add: # mocked_add.return_value = DeviceId() # rw = client.post('/device/add', data=DEVICE_EMU, follow_redirects=True) # assert b'success' in rw.data # def test_device_delete_action(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # with mock.patch('webui.service.device.routes.device_client.DeleteDevice') as mocked_delete,\ # mock.patch('webui.service.device.routes.context_client.ListDevices') as mocked_list: # mocked_list.return_value = DeviceList() # returns an empty list # rw = client.get('/device/EMULATED/delete', follow_redirects=True) # mocked_list.assert_called() # mocked_delete.assert_called() # assert b'success' in rw.data Loading
src/webui/requirements.in +1 −1 Original line number Diff line number Diff line flask flask-wtf flask-healthz flask-unittest grpcio grpcio-health-checking prometheus-client pytest pytest-benchmark lorem-text redis
src/webui/service/main/routes.py +6 −4 Original line number Diff line number Diff line Loading @@ -6,14 +6,16 @@ from webui.service.main.forms import ContextForm main = Blueprint('main', __name__) context_client: ContextClient = ContextClient(CONTEXT_SERVICE_ADDRESS, CONTEXT_SERVICE_PORT) @main.route('/', methods=['GET', 'POST']) def home(): # flash('This is an info message', 'info') # flash('This is a danger message', 'danger') client: ContextClient = ContextClient(CONTEXT_SERVICE_ADDRESS, CONTEXT_SERVICE_PORT) client.connect() response = client.ListContextIds(Empty()) client.close() context_client.connect() response = context_client.ListContextIds(Empty()) context_client.close() context_form: ContextForm = ContextForm() context_form.context.choices.append(('', 'Select...')) for context in response.context_ids: Loading
src/webui/service/service/routes.py +6 −3 Original line number Diff line number Diff line Loading @@ -6,17 +6,20 @@ from webui.proto.context_pb2 import ContextId, ServiceList, ServiceTypeEnum, Ser service = Blueprint('service', __name__, url_prefix='/service') context_client: ContextClient = ContextClient(CONTEXT_SERVICE_ADDRESS, CONTEXT_SERVICE_PORT) @service.get('/') def home(): # flash('This is an info message', 'info') # flash('This is a danger message', 'danger') client: ContextClient = ContextClient(CONTEXT_SERVICE_ADDRESS, CONTEXT_SERVICE_PORT) request: ContextId = ContextId() request.context_uuid.uuid = session['context_uuid'] service_list: ServiceList = client.ListServices(request) context_client.connect() service_list: ServiceList = context_client.ListServices(request) # print(service_list) client.close() context_client.close() return render_template('service/home.html', services=service_list.services, ste=ServiceTypeEnum, sse=ServiceStatusEnum) Loading
src/webui/tests/test_unitary.py +138 −101 Original line number Diff line number Diff line import pytest # import pytest from flask_unittest import ClientTestCase from unittest import mock from flask.testing import FlaskClient from flask.app import Flask from flask.helpers import url_for # from device.client.DeviceClient import DeviceClient from webui.service import create_app from webui.proto.context_pb2 import Empty, DeviceId, DeviceList # @pytest.fixture(scope='session') # def device_client(device_service): # _client = DeviceClient(address='127.0.0.1', port=100) # yield _client # _client.close() @pytest.fixture def app(): class TestWebUI(ClientTestCase): app = create_app(use_config={'TESTING': True, 'SERVER_NAME': 'localhost.localdomain', 'SECRET_KEY': '>s&}24@{]]#k3&^5$f3#?6?h3{W@[}/7z}2pa]>{3&5%RP<)[(', 'WTF_CSRF_ENABLED': False}) yield app def setUp(self, client: FlaskClient) -> None: @pytest.fixture def client(app): with app.test_client() as client: yield client # return app.test_client() self.mocker_delete_device = mock.patch('webui.service.device.routes.device_client.DeleteDevice') self.mocker_delete_device.return_value = Empty() self.mocker_delete_device.start() self.addCleanup(self.mocker_delete_device.stop) def test_routes(app): with app.app_context(): url = url_for('main.home') url = url_for('service.home') url = url_for('device.home') self.mocker_list_devices = mock.patch('webui.service.device.routes.context_client.ListDevices') self.mocker_list_devices.return_value = DeviceList() # returns an empty list self.mocker_list_devices.start() self.addCleanup(self.mocker_list_devices.stop) def test_service_up(client): rw = client.get('/') assert rw.status_code == 200, 'Service is not up!' self.mocker_add_device = mock.patch('webui.service.device.routes.device_client.AddDevice') self.mocker_add_device.return_value = DeviceId() self.mocker_add_device.start() self.addCleanup(self.mocker_add_device.stop) def test_home_page(client): rw = client.get('/') assert rw.status_code == 200, 'Error in the home page!' assert b'Select the working context' in rw.data return super().setUp(client) def test_service_home_page(client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' rw = client.get('/service/') assert rw.status_code == 200 assert b'Services' in rw.data assert b'Add New Service' in rw.data def tearDown(self, client: FlaskClient) -> None: mock.patch.stopall() return super().tearDown(client) def test_device_home_page(client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' rw = client.get('/device/') assert rw.status_code == 200 assert b'Devices' in rw.data assert b'Add New Device' in rw.data @pytest.mark.parametrize('device_id', ( 'DEV1', 'DEV2', 'DEV3', )) def test_device_detail_page(client, device_id): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' rw = client.get(f'/device/detail/{device_id}') assert rw.status_code == 200 assert b'Device' in rw.data assert device_id in rw.data.decode() assert b'Endpoints' in rw.data, 'Missing endpoint information on the device detail page.' # assert b'Add New Device' in rw.data def test_routes(self, client): with self.app.app_context(): url_for('main.home') url_for('service.home') url_for('device.home') def test_device_add_page(client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' rw = client.get('/device/add') assert rw.status_code == 200 assert b'Add New Device' in rw.data assert b'Operational Status' in rw.data, 'Form is not correctly implemented.' assert b'Type' in rw.data, 'Form is not correctly implemented.' assert b'Configurations' in rw.data, 'Form is not correctly implemented.' assert b'Drivers' in rw.data, 'Form is not correctly implemented.' def test_device_add_action(client): def test_device_add_action_success(self, client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' DEVICE_EMU = { Loading @@ -96,18 +54,97 @@ def test_device_add_action(client): 'device_drivers': 0, 'device_endpoints': [], } with mock.patch('webui.service.device.routes.device_client.AddDevice') as mocked_add: mocked_add.return_value = DeviceId() rw = client.post('/device/add', data=DEVICE_EMU, follow_redirects=True) assert b'success' in rw.data rv = client.post('/device/add', data=DEVICE_EMU, follow_redirects=True) self.assertInResponse(b'success', rv) def test_device_delete_action(client): def test_device_delete_action(self, client): with client.session_transaction() as sess: sess['context_uuid'] = 'admin' with mock.patch('webui.service.device.routes.device_client.DeleteDevice') as mocked_delete,\ mock.patch('webui.service.device.routes.context_client.ListDevices') as mocked_list: mocked_list.return_value = DeviceList() # returns an empty list rw = client.get('/device/EMULATED/delete', follow_redirects=True) mocked_list.assert_called() mocked_delete.assert_called() assert b'success' in rw.data rv = client.get('/device/EMULATED/delete', follow_redirects=True) # mocked_list.assert_called() # mocked_delete.assert_called() self.assertInResponse(b'success', rv) def test_service_up(self, client): pass # def test_service_up(client): # rw = client.get('/') # assert rw.status_code == 200, 'Service is not up!' # def test_home_page(client): # rw = client.get('/') # assert rw.status_code == 200, 'Error in the home page!' # assert b'Select the working context' in rw.data # def test_service_home_page(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # rw = client.get('/service/') # assert rw.status_code == 200 # assert b'Services' in rw.data # assert b'Add New Service' in rw.data # def test_device_home_page(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # rw = client.get('/device/') # assert rw.status_code == 200 # assert b'Devices' in rw.data # assert b'Add New Device' in rw.data # @pytest.mark.parametrize('device_id', ( # 'DEV1', # 'DEV2', # 'DEV3', # )) # def test_device_detail_page(client, device_id): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # rw = client.get(f'/device/detail/{device_id}') # assert rw.status_code == 200 # assert b'Device' in rw.data # assert device_id in rw.data.decode() # assert b'Endpoints' in rw.data, 'Missing endpoint information on the device detail page.' # # assert b'Add New Device' in rw.data # def test_device_add_page(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # rw = client.get('/device/add') # assert rw.status_code == 200 # assert b'Add New Device' in rw.data # assert b'Operational Status' in rw.data, 'Form is not correctly implemented.' # assert b'Type' in rw.data, 'Form is not correctly implemented.' # assert b'Configurations' in rw.data, 'Form is not correctly implemented.' # assert b'Drivers' in rw.data, 'Form is not correctly implemented.' # def test_device_add_action(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # DEVICE_EMU = { # 'device_id': 'EMULATED', # 'device_type': 'emulated', # 'device_config': '', # 'operational_status': 1, # 'device_drivers': 0, # 'device_endpoints': [], # } # with mock.patch('webui.service.device.routes.device_client.AddDevice') as mocked_add: # mocked_add.return_value = DeviceId() # rw = client.post('/device/add', data=DEVICE_EMU, follow_redirects=True) # assert b'success' in rw.data # def test_device_delete_action(client): # with client.session_transaction() as sess: # sess['context_uuid'] = 'admin' # with mock.patch('webui.service.device.routes.device_client.DeleteDevice') as mocked_delete,\ # mock.patch('webui.service.device.routes.context_client.ListDevices') as mocked_list: # mocked_list.return_value = DeviceList() # returns an empty list # rw = client.get('/device/EMULATED/delete', follow_redirects=True) # mocked_list.assert_called() # mocked_delete.assert_called() # assert b'success' in rw.data