squash. add federation test

scripts/irods/test/test_federation.py

@@ -2379,3 +2379,95 @@ def test_sufficient_permissions_results_in_replication_and_no_error(self):
 							'DATA_REPL_STATUS', str(0)
 						])
 					self.user0.assert_icommand(['irm', '-f', logical_path])
+
+
+class Test_IQuery(SessionsMixin, unittest.TestCase):
+
+	# This test suite expects tempZone to contain the following users:
+	#
+	#   - rods#tempZone
+	#   - zonehopper#tempZone (created by the irods_testing_environment)
+	#   - zonehopper#otherZone (created by the irods_testing_environment)
+	#
+	# otherZone must contain the following users:
+	#
+	#   - rods#otherZone
+	#
+	# The test suite must be launched from a server in otherZone. That means tempZone
+	# is identified as the remote federated zone.
+	#
+	# Just before the tests are run, the base class will create three additional users
+	# in otherZone. The following users will appear:
+	#
+	#   - admin#otherZone
+	#   - zonehopper#otherZone
+	#   - zonehopper#tempZone (created by setUp()) TODO Remove
+
+	def setUp(self):
+		super(Test_IQuery, self).setUp()
+
+		# session.make_sessions_mixin() creates admins and users that connect to the host
+		# identified by lib.get_hostname().
+		#
+		# For this particular set of tests, that means these users connect to the host where
+		# "otherZone" runs.
+		self.local_admin = self.admin_sessions[0] # admin#otherZone
+		self.local_user = self.user_sessions[0]   # zonehopper#otherZone
+
+		# Create session for zonehopper#tempZone. The session will be connected to tempZone.
+		password = test.settings.FEDERATION.RODSUSER_NAME_PASSWORD_LIST[0][1]
+		host = test.settings.FEDERATION.REMOTE_HOST
+		zone = test.settings.FEDERATION.REMOTE_ZONE
+		self.remote_user = session.make_session_for_existing_user(self.local_user.username, password, host, zone)
+
+		# Create zonehopper#tempZone in otherZone. This must be handled by the test suite rather
+		# than the session.make_sessions_mixin() because that function only knows about the local zone.
+		self.remote_user_home_collection = self.remote_user.remote_home_collection(self.local_user.zone_name)
+		#self.local_admin.assert_icommand(['iadmin', 'mkuser', self.remote_user.qualified_username, 'rodsuser'])
+
+		# Validity check: If the remote user's home collection does not exist in the zone these tests
+		# are run from, then the testing environment is in a bad state.
+		#self.local_admin.assert_icommand(['ils', self.remote_user_home_collection], 'STDOUT', [self.remote_user.qualified_username])
+
+	def tearDown(self):
+		self.remote_user.__exit__()
+		#self.local_admin.run_icommand(['iadmin', 'rmuser', self.remote_user.qualified_username])
+		super(Test_IQuery, self).tearDown()
+
+	def test_iquery_correctly_honors_permissions_when_users_share_the_same_username_but_different_zone_names__issue_7570(self):
+		import json
+
+		remote_zone = test.settings.FEDERATION.REMOTE_ZONE
+
+		# Show the local user can find their remote home collection using iquery.
+		local_user_remote_home_collection = self.local_user.remote_home_collection(remote_zone)
+		query = f"select COLL_NAME where COLL_NAME = '{local_user_remote_home_collection}'"
+		self.local_user.assert_icommand(['iquery', '-z', remote_zone, query], 'STDOUT', [local_user_remote_home_collection])
+
+		# Show the local user cannot find the remote home collection of the remote user using iquery.
+		# This proves that iquery understands the permission model for collections even in federated environments.
+		query = f"select COLL_NAME where COLL_NAME = '{self.remote_user.home_collection}'"
+		self.local_user.assert_icommand(['iquery', '-z', remote_zone, query], 'STDOUT', ['[]'])
+
+		#
+		# Demonstrate the same thing using data objects.
+		#
+
+		# Create a data object in the remote user's home collection.
+		data_object = f'{self.remote_user.home_collection}/issue_7570.txt'
+		self.remote_user.assert_icommand(['itouch', data_object])
+
+		# Give the local user permission to read the contents of the remote user's remote home collection.
+		# This allows the local user to list the contents of the remote user's remote home collection.
+		self.remote_user.assert_icommand(['ichmod', 'read_object', self.local_user.qualified_username, self.remote_user.home_collection])
+
+		# Show the remote user can find the data object using iquery.
+		remote_coll_name = self.remote_user.home_collection
+		remote_data_name = os.path.basename(data_object)
+		query = f"select COLL_NAME, DATA_NAME where COLL_NAME = '{remote_coll_name}' and DATA_NAME = '{remote_data_name}'"
+		self.remote_user.assert_icommand(['iquery', query], 'STDOUT', [json.dumps([[remote_coll_name, remote_data_name]], separators=(',', ':'))])
+
+		# Show the local user cannot find the remote data object of the remote user using iquery.
+		# This proves that iquery understands the permission model for data objects even in federated environments.
+		query = f"select COLL_NAME, DATA_NAME where COLL_NAME = '{remote_coll_name}' and DATA_NAME = '{remote_data_name}'"
+		self.local_user.assert_icommand(['iquery', '-z', remote_zone, query], 'STDOUT', ['[]'])