[ 7444, 7465] Add tests for irepl at various permission levels

There is already a test for admin users being able to replicate with
insufficient permissions: test_irepl_data_object_with_no_permission__4479

This commit adds a test for a regular rodsuser attempting to replicate
a data object on which they have all the various permission levels
available. There is a set that does not allow the user to even see the
data object exists, a level that lets the user see it but not replicate,
and a level that lets the user replicate. Each permission level exists in
a distinct unittest.subTest.

scripts/core_tests_list.json

@@ -74,6 +74,7 @@
     "test_ireg.test_ireg_options",
     "test_ireg.test_ireg_replica",
     "test_irepl.Test_Irepl",
+    "test_irepl.test_all_permission_levels__issue__7444_7465",
     "test_irepl.test_invalid_parameters",
     "test_irepl.test_irepl_repl_status",
     "test_irepl.test_irepl_replication_hierarchy",

scripts/irods/test/test_irepl.py

@@ -1878,3 +1878,139 @@ def test_irepl_Sf_Rd_foo(self):
            #{'start':{'a':'X', 'b':'X', 'c':'X', 'f':'-'}, 'end':{'a':'X', 'b':'X', 'c':'X', 'f':'-'}, 'output':{'out':None, 'err':None, 'rc':None}}, # dXXXf-
            #{'start':{'a':'X', 'b':'X', 'c':'X', 'f':'&'}, 'end':{'a':'X', 'b':'X', 'c':'X', 'f':'&'}, 'output':{'out':None, 'err':None, 'rc':None}}, # dXXXf&
            #{'start':{'a':'X', 'b':'X', 'c':'X', 'f':'X'}, 'end':{'a':'X', 'b':'X', 'c':'X', 'f':'X'}, 'output':{'out':None, 'err':None, 'rc':None}}  # dXXXfX
