Merge pull request #155 from lehtiolab/154-small-things-collected

Accumulated fixes for smaller problems

Author: glormph

src/backend/analysis/tests.py

@@ -857,11 +857,16 @@ def test_new_analysis_and_run_and_purge(self):
         self.assertFalse(os.path.exists(webfn))
         self.assertTrue(os.path.exists(anafn))
         self.assertTrue(os.path.exists(nfrunfn))
+        rmdirjobs = jm.Job.objects.filter(funcname='delete_empty_directory')
+        self.assertEqual(rmdirjobs.filter(state=jj.Jobstates.PENDING).count(), 2)
         # Two purge jobs and two delete dir jobs, this test is getting slow
         self.run_job() # purge files
         self.run_job() # rm dir
         self.run_job() # purge
         self.run_job() # rm dir
+        # Second rmdir job is executed but status is not resolved 
+        self.assertEqual(rmdirjobs.filter(state=jj.Jobstates.DONE).count(), 1)
+        self.assertEqual(rmdirjobs.filter(state=jj.Jobstates.PROCESSING).count(), 1)
         self.assertFalse(os.path.exists(anafn))
         self.assertFalse(os.path.exists(nfrunfn))
 

src/backend/analysis/views.py

@@ -1374,9 +1374,9 @@ def do_analysis_deletion(analysis):
     # FIXME analysisdeleted tracking no longer needed when we backup etc
     # implement a log instead
     am.AnalysisDeleted.objects.update_or_create(analysis=analysis)
-    if jobq := jm.Job.objects.filter(nextflowsearch__analysis=analysis).exclude(state__in=[
-                jj.Jobstates.ERROR, jj.Jobstates.DONE, jj.Jobstates.REVOKING, jj.Jobstates.CANCELED]):
-        jv.cancel_or_revoke_job(jobq)
+    jm.Job.objects.filter(nextflowsearch__analysis=analysis).exclude(state__in=[
+                jj.Jobstates.DONE, jj.Jobstates.REVOKING, jj.Jobstates.CANCELED]).update(
+                state=jj.Jobstates.REVOKING)
     if analysis.success_completed:
         # Back up files if that for some reason (old analysis) hasnt happened earlier
         backup_jobs = []
@@ -1431,13 +1431,17 @@ def do_analysis_deletion(analysis):
                 return f'error trying to queue delete files job: {joberror}'
             rmdirkw = {'sfloc_ids': sfloc_ids, 'path': path, 'share_id': shareid}
             rmdirjobs.append(rmdirkw)
-            if joberror := check_job_error('delete_empty_directory', sfloc_ids=sfloc_ids):
+            if joberror := check_job_error('delete_empty_directory', **rmdirkw):
                 return f'error trying to queue delete files job: {joberror}'
         for sfloc_ids in purgejobs:
             create_job_without_check('purge_files', sfloc_ids=sfloc_ids)
             rm.StoredFileLoc.objects.filter(pk__in=sfloc_ids).update(active=False)
+        jwrap = jobmap['delete_empty_directory'](False)
         for rmdirkw in rmdirjobs:
-            create_job_without_check('delete_empty_directory', **rmdirkw)
+            # Check first if not files exist in dir, e.g. in case of multiple
+            # analyses. Skip delete dir in that case
+            if not jwrap.files_in_dir_existing(**rmdirkw):
+                create_job_without_check('delete_empty_directory', **rmdirkw)
         sfiles.update(deleted=True)
 
     # No more errors, mark for deletion
@@ -1483,8 +1487,7 @@ def stop_analysis(request):
     if not request.user.is_superuser and not anaq.exists():
         return JsonResponse({'error': 'Analysis does not exist or you dont have permission'}, status=403)
     anaq.update(editable=True)
-    jobq = jm.Job.objects.filter(nextflowsearch__analysis_id=req['item_id'])
-    jv.cancel_or_revoke_job(jobq)
+    jm.Job.objects.filter(nextflowsearch__analysis_id=req['item_id']).update(state=jj.Jobstates.REVOKING)
     return JsonResponse({})
 
 
@@ -1606,7 +1609,7 @@ def upload_servable_file(request):
 @login_required
 def find_datasets(request):
     searchterms = [x for x in request.GET['q'].split() if x != '']
