From e7860ea03b006657f42149d1d5031651b4aa69ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Simon=20Bri=C3=A8re?= Date: Fri, 1 Apr 2022 11:06:02 -0400 Subject: [PATCH] Refs #79. Added UserQueryTests API and related tests. --- .../modules/DatabaseModule/DBManager.py | 5 + .../DatabaseModule/DBManagerTeraUserAccess.py | 43 +++-- .../FlaskModule/API/user/UserQueryAssets.py | 1 - .../FlaskModule/API/user/UserQueryTestType.py | 45 ++++- .../FlaskModule/API/user/UserQueryTests.py | 123 ++++++++++++ .../python/modules/FlaskModule/FlaskModule.py | 2 + .../python/opentera/db/models/TeraSession.py | 3 +- .../python/opentera/db/models/TeraTest.py | 24 ++- .../python/opentera/db/models/TeraTestType.py | 22 +++ .../API/user/test_UserQueryTestTypes.py | 28 +++ .../API/user/test_UserQueryTests.py | 176 ++++++++++++++++++ 11 files changed, 441 insertions(+), 31 deletions(-) create mode 100644 teraserver/python/modules/FlaskModule/API/user/UserQueryTests.py create mode 100644 teraserver/python/tests/modules/FlaskModule/API/user/test_UserQueryTests.py diff --git a/teraserver/python/modules/DatabaseModule/DBManager.py b/teraserver/python/modules/DatabaseModule/DBManager.py index 23a38dffd..95a38e94f 100755 --- a/teraserver/python/modules/DatabaseModule/DBManager.py +++ b/teraserver/python/modules/DatabaseModule/DBManager.py @@ -43,6 +43,7 @@ from opentera.db.models.TeraTestType import TeraTestType from opentera.db.models.TeraTestTypeSite import TeraTestTypeSite from opentera.db.models.TeraTestTypeProject import TeraTestTypeProject +from opentera.db.models.TeraTest import TeraTest from opentera.config.ConfigManager import ConfigManager from modules.FlaskModule.FlaskModule import flask_app @@ -261,6 +262,10 @@ def create_defaults(self, config: ConfigManager, test=False): TeraTestTypeSite.create_defaults(test) TeraTestTypeProject.create_defaults(test) + if TeraTest.get_count() == 0: + print('No test - creating defaults') + TeraTest.create_defaults(test) + def setup_events(self): # TODO Add events that need to be sent through redis # TODO Useful to specify event name, always get_model_name() ? diff --git a/teraserver/python/modules/DatabaseModule/DBManagerTeraUserAccess.py b/teraserver/python/modules/DatabaseModule/DBManagerTeraUserAccess.py index 33ff3d5e1..4e0b57617 100644 --- a/teraserver/python/modules/DatabaseModule/DBManagerTeraUserAccess.py +++ b/teraserver/python/modules/DatabaseModule/DBManagerTeraUserAccess.py @@ -1291,7 +1291,6 @@ def query_asset(self, asset_id: int = None, asset_uuid: str = None): if not asset_id and not asset_uuid: return None - if asset_id: asset: TeraAsset = TeraAsset.get_asset_by_id(asset_id) elif asset_uuid: @@ -1306,24 +1305,28 @@ def query_asset(self, asset_id: int = None, asset_uuid: str = None): if asset.asset_service_uuid not in [service.service_uuid for service in self.get_accessible_services()]: return None - # If a user has access to a session, it should have access to its assets - # session_ids = self.get_accessible_sessions_ids() - # device_ids = self.get_accessible_devices_ids() - # participant_ids = self.get_accessible_participants_ids() - # user_ids = self.get_accessible_users_ids() - # service_ids = self.get_accessible_services_ids() - # - # query = TeraAsset.query.filter(TeraAsset.id_session.in_(session_ids))\ - # .filter(or_(TeraAsset.id_service.in_(service_ids), TeraAsset.id_service == None)) - # - # if asset_id: - # query = query.filter(TeraAsset.id_asset == asset_id) - # elif asset_uuid: - # query = query.filter(TeraAsset.asset_uuid == asset_uuid) - # - # return query.all() + return [asset] - # .filter(or_(TeraAsset.id_device.in_(device_ids), TeraAsset.id_device == None)) \ - # .filter(or_(TeraAsset.id_participant.in_(participant_ids), TeraAsset.id_participant == None)) \ - # .filter(or_(TeraAsset.id_user.in_(user_ids), TeraAsset.id_user == None)) \ + def query_test(self, test_id: int = None, test_uuid: str = None): + from opentera.db.models.TeraTest import TeraTest + + if not test_id and not test_uuid: + return None + + if test_id: + test: TeraTest = TeraTest.get_test_by_id(test_id) + elif test_uuid: + asset: TeraTest = TeraTest.get_test_by_uuid(test_uuid) + else: + return None + + test_session = self.query_session(test.id_session) + if not test_session: + # No access to asset session + return None + + if test.test_test_type.id_service not in [service.id_service for service in self.get_accessible_services()]: + return None + + return test diff --git a/teraserver/python/modules/FlaskModule/API/user/UserQueryAssets.py b/teraserver/python/modules/FlaskModule/API/user/UserQueryAssets.py index 595f3cc82..0f414930d 100644 --- a/teraserver/python/modules/FlaskModule/API/user/UserQueryAssets.py +++ b/teraserver/python/modules/FlaskModule/API/user/UserQueryAssets.py @@ -127,7 +127,6 @@ def get(self): # if args['with_only_token']: # return {'access_token': access_token} - for asset in assets: if args['with_only_token']: asset_json = {'asset_uuid': asset.asset_uuid} diff --git a/teraserver/python/modules/FlaskModule/API/user/UserQueryTestType.py b/teraserver/python/modules/FlaskModule/API/user/UserQueryTestType.py index 0fc1709ba..a1fe0825a 100644 --- a/teraserver/python/modules/FlaskModule/API/user/UserQueryTestType.py +++ b/teraserver/python/modules/FlaskModule/API/user/UserQueryTestType.py @@ -2,9 +2,7 @@ from flask_restx import Resource, reqparse, inputs from modules.LoginModule.LoginModule import user_multi_auth, current_user from modules.FlaskModule.FlaskModule import user_api_ns as api -from opentera.db.models.TeraUser import TeraUser from opentera.db.models.TeraTestType import TeraTestType -from opentera.db.models.TeraTestTypeSite import TeraTestTypeSite from opentera.db.models.TeraServiceSite import TeraServiceSite from opentera.db.models.TeraTestTypeProject import TeraTestTypeProject from modules.DatabaseModule.DBManager import DBManager @@ -12,12 +10,17 @@ from sqlalchemy import exc from flask_babel import gettext +from opentera.redis.RedisVars import RedisVars + # Parser definition(s) get_parser = api.parser() get_parser.add_argument('id_test_type', type=int, help='ID of the test type to query') get_parser.add_argument('id_project', type=int, help='ID of the project to get test types for') get_parser.add_argument('id_site', type=int, help='ID of the site to get test types for') get_parser.add_argument('list', type=inputs.boolean, help='Flag that limits the returned data to minimal information') +get_parser.add_argument('with_urls', type=inputs.boolean, help='Also include test types urls') +get_parser.add_argument('with_only_token', type=inputs.boolean, help='Only includes the access token. ' + 'Will ignore with_urls if specified.') # post_parser = reqparse.RequestParser() # post_parser.add_argument('session_type', type=str, location='json', help='Session type to create / update', @@ -64,13 +67,41 @@ def get(self): try: test_types_list = [] + servername = self.module.config.server_config['hostname'] + port = self.module.config.server_config['port'] + if 'X_EXTERNALSERVER' in request.headers: + servername = request.headers['X_EXTERNALSERVER'] + + if 'X_EXTERNALPORT' in request.headers: + port = request.headers['X_EXTERNALPORT'] + for tt in test_types: - if args['list'] is None: - tt_json = tt.to_json() - test_types_list.append(tt_json) + if args['with_only_token']: + tt_json = {'test_type_uuid': tt.test_type_uuid} else: - tt_json = tt.to_json(minimal=True) - test_types_list.append(tt_json) + tt_json = tt.to_json(minimal=args['list']) + + if args['with_urls'] or args['with_only_token']: + # Access token + token_key = self.module.redisGet(RedisVars.RedisVar_ServiceTokenAPIKey) + projects_ids = [proj.id_project for proj in + TeraTestTypeProject.get_projects_for_test_type(tt.id_test_type)] + admin_projects_ids = user_access.get_accessible_projects_ids(admin_only=True) + + # Is project admin in at least one of the related project? If so, can edit the test type + is_project_admin = len(set(projects_ids).difference(admin_projects_ids)) != len(projects_ids) + + access_token = TeraTestType.get_access_token(test_type_uuids=tt.test_type_uuid, + token_key=token_key, + requester_uuid=current_user.user_uuid, + can_edit=is_project_admin, + expiration=1800) + tt_json['access_token'] = access_token + + if args['with_urls']: + tt_json.update(tt.get_service_urls(server_url=servername, server_port=port)) + + test_types_list.append(tt_json) return test_types_list diff --git a/teraserver/python/modules/FlaskModule/API/user/UserQueryTests.py b/teraserver/python/modules/FlaskModule/API/user/UserQueryTests.py new file mode 100644 index 000000000..08c521f07 --- /dev/null +++ b/teraserver/python/modules/FlaskModule/API/user/UserQueryTests.py @@ -0,0 +1,123 @@ +from flask import session, request +from flask_restx import Resource, inputs +from flask_babel import gettext +from modules.LoginModule.LoginModule import user_multi_auth, current_user +from modules.FlaskModule.FlaskModule import user_api_ns as api +from opentera.db.models.TeraTest import TeraTest +from opentera.db.models.TeraService import TeraService + +from modules.DatabaseModule.DBManager import DBManager +from opentera.redis.RedisVars import RedisVars + +# Parser definition(s) +# GET +get_parser = api.parser() +get_parser.add_argument('id_test', type=int, help='Specific ID of test to query information.') +get_parser.add_argument('test_uuid', type=str, help='Specific UUID of test to query information.') +get_parser.add_argument('id_device', type=int, help='ID of the device from which to request all tests') +get_parser.add_argument('id_session', type=int, help='ID of session from which to request all tests') +get_parser.add_argument('id_participant', type=int, help='ID of participant from which to request all tests') +get_parser.add_argument('id_user', type=int, help='ID of the user from which to request all tests.') + +get_parser.add_argument('with_urls', type=inputs.boolean, help='Also include tests results url') +get_parser.add_argument('with_only_token', type=inputs.boolean, help='Only includes the access token. ' + 'Will ignore with_urls if specified.') +get_parser.add_argument('full', type=inputs.boolean, help='Also include names of sessions, users, services, ... in the ' + 'reply') + + +class UserQueryTests(Resource): + + def __init__(self, _api, *args, **kwargs): + Resource.__init__(self, _api, *args, **kwargs) + self.module = kwargs.get('flaskModule', None) + self.test = kwargs.get('test', False) + + @user_multi_auth.login_required + @api.expect(get_parser) + @api.doc(description='Get test information. Only one of the ID parameter is supported at once', + responses={200: 'Success - returns list of assets', + 400: 'Required parameter is missing', + 403: 'Logged user doesn\'t have permission to access the requested data'}) + def get(self): + user_access = DBManager.userAccess(current_user) + args = get_parser.parse_args() + + # At least one argument required + if not any(args.values()): + return gettext('No arguments specified'), 400 + elif args['id_device']: + if args['id_device'] not in user_access.get_accessible_devices_ids(): + return gettext('Device access denied'), 403 + tests = TeraTest.get_tests_for_device(device_id=args['id_device']) + elif args['id_session']: + if not user_access.query_session(session_id=args['id_session']): + return gettext('Session access denied'), 403 + tests = TeraTest.get_tests_for_session(session_id=args['id_session']) + elif args['id_participant']: + if args['id_participant'] not in user_access.get_accessible_participants_ids(): + return gettext('Participant access denied'), 403 + tests = TeraTest.get_tests_for_participant(part_id=args['id_participant']) + elif args['id_user']: + if args['id_user'] not in user_access.get_accessible_users_ids(): + return gettext("User access denied"), 403 + tests = TeraTest.get_tests_for_user(user_id=args['id_user']) + elif args['id_test']: + tests = [user_access.query_test(test_id=args['id_test'])] + elif args['test_uuid']: + tests = [user_access.query_test(test_uuid=args['test_uuid'])] + else: + return gettext('Missing argument'), 400 + + if not tests: + return [] + + tests_list = [] + servername = self.module.config.server_config['hostname'] + port = self.module.config.server_config['port'] + if 'X_EXTERNALSERVER' in request.headers: + servername = request.headers['X_EXTERNALSERVER'] + + if 'X_EXTERNALPORT' in request.headers: + port = request.headers['X_EXTERNALPORT'] + + for test in tests: + if test is None: + continue + + if args['with_only_token']: + test_json = {'test_uuid': test.test_uuid} + else: + test_json = test.to_json(minimal=not args['full']) + + # Access token + if args['with_urls'] or args['with_only_token']: + # Access token + token_key = self.module.redisGet(RedisVars.RedisVar_ServiceTokenAPIKey) + access_token = TeraTest.get_access_token(test_uuids=test.test_uuid, + token_key=token_key, + requester_uuid=current_user.user_uuid, + expiration=1800) + test_json['access_token'] = access_token + + if args['with_urls']: + # We have previously verified that the service is available to the user + test_json.update(test.get_service_url(server_url=servername, server_port=port)) + + tests_list.append(test_json) + + return tests_list + + @user_multi_auth.login_required + @api.doc(description='Delete test.', + responses={501: 'Unable to update test from here - use service!'}) + def post(self): + return gettext('Test information update and creation must be done directly into a service (such as ' + 'Test service)'), 501 + + @user_multi_auth.login_required + @api.doc(description='Delete test.', + responses={501: 'Unable to delete test from here'}) + def delete(self): + return gettext('Test deletion must be done directly into a service (such as ' + 'Test service)'), 501 diff --git a/teraserver/python/modules/FlaskModule/FlaskModule.py b/teraserver/python/modules/FlaskModule/FlaskModule.py index 7eb687fd3..8281233b5 100755 --- a/teraserver/python/modules/FlaskModule/FlaskModule.py +++ b/teraserver/python/modules/FlaskModule/FlaskModule.py @@ -170,6 +170,7 @@ def init_user_api(self): from modules.FlaskModule.API.user.UserQueryTestTypeSites import UserQueryTestTypeSites from modules.FlaskModule.API.user.UserQueryTestTypeProjects import UserQueryTestTypeProjects from modules.FlaskModule.API.user.UserQueryTestType import UserQueryTestTypes + from modules.FlaskModule.API.user.UserQueryTests import UserQueryTests # Resources user_api_ns.add_resource(UserQueryAssets, '/assets', resource_class_kwargs=kwargs) @@ -207,6 +208,7 @@ def init_user_api(self): user_api_ns.add_resource(UserQuerySites, '/sites', resource_class_kwargs=kwargs) user_api_ns.add_resource(UserQuerySiteAccess, '/siteaccess', resource_class_kwargs=kwargs) user_api_ns.add_resource(UserQueryUserStats, '/stats', resource_class_kwargs=kwargs) + user_api_ns.add_resource(UserQueryTests, '/tests', resource_class_kwargs=kwargs) user_api_ns.add_resource(UserQueryTestTypes, '/testtypes', resource_class_kwargs=kwargs) user_api_ns.add_resource(UserQueryTestTypeProjects, '/testtypes/projects', resource_class_kwargs=kwargs) user_api_ns.add_resource(UserQueryTestTypeSites, '/testtypes/sites', resource_class_kwargs=kwargs) diff --git a/teraserver/python/opentera/db/models/TeraSession.py b/teraserver/python/opentera/db/models/TeraSession.py index a6135511e..7f2a7a8f1 100644 --- a/teraserver/python/opentera/db/models/TeraSession.py +++ b/teraserver/python/opentera/db/models/TeraSession.py @@ -48,6 +48,7 @@ class TeraSession(db.Model, BaseModel): session_session_type = db.relationship('TeraSessionType', back_populates='session_type_sessions', lazy='joined') session_events = db.relationship('TeraSessionEvent', cascade="delete", back_populates='session_event_session') session_assets = db.relationship('TeraAsset', cascade='delete', back_populates='asset_session') + session_tests = db.relationship('TeraTest', cascade='delete', back_populates='test_session') def to_json(self, ignore_fields=None, minimal=False): if ignore_fields is None: @@ -55,7 +56,7 @@ def to_json(self, ignore_fields=None, minimal=False): ignore_fields.extend(['session_participants', 'session_creator_user', 'session_creator_device', 'session_creator_participant', 'session_creator_service', 'session_session_type', - 'session_events', 'session_users', 'session_devices', 'session_assets']) + 'session_events', 'session_users', 'session_devices', 'session_assets', 'session_tests']) if minimal: ignore_fields.extend(['session_comments', 'session_parameters']) diff --git a/teraserver/python/opentera/db/models/TeraTest.py b/teraserver/python/opentera/db/models/TeraTest.py index 4811b6a4d..7c2187f86 100644 --- a/teraserver/python/opentera/db/models/TeraTest.py +++ b/teraserver/python/opentera/db/models/TeraTest.py @@ -28,7 +28,7 @@ class TeraTest(db.Model, BaseModel): test_status = db.Column(db.Integer, nullable=False, default=0) test_summary = db.Column(db.String, nullable=True) # This contains a json formatted summary for results display - test_session = db.relationship("TeraSession", back_populates='test_assets') + test_session = db.relationship("TeraSession", back_populates='session_tests') test_device = db.relationship("TeraDevice") test_user = db.relationship("TeraUser") test_participant = db.relationship("TeraParticipant") @@ -95,6 +95,7 @@ def create_defaults(test=False): new_test.test_name = "Test #" + str(i) new_test.test_session = session2 new_test.test_uuid = str(uuid.uuid4()) + new_test.test_datetime = session2.session_start_datetime if i == 0: new_test.id_test_type = pretesttype.id_test_type new_test.id_participant = TeraParticipant.get_participant_by_name('Participant #1').id_participant @@ -106,6 +107,7 @@ def create_defaults(test=False): new_test.id_device = TeraDevice.get_device_by_id(1).id_device if i == 3: new_test.test_session = session3 + new_test.test_datetime = session3.session_start_datetime new_test.id_test_type = pretesttype.id_test_type new_test.id_participant = TeraParticipant.get_participant_by_name('Participant #1').id_participant db.session.add(new_test) @@ -121,7 +123,7 @@ def get_test_by_uuid(test_uuid: str): return TeraTest.query.filter_by(test_uuid=test_uuid).first() @staticmethod - def gettests_for_device(device_id: int): + def get_tests_for_device(device_id: int): return TeraTest.query.filter_by(id_device=device_id).all() @staticmethod @@ -157,6 +159,24 @@ def get_access_token(test_uuids: list, token_key: str, requester_uuid: str, expi return jwt.encode(payload, token_key, algorithm='HS256') + def get_service_url(self, server_url: str, server_port: int) -> dict: + urls = {'test_answers_url': None, + 'test_answers_web_url': None, + } + + if not self.test_test_type.test_type_service.service_enabled: + return urls # Service disabled = no Urls! + + service_endpoint = self.test_test_type.test_type_service.service_clientendpoint + base_url = 'https://' + server_url + ':' + str(server_port) + service_endpoint + + urls['test_answers_url'] = base_url + '/api/tests/answers' + + if self.test_test_type.test_type_has_web_format: + urls['test_answers_web_url'] = base_url + '/api/tests/answers/web' + + return urls + @classmethod def insert(cls, test): # Generate UUID diff --git a/teraserver/python/opentera/db/models/TeraTestType.py b/teraserver/python/opentera/db/models/TeraTestType.py index 9232603c6..ecb1241a7 100644 --- a/teraserver/python/opentera/db/models/TeraTestType.py +++ b/teraserver/python/opentera/db/models/TeraTestType.py @@ -98,6 +98,28 @@ def get_access_token(test_type_uuids: list, token_key: str, requester_uuid: str, return jwt.encode(payload, token_key, algorithm='HS256') + def get_service_urls(self, server_url: str, server_port: int) -> dict: + urls = {'test_type_json_url': None, + 'test_type_web_url': None, + 'test_type_web_editor_url': None} + + if not self.test_type_service.service_enabled: + return urls # Service disabled = no Urls! + + service_endpoint = self.test_type_service.service_clientendpoint + base_url = 'https://' + server_url + ':' + str(server_port) + service_endpoint + + if self.test_type_has_json_format: + urls['test_type_json_url'] = base_url + '/api/testtypes/form' + + if self.test_type_has_web_format: + urls['test_type_web_url'] = base_url + '/api/testtypes/web' + + if self.test_type_has_web_editor: + urls['test_type_web_editor_url'] = base_url + '/api/testtypes/web/edit' + + return urls + @classmethod def insert(cls, test_type): # Generate UUID diff --git a/teraserver/python/tests/modules/FlaskModule/API/user/test_UserQueryTestTypes.py b/teraserver/python/tests/modules/FlaskModule/API/user/test_UserQueryTestTypes.py index fc6cd95d3..3ca5da2c4 100644 --- a/teraserver/python/tests/modules/FlaskModule/API/user/test_UserQueryTestTypes.py +++ b/teraserver/python/tests/modules/FlaskModule/API/user/test_UserQueryTestTypes.py @@ -224,6 +224,34 @@ def test_post_and_delete(self): endpoint='/api/user/services/projects', client=self.test_client) self.assertEqual(response.status_code, 200, msg='Back to default state!') + def test_query_with_urls(self): + response = self._get_with_user_http_auth(username='admin', password='admin', client=self.test_client, + params={'with_urls': True}) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.headers['Content-Type'], 'application/json') + json_data = response.json + self.assertEqual(len(json_data), 3) + + for data_item in json_data: + self._checkJson(data_item) + self.assertTrue(data_item.__contains__('test_type_json_url')) + self.assertTrue(data_item.__contains__('test_type_web_url')) + self.assertTrue(data_item.__contains__('test_type_web_editor_url')) + self.assertTrue(data_item.__contains__('access_token')) + + def test_query_access_token_only(self): + response = self._get_with_user_http_auth(username='admin', password='admin', client=self.test_client, + params={'with_only_token': True}) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.headers['Content-Type'], 'application/json') + json_data = response.json + self.assertEqual(len(json_data), 3) + + for data_item in json_data: + self.assertEqual(len(data_item), 2) + self.assertTrue(data_item.__contains__('test_type_uuid')) + self.assertTrue(data_item.__contains__('access_token')) + def _checkJson(self, json_data, minimal=False): self.assertGreater(len(json_data), 0) self.assertTrue(json_data.__contains__('id_test_type')) diff --git a/teraserver/python/tests/modules/FlaskModule/API/user/test_UserQueryTests.py b/teraserver/python/tests/modules/FlaskModule/API/user/test_UserQueryTests.py new file mode 100644 index 000000000..d510959b7 --- /dev/null +++ b/teraserver/python/tests/modules/FlaskModule/API/user/test_UserQueryTests.py @@ -0,0 +1,176 @@ +from tests.modules.FlaskModule.API.user.BaseUserAPITest import BaseUserAPITest +from modules.FlaskModule.FlaskModule import flask_app + + +class UserQueryTestsTest(BaseUserAPITest): + test_endpoint = '/api/user/tests' + + def setUp(self): + super().setUp() + from modules.FlaskModule.FlaskModule import user_api_ns + from BaseUserAPITest import FakeFlaskModule + # Setup minimal API + from modules.FlaskModule.API.user.UserQueryTests import UserQueryTests + kwargs = {'flaskModule': FakeFlaskModule(config=BaseUserAPITest.getConfig())} + user_api_ns.add_resource(UserQueryTests, '/tests', resource_class_kwargs=kwargs) + + # Create test client + self.test_client = flask_app.test_client() + + def tearDown(self): + super().tearDown() + + def test_no_auth(self): + response = self._get_with_user_http_auth(client=self.test_client) + self.assertEqual(response.status_code, 401) + + def test_post_no_auth(self): + response = self._post_with_user_http_auth(client=self.test_client) + self.assertEqual(response.status_code, 401) + + def test_delete_no_auth(self): + response = self._delete_with_user_http_auth(client=self.test_client) + self.assertEqual(response.status_code, 401) + + def test_query_no_params_as_admin(self): + response = self._get_with_user_http_auth(username='admin', password='admin', client=self.test_client) + self.assertEqual(response.status_code, 400) + + def test_query_bad_params_as_admin(self): + params = {'id_invalid': 1} + response = self._get_with_user_http_auth(username='admin', password='admin', params=params, + client=self.test_client) + self.assertEqual(response.status_code, 400) + + def test_query_device_tests_as_admin(self): + payload = {'id_device': 1, 'with_urls': True} + response = self._get_with_user_http_auth(username='admin', password='admin', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 200) + + json_data = response.json + self.assertEqual(len(json_data), 1) + + for data_item in json_data: + self._checkJson(json_data=data_item) + + def test_query_device_tests_no_access(self): + payload = {'id_device': 1} + response = self._get_with_user_http_auth(username='user4', password='user4', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 403) + + def test_query_session_tests_as_admin(self): + payload = {'id_session': 2, 'with_urls': True} + response = self._get_with_user_http_auth(username='admin', password='admin', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 200) + + json_data = response.json + self.assertTrue(len(json_data), 3) + + for data_item in json_data: + self._checkJson(json_data=data_item) + + def test_query_session_tests_no_access(self): + payload = {'id_session': 2} + response = self._get_with_user_http_auth(username='user4', password='user4', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 403) + + def test_query_participant_tests_as_admin(self): + payload = {'id_participant': 1, 'with_urls': True} + response = self._get_with_user_http_auth(username='admin', password='admin', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 200) + + json_data = response.json + self.assertTrue(len(json_data), 2) + + for data_item in json_data: + self._checkJson(json_data=data_item) + + def test_query_participant_tests_no_access(self): + payload = {'id_participant': 1} + response = self._get_with_user_http_auth(username='user4', password='user4', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 403) + + def test_query_user_tests_as_admin(self): + payload = {'id_user': 2} + response = self._get_with_user_http_auth(username='admin', password='admin', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 200) + + json_data = response.json + self.assertEqual(len(json_data), 1) + + for data_item in json_data: + self._checkJson(json_data=data_item, minimal=True) + + def test_query_user_tests_no_access(self): + payload = {'id_user': 1} + response = self._get_with_user_http_auth(username='user4', password='user4', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 403) + + def test_query_test_as_admin(self): + payload = {'id_test': 1, 'with_urls': True} + response = self._get_with_user_http_auth(username='admin', password='admin', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 200) + + json_data = response.json + self.assertTrue(len(json_data), 1) + + for data_item in json_data: + self._checkJson(json_data=data_item) + + def test_query_test_no_access(self): + payload = {'id_test': 1} + response = self._get_with_user_http_auth(username='user4', password='user4', params=payload, + client=self.test_client) + json_data = response.json + self.assertEqual(len(json_data), 0) + self.assertEqual(response.status_code, 200) + + def test_post_as_admin(self): + response = self._post_with_user_http_auth(username='admin', password='admin', + json={}, client=self.test_client) + self.assertEqual(response.status_code, 501) + + def test_delete_as_admin(self): + response = self._delete_with_user_http_auth(username='admin', password='admin', params={'id_todel': 1}, + client=self.test_client) + self.assertEqual(response.status_code, 501) + + def test_query_session_tests_as_admin_token_only(self): + payload = {'id_session': 2, 'with_urls': True, 'with_only_token': True} + response = self._get_with_user_http_auth(username='admin', password='admin', params=payload, + client=self.test_client) + self.assertEqual(response.status_code, 200) + + json_data = response.json + self.assertEqual(len(json_data), 3) + for data_item in json_data: + self.assertFalse(data_item.__contains__("test_name")) + self.assertTrue(data_item.__contains__("test_uuid")) + self.assertTrue(data_item.__contains__("access_token")) + + def _checkJson(self, json_data, minimal=False): + self.assertGreater(len(json_data), 0) + self.assertTrue(json_data.__contains__('id_test')) + self.assertTrue(json_data.__contains__('id_session')) + self.assertTrue(json_data.__contains__('id_device')) + self.assertTrue(json_data.__contains__('id_participant')) + self.assertTrue(json_data.__contains__('id_user')) + self.assertTrue(json_data.__contains__('id_service')) + self.assertTrue(json_data.__contains__('test_name')) + self.assertTrue(json_data.__contains__('test_uuid')) + self.assertTrue(json_data.__contains__('test_summary')) + self.assertTrue(json_data.__contains__('test_status')) + self.assertTrue(json_data.__contains__('test_datetime')) + if not minimal: + self.assertTrue(json_data.__contains__('test_answers_url')) + self.assertTrue(json_data.__contains__('test_answers_web_url')) + self.assertTrue(json_data.__contains__('access_token'))