+
+
+class test_all_permission_levels__issue__7444_7465(unittest.TestCase):
+    @classmethod
+    def setUpClass(self):
+        self.user0 = session.mkuser_and_return_session('rodsuser', 'smeagol', 'spass', lib.get_hostname())
+        self.user1 = session.mkuser_and_return_session('rodsuser', 'bilbo', 'bpass', lib.get_hostname())
+
+        # Give other user ownership of the session collection so we can focus on object permissions.
+        self.user0.assert_icommand(['ichmod', 'own', self.user1.username, self.user0.session_collection])
+
+        self.target_resource = 'target_resource'
+        self.other_resource = 'other_resource'
+
+        with session.make_session_for_existing_admin() as admin_session:
+            lib.create_ufs_resource(admin_session, self.target_resource, hostname=test.settings.HOSTNAME_2)
+            lib.create_ufs_resource(admin_session, self.other_resource, hostname=test.settings.HOSTNAME_3)
+
+    @classmethod
+    def tearDownClass(self):
+        self.user0.__exit__()
+        self.user1.__exit__()
+
+        with session.make_session_for_existing_admin() as admin_session:
+            admin_session.assert_icommand(['iadmin', 'rmuser', self.user0.username])
+            admin_session.assert_icommand(['iadmin', 'rmuser', self.user1.username])
+            lib.remove_resource(admin_session, self.target_resource)
+            lib.remove_resource(admin_session, self.other_resource)
+
+    def test_permissions_that_do_not_allow_user_to_see_object_results_in_no_replication_and_an_error(self):
+        logical_path = os.path.join(self.user0.session_collection, 'this_object_is_invisible')
+        permissions = ['null', 'read_metadata']
+
+        for permission in permissions:
+            with self.subTest(permission):
+                try:
+                    # Create a data object and make sure it is good.
+                    self.user0.assert_icommand(['itouch', '-R', self.target_resource, logical_path])
+                    self.assertTrue(lib.replica_exists_on_resource(self.user0, logical_path, self.target_resource))
+                    self.assertEqual(str(1), lib.get_replica_status(self.user0, os.path.basename(logical_path), 0))
+
+                    self.user0.assert_icommand(['ichmod', permission, self.user1.username, logical_path])
+
+                    # Try to replicate the data object and fail because user1 cannot even see the data object.
+                    self.user1.assert_icommand(
+                        ['irepl', '-R', self.other_resource, logical_path], 'STDERR', 'does not exist')
+                    self.assertTrue(lib.replica_exists_on_resource(self.user0, logical_path, self.target_resource))
+                    self.assertEqual(str(1), lib.get_replica_status(self.user0, os.path.basename(logical_path), 0))
+                    self.assertFalse(lib.replica_exists_on_resource(self.user0, logical_path, self.other_resource))
+
+                finally:
+                    # In the case where this test does not pass (i.e. REGRESSION) it is possible for the replicas to be
+                    # stuck in the intermediate or write-locked status. We set the status at the end here to ensure
+                    # that the object can be removed.
+                    with session.make_session_for_existing_admin() as admin_session:
+                        admin_session.assert_icommand(['ils', '-L', os.path.dirname(logical_path)], 'STDOUT')
+                        for replica_number in range(2):
+                            admin_session.run_icommand([
+                                'iadmin', 'modrepl',
+                                'logical_path', logical_path,
+                                'replica_number', str(replica_number),
+                                'DATA_REPL_STATUS', str(0)
+                            ])
+                    self.user0.assert_icommand(['irm', '-f', logical_path])
+
+    def test_insufficient_permissions_results_in_no_replication_and_an_error(self):
+        logical_path = os.path.join(self.user0.session_collection, 'this_object_will_not_be_replicated')
+        permissions = ['read_object', 'create_metadata', 'modify_metadata', 'delete_metadata', 'create_object']
+
+        for permission in permissions:
+            with self.subTest(permission):
+                try:
+                    # Create a data object and make sure it is good.
+                    self.user0.assert_icommand(['itouch', '-R', self.target_resource, logical_path])
+                    self.assertTrue(lib.replica_exists_on_resource(self.user0, logical_path, self.target_resource))
+                    self.assertEqual(str(1), lib.get_replica_status(self.user0, os.path.basename(logical_path), 0))
+
+                    self.user0.assert_icommand(['ichmod', permission, self.user1.username, logical_path])
+
+                    # Try to replicate the data object and ensure that it fails with an error.
+                    self.user1.assert_icommand(
+                        ['irepl', '-R', self.other_resource, logical_path], 'STDERR', 'SYS_USER_NO_PERMISSION')
+                    self.assertTrue(lib.replica_exists_on_resource(self.user0, logical_path, self.target_resource))
+                    self.assertEqual(str(1), lib.get_replica_status(self.user0, os.path.basename(logical_path), 0))
+                    self.assertFalse(lib.replica_exists_on_resource(self.user0, logical_path, self.other_resource))
+
+                finally:
+                    # In the case where this test does not pass (i.e. REGRESSION) it is possible for the replicas to be
+                    # stuck in the intermediate or write-locked status. We set the status at the end here to ensure
+                    # that the object can be removed.
+                    with session.make_session_for_existing_admin() as admin_session:
+                        admin_session.assert_icommand(['ils', '-L', os.path.dirname(logical_path)], 'STDOUT')
+                        for replica_number in range(2):
+                            admin_session.run_icommand([
+                                'iadmin', 'modrepl',
+                                'logical_path', logical_path,
+                                'replica_number', str(replica_number),
+                                'DATA_REPL_STATUS', str(0)
+                            ])
+                    self.user0.assert_icommand(['irm', '-f', logical_path])
+
+    def test_sufficient_permissions_results_in_replication_and_no_error(self):
+        logical_path = os.path.join(self.user0.session_collection, 'this_object_will_be_replicated')
+        permissions = ['modify_object', 'delete_object', 'own']
+
+        for permission in permissions:
+            with self.subTest(permission):
+                try:
+                    # Create a data object and make sure it is good.
+                    self.user0.assert_icommand(['itouch', '-R', self.target_resource, logical_path])
+                    self.assertTrue(lib.replica_exists_on_resource(self.user0, logical_path, self.target_resource))
+                    self.assertEqual(str(1), lib.get_replica_status(self.user0, os.path.basename(logical_path), 0))
+
+                    self.user0.assert_icommand(['ichmod', permission, self.user1.username, logical_path])
+
+                    # Try to replicate the data object and ensure that it completes successfully.
+                    self.user1.assert_icommand(['irepl', '-R', self.other_resource, logical_path])
+                    self.assertTrue(lib.replica_exists_on_resource(self.user0, logical_path, self.target_resource))
+                    self.assertEqual(str(1), lib.get_replica_status(self.user0, os.path.basename(logical_path), 0))
+                    self.assertTrue(lib.replica_exists_on_resource(self.user0, logical_path, self.other_resource))
+                    self.assertEqual(str(1), lib.get_replica_status(self.user0, os.path.basename(logical_path), 1))
+
+                finally:
+                    # In the case where this test does not pass (i.e. REGRESSION) it is possible for the replicas to be
+                    # stuck in the intermediate or write-locked status. We set the status at the end here to ensure
+                    # that the object can be removed.
+                    with session.make_session_for_existing_admin() as admin_session:
+                        admin_session.assert_icommand(['ils', '-L', os.path.dirname(logical_path)], 'STDOUT')
+                        for replica_number in range(2):
+                            admin_session.run_icommand([
+                                'iadmin', 'modrepl',
+                                'logical_path', logical_path,
+                                'replica_number', str(replica_number),
+                                'DATA_REPL_STATUS', str(0)
+                            ])
+                    self.user0.assert_icommand(['irm', '-f', logical_path])