Skip to content

Some test and test infrastructure fixes and improvements #601

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Aug 1, 2024
Merged

Some test and test infrastructure fixes and improvements #601

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
199c84d
[#566] Rename login_auth_test.py to prevent running with full suite
alanking Aug 1, 2024
e3c870b
[#597] genquery2_test: Replace Postgres-specific assertions
alanking Aug 1, 2024
313fc2b
[#600] Comment out test_files_query_case_sensitive BETWEEN tests
alanking Aug 1, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 17 additions & 24 deletions irods/test/genquery2_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,8 @@

class TestGenQuery2(unittest.TestCase):

def setUp(self):
@classmethod
def setUpClass(self):
self.sess = helpers.make_session()

if self.sess.server_version < (4, 3, 2):
Expand All @@ -19,71 +20,63 @@ def setUp(self):
self.sess.collections.create(self.coll_path_a)
self.sess.collections.create(self.coll_path_b)

def tearDown(self):
@classmethod
def tearDownClass(self):
'''Remove test data and close connections
'''
self.sess.collections.remove(self.coll_path_a, force=True)
self.sess.collections.remove(self.coll_path_b, force=True)
self.sess.cleanup()

def test_select(self):
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}'".format(
self.coll_path_a)
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}'".format(self.coll_path_a)
q = self.sess.genquery2_object()
query_result = q.execute(query)
query_sql = q.get_sql(query)
self.assertIn([self.coll_path_a], query_result)
self.assertEqual(len(query_result), 1)
# This assumes the iCAT database runs on PostgreSQL
self.assertEqual(query_sql, "select distinct t0.coll_name from R_COLL_MAIN t0 inner join R_OBJT_ACCESS pcoa on t0.coll_id = pcoa.object_id inner join R_TOKN_MAIN pct on pcoa.access_type_id = pct.token_id inner join R_USER_MAIN pcu on pcoa.user_id = pcu.user_id where t0.coll_name = ? and pcoa.access_type_id >= 1000 fetch first 256 rows only")
# Use upper() here in case GenQuery2 returns lowercase table names in a future implementation.
self.assertIn('R_COLL_MAIN', q.get_sql(query).upper())

def test_select_with_explicit_zone(self):
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}'".format(
self.coll_path_a)
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}'".format(self.coll_path_a)
q = self.sess.genquery2_object()
query_result = q.execute(query, zone=self.sess.zone)
query_sql = q.get_sql(query, zone=self.sess.zone)
self.assertIn([self.coll_path_a], query_result)
self.assertEqual(len(query_result), 1)
# This assumes the iCAT database runs on PostgreSQL
self.assertEqual(query_sql, "select distinct t0.coll_name from R_COLL_MAIN t0 inner join R_OBJT_ACCESS pcoa on t0.coll_id = pcoa.object_id inner join R_TOKN_MAIN pct on pcoa.access_type_id = pct.token_id inner join R_USER_MAIN pcu on pcoa.user_id = pcu.user_id where t0.coll_name = ? and pcoa.access_type_id >= 1000 fetch first 256 rows only")
# Use upper() here in case GenQuery2 returns lowercase table names in a future implementation.
self.assertIn('R_COLL_MAIN', q.get_sql(query).upper())

def test_select_with_shorthand(self):
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}'".format(
self.coll_path_a)
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}'".format(self.coll_path_a)
query_result = self.sess.genquery2(query)
self.assertIn([self.coll_path_a], query_result)
self.assertEqual(len(query_result), 1)

def test_select_with_shorthand_and_explicit_zone(self):
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}'".format(
self.coll_path_a)
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}'".format(self.coll_path_a)
query_result = self.sess.genquery2(query, zone=self.sess.zone)
self.assertIn([self.coll_path_a], query_result)
self.assertEqual(len(query_result), 1)

