|
1 | 1 | # -*- coding: utf-8 -*-
|
2 |
| -"""QGIS Unit tests for the postgres provider. |
| 2 | +"""QGIS Unit tests for QgsSpatialiteProvider |
3 | 3 |
|
4 | 4 | .. note:: This program is free software; you can redistribute it and/or modify
|
5 | 5 | it under the terms of the GNU General Public License as published by
|
6 | 6 | the Free Software Foundation; either version 2 of the License, or
|
7 | 7 | (at your option) any later version.
|
8 | 8 | """
|
9 |
| -__author__ = 'Matthias Kuhn' |
10 |
| -__date__ = '2015-04-23' |
11 |
| -__copyright__ = 'Copyright 2015, The QGIS Project' |
| 9 | +__author__ = 'Vincent Mora' |
| 10 | +__date__ = '09/07/2013' |
| 11 | +__copyright__ = 'Copyright 2013, The QGIS Project' |
12 | 12 | # This will get replaced with a git SHA1 when you do a git archive
|
13 | 13 | __revision__ = '$Format:%H$'
|
14 | 14 |
|
15 | 15 | import qgis
|
16 | 16 | import os
|
| 17 | +import tempfile |
| 18 | +import sys |
| 19 | + |
| 20 | +from qgis.core import QgsVectorLayer, QgsPoint, QgsFeature |
17 | 21 |
|
18 |
| -from qgis.core import QgsVectorLayer, QgsFeatureRequest, QgsFeature, QgsProviderRegistry |
19 |
| -from PyQt4.QtCore import QSettings |
20 | 22 | from utilities import (unitTestDataPath,
|
21 | 23 | getQgisTestApp,
|
22 |
| - unittest, |
23 |
| - TestCase |
| 24 | + TestCase, |
| 25 | + unittest |
24 | 26 | )
|
25 | 27 | from providertestbase import ProviderTestCase
|
26 | 28 |
|
| 29 | +try: |
| 30 | + from pyspatialite import dbapi2 as sqlite3 |
| 31 | +except ImportError: |
| 32 | + print "You should install pyspatialite to run the tests" |
| 33 | + raise ImportError |
| 34 | + |
| 35 | +# Convenience instances in case you may need them |
27 | 36 | QGISAPP, CANVAS, IFACE, PARENT = getQgisTestApp()
|
28 | 37 | TEST_DATA_DIR = unitTestDataPath()
|
29 | 38 |
|
30 |
| -class TestPyQgsPostgresProvider(TestCase, ProviderTestCase): |
| 39 | + |
| 40 | +def die(error_message): |
| 41 | + raise Exception(error_message) |
| 42 | + |
| 43 | +class TestQgsSpatialiteProvider(TestCase, ProviderTestCase): |
| 44 | + |
31 | 45 | @classmethod
|
32 | 46 | def setUpClass(cls):
|
33 | 47 | """Run before all tests"""
|
34 |
| - # Create test layer |
| 48 | + # setup provider for base tests |
35 | 49 | cls.vl = QgsVectorLayer( 'dbname=\'{}/provider/spatialite.db\' table="somedata" (geometry) sql='.format(TEST_DATA_DIR), 'test', 'spatialite' )
|
36 | 50 | assert(cls.vl.isValid())
|
37 | 51 | cls.provider = cls.vl.dataProvider()
|
38 | 52 |
|
| 53 | + # create test db |
| 54 | + cls.dbname = os.path.join( tempfile.gettempdir(), "test.sqlite" ) |
| 55 | + if os.path.exists( cls.dbname ): |
| 56 | + os.remove( cls.dbname ) |
| 57 | + con = sqlite3.connect(cls.dbname, isolation_level=None) |
| 58 | + cur = con.cursor() |
| 59 | + cur.execute( "BEGIN" ) |
| 60 | + sql = "SELECT InitSpatialMetadata()" |
| 61 | + cur.execute(sql) |
| 62 | + |
| 63 | + # simple table with primary key |
| 64 | + sql = "CREATE TABLE test_pg (id INTEGER NOT NULL PRIMARY KEY, name TEXT NOT NULL)" |
| 65 | + cur.execute(sql) |
| 66 | + sql = "SELECT AddGeometryColumn('test_pg', 'geometry', 4326, 'POLYGON', 'XY')" |
| 67 | + cur.execute(sql) |
| 68 | + sql = "INSERT INTO test_pg (id, name, geometry) " |
| 69 | + sql += "VALUES (1, 'toto', GeomFromText('POLYGON((0 0,1 0,1 1,0 1,0 0))', 4326))" |
| 70 | + cur.execute(sql) |
| 71 | + |
| 72 | + # table with multiple column primary key |
| 73 | + sql = "CREATE TABLE test_pg_mk (id INTEGER NOT NULL, name TEXT NOT NULL, PRIMARY KEY(id,name))" |
| 74 | + cur.execute(sql) |
| 75 | + sql = "SELECT AddGeometryColumn('test_pg_mk', 'geometry', 4326, 'POLYGON', 'XY')" |
| 76 | + cur.execute(sql) |
| 77 | + sql = "INSERT INTO test_pg_mk (id, name, geometry) " |
| 78 | + sql += "VALUES (1, 'toto', GeomFromText('POLYGON((0 0,1 0,1 1,0 1,0 0))', 4326))" |
| 79 | + cur.execute(sql) |
| 80 | + |
| 81 | + # simple table with primary key |
| 82 | + sql = "CREATE TABLE test_q (id INTEGER NOT NULL PRIMARY KEY, name TEXT NOT NULL)" |
| 83 | + cur.execute(sql) |
| 84 | + sql = "SELECT AddGeometryColumn('test_q', 'geometry', 4326, 'POLYGON', 'XY')" |
| 85 | + cur.execute(sql) |
| 86 | + sql = "INSERT INTO test_q (id, name, geometry) " |
| 87 | + sql += "VALUES (11, 'toto', GeomFromText('POLYGON((0 0,1 0,1 1,0 1,0 0))', 4326))" |
| 88 | + cur.execute(sql) |
| 89 | + sql = "INSERT INTO test_q (id, name, geometry) " |
| 90 | + sql += "VALUES (21, 'toto', GeomFromText('POLYGON((0 0,1 0,1 1,0 1,0 0))', 4326))" |
| 91 | + cur.execute(sql) |
| 92 | + |
| 93 | + # simple table with a geometry column named 'Geometry' |
| 94 | + sql = "CREATE TABLE test_n (Id INTEGER NOT NULL PRIMARY KEY, name TEXT NOT NULL)" |
| 95 | + cur.execute(sql) |
| 96 | + sql = "SELECT AddGeometryColumn('test_n', 'Geometry', 4326, 'POLYGON', 'XY')" |
| 97 | + cur.execute(sql) |
| 98 | + sql = "INSERT INTO test_n (id, name, geometry) " |
| 99 | + sql += "VALUES (1, 'toto', GeomFromText('POLYGON((0 0,1 0,1 1,0 1,0 0))', 4326))" |
| 100 | + cur.execute(sql) |
| 101 | + sql = "INSERT INTO test_n (id, name, geometry) " |
| 102 | + sql += "VALUES (2, 'toto', GeomFromText('POLYGON((0 0,1 0,1 1,0 1,0 0))', 4326))" |
| 103 | + cur.execute(sql) |
| 104 | + |
| 105 | + cur.execute( "COMMIT" ) |
| 106 | + con.close() |
| 107 | + |
39 | 108 | @classmethod
|
40 | 109 | def tearDownClass(cls):
|
41 | 110 | """Run after all tests"""
|
| 111 | + # for the time being, keep the file to check with qgis |
| 112 | + #if os.path.exists(cls.dbname) : |
| 113 | + # os.remove(cls.dbname) |
| 114 | + pass |
| 115 | + |
| 116 | + def setUp(self): |
| 117 | + """Run before each test.""" |
| 118 | + pass |
| 119 | + |
| 120 | + def tearDown(self): |
| 121 | + """Run after each test.""" |
| 122 | + pass |
| 123 | + |
| 124 | + def test_SplitFeature(self): |
| 125 | + """Create spatialite database""" |
| 126 | + layer = QgsVectorLayer("dbname=%s table=test_pg (geometry)" % self.dbname, "test_pg", "spatialite") |
| 127 | + assert(layer.isValid()) |
| 128 | + assert(layer.hasGeometryType()) |
| 129 | + layer.startEditing() |
| 130 | + layer.splitFeatures([QgsPoint(0.5, -0.5), QgsPoint(0.5, 1.5)], 0)==0 or die("error in split") |
| 131 | + layer.splitFeatures([QgsPoint(-0.5, 0.5), QgsPoint(1.5, 0.5)], 0)==0 or die("error in split") |
| 132 | + if not layer.commitChanges(): |
| 133 | + die("this commit should work") |
| 134 | + layer.featureCount() == 4 or die("we should have 4 features after 2 split") |
| 135 | + |
| 136 | + def xtest_SplitFeatureWithFailedCommit(self): |
| 137 | + """Create spatialite database""" |
| 138 | + layer = QgsVectorLayer("dbname=%s table=test_pg_mk (geometry)" % self.dbname, "test_pg_mk", "spatialite") |
| 139 | + assert(layer.isValid()) |
| 140 | + assert(layer.hasGeometryType()) |
| 141 | + layer.startEditing() |
| 142 | + layer.splitFeatures([QgsPoint(0.5, -0.5), QgsPoint(0.5, 1.5)], 0)==0 or die("error in split") |
| 143 | + layer.splitFeatures([QgsPoint(-0.5, 0.5), QgsPoint(1.5, 0.5)], 0)==0 or die("error in split") |
| 144 | + if layer.commitChanges(): |
| 145 | + die("this commit should fail") |
| 146 | + layer.rollBack() |
| 147 | + feat = QgsFeature() |
| 148 | + it=layer.getFeatures() |
| 149 | + it.nextFeature(feat) |
| 150 | + ref = [[(0,0), (1,0), (1,1), (0,1), (0,0)]] |
| 151 | + res = feat.geometry().asPolygon() |
| 152 | + for ring1, ring2 in zip(ref, res): |
| 153 | + for p1, p2 in zip(ring1, ring2): |
| 154 | + for c1, c2 in zip(p1, p2): |
| 155 | + c1 == c2 or die("polygon has been altered by failed edition") |
| 156 | + |
| 157 | + def test_queries(self): |
| 158 | + """Test loading of query-based layers""" |
| 159 | + |
| 160 | + # a query with a geometry, but no unique id |
| 161 | + # the id will be autoincremented |
| 162 | + l = QgsVectorLayer("dbname=%s table='(select * from test_q)' (geometry)" % self.dbname, "test_pg_query1", "spatialite") |
| 163 | + assert(l.isValid()) |
| 164 | + # the id() is autoincremented |
| 165 | + sum_id1 = sum(f.id() for f in l.getFeatures()) |
| 166 | + # the attribute 'id' works |
| 167 | + sum_id2 = sum(f.attributes()[0] for f in l.getFeatures()) |
| 168 | + assert(sum_id1 == 3) # 1+2 |
| 169 | + assert(sum_id2 == 32) # 11 + 21 |
| 170 | + |
| 171 | + # and now with an id declared |
| 172 | + l = QgsVectorLayer("dbname=%s table='(select * from test_q)' (geometry) key='id'" % self.dbname, "test_pg_query1", "spatialite") |
| 173 | + assert(l.isValid()) |
| 174 | + sum_id1 = sum(f.id() for f in l.getFeatures()) |
| 175 | + sum_id2 = sum(f.attributes()[0] for f in l.getFeatures()) |
| 176 | + assert(sum_id1 == 32) |
| 177 | + assert(sum_id2 == 32) |
| 178 | + |
| 179 | + # a query, but no geometry |
| 180 | + l = QgsVectorLayer("dbname=%s table='(select id,name from test_q)' key='id'" % self.dbname, "test_pg_query1", "spatialite") |
| 181 | + assert(l.isValid()) |
| 182 | + sum_id1 = sum(f.id() for f in l.getFeatures()) |
| 183 | + sum_id2 = sum(f.attributes()[0] for f in l.getFeatures()) |
| 184 | + assert(sum_id1 == 32) |
| 185 | + assert(sum_id2 == 32) |
| 186 | + |
| 187 | + def test_case(self): |
| 188 | + """Test case sensitivity issues""" |
| 189 | + l = QgsVectorLayer("dbname=%s table='test_n' (geometry) key='id'" % self.dbname, "test_n1", "spatialite") |
| 190 | + assert(l.isValid()) |
| 191 | + assert(l.dataProvider().fields().count() == 2) |
| 192 | + fields = [f.name() for f in l.dataProvider().fields()] |
| 193 | + assert('Geometry' not in fields) |
| 194 | + |
42 | 195 |
|
43 | 196 | if __name__ == '__main__':
|
44 | 197 | unittest.main()
|
0 commit comments