connector_test.py 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778
  1. """
  2. ***************************************************************************
  3. connector_test.py
  4. ---------------------
  5. Date : May 2017
  6. Copyright : (C) 2017, Sandro Santilli
  7. Email : strk at kbt dot io
  8. ***************************************************************************
  9. * *
  10. * This program is free software; you can redistribute it and/or modify *
  11. * it under the terms of the GNU General Public License as published by *
  12. * the Free Software Foundation; either version 2 of the License, or *
  13. * (at your option) any later version. *
  14. * *
  15. ***************************************************************************
  16. """
  17. __author__ = 'Sandro Santilli'
  18. __date__ = 'May 2017'
  19. __copyright__ = '(C) 2017, Sandro Santilli'
  20. import os
  21. import qgis
  22. import unittest
  23. from qgis.testing import start_app, QgisTestCase
  24. from qgis.core import QgsDataSourceUri
  25. from qgis.utils import iface
  26. from qgis.PyQt.QtCore import QObject
  27. start_app()
  28. from db_manager.db_plugins.postgis.connector import PostGisDBConnector
  29. class TestDBManagerPostgisConnector(QgisTestCase):
  30. # def setUpClass():
  31. def _getUser(self, connector):
  32. r = connector._execute(None, "SELECT USER")
  33. val = connector._fetchone(r)[0]
  34. connector._close_cursor(r)
  35. return val
  36. def _getDatabase(self, connector):
  37. r = connector._execute(None, "SELECT current_database()")
  38. val = connector._fetchone(r)[0]
  39. connector._close_cursor(r)
  40. return val
  41. # See https://github.com/qgis/QGIS/issues/24525
  42. # and https://github.com/qgis/QGIS/issues/19005
  43. def test_dbnameLessURI(self):
  44. obj = QObject() # needs to be kept alive
  45. obj.connectionName = lambda: 'fake'
  46. obj.providerName = lambda: 'postgres'
  47. c = PostGisDBConnector(QgsDataSourceUri(), obj)
  48. self.assertIsInstance(c, PostGisDBConnector)
  49. uri = c.uri()
  50. # No username was passed, so we expect it to be taken
  51. # from PGUSER or USER environment variables
  52. expected_user = os.environ.get('PGUSER') or os.environ.get('USER')
  53. actual_user = self._getUser(c)
  54. self.assertEqual(actual_user, expected_user)
  55. # No database was passed, so we expect it to be taken
  56. # from PGDATABASE or expected user
  57. expected_db = os.environ.get('PGDATABASE') or expected_user
  58. actual_db = self._getDatabase(c)
  59. self.assertEqual(actual_db, expected_db)
  60. # TODO: add service-only test (requires a ~/.pg_service.conf file)
  61. if __name__ == '__main__':
  62. unittest.main()