Refs #79. Added Service API for test types and tests

teraserver/python/modules/DatabaseModule/DBManagerTeraServiceAccess.py

@@ -155,6 +155,17 @@ def get_accessible_sessions_types(self):
     def get_accessible_sessions_types_ids(self):
         return [st.id_session_type for st in self.get_accessible_sessions_types()]
 
+    def get_accessible_tests_types(self):
+        from opentera.db.models.TeraTestType import TeraTestType
+
+        query = TeraTestType.query.filter(TeraTestType.id_service == self.service.id_service)\
+            .order_by(TeraTestType.test_type_name.asc())
+
+        return query.all()
+
+    def get_accessible_tests_types_ids(self):
+        return [tt.id_test_type for tt in self.get_accessible_tests_types()]
+
     def get_site_role(self, site_id: int, uuid_user: str):
         user = self.get_user_with_uuid(uuid_user)
         sites_roles = user.get_sites_roles()
@@ -205,3 +216,42 @@ def query_asset(self, asset_id: int):
         return TeraAsset.query.filter(TeraAsset.id_session.in_(session_ids)).filter(TeraAsset.id_asset == asset_id)\
             .all()
         # .filter(or_(TeraAsset.id_service.in_(service_ids), TeraAsset.id_service == 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
+
+        test = None
+        if test_id:
+            test = TeraTest.get_test_by_id(test_id)
+        elif test_uuid:
+            test = TeraTest.get_test_by_uuid(test_uuid)
+
+        if not test:
+            return None
+
+        test_session = self.query_session(test.id_session)
+        if not test_session:
+            # No access to asset session
+            return None
+
+        return test
+
+    def query_session(self, session_id: int):
+        from opentera.db.models.TeraSession import TeraSession
+
+        session = TeraSession.get_session_by_id(session_id)
+
+        if session:
+            # Check if we are the creator of that session
+            if session.id_creator_service == self.service.id_service:
+                return session
+
+            # Check if we have access to the project of that session
+            accessible_projects = self.get_accessible_projects_ids()
+            if session.get_associated_project_id() in accessible_projects:
+                return session
+
+        return None

teraserver/python/modules/DatabaseModule/DBManagerTeraUserAccess.py

@@ -1317,7 +1317,7 @@ def query_test(self, test_id: int = None, test_uuid: str = 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)
+            test: TeraTest = TeraTest.get_test_by_uuid(test_uuid)
         else:
             return None
 

teraserver/python/modules/FlaskModule/API/service/ServiceQueryAssets.py

@@ -110,15 +110,6 @@ def get(self):
             services_infos = {service.service_uuid: service.service_clientendpoint
                               for service in TeraService.query.filter(TeraService.service_enabled == True).all()}
 
-            # Access token
-            # from opentera.redis.RedisVars import RedisVars
-            # token_key = self.module.redisGet(RedisVars.RedisVar_ServiceTokenAPIKey)
-            # access_token = TeraAsset.get_access_token(asset_uuids=[asset.asset_uuid for asset in assets],
-            #                                           token_key=token_key, requester_uuid=current_service.service_uuid,
-            #                                           expiration=1800)
-            # 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}

teraserver/python/modules/FlaskModule/API/service/ServiceQuerySessionTypes.py

@@ -3,9 +3,15 @@
 from modules.LoginModule.LoginModule import LoginModule, current_service
 from modules.FlaskModule.FlaskModule import service_api_ns as api
 from modules.DatabaseModule.DBManager import DBManager
+from opentera.db.models.TeraSessionTypeProject import TeraSessionTypeProject
+from opentera.db.models.TeraSessionTypeSite import TeraSessionTypeSite
+from opentera.db.models.TeraParticipant import TeraParticipant
 
 # Parser definition(s)
 get_parser = api.parser()
+get_parser.add_argument('id_site', type=int, help='ID of the site to query session types for')
+get_parser.add_argument('id_project', type=int, help='ID of the project to query session types for')
+get_parser.add_argument('id_participant', type=int, help='ID of the participant to query types for')
 
 
 class ServiceQuerySessionTypes(Resource):
