[#534] implement replica truncate

Co-authored-by: Alan King <alanking@renci.org>

Author: d-w-moore

irods/api_number.py

@@ -158,6 +158,7 @@
     "OPEN_COLLECTION201_AN": 712,
 
     "GET_LIBRARY_FEATURES_AN": 801,
+    "REPLICA_TRUNCATE_AN": 802,
 
     # 1000 - 1059 - NETCDF API calls
     "NC_OPEN_AN": 1000,

irods/data_object.py

@@ -115,6 +115,9 @@ def unregister(self, **options):
     def truncate(self, size):
         self.manager.truncate(self.path, size)
 
+    def replica_truncate(self, size, **options):
+        return self.manager.replica_truncate(self.path, size, **options)
+
     def replicate(self, resource = None, **options):
         self.manager.replicate(self.path, resource = resource, **options)
 

irods/exception.py

@@ -69,7 +69,7 @@ class MultipleResultsFound(QueryException):
     pass
 
 
-class NotImplementedInIRODSServer(RuntimeError):
+class NotImplementedInIRODSServer(PycommandsException):
     def __init__(self, feature_description, required_iRODS_version = ()):
         super(NotImplementedInIRODSServer,self).__init__(feature_description + ': Not supported by the connected iRODS server.')
         self.required_iRODS_version = required_iRODS_version
@@ -641,6 +641,10 @@ class SYS_REPLICA_DOES_NOT_EXIST(iRODSException):
     code = -164000
 
 
+class SYS_REPLICA_INACCESSIBLE(iRODSException):
+    code = -168000
+
+
 class SYS_NOT_ALLOWED(iRODSException):
     code = -169000
 

irods/manager/data_object_manager.py

@@ -534,6 +534,37 @@ def make_FileOpenRequest(**extra_opts):
             ret_value.seek(0,io.SEEK_END)
         return ret_value
 
+    def replica_truncate(self, path, desired_size, **options):
+
+        if self.sess.server_version == (4,3,2):
+            message = 'replica_truncate responses can fail to parse with iRODS 4.3.2 due to routine omission of the JSON response string, so this method is not supported for iRODS 4.3.2.'
+            raise ex.OperationNotSupported(message)
+        else:
+            required_server_version = (4,3,3)
+            if self.sess.server_version < required_server_version:
+                raise ex.NotImplementedInIRODSServer('replica_truncate', required_server_version)
+
+        message_body = FileOpenRequest(
+            objPath=path,
+            createMode=0,
+            openFlags=0,
+            offset=0,
+            dataSize=desired_size,
+            numThreads=self.sess.numThreads,
+            oprType=0,
+            KeyValPair_PI=StringStringMap(options),
+        )
+        message = iRODSMessage('RODS_API_REQ',
+                               msg=message_body,
+                               int_info=api_number["REPLICA_TRUNCATE_AN"])
+
+        with self.sess.pool.get_connection() as conn:
+            conn.send(message)
+            response = conn.recv()
+            msg = response.get_main_message( STR_PI )
+
+        return json.loads(msg.myStr)
+
     def trim(self, path, **options):
 
         try:

irods/test/data_obj_test.py

@@ -3,7 +3,8 @@
 from datetime import datetime
 import base64
 import concurrent.futures
-import contextlib  # check if redundant
+import contextlib
+import errno
 import hashlib
 import io
 import itertools
@@ -2362,6 +2363,84 @@ def _update(n):
 
         self.assertEqual(pbar.percent_done(), 100)
 
+    def test_replica_truncate_related_errors__issue_534(self):
+        sess = self.sess
+        data_objs = self.sess.data_objects
+        truncated_size = 16
+
+        try:
+            # Set path for a new, as yet nonexisting data object.
+            data_path = '{}/{}'.format(
+                helpers.home_collection(sess),
+                unique_name(my_function_name(), datetime.now(), truncated_size) + "issue_534_errors")
+            self.assertFalse (data_objs.exists(data_path), "Data object should not yet exist.")
+
+            # Assert appropriate error is raised for the nonexisting object.
+            with self.assertRaises(ex.OBJ_PATH_DOES_NOT_EXIST):
+                data_objs.replica_truncate(data_path, truncated_size)
+
+            d = data_objs.create(data_path)
+
+            # Assert appropriate error is raised for an existing object without a replica on the given resource.
+            with self.create_simple_resc() as newResc:
+                with self.assertRaises(ex.SYS_REPLICA_INACCESSIBLE):
+                    d.replica_truncate(truncated_size, **{kw.RESC_NAME_KW: newResc})
+
+            # Assert appropriate error is raised for an invalid (negative) size argument.
+            try:
+                data_objs.replica_truncate(data_path, -truncated_size)
+            except Exception as e:
+                self.assertIsInstance(e, ex.UNIX_FILE_TRUNCATE_ERR)
+                self.assertEqual('EINVAL', errno.errorcode[abs(e.code) % 1000])
+
+        finally:
+            if data_objs.exists(data_path):
+                data_objs.unlink(data_path, force = True)
+
+    def test_replica_truncate__issue_534(self):
+        sess = self.sess
+        data_objs = self.sess.data_objects
+        original_size = 16
+        original_content = b'_' * original_size
+        with self.create_simple_resc() as newResc:
+            for truncated_size in (original_size//2, original_size + 1024):
+                try:
+                    data_path = '{}/{}'.format(
+                        helpers.home_collection(sess),
+                        unique_name(my_function_name(), datetime.now(), truncated_size) + "_issue_534")
+
+                    # Create a data object with size original_size.
+                    with data_objs.open(data_path,'w') as f:
+                        f.write(original_content)
+                    d = data_objs.get(data_path)
+
+                    # Ensure there are two replicas, one on newResc as well as one on 'demoResc'
+                    d.replicate(resource = newResc)
+                    response = d.replica_truncate(truncated_size, **{kw.RESC_NAME_KW: newResc})
+
+                    # Stat data object again.
+                    d = data_objs.get(data_path)
+
+                    # Check that returned resource hierarchy and replica number match expectations.
+                    self.assertEqual(response['replica_number'],
+                        [_ for _ in data_objs.get(data_path).replicas if _.resource_name == newResc][0].number)
+                    self.assertEqual(response['resource_hierarchy'], newResc)
+
+                    # Make sure sizes are as expected.
+                    self.assertEqual([_ for _ in d.replicas if _.resource_name == newResc][0].size, truncated_size)
+                    self.assertEqual([_ for _ in d.replicas if _.resource_name != newResc][0].size, original_size)
+
+                    # Check that content of truncated replicas is as expected.
+                    if truncated_size <= original_size:
+                        self.assertEqual( d.open('r').read(), original_content[:truncated_size])
+                    else:
+                        self.assertEqual( d.open('r').read(), original_content + b'\0' * (truncated_size - original_size))
+
+                finally:
+                    if data_objs.exists(data_path):
+                        data_objs.unlink(data_path, force = True)
+
+
 if __name__ == '__main__':
     # let the tests find the parent irods lib
     sys.path.insert(0, os.path.abspath('../..'))