mirror of
https://github.com/qgis/QGIS.git
synced 2025-04-14 00:07:35 -04:00
Add DBManager test for service-only URI
Ref #16626 Also fixes runs with non-standard QGIS_PGTEST_DB env variable set
This commit is contained in:
parent
de9e70e6ba
commit
362d4f0697
@ -41,42 +41,89 @@ from db_manager.db_plugins.postgis.connector import PostGisDBConnector
|
||||
|
||||
class TestDBManagerPostgisPlugin(unittest.TestCase):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(self):
|
||||
self.old_pgdatabase_env = os.environ.get('PGDATABASE')
|
||||
self.testdb = os.environ.get('QGIS_PGTEST_DB') or 'qgis_test'
|
||||
os.environ['PGDATABASE'] = self.testdb
|
||||
|
||||
# Create temporary service file
|
||||
self.old_pgservicefile_env = os.environ.get('PGSERVICEFILE')
|
||||
self.tmpservicefile = '/tmp/qgis-test-{}-pg_service.conf'.format(os.getpid())
|
||||
os.environ['PGSERVICEFILE'] = self.tmpservicefile
|
||||
|
||||
f = open(self.tmpservicefile, "w")
|
||||
f.write("[dbmanager]\ndbname={}\n".format(self.testdb))
|
||||
# TODO: add more things if PGSERVICEFILE was already set ?
|
||||
f.close()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(self):
|
||||
# Restore previous env variables if needed
|
||||
if self.old_pgdatabase_env:
|
||||
os.environ['PGDATABASE'] = self.old_pgdatabase_env
|
||||
if self.old_pgservicefile_env:
|
||||
os.environ['PGSERVICEFILE'] = self.old_pgservicefile_env
|
||||
# Remove temporary service file
|
||||
os.unlink(self.tmpservicefile)
|
||||
|
||||
# See https://issues.qgis.org/issues/16625
|
||||
# and https://issues.qgis.org/issues/10600
|
||||
|
||||
def test_rasterTable(self):
|
||||
def test_rasterTableGdalURI(self):
|
||||
|
||||
testdb = os.environ.get('QGIS_PGTEST_DB') or 'qgis_test'
|
||||
os.environ['PGDATABASE'] = testdb
|
||||
def check_rasterTableGdalURI(expected_dbname):
|
||||
tables = database.tables()
|
||||
raster_tables_count = 0
|
||||
for tab in tables:
|
||||
if tab.type == Table.RasterType:
|
||||
raster_tables_count += 1
|
||||
gdalUri = tab.gdalUri()
|
||||
m = re.search(' dbname=([^ ]*) ', gdalUri)
|
||||
self.assertTrue(m)
|
||||
actual_dbname = m.group(1)
|
||||
self.assertEqual(actual_dbname, expected_dbname)
|
||||
#print(tab.type)
|
||||
#print(tab.quotedName())
|
||||
#print(tab)
|
||||
|
||||
# We need to make sure a database is created with at
|
||||
# least one raster table !
|
||||
self.assertEqual(raster_tables_count, 1)
|
||||
|
||||
obj = QObject() # needs to be kept alive
|
||||
|
||||
# Test for empty URI
|
||||
# See https://issues.qgis.org/issues/16625
|
||||
# and https://issues.qgis.org/issues/10600
|
||||
|
||||
expected_dbname = self.testdb
|
||||
os.environ['PGDATABASE'] = expected_dbname
|
||||
|
||||
database = PGDatabase(obj, QgsDataSourceUri())
|
||||
self.assertIsInstance(database, PGDatabase)
|
||||
|
||||
uri = database.uri()
|
||||
self.assertEqual(uri.host(), '')
|
||||
self.assertEqual(uri.username(), '')
|
||||
expected_user = os.environ.get('PGUSER') or os.environ.get('USER')
|
||||
expected_dbname = os.environ.get('PGDATABASE') or expected_user
|
||||
self.assertEqual(uri.database(), expected_dbname)
|
||||
self.assertEqual(uri.service(), '')
|
||||
|
||||
tables = database.tables()
|
||||
raster_tables_count = 0
|
||||
for tab in tables:
|
||||
if tab.type == Table.RasterType:
|
||||
raster_tables_count += 1
|
||||
gdalUri = tab.gdalUri()
|
||||
m = re.search(' dbname=([^ ]*) ', gdalUri)
|
||||
self.assertTrue(m)
|
||||
actual_dbname = m.group(1)
|
||||
self.assertEqual(actual_dbname, expected_dbname)
|
||||
#print(tab.type)
|
||||
#print(tab.quotedName())
|
||||
#print(tab)
|
||||
check_rasterTableGdalURI(expected_dbname)
|
||||
|
||||
# We need to make sure a database is created with at
|
||||
# least one raster table !
|
||||
self.assertEqual(raster_tables_count, 1)
|
||||
# Test for service-only URI
|
||||
# See https://issues.qgis.org/issues/16626
|
||||
|
||||
os.environ['PGDATABASE'] = 'fake'
|
||||
database = PGDatabase(obj, QgsDataSourceUri('service=dbmanager'))
|
||||
self.assertIsInstance(database, PGDatabase)
|
||||
|
||||
uri = database.uri()
|
||||
self.assertEqual(uri.host(), '')
|
||||
self.assertEqual(uri.username(), '')
|
||||
self.assertEqual(uri.database(), '')
|
||||
self.assertEqual(uri.service(), 'dbmanager')
|
||||
|
||||
check_rasterTableGdalURI(expected_dbname)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
Loading…
x
Reference in New Issue
Block a user