Merge pull request #161 from mcneilco/ACAS-745

ACAS-745: Add raise_on_linked_data parameter to delete_lot method

Author: brianbolt

acasclient/acasclient.py

@@ -2247,11 +2247,12 @@ def get_lot_dependencies(self, lot_corp_name, include_linked_lots=True, include_
         resp.raise_for_status()
         return resp.json()
 
-    def delete_lot(self, lot_corp_name):
+    def delete_lot(self, lot_corp_name, raise_on_linked_data=True):
         """Delete a lot
 
         Args:
             lot_corp_name (str): Corp name of lot to delete
+            raise_on_linked_data (bool): Whether to raise an error if there is linked data, default True
 
         Returns:
             A dict with "success": true if successful. For example
@@ -2261,7 +2262,19 @@ def delete_lot(self, lot_corp_name):
             Or None if there was an error
         Raises:
             HTTPError: If permission denied
+            ValueError: If there is linked data and raise_on_linked_data is True
         """
+
+        if raise_on_linked_data:
+            # Check for any lot dependencies
+            lot_dependencies_return = self.get_lot_dependencies(lot_corp_name, include_linked_lots=False, include_analysis_group_values=False)
+
+            # Only proceed if there is not any linked data
+            if lot_dependencies_return["linkedDataExists"]:
+                linked_experiments = lot_dependencies_return["linkedExperiments"]
+                error_message = f"Refusing to delete lot {lot_corp_name}. Re-run with raise_on_linked_data=False to delete lot and experimental data linked to lot (deletes assay data linked to lot data not full experiments). Linked experiments: {linked_experiments}."
+                raise ValueError(error_message)
+
         resp = self.session.delete("{}/cmpdReg/metalots/corpName/{}"
                                 .format(self.url, lot_corp_name))
         if resp.status_code == 500:

tests/test_acasclient.py

@@ -293,7 +293,7 @@ def delete_all_lots_and_experiments(self):
         if len(lots) > 0:
             # Delete any remaining lots not from bulk loads
             for lot in lots:
-                self.client.delete_lot(lot['lotCorpName'])
+                self.client.delete_lot(lot['lotCorpName'], raise_on_linked_data=False)
 
 def requires_basic_cmpd_reg_load(func):
     """
@@ -3314,13 +3314,13 @@ def test_004_delete_lot(self, experiment):
         acas_admin = self.create_and_connect_backdoor_user(prefix="acas-admin-", acas_user=True, acas_admin=True, creg_user=False, creg_admin=False)
         cmpdreg_admin = self.create_and_connect_backdoor_user(prefix="cmpdreg-admin-", acas_user=False, acas_admin=False, creg_user=True, creg_admin=True)
 
-        def can_delete_lot(self, user_client, lot_corp_name, set_owner_first=True):
+        def can_delete_lot(self, user_client, lot_corp_name, set_owner_first=True, raise_on_linked_data=False):
             if set_owner_first:
                 meta_lot = self.client.get_meta_lot(lot_corp_name)
                 meta_lot["lot"]["chemist"] = user_client.username
                 self.client.save_meta_lot(meta_lot)
             try:
-                response = user_client.delete_lot(lot_corp_name)
+                response = user_client.delete_lot(lot_corp_name, raise_on_linked_data=raise_on_linked_data)
             except requests.HTTPError:
                 return False
             self.assertIn("success", response)
@@ -3364,7 +3364,12 @@ def can_delete_lot(self, user_client, lot_corp_name, set_owner_first=True):
         self.assertFalse(can_delete_lot(self, acas_admin, restricted_lot_corp_name, set_owner_first=True))
 
         # Allow rule: CregAdmin/ACASAdmin can delete the lot and assay data
-        self.assertTrue(can_delete_lot(self, self.client, restricted_lot_corp_name, set_owner_first=True))
+        # First verify that raise_on_linked_data=True will raise an error
+        with self.assertRaises(ValueError) as context:
+            can_delete_lot(self, self.client, restricted_lot_corp_name, set_owner_first=True, raise_on_linked_data=True)
+        self.assertIn('Refusing to delete lot', str(context.exception))
+        # Now actually delete the lot
+        self.assertTrue(can_delete_lot(self, self.client, restricted_lot_corp_name, set_owner_first=True, raise_on_linked_data=False))
 
         # Verify lot is actually deleted
         meta_lot = self.client.get_meta_lot(restricted_lot_corp_name)