[#525] Add support for touch API operation

irods/api_number.py

@@ -182,6 +182,7 @@
     "ATOMIC_APPLY_METADATA_OPERATIONS_APN": 20002,
     "GET_FILE_DESCRIPTOR_INFO_APN": 20000,
     "REPLICA_CLOSE_APN": 20004,
+    "TOUCH_APN": 20007,
 
     "AUTH_PLUG_REQ_AN": 1201
 }

irods/data_object.py

@@ -70,7 +70,9 @@ def __init__(self, manager, parent=None, results=None):
                 r[DataObject.resc_hier],
                 checksum=r[DataObject.checksum],
                 size=r[DataObject.size],
-                comments=r[DataObject.comments]
+                comments=r[DataObject.comments],
+                create_time=r[DataObject.create_time],
+                modify_time=r[DataObject.modify_time]
             ) for r in replicas]
         self._meta = None
 

irods/exception.py

@@ -23,6 +23,10 @@ class DoesNotExist(PycommandsException):
     pass
 
 
+class InvalidInputArgument(PycommandsException):
+    pass
+
+
 class DataObjectDoesNotExist(DoesNotExist):
     pass
 

irods/manager/_internal/_api_impl.py

@@ -0,0 +1,12 @@
+from irods.api_number import api_number
+from irods.message import iRODSMessage, JSON_Message
+
+def _touch_impl(session, path, **options):
+    with session.pool.get_connection() as conn:
+        message_body = JSON_Message(
+            {'logical_path': path, 'options': options},
+            conn.server_version)
+        message = iRODSMessage('RODS_API_REQ', msg=message_body,
+                               int_info=api_number['TOUCH_APN'])
+        conn.send(message)
+        response = conn.recv()

irods/manager/_internal/_logical_path.py

@@ -0,0 +1,19 @@
+from irods.exception import CollectionDoesNotExist
+
+def _is_collection(session, path):
+    """Return True if the logical path points to a collection, else False.
+
+    Parameters
+    ----------
+    session: iRODSSession
+        The session object.
+
+    path: string
+        The absolute logical path to a collection.
+    """
+    try:
+        session.collections.get(path)
+        return True
+    except CollectionDoesNotExist:
+        pass
+    return False

irods/manager/collection_manager.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import
 from irods.models import Collection, DataObject
 from irods.manager import Manager
+from irods.manager._internal import _api_impl
 from irods.message import iRODSMessage, CollectionRequest, FileOpenRequest, ObjCopyRequest, StringStringMap
 from irods.exception import CollectionDoesNotExist, NoResultFound
 from irods.api_number import api_number
@@ -150,3 +151,40 @@ def register(self, dir_path, coll_path, **options):
         with self.sess.pool.get_connection() as conn:
             conn.send(message)
             response = conn.recv()
+
+    def touch(self, path, **options):
+        """Change the mtime of an existing collection.
+
+        Parameters
+        ----------
+        path: string
+            The absolute logical path of a collection.
+
+        seconds_since_epoch: integer, optional
+            The number of seconds since epoch representing the new mtime. Cannot
+            be used with "reference" parameter.
+
+        reference: string, optional
+            Use the mtime of the logical path to the data object or collection
+            identified by this option. Cannot be used with "seconds_since_epoch"
+            parameter.
+
+        Raises
+        ------
+        CollectionDoesNotExist
+            If the target collection does not exist or does not point to a
+            collection.
+        """
+        # Attempt to lookup the collection. If it does not exist, the call
+        # will raise an exception.
+        #
+        # Enforces the requirement that collections must exist before this
+        # operation is invoked.
+        self.get(path)
+
+        # The following options to the touch API are not allowed for collections.
+        options.pop('no_create', None)
+        options.pop('replica_number', None)
+        options.pop('leaf_resource_name', None)
+
+        _api_impl._touch_impl(self.sess, path, no_create=True, **options)

irods/manager/data_object_manager.py

@@ -3,6 +3,7 @@
 import io
 from irods.models import DataObject, Collection
 from irods.manager import Manager
