Toolbox cleanup, add associated user

.github/workflows/build_image_from_dev.yml

@@ -100,7 +100,6 @@ jobs:
         uses: docker/build-push-action@v6
         with:
           context: .
-          ref: dev
           push: true
           tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}

config/default.py

@@ -27,6 +27,7 @@
 PRODUCTION = (environ.get('PRODUCTION', default="false") == "true")
 TEST_UID = environ.get('TEST_UID', default="dhf8r")
 ADMIN_UIDS = re.split(r',\s*', environ.get('ADMIN_UIDS', default="dhf8r,kcm4zc,cah3us"))
+SUPERUSER_UIDS = re.split(r',\s*', environ.get('SUPERUSER_UIDS', default="dhf8r,kcm4zc,cah3us"))
 DEFAULT_UID = environ.get('DEFAULT_UID', default="dhf8r")
 
 # Sentry flag

crc/models/user.py

@@ -20,6 +20,11 @@ def is_admin(self):
         # may change in the future.
         return self.uid in app.config['ADMIN_UIDS']
 
+    def is_superuser(self):
+        # Currently superuser abilities are set in the configuration, but this
+        # may change in the future.
+        return self.uid in app.config['SUPERUSER_UIDS']
+
     def encode_auth_token(self):
         """
         Generates the Auth Token
@@ -60,13 +65,18 @@ class Meta:
         include_relationships = True
     uid = fields.String()
     is_admin = fields.Method('get_is_admin', dump_only=True)
+    is_superuser = fields.Method('get_is_superuser', dump_only=True)
     ldap_info = fields.Nested(LdapSchema)
     impersonator = fields.Nested('self', many=False, allow_none=True)
 
     @staticmethod
     def get_is_admin(user):
         return user.is_admin()
 
+    @staticmethod
+    def get_is_superuser(user):
+        return user.is_superuser()
+
 
 class AdminSessionModel(db.Model):
     __tablename__ = 'admin_session'

crc/scripts/add_admin_user.py

@@ -0,0 +1,44 @@
+from crc import app, db
+from crc.api.common import ApiError
+from crc.models.study import StudyAssociated
+from crc.scripts.script import Script
+from crc.services.ldap_service import LdapService
+
+from sqlalchemy import text
+
+
+class AddAdminUser(Script):
+    scripts = {}
+
+    def get_description(self):
+        return """Add a user to the study_associated_user table (for all studies)"""
+
+    def do_task_validate_only(self, task, study_id, workflow_id, *args, **kwargs):
+        return self.do_task(task, study_id, workflow_id, *args, **kwargs)
+
+    def do_task(self, task, study_id, workflow_id, *args, **kwargs):
+        associated_user = kwargs.get('associated_user', None)
+        associated_group = kwargs.get('associated_group', None)
+        scripts = self.generate_augmented_list(task, study_id, workflow_id)
+        self.scripts = scripts
+        try:
+            ldap_user = self.scripts['ldap'](associated_user)
+        except ApiError as ae:
+            return {"error": str(ae), 'message': f"User {associated_user} not found in LDAP"}
+
+        if 'uid' in ldap_user and ldap_user['uid'] == associated_user:
+            # study_associated = StudyAssociated()
+
+            # using 'lje5u' because they are in both CTO and CTO Finance groups
+            # sql_string = f"select study_id, '{associated_user}', 'CTO', false, true from study_associated_user where uid='lje5u' and role='{associated_group}'"
+
+            sql_string = """insert into study_associated_user
+            (study_id, uid, role, send_email, access)
+            select study_id, '%s', 'CTO', false, true 
+            from study_associated_user 
+            where uid='lje5u' and role='%s'""" % (associated_user, associated_group)
+
+            sql = text(sql_string)
+            result = db.engine.execute(sql)
+            print(result.rowcount)
+            return {"message": f"User {associated_user} added to {associated_group} group"}

crc/scripts/get_top_level_data.py

@@ -0,0 +1,26 @@
+from crc import session
+from crc.api.common import ApiError
+from crc.models.study import StudyModel
+from crc.scripts.script import Script
+from crc.services.workflow_processor import WorkflowProcessor
+from crc.services.workflow_spec_service import WorkflowSpecService
+
+
+class GetTopLevelData(Script):
+
+    def get_description(self):
+        return """This is my description"""
+
+    def do_task_validate_only(self, task, study_id, workflow_id, *args, **kwargs):
+        return self.do_task(task, study_id, workflow_id, *args, **kwargs)
+
+    def do_task(self, task, study_id, workflow_id, *args, **kwargs):
+        spec_service = WorkflowSpecService()
+        study_model = session.query(StudyModel).filter(StudyModel.id == study_id).first()
+
+        try:
+            master_workflow_results = WorkflowProcessor.run_master_spec(spec_service.master_spec, study_model)
+        except Exception as e:
+            raise ApiError("error_running_master_spec", f"Error running master spec: {str(e)}")
+
+        return master_workflow_results