Add modality tests

Author: nagem

api/dao/basecontainerstorage.py

@@ -160,8 +160,20 @@ def update_el(self, _id, payload, unset_payload=None, recursive=False, r_payload
                 raise APIStorageException(e.message)
         if recursive and r_payload is not None:
             containerutil.propagate_changes(self.cont_name, _id, {}, {'$set': util.mongo_dict(r_payload)})
+
+        config.log.warning(update)
         return self.dbc.update_one({'_id': _id}, update)
 
+    def replace_el(self, _id, payload):
+        if self.use_object_id:
+            try:
+                _id = bson.ObjectId(_id)
+            except bson.errors.InvalidId as e:
+                raise APIStorageException(e.message)
+        payload['_id'] = _id
+        return self.dbc.replace_one({'_id': _id}, payload)
+
+
     def delete_el(self, _id):
         if self.use_object_id:
             try:

api/handlers/modalityhandler.py

@@ -30,10 +30,7 @@ def post(self):
         #validate_data(payload, 'modality.json', 'input', 'POST', optional=True)
 
         result = self.storage.create_el(payload)
-        if result.acknowledged:
-            return {'_id': result.inserted_id}
-        else:
-            self.abort(400, 'Modality not inserted')
+        return {'_id': result.inserted_id}
 
     @require_admin
     def put(self, modality_name):
@@ -42,18 +39,14 @@ def put(self, modality_name):
         # POST unnecessary, used to avoid run-time modification of schema
         #validate_data(payload, 'modality.json', 'input', 'POST', optional=True)
 
-        result = self.storage.update_el(modality_name, payload)
-        if result.matched_count == 1:
-            return {'modified': result.modified_count}
-        else:
+        result = self.storage.replace_el(modality_name, payload)
+        if result.matched_count != 1:
             raise APINotFoundException('Modality with name {} not found, modality not updated'.format(modality_name))
 
     @require_admin
     def delete(self, modality_name):
         result = self.storage.delete_el(modality_name)
-        if result.deleted_count == 1:
-            return {'deleted': result.deleted_count}
-        else:
+        if result.deleted_count != 1:
             raise APINotFoundException('Modality with name {} not found, modality not deleted'.format(modality_name))
 
 

tests/integration_tests/python/test_classification.py

@@ -0,0 +1,60 @@
+def test_modalities(data_builder, as_admin, as_user):
+
+    payload = {
+        '_id': 'MR',
+        'classification': {
+            'Intent': ["Structural", "Functional", "Localizer"],
+            'Contrast': ["B0", "B1", "T1", "T2"]
+        }
+    }
+
+    # test adding new modality
+    r = as_admin.post('/modalities', json=payload)
+    assert r.ok
+    assert r.json()['_id'] == payload['_id']
+    modality1 = payload['_id']
+
+    # get specific modality
+    r = as_user.get('/modalities/' + modality1)
+    assert r.ok
+    assert r.json() == payload
+
+    # try replacing existing modality via POST
+    r = as_admin.post('/modalities', json=payload)
+    assert r.status_code == 409
+
+    # list modalities as non-admin
+    r = as_user.get('/modalities')
+    assert r.ok
+    modalities = r.json()
+    assert len(modalities) == 1
+    assert modalities[0]['_id'] == modality1
+
+    # replace existing modality
+    update = {
+        'classification': {
+            'Intent': ["new", "stuff"]
+        }
+    }
+    r = as_admin.put('/modalities/' + modality1, json=update)
+    assert r.ok
+    r = as_admin.get('/modalities/' + modality1)
+    assert r.ok
+    assert r.json()['classification'] == update['classification']
+
+    # try to replace missing modality
+    r = as_admin.put('/modalities/' + 'madeup', json=update)
+    assert r.status_code == 404
+
+    # delete modality
+    r = as_admin.delete('/modalities/' + modality1)
+    assert r.ok
+
+    # try to delete missing modality
+    r = as_admin.delete('/modalities/' + modality1)
+    assert r.status_code == 404
+
+
+
+
+

tests/integration_tests/python/test_rules.py

@@ -437,5 +437,3 @@ def test_rules(randstr, data_builder, file_form, as_root, as_admin, with_user, a
     # delete rule
     r = as_admin.delete('/projects/' + project + '/rules/' + rule3)
     assert r.ok
-
-    # TODO add and test 'new-style' rules