-    dbdsets = dm.Dataset.query_creator(searchterms).filter(deleted=False)
+    dbdsets = dm.Dataset.objects.filter(dm.Dataset.query_creator(searchterms), deleted=False)
     dsets = {}
     qpid = dm.Datatype.get_quantprot_id()
     for d in dbdsets.select_related('runname__experiment__project', 'datatype',

src/backend/datasets/jobs.py

@@ -29,12 +29,18 @@ def process(self, **kwargs):
             raise RuntimeError('Dataset source files are spread over more than one location, please '
                     'contact admin to make sure files are consolicated before renaming path')
 
-        srcloc = srcloc.get()
-        fss = FileserverShare.objects.filter(share=srcloc['servershare__pk'], server__active=True
-                ).values('server__name', 'path').first()
-        self.queue = self.get_server_based_queue(fss['server__name'], settings.QUEUE_FASTSTORAGE)
-        self.run_tasks = [(fss['path'], srcloc['path'], kwargs['newpath'], 
-            [x['pk'] for x in srcsfs.values('pk')], kwargs['dss_id'])]
+        elif srcloc.count():
+            srcloc = srcloc.get()
+            fss = FileserverShare.objects.filter(share=srcloc['servershare__pk'], server__active=True
+                    ).values('server__name', 'path').first()
+            self.queue = self.get_server_based_queue(fss['server__name'], settings.QUEUE_FASTSTORAGE)
+            self.run_tasks = [(fss['path'], srcloc['path'], kwargs['newpath'], 
+                [x['pk'] for x in srcsfs.values('pk')], kwargs['dss_id'])]
+        else:
+            # There are no sfloc files in this job, e.g. empty dataset, so we will not rename
+            # on disk
+            dss = DatasetServer.objects.filter(pk=kwargs['dss_id'])
+            dss.update(storage_loc=kwargs['newpath'])
 
 
 class RsyncDatasetServershare(DatasetJob):

src/backend/datasets/models.py

@@ -6,6 +6,8 @@
 
 
 class PrincipalInvestigator(models.Model):
+    # Not for actual accounting (that would be external reference on the project table)
+    # more for staff to know who to contact for an active project
     name = models.TextField(max_length=100)
 
     def __str__(self):
@@ -117,6 +119,34 @@ class Dataset(models.Model):
     purged = models.BooleanField(default=False) # for UI only, indicate permanent deleted from cold storage too
     locked = models.BooleanField(default=False)
 
+    @staticmethod
+    def query_creator(searchterms):
+        query = Q(runname__name__icontains=searchterms[0])
+        query |= Q(runname__experiment__name__icontains=searchterms[0])
+        query |= Q(runname__experiment__project__name__icontains=searchterms[0])
+        query |= Q(datatype__name__icontains=searchterms[0])
+        try:
+            float(searchterms[0])
+        except ValueError:
+            pass
+        else:
+            query |= Q(prefractionationdataset__hiriefdataset__hirief__start=searchterms[0])
+            query |= Q(prefractionationdataset__hiriefdataset__hirief__end=searchterms[0])
+        for term in searchterms[1:]:
+            subquery = Q(runname__name__icontains=term)
+            subquery |= Q(runname__experiment__name__icontains=term)
+            subquery |= Q(runname__experiment__project__name__icontains=term)
+            subquery |= Q(datatype__name__icontains=term)
+            try:
+                float(term)
+            except ValueError:
+                pass
+            else:
+                subquery |= Q(prefractionationdataset__hiriefdataset__hirief__start=term)
+                subquery |= Q(prefractionationdataset__hiriefdataset__hirief__end=term)
+            query &= subquery
+        return query
+
 
 class DatasetServer(models.Model):
     '''
@@ -154,35 +184,6 @@ class Meta:
                 models.UniqueConstraint(fields=['dataset', 'storageshare'], name='uni_dsshare'),
                 ]
 
-    @classmethod
-    def query_creator(cls, searchterms):
-        query = Q(runname__name__icontains=searchterms[0])
-        query |= Q(runname__experiment__name__icontains=searchterms[0])
-        query |= Q(runname__experiment__project__name__icontains=searchterms[0])
-        query |= Q(datatype__name__icontains=searchterms[0])
-        try:
-            float(searchterms[0])
-        except ValueError:
-            pass
-        else:
-            query |= Q(prefractionationdataset__hiriefdataset__hirief__start=searchterms[0])
-            query |= Q(prefractionationdataset__hiriefdataset__hirief__end=searchterms[0])
-        for term in searchterms[1:]:
-            subquery = Q(runname__name__icontains=term)
-            subquery |= Q(runname__experiment__name__icontains=term)
-            subquery |= Q(runname__experiment__project__name__icontains=term)
-            subquery |= Q(datatype__name__icontains=term)
-            try:
-                float(term)
-            except ValueError:
-                pass
-            else:
-                subquery |= Q(prefractionationdataset__hiriefdataset__hirief__start=term)
-                subquery |= Q(prefractionationdataset__hiriefdataset__hirief__end=term)
-            query &= subquery
-        return cls.objects.filter(query)
-
-
 
 class DatasetOwner(models.Model):
     dataset = models.ForeignKey(Dataset, on_delete=models.CASCADE)