@@ -24,8 +30,24 @@ def __init__(self, _api, *args, **kwargs):
                         403: 'Logged user doesn\'t have permission to access the requested data'})
     def get(self):
         parser = get_parser
-        args = parser.parse_args(strict=True)
-
+        args = parser.parse_args()
         service_access = DBManager.serviceAccess(current_service)
 
-        return [st.to_json() for st in service_access.get_accessible_sessions_types()]
+        session_types = []
+        if args['id_site']:
+            if args['id_site'] in service_access.get_accessibles_sites_ids():
+                session_types = [st.session_type_site_session_type for st in
+                                 TeraSessionTypeSite.get_sessions_types_for_site(args['id_site'])]
+        elif args['id_project']:
+            if args['id_project'] in service_access.get_accessible_projects_ids():
+                session_types = [st.session_type_project_session_type for st in
+                                 TeraSessionTypeProject.get_sessions_types_for_project(args['id_project'])]
+        elif args['id_participant']:
+            if args['id_participant'] in service_access.get_accessible_participants_ids():
+                part_info = TeraParticipant.get_participant_by_id(args['id_participant'])
+                session_types = [st.session_type_project_session_type for st in
+                                 TeraSessionTypeProject.get_sessions_types_for_project(part_info.id_project)]
+        else:
+            session_types = service_access.get_accessible_sessions_types()
+
+        return [st.to_json() for st in session_types]

teraserver/python/modules/FlaskModule/API/service/ServiceQueryTestTypes.py

@@ -0,0 +1,55 @@
+from flask_restx import Resource
+from flask_babel import gettext
+from modules.LoginModule.LoginModule import LoginModule, current_service
+from modules.FlaskModule.FlaskModule import service_api_ns as api
+from modules.DatabaseModule.DBManager import DBManager
+from opentera.db.models.TeraTestTypeProject import TeraTestTypeProject
+from opentera.db.models.TeraTestTypeSite import TeraTestTypeSite
+from opentera.db.models.TeraParticipant import TeraParticipant
+
+
+# Parser definition(s)
+get_parser = api.parser()
+get_parser.add_argument('id_site', type=int, help='ID of the site to query test types for')
+get_parser.add_argument('id_project', type=int, help='ID of the project to query test types for')
+get_parser.add_argument('id_participant', type=int, help='ID of the participant to query types for')
+
+
+class ServiceQueryTestTypes(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)
+
+    @LoginModule.service_token_or_certificate_required
+    @api.expect(get_parser)
+    @api.doc(description='Return test types information for the current service',
+             responses={200: 'Success',
+                        500: 'Required parameter is missing',
+                        501: 'Not implemented.',
+                        403: 'Logged user doesn\'t have permission to access the requested data'})
+    def get(self):
+        parser = get_parser
+        args = parser.parse_args(strict=True)
+
+        service_access = DBManager.serviceAccess(current_service)
+
+        test_types = []
+        if args['id_site']:
+            if args['id_site'] in service_access.get_accessibles_sites_ids():
+                test_types = [tt.test_type_site_test_type for tt in
+                              TeraTestTypeSite.get_tests_types_for_site(args['id_site'])]
+        elif args['id_project']:
+            if args['id_project'] in service_access.get_accessible_projects_ids():
+                test_types = [tt.test_type_project_test_type for tt in
+                              TeraTestTypeProject.get_tests_types_for_project(args['id_project'])]
+        elif args['id_participant']:
+            if args['id_participant'] in service_access.get_accessible_participants_ids():
+                part_info = TeraParticipant.get_participant_by_id(args['id_participant'])
+                test_types = [tt.test_type_project_test_type for tt in
+                              TeraTestTypeProject.get_tests_types_for_project(part_info.id_project)]
+        else:
+            test_types = service_access.get_accessible_tests_types()
+
+        return [tt.to_json() for tt in test_types]