add a flag to not auto-complete filtering datasets

iceprod/rest/handlers/datasets.py

@@ -119,6 +119,7 @@ async def post(self):
             'priority': float,
             'debug': bool,
             'jobs_immutable': bool,
+            'always_active': bool,
             'status': str,
             'auth_groups_read': list,
         }
@@ -168,6 +169,8 @@ async def post(self):
         if 'jobs_immutable' not in data:
             data['jobs_immutable'] = False
         data['truncated'] = False
+        if 'always_active' not in data:
+            data['always_active'] = False
 
         # insert
         ret = await self.db.datasets.insert_one(data)

iceprod/scheduled_tasks/dataset_completion.py

@@ -45,6 +45,9 @@ async def run(rest_client, debug=False):
         # test if dataset is complete / failed
         try:
             dataset = await rest_client.request('GET', f'/datasets/{dataset_id}')
+            if dataset.get('always_active', False):
+                logger.info('dataset %s always active, skipping', dataset_id)
+                continue
             jobs = await rest_client.request('GET', f'/datasets/{dataset_id}/job_counts/status')
             if dataset.get('truncated', False) or sum(jobs[s] for s in jobs) >= dataset['jobs_submitted']:
                 # jobs are all buffered / materialized

tests/rest/datasets_test.py

@@ -55,6 +55,29 @@ async def test_rest_datasets_post_bad_role(server):
     assert exc_info.value.response.status_code == 403
 
 
+async def test_rest_datasets_post_always_active(server):
+    client = server(roles=['user'])
+
+    data = {
+        'description': 'blah',
+        'tasks_per_job': 4,
+        'jobs_submitted': 1,
+        'tasks_submitted': 4,
+        'group': 'users',
+        'always_active': True
+    }
+    ret = await client.request('POST', '/datasets', data)
+    dataset_id = ret['result'].split('/')[-1]
+
+    ret = await client.request('GET', '/datasets')
+    assert dataset_id in ret
+
+    ret = await client.request('GET', f'/datasets/{dataset_id}')
+    for k in data:
+        assert k in ret
+        assert data[k] == ret[k]
+
+
 async def test_rest_datasets_err(server):
     client = server(roles=['system'])
     with pytest.raises(requests.exceptions.HTTPError) as exc_info:

tests/scheduled_tasks/dataset_completion_test.py

@@ -92,3 +92,24 @@ async def client(method, url, args=None):
 
     # check it normally hides the error
     await dataset_completion.run(rc, debug=False)
+
+
+
+async def test_always_active():
+    rc = MagicMock()
+    dataset_summaries = {'processing':['foo']}
+    async def client(method, url, args=None):
+        logger.info('REST: %s, %s', method, url)
+        if url.startswith('/dataset_summaries'):
+            return dataset_summaries
+        elif url == '/config/foo':
+            return {}
+        elif url == '/datasets/foo':
+            return {'jobs_submitted':2, 'tasks_submitted':2, 'always_active': True}
+        else:
+            raise Exception()
+    rc.request = client
+    client.called = False
+
+    await dataset_completion.run(rc, debug=True)
+    assert client.called