+from irods.manager._internal import _api_impl, _logical_path
 from irods.message import (
     iRODSMessage, FileOpenRequest, ObjCopyRequest, StringStringMap, DataObjInfo, ModDataObjMeta,
     DataObjChksumRequest, DataObjChksumResponse, RErrorStack, STR_PI
@@ -726,3 +727,48 @@ def modDataObjMeta(self, data_obj_info, meta_dict, **options):
         with self.sess.pool.get_connection() as conn:
             conn.send(message)
             response = conn.recv()
+
+    def touch(self, path, **options):
+        """Change the mtime of a data object.
+
+        A path argument that does not exist will be created as an empty data
+        object, unless "no_create=True" is supplied.
+
+        Parameters
+        ----------
+        path: string
+            The absolute logical path of a data object.
+
+        no_create: boolean, optional
+            Instructs the system not to create a data object when it does not
+            exist.
+
+        replica_number: integer, optional
+            The replica number of the replica to update. Replica numbers cannot
+            be used to create data objects or additional replicas. Cannot be used
+            with "leaf_resource_name".
+
+        leaf_resource_name: string, optional
+            The name of the leaf resource containing the replica to update. If
+            the object identified by the "path" parameter does not exist and this
+            parameter holds a valid resource, the data object will be created at
+            the specified resource. Cannot be used with "replica_number" parameter.
+
+        seconds_since_epoch: integer, optional
+            The number of seconds since epoch representing the new mtime. Cannot
+            be used with "reference" parameter.
+
+        reference: string, optional
+            Use the mtime of the logical path to the data object or collection
+            identified by this option. Cannot be used with "seconds_since_epoch"
+            parameter.
+
+        Raises
+        ------
+        InvalidInputArgument
+            If the path points to a collection.
+        """
+        if _logical_path._is_collection(self.sess, path):
+            raise ex.InvalidInputArgument()
+
+        _api_impl._touch_impl(self.sess, path, **options)

irods/test/collection_test.py

@@ -1,5 +1,6 @@
 #! /usr/bin/env python
 from __future__ import absolute_import
+from datetime import datetime
 import os
 import sys
 import socket
@@ -15,13 +16,31 @@
 from irods.test.helpers import my_function_name, unique_name
 from irods.collection import iRODSCollection
 
+RODSUSER = 'nonadmin'
 
 class TestCollection(unittest.TestCase):
 
+    class WrongUserType(RuntimeError): pass
+
+    @classmethod
+    def setUpClass(cls):
+        adm = helpers.make_session()
+        if adm.users.get(adm.username).type != 'rodsadmin':
+            raise cls.WrongUserType('Must be an iRODS admin to run tests in class {0.__name__}'.format(cls))
+        cls.logins = helpers.iRODSUserLogins(adm)
+        cls.logins.create_user(RODSUSER, 'abc123')
+
+
+    @classmethod
+    def tearDownClass(cls):
+        # TODO(#553): Skipping this will result in an interpreter seg fault for Py3.6 but not 3.11; why?
+        del cls.logins
+
+
     def setUp(self):
         self.sess = helpers.make_session()
-        self.test_coll_path = '/{}/home/{}/test_dir'.format(self.sess.zone, self.sess.username)
 
+        self.test_coll_path = '/{}/home/{}/test_dir'.format(self.sess.zone, self.sess.username)
         self.test_coll = self.sess.collections.create(self.test_coll_path)
 
 
@@ -380,6 +399,96 @@ def test_object_paths_with_dot_and_dotdot__323(self):
         home2 = normalize('/zone','holmes','..','home','public','..','user')
         self.assertEqual(home2, '/zone/home/user')
 
+    def test_update_mtime_of_collection_using_touch_operation_as_non_admin__525(self):
+        user_session = self.logins.session_for_user(RODSUSER)
+
+        # Capture mtime of the home collection.
+        home_collection_path = helpers.home_collection(user_session)
+        collection = user_session.collections.get(home_collection_path)
+        old_mtime = collection.modify_time
+
+        # Set the mtime to an earlier time.
+        new_mtime = 1400000000
+        user_session.collections.touch(home_collection_path, seconds_since_epoch=new_mtime)
+
+        # Compare mtimes for correctness.
+        collection = user_session.collections.get(home_collection_path)
+        self.assertEqual(datetime.utcfromtimestamp(new_mtime), collection.modify_time)
+        self.assertGreater(old_mtime, collection.modify_time)
+
+    def test_touch_operation_does_not_create_new_collections__525(self):
+        user_session = self.logins.session_for_user(RODSUSER)
+
+        # The collection should not exist.
+        collection_path = f'{helpers.home_collection(user_session)}/test_touch_operation_does_not_create_new_collections__525'
+        with self.assertRaises(CollectionDoesNotExist):
+            user_session.collections.get(collection_path)
+
+        # Show the touch operation throws an exception if the target collection
+        # does not exist.
+        with self.assertRaises(CollectionDoesNotExist):
+            user_session.collections.touch(collection_path)
+
+        # Show the touch operation did not create a new collection.
+        with self.assertRaises(CollectionDoesNotExist):
+            user_session.collections.get(collection_path)
+
+    def test_touch_operation_does_not_work_when_given_a_data_object__525(self):
+        try:
+            user_session = self.logins.session_for_user(RODSUSER)
+
+            # Create a data object.
+            data_object_path = f'{helpers.home_collection(user_session)}/test_touch_operation_does_not_work_when_given_a_data_object__525.txt'
+            self.assertFalse(user_session.data_objects.exists(data_object_path))
+            user_session.data_objects.touch(data_object_path)
+            self.assertTrue(user_session.data_objects.exists(data_object_path))
+
+            # Show the touch operation for collections throws an exception when
+            # given a path pointing to a data object.
+            with self.assertRaises(CollectionDoesNotExist):
+                user_session.collections.touch(data_object_path)
+
+        finally:
+            user_session.data_objects.unlink(data_object_path, force=True)
+
+    def test_touch_operation_ignores_unsupported_options__525(self):
+        user_session = self.logins.session_for_user(RODSUSER)
+        path = f'{helpers.home_collection(user_session)}/test_touch_operation_ignores_unsupported_options__525'
+
+        try:
+            # Capture mtime of the home collection.
+            collection = user_session.collections.create(path)
+            old_mtime = collection.modify_time
+
+            # Capture the current time.
+            time.sleep(2) # Guarantees the mtime is different.
+            new_mtime = int(time.time())
+
+            # The touch API for the iRODS server will attempt to create a new data object
+            # if the "no_create" option is set to false. The PRC's collection interface will
+            # ignore that option if passed.
+            #
+            # The following arguments don't make sense for collections and will also be ignored.
+            #
+            #   - replica_number
+            #   - leaf_resource_name
+            #
+            # They are included to prove the PRC handles them appropriately (i.e. unsupported
+            # parameters are removed from the request).
+            user_session.collections.touch(path,
+                                           no_create=False,
+                                           replica_number=525,
+                                           seconds_since_epoch=new_mtime,
+                                           leaf_resource_name='ufs525')
+
+            # Compare mtimes for correctness.
+            collection = user_session.collections.get(path)
+            self.assertEqual(datetime.utcfromtimestamp(int(new_mtime)), collection.modify_time)
+
+        finally:
+            if collection:
+                user_session.collections.remove(path, recurse=True, force=True)
+
 
 if __name__ == "__main__":
     # let the tests find the parent irods lib

irods/test/data_obj_test.py

@@ -1,5 +1,6 @@
 #! /usr/bin/env python
 from __future__ import absolute_import
+from datetime import datetime
 import base64
 import concurrent.futures
 import contextlib  # check if redundant
@@ -2044,6 +2045,42 @@ def test_append_mode_will_append_to_data_object__issue_495(self):
             if data.exists(testfile):
                 data.unlink(testfile,force=True)
 
+    def test_update_mtime_of_data_object_using_touch_operation_as_non_admin__525(self):
+        try:
+            user_session = self.logins.session_for_user(RODSUSER)
+
+            # Create a data object.
+            data_object_path = f'{helpers.home_collection(user_session)}/test_update_mtime_of_data_object_using_touch_operation__525.txt'
+            self.assertFalse(user_session.data_objects.exists(data_object_path))
+            user_session.data_objects.touch(data_object_path)
+            self.assertTrue(user_session.data_objects.exists(data_object_path))
+
+            # Capture mtime of data object.
+            data_object = user_session.data_objects.get(data_object_path)
+            old_mtime = data_object.replicas[0].modify_time
+
+            # Set the mtime to an earlier time.
+            new_mtime = 1400000000
+            user_session.data_objects.touch(data_object_path, seconds_since_epoch=new_mtime)
+
+            # Compare mtimes for correctness.
+            data_object = user_session.data_objects.get(data_object_path)
+            self.assertEqual(datetime.utcfromtimestamp(int(new_mtime)), data_object.replicas[0].modify_time)
+            self.assertGreater(old_mtime, data_object.replicas[0].modify_time)
+
+        finally:
+            if data_object:
+                user_session.data_objects.unlink(data_object.path, force=True)
+
+    def test_touch_operation_does_not_work_when_given_a_collection__525(self):
+        user_session = self.logins.session_for_user(RODSUSER)
+
+        # Show the touch operation for data objects throws an exception when
+        # given a path pointing to a collection.
+        home_collection_path = helpers.home_collection(user_session)
+        with self.assertRaises(ex.InvalidInputArgument):
+            user_session.data_objects.touch(home_collection_path)
+
 
 if __name__ == '__main__':
     # let the tests find the parent irods lib