Skip to content

Commit

Permalink
[#597] genquery2_test: Replace Postgres-specific assertions
Browse files Browse the repository at this point in the history 
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.
  • Loading branch information
@alanking
alanking committed Aug 1, 2024
1 parent 459e643 commit 95c2ee4
Showing 1 changed file with 17 additions and 24 deletions.
41 changes: 17 additions & 24 deletions irods/test/genquery2_test.py
View file
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 comments on commit 95c2ee4

Please sign in to comment.