Add tests

test/integration_tests/python/test_batch.py

@@ -246,3 +246,67 @@ def test_batch(data_builder, as_user, as_admin, as_root):
     # test batch is complete
     r = as_admin.get('/batch/' + batch_id)
     assert r.json()['state'] == 'failed'
+
+def test_no_input_batch(data_builder, as_admin, as_root):
+    project = data_builder.create_project()
+    session = data_builder.create_session(project=project)
+    acquisition = data_builder.create_acquisition(session=session)
+
+    gear_doc = default_payload['gear']['gear']
+    gear_doc['inputs'] = {
+        'api_key': {
+            'base': 'api-key'
+        }
+    }
+    gear = data_builder.create_gear(gear=gear_doc)
+
+    # create a batch w/o inputs targeting session
+    r = as_admin.post('/batch', json={
+        'gear_id': gear,
+        'targets': [{'type': 'session', 'id': session}]
+    })
+    assert r.ok
+    batch1 = r.json()
+    assert len(batch1['matched']) == 1
+    assert batch1['matched'][0]['id'] == session
+
+    # create a batch w/o inputs targeting acquisition
+    r = as_admin.post('/batch', json={
+        'gear_id': gear,
+        'targets': [{'type': 'acquisition', 'id': acquisition}]
+    })
+    assert r.ok
+    batch2 = r.json()
+    assert len(batch2['matched']) == 1
+    assert batch2['matched'][0]['id'] == session
+
+    # create a batch w/o inputs targeting project
+    r = as_admin.post('/batch', json={
+        'gear_id': gear,
+        'targets': [{'type': 'project', 'id': project}]
+    })
+    assert r.ok
+    batch3 = r.json()
+    assert len(batch3['matched']) == 1
+    assert batch3['matched'][0]['id'] == session
+
+    batch_id = batch1['_id']
+
+    # run batch
+    r = as_admin.post('/batch/' + batch_id['_id'] + '/run')
+    assert r.ok
+
+    # test batch.state after calling run
+    r = as_admin.get('/batch/' + batch_id)
+    assert r.json()['state'] == 'running'
+
+    for job in r.json()['jobs']:
+        # set jobs to failed
+        r = as_root.put('/jobs/' + job, json={'state': 'running'})
+        assert r.ok
+        r = as_root.put('/jobs/' + job, json={'state': 'complete'})
+        assert r.ok
+
+    # test batch is complete
+    r = as_admin.get('/batch/' + batch_id)
+    assert r.json()['state'] == 'complete'