Refs #258. Work in progress on tests for user sessiontype-service API

teraserver/python/modules/FlaskModule/API/user/UserQuerySessionTypeServices.py

@@ -105,7 +105,118 @@ def post(self):
         """
         Create / update session types - service additional association
         """
-        return 501
+        user_access = DBManager.userAccess(current_user)
+        accessible_service_ids = user_access.get_accessible_services_ids(admin_only=True)
+        if 'session_type' in request.json:
+            # We have a session_type. Get list of items
+            if 'id_session_type' not in request.json['session_type']:
+                return gettext('Missing id_session_type'), 400
+            if 'services' not in request.json['session_type']:
+                return gettext('Missing services'), 400
+            id_session_type = request.json['session_type']['id_session_type']
+
+            if id_session_type not in user_access.get_accessible_session_types_ids(admin_only=True):
+                return gettext("Access denied"), 403
+
+            # Get all current association for session type
+            current_services = TeraSessionTypeServices.get_services_for_session_type(session_type_id=id_session_type)
+            current_services_ids = [service.id_service for service in current_services]
+            received_service_ids = [service['id_service'] for service in request.json['session_type']['services']]
+            # Difference - we must delete services not anymore in the list
+            todel_ids = set(current_services_ids).difference(received_service_ids)
+            # Also filter services already there
+            # received_service_ids = set(received_service_ids).difference(current_services_ids)
+            try:
+                for service_id in todel_ids:
+                    if service_id in accessible_service_ids:  # Don't remove from the list if not admin for that project!
+                        TeraSessionTypeServices.delete_with_ids(session_type_id=id_session_type, service_id=service_id,
+                                                                autocommit=False)
+                TeraSessionTypeServices.commit()
+            except exc.IntegrityError as e:
+                self.module.logger.log_warning(self.module.module_name, UserQuerySessionTypeServices.__name__, 'delete',
+                                               500, 'Integrity error', str(e))
+                return gettext('Can\'t remove associated service from session type: please delete all sessions using '
+                               'that type before deleting.'), 500
+            # Build services association to add
+            json_sts = [{'id_session_type': id_session_type, 'id_service': service_id}
+                        for service_id in received_service_ids]
+        elif 'service' in request.json:
+            # We have a service. Get list of items
+            if 'id_service' not in request.json['service']:
+                return gettext('Missing service ID'), 400
+            if 'sessiontypes' not in request.json['service']:
+                return gettext('Missing session types'), 400
+            id_service = request.json['service']['id_service']
+
+            # Check if admin for that project
+            if id_service not in accessible_service_ids:
+                return gettext('Access denied'), 403
+
+            # Get all current service association for session type
+            current_session_types = TeraSessionTypeServices.get_sessions_types_for_service(service_id=id_service)
+            current_session_types_ids = [st.id_session_type for st in current_session_types]
+            received_st_ids = [st['id_session_type'] for st in request.json['service']['sessiontypes']]
+            # Difference - we must delete types not anymore in the list
+            todel_ids = set(current_session_types_ids).difference(received_st_ids)
+            # Also filter session types already there
+            received_st_ids = set(received_st_ids).difference(current_session_types_ids)
+            try:
+                for st_id in todel_ids:
+                    TeraSessionTypeServices.delete_with_ids(session_type_id=st_id, service_id=id_service,
+                                                            autocommit=False)
+                TeraSessionTypeServices.commit()
+            except exc.IntegrityError as e:
+                self.module.logger.log_warning(self.module.module_name, UserQuerySessionTypeServices.__name__, 'delete',
+                                               500, 'Integrity error', str(e))
+                return gettext('Can\'t remove session type association from service: please delete all sessions using '
+                               'that type before deleting.'), 500
+
+            # Build associations to add
+            json_sts = [{'id_session_type': st_id, 'id_service': id_service} for st_id in received_st_ids]
+        elif 'session_type_service' in request.json:
+            json_sts = request.json['session_type_service']
+            if not isinstance(json_sts, list):
+                json_sts = [json_sts]
+        else:
+            return gettext('Unknown format'), 400
+
+        # Validate if we have an id and access
+        for json_st in json_sts:
+            if 'id_session_type' not in json_st or 'id_service' not in json_st:
+                return gettext('Badly formatted request'), 400
+
+            if json_st['id_service'] not in accessible_service_ids:
+                return gettext('Forbidden'), 403
+
+            if 'id_session_type_service' not in json_st:
+                # Check if already exists
+                st = TeraSessionTypeServices. \
+                    get_session_type_service_for_session_type_service(service_id=int(json_st['id_service']),
+                                                                      session_type_id=int(json_st['id_session_type']))
+                if st:
+                    json_st['id_session_type_service'] = st.id_session_type_service
+                else:
+                    json_st['id_session_type_service'] = 0
+
+            # Do the update!
+            if int(json_st['id_session_type_service']) > 0:
+                pass
+            else:
+                try:
+                    new_sts = TeraSessionTypeServices()
+                    new_sts.from_json(json_st)
+                    new_sts = TeraSessionTypeServices.insert(new_sts)
+                    # Update ID for further use
+                    json_st['id_session_type_service'] = new_sts.id_session_type_service
+                except exc.SQLAlchemyError as e:
+                    import sys
+                    print(sys.exc_info())
+                    self.module.logger.log_error(self.module.module_name,
+                                                 UserQuerySessionTypeServices.__name__,
+                                                 'post', 500, 'Database error', str(e))
+                    return gettext('Database error'), 500
+
+        return json_sts
 
     @api.doc(description='Delete a specific session-type - service additional association.',
              responses={200: 'Success',

teraserver/python/opentera/db/models/TeraSessionTypeServices.py

@@ -83,6 +83,12 @@ def get_sessions_types_for_service(service_id: int, with_deleted: bool = False):
         return TeraSessionTypeServices.query.execution_options(include_deleted=with_deleted)\
             .filter_by(id_service=service_id).all()
 
+    @staticmethod
+    def get_session_type_service_for_session_type_service(service_id: int, session_type_id: int,
+                                                          with_deleted: bool = False):
+        return TeraSessionTypeServices.query.execution_options(include_deleted=with_deleted) \
+            .filter_by(id_service=service_id, id_session_type=session_type_id).first()
+
     @staticmethod
     def check_integrity(obj_to_check):
         # Make sure service is associated to sites of that session type