-
-
Notifications
You must be signed in to change notification settings - Fork 3k
/
plugin_test.py
130 lines (103 loc) · 4.63 KB
/
plugin_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
# -*- coding: utf-8 -*-
"""
***************************************************************************
plugin_test.py
---------------------
Date : May 2017
Copyright : (C) 2017, Sandro Santilli
Email : strk at kbt dot io
***************************************************************************
* *
* This program is free software; you can redistribute it and/or modify *
* it under the terms of the GNU General Public License as published by *
* the Free Software Foundation; either version 2 of the License, or *
* (at your option) any later version. *
* *
***************************************************************************
"""
__author__ = 'Sandro Santilli'
__date__ = 'May 2017'
__copyright__ = '(C) 2017, Sandro Santilli'
# This will get replaced with a git SHA1 when you do a git archive
__revision__ = '$Format:%H$'
import os
import re
import qgis
from qgis.testing import start_app, unittest
from qgis.core import QgsDataSourceURI
from qgis.utils import iface
from qgis.PyQt.QtCore import QObject
start_app()
from db_manager.db_plugins.postgis.plugin import PostGisDBPlugin, PGRasterTable
from db_manager.db_plugins.postgis.plugin import PGDatabase
from db_manager.db_plugins.plugin import Table
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
def test_rasterTableGdalURI(self):
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(), '')
self.assertEqual(uri.database(), expected_dbname)
self.assertEqual(uri.service(), '')
check_rasterTableGdalURI(expected_dbname)
# 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__':
unittest.main()