def test_select_or(self):
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}' OR COLL_NAME = '{}'".format(
self.coll_path_a, self.coll_path_b)
query = "SELECT COLL_NAME WHERE COLL_NAME = '{}' OR COLL_NAME = '{}'".format(self.coll_path_a, self.coll_path_b)
q = self.sess.genquery2_object()
query_result = q.execute(query)
query_sql = q.get_sql(query)
self.assertIn([self.coll_path_a], query_result)
self.assertIn([self.coll_path_b], query_result)
self.assertEqual(len(query_result), 2)
# This assumes the iCAT database runs on PostgreSQL
self.assertEqual(query_sql, "select distinct t0.coll_name from R_COLL_MAIN t0 inner join R_OBJT_ACCESS pcoa on t0.coll_id = pcoa.object_id inner join R_TOKN_MAIN pct on pcoa.access_type_id = pct.token_id inner join R_USER_MAIN pcu on pcoa.user_id = pcu.user_id where t0.coll_name = ? or t0.coll_name = ? and pcoa.access_type_id >= 1000 fetch first 256 rows only")
# Use upper() here in case GenQuery2 returns lowercase table names in a future implementation.
self.assertIn('R_COLL_MAIN', q.get_sql(query).upper())

def test_select_and(self):
query = "SELECT COLL_NAME WHERE COLL_NAME LIKE '{}' AND COLL_NAME LIKE '{}'".format(
"%test_query2_coll%", "%query2_coll_a%")
q = self.sess.genquery2_object()
query_result = q.execute(query)
query_sql = q.get_sql(query)
self.assertIn([self.coll_path_a], query_result)
self.assertEqual(len(query_result), 1)
# This assumes the iCAT database runs on PostgreSQL
self.assertEqual(query_sql, "select distinct t0.coll_name from R_COLL_MAIN t0 inner join R_OBJT_ACCESS pcoa on t0.coll_id = pcoa.object_id inner join R_TOKN_MAIN pct on pcoa.access_type_id = pct.token_id inner join R_USER_MAIN pcu on pcoa.user_id = pcu.user_id where t0.coll_name like ? and t0.coll_name like ? and pcoa.access_type_id >= 1000 fetch first 256 rows only")
# Use upper() here in case GenQuery2 returns lowercase table names in a future implementation.
self.assertIn('R_COLL_MAIN', q.get_sql(query).upper())

def test_column_mappings(self):
q = self.sess.genquery2_object()
Expand Down
0 irods/test/login_auth_test.py → ...test/login_auth_test_must_run_manually.py
View file
Open in desktop
File renamed without changes.
39 changes: 21 additions & 18 deletions irods/test/query_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,24 +180,27 @@ def test_files_query_case_sensitive(self):

# BETWEEN tests

result13 = self.sess.query(DataObject.name).filter(
Collection.name == self.coll_path).filter(
Between(DataObject.name, [self.case_sensitive_obj_name1,
self.case_sensitive_obj_name1 + "_"])).all()
self.assertTrue(result13.has_value(self.case_sensitive_obj_name1))
self.assertEqual(len(result13), 1)

result14 = self.sess.query(DataObject.name).filter(
Collection.name == self.coll_path).filter(
Between(DataObject.name, [str.lower(self.case_sensitive_obj_name1),
str.lower(self.case_sensitive_obj_name1) + "_"])).all()
self.assertEqual(len(result14), 0)

result15 = self.sess.query(DataObject.name).filter(
Collection.name == self.coll_path).filter(
Between(DataObject.name, [str.upper(self.case_sensitive_obj_name1),
str.upper(self.case_sensitive_obj_name1) + "_"])).all()
self.assertEqual(len(result15), 0)
# TODO(#600): Uncomment these lines and/or make a new test when database flavor can be detected.
# The resultset for BETWEEN queries can differ from database to database.

#result13 = self.sess.query(DataObject.name).filter(
# Collection.name == self.coll_path).filter(
# Between(DataObject.name, [self.case_sensitive_obj_name1,
# self.case_sensitive_obj_name1 + "_"])).all()
#self.assertTrue(result13.has_value(self.case_sensitive_obj_name1))
#self.assertEqual(len(result13), 1)

#result14 = self.sess.query(DataObject.name).filter(
# Collection.name == self.coll_path).filter(
# Between(DataObject.name, [str.lower(self.case_sensitive_obj_name1),
# str.lower(self.case_sensitive_obj_name1) + "_"])).all()
#self.assertEqual(len(result14), 0)

#result15 = self.sess.query(DataObject.name).filter(
# Collection.name == self.coll_path).filter(
# Between(DataObject.name, [str.upper(self.case_sensitive_obj_name1),
# str.upper(self.case_sensitive_obj_name1) + "_"])).all()
#self.assertEqual(len(result15), 0)

def test_files_query_case_insensitive(self):
# This tests that GenQueries are case-insensitive when the case_sensitive
Expand Down