From 95c2ee492a4fed929c8ed26f32d5b057f1c99009 Mon Sep 17 00:00:00 2001 From: Alan King Date: Thu, 1 Aug 2024 10:59:58 -0400 Subject: [PATCH] [#597] genquery2_test: Replace Postgres-specific assertions Many of the tests in genquery2_tests.py include assertions on the specific SQL strings generated by GenQuery2. These assertions assume a Postgres database, causing the tests to fail on non-Postgres databases which are nonetheless supported by the iRODS server. These assertions have been replaced. This changes the Postgres-specific assertions in many of the genquery2_test tests to just assert that a table name is returned. We cannot assert the specific contents without reaching out to the server to detect the version and flavor of the database. Even then, the specific results may vary over time. Testing the generated SQL is more the responsibility of the GenQuery2 parser and API in the server anyway, so this test should just be asserting that a result that sort of looks like what we want is being returned by the library. --- irods/test/genquery2_test.py | 41 +++++++++++++++--------------------- 1 file changed, 17 insertions(+), 24 deletions(-) diff --git a/irods/test/genquery2_test.py b/irods/test/genquery2_test.py index 334293e5..136a0580 100644 --- a/irods/test/genquery2_test.py +++ b/irods/test/genquery2_test.py @@ -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): @@ -19,7 +20,8 @@ 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) @@ -27,63 +29,54 @@ def tearDown(self): 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()