/[thuban]/branches/WIP-pyshapelib-bramz/test/test_postgis_db.py
ViewVC logotype

Annotation of /branches/WIP-pyshapelib-bramz/test/test_postgis_db.py

Parent Directory Parent Directory | Revision Log Revision Log


Revision 1638 - (hide annotations)
Fri Aug 22 18:19:14 2003 UTC (21 years, 6 months ago) by bh
Original Path: trunk/thuban/test/test_postgis_db.py
File MIME type: text/x-python
File size: 16288 byte(s)
Implement saving a session with a postgis connection

* Resources/XML/thuban-0.9.dtd (dbconnection, dbshapesource) New
elements for database connections and shapestores using db
connections
(session): Add the dbconnections to the content model

* Thuban/Model/save.py (SessionSaver.write_db_connections): New.
Write the db connections
(SessionSaver.write_session): Call write_db_connections to write
the connection before the data sources
(SessionSaver.write_data_containers): Handle postgis shapestores

* test/test_save.py (SaveSessionTest.thubanids)
(SaveSessionTest.thubanidrefs): Update for new DTD
(SaveSessionTest.test_save_postgis): New. Test saving a session
with postgis connections

* Thuban/Model/postgisdb.py (PostGISTable.DBConnection)
(PostGISTable.TableName): New accessor methods for the connection
and table name

* test/test_postgis_db.py (TestPostGISTable.test_dbconn)
(TestPostGISTable.test_dbname): New methods to test the new
PostGISConnection methods

1 bh 1605 # Copyright (C) 2003 by Intevation GmbH
2     # Authors:
3     # Bernhard Herzog <[email protected]>
4     #
5     # This program is free software under the GPL (>=v2)
6     # Read the file COPYING coming with the software for details.
7    
8     """Test for Thuban.Model.postgisdb"""
9    
10     import os
11     import unittest
12    
13    
14     try:
15     import psycopg
16     except ImportError:
17     # No psycopg available. Nothing to be done here because the
18     # postgis.py support module determines this too and the tests will
19     # be skipped completely.
20     pass
21    
22     import postgissupport
23    
24     import support
25     support.initthuban()
26    
27 bh 1636 from Thuban.Model.postgisdb import ConnectionError, PostGISConnection, \
28     PostGISTable, PostGISShapeStore
29 bh 1605 from Thuban.Model.table import FIELDTYPE_INT, FIELDTYPE_STRING, \
30     FIELDTYPE_DOUBLE
31     from Thuban.Model.data import SHAPETYPE_POINT, SHAPETYPE_POLYGON, \
32     SHAPETYPE_ARC, RAW_WKT
33    
34    
35 bh 1620 class NonConnection(PostGISConnection):
36    
37     """Special connection class that doesn't actually connect"""
38    
39     def connect(self):
40     pass
41    
42    
43    
44     class TestBriefDescription(unittest.TestCase):
45    
46     def test(self):
47     """Test PostGISConnection.BriefDescription()"""
48     self.assertEquals(NonConnection("somedb").BriefDescription(),
49     "postgis://@:/somedb")
50     self.assertEquals(NonConnection("db", host="here",
51     port="123").BriefDescription(),
52     "postgis://@here:123/db")
53     self.assertEquals(NonConnection("db", user="me",
54     port="123").BriefDescription(),
55     "postgis://me@:123/db")
56    
57    
58 bh 1605 class TestPostGISConnection(unittest.TestCase):
59    
60     def setUp(self):
61     """Start the server and create a database.
62    
63     The database name will be stored in self.dbname, the server
64     object in self.server and the db object in self.db.
65     """
66     postgissupport.skip_if_no_postgis()
67     self.server = postgissupport.get_test_server()
68     self.dbname = ".".join(self.id().split(".")[-2:])[-31:]
69     self.db = self.server.new_postgis_db(self.dbname)
70    
71     def test_gis_tables_empty(self):
72     """Test PostGISConnection.GISTables() on empty DB"""
73 bh 1634 db = PostGISConnection(dbname = self.dbname,
74     **self.server.connection_params("user"))
75 bh 1605
76     # An empty database doesn't have any GIS tables
77     self.assertEquals(db.GeometryTables(), [])
78    
79     def test_gis_tables_non_empty(self):
80     """Test PostGISConnection.GISTables() on non-empty DB"""
81 bh 1634 db = PostGISConnection(dbname = self.dbname,
82     **self.server.connection_params("user"))
83 bh 1605
84 bh 1634 conn = psycopg.connect("dbname=%s " % self.dbname
85     + self.server.connection_string("admin"))
86 bh 1605 cursor = conn.cursor()
87     cursor.execute("CREATE TABLE test (A INT);")
88     cursor.execute("SELECT AddGeometryColumn(%(dbname)s, 'test',"
89     " 'geometry', -1, 'POINT', 2);",
90     {"dbname": self.dbname})
91     conn.commit()
92    
93     # An empty database doesn't have any GIS tables
94     self.assertEquals(db.GeometryTables(), ["test"])
95    
96    
97 bh 1636 class TestPostgisDBExceptions(unittest.TestCase):
98    
99     def setUp(self):
100     """Start the postgis server and switch on authentication"""
101     postgissupport.skip_if_no_postgis()
102     self.server = postgissupport.get_test_server()
103     self.postgisdb = self.server.get_default_static_data_db()
104     self.server.require_authentication(True)
105    
106     def tearDown(self):
107     """Extend the inherited method to switch off postgresql authentication
108     """
109     self.server.require_authentication(False)
110    
111     def test_no_password(self):
112     """Test PostGISConnection with omitted but required password"""
113     connection_params = self.server.connection_params("user")
114     # remove the password deliberately
115     del connection_params["password"]
116    
117     self.assertRaises(ConnectionError,
118     PostGISConnection, dbname = self.postgisdb.dbname,
119     **connection_params)
120    
121    
122 bh 1605 class PostGISStaticTests(unittest.TestCase, support.FloatComparisonMixin):
123    
124     """Base class for PostGIS tests with static data."""
125    
126     def setUp(self):
127     """Start the server and create a database with static data
128    
129     This method sets the following instance attributes:
130    
131     dbname -- the name of the database
132    
133     server -- The server object
134    
135     db -- the PostGISConnection object
136     """
137     postgissupport.skip_if_no_postgis()
138     self.server = postgissupport.get_test_server()
139     self.postgisdb = self.server.get_default_static_data_db()
140     self.db = PostGISConnection(dbname = self.postgisdb.dbname,
141 bh 1634 **self.server.connection_params("user"))
142 bh 1605
143     def tearDown(self):
144     """Close the database connection"""
145     self.db.Close()
146    
147    
148     class TestPostGISTable(PostGISStaticTests):
149    
150     def setUp(self):
151     """Extend inherited method to set self.table to a PostGISTable"""
152     PostGISStaticTests.setUp(self)
153     self.table = PostGISTable(self.db, "landmarks")
154    
155 bh 1638 def test_dbconn(self):
156     """Test PostGISTable.DBConnection()"""
157     self.failUnless(self.table.DBConnection() is self.db)
158    
159     def test_dbname(self):
160     """Test PostGISTable.TableName()"""
161     self.assertEquals(self.table.TableName(), "landmarks")
162    
163 bh 1605 def test_dependencies(self):
164     """Test PostGISTable.Dependencies()"""
165     # A PostGISTable depends on no other data container
166     self.assertEquals(self.table.Dependencies(), ())
167    
168     def test_num_rows(self):
169     """Test PostGISTable.NumRows()"""
170     self.assertEquals(self.table.NumRows(), 34)
171    
172     def test_num_columns(self):
173     """Test PostGISTable.NumColumns()"""
174     # The table in the postgis db has one additional column, "gid",
175     # so there's one more column in the PostGISTable than in the DBF
176     self.assertEquals(self.table.NumColumns(), 7)
177    
178     def test_columns(self):
179     """Test PostGISTable.Columns()"""
180     self.assertEquals(len(self.table.Columns()), 7)
181     self.assertEquals(self.table.Columns()[0].name, "gid")
182     self.assertEquals(self.table.Columns()[0].type, FIELDTYPE_INT)
183     self.assertEquals(self.table.Columns()[0].index, 0)
184     self.assertEquals(self.table.Columns()[1].name, "area")
185     self.assertEquals(self.table.Columns()[1].type, FIELDTYPE_DOUBLE)
186     self.assertEquals(self.table.Columns()[1].index, 1)
187     self.assertEquals(self.table.Columns()[5].name, "clptlabel")
188     self.assertEquals(self.table.Columns()[5].type, FIELDTYPE_STRING)
189     self.assertEquals(self.table.Columns()[5].index, 5)
190    
191     def test_column(self):
192     """Test PostGISTable.Column()"""
193     self.assertEquals(self.table.Column("area").name, "area")
194     self.assertEquals(self.table.Column("area").type, FIELDTYPE_DOUBLE)
195     self.assertEquals(self.table.Column("area").index, 1)
196    
197     def test_has_column(self):
198     """Test PostGISTable.HasColumn()"""
199     self.assert_(self.table.HasColumn("area"))
200     self.failIf(self.table.HasColumn("foo"))
201    
202     def test_read_row_as_dict(self):
203     """Test PostGISTable.ReadRowAsDict()"""
204     self.assertEquals(self.table.ReadRowAsDict(3),
205     {"gid": 3,
206     "area": 0.0,
207     "perimeter": 0.0,
208     "clpoint_": 4,
209     "clpoint_id": 24,
210     "clptlabel": "RUINS",
211     "clptflag": 0})
212    
213     def test_read_value(self):
214     """Test PostGISTable.ReadValue()"""
215     self.assertEquals(self.table.ReadValue(3, 4), 24)
216     self.assertEquals(self.table.ReadValue(3, "clpoint_id"), 24)
217    
218     def test_value_range(self):
219     """Test PostGISTable.ValueRange()"""
220     self.assertEquals(self.table.ValueRange("clpoint_id"), (21, 74))
221    
222     def test_unique_values(self):
223     """Test PostGISTable.UniqueValues()"""
224     values = self.table.UniqueValues("clptlabel")
225     values.sort()
226     self.assertEquals(values, ["BUILDING", "FARM", "HUT","LIGHTHOUSE",
227     "OTHER/UNKNOWN", "RUINS"])
228    
229     def test_simple_query(self):
230     """Test PostGISTable.SimpleQuery()"""
231     table = self.table
232     self.assertEquals(table.SimpleQuery(table.Column("clptlabel"),
233     "==", "FARM"),
234     [6])
235     self.assertEquals(table.SimpleQuery(table.Column("clpoint_id"),
236     ">", 70),
237     [24, 25, 26])
238     self.assertEquals(table.SimpleQuery(table.Column("clpoint_id"),
239     "<", table.Column("clpoint_")),
240     [28, 29, 30, 31, 32, 33])
241    
242    
243     class TestPostGISShapestorePoint(PostGISStaticTests):
244    
245     """Tests for PostGISShapeStore objects with POINT data"""
246    
247     def setUp(self):
248     """Extend inherited method to set self.table to a PostGISShapeStore"""
249     PostGISStaticTests.setUp(self)
250     self.store = PostGISShapeStore(self.db, "landmarks")
251    
252     #
253     # First, some tests that should be independend of the shapetype, so
254     # it shouldn't be necessary to repeat them for other shapetypes
255     #
256    
257     def test_dependencies(self):
258     """Test PostGISShapeStore.Dependencies()"""
259     # A PostGISShapeStore depends on no other data container
260     self.assertEquals(self.store.Dependencies(), ())
261    
262     def test_table(self):
263     """Test PostGISShapeStore.Table() with POINT shapes"""
264     # A PostGISShapeStore is its own table
265     self.assert_(self.store.Table() is self.store)
266    
267     def test_orig_shapestore(self):
268     """Test PostGISShapeStore.OrigShapeStore() with POINT shapes"""
269     # A PostGISShapeStore is not derived from another shape store
270     self.assert_(self.store.OrigShapeStore() is None)
271    
272     def test_raw_format(self):
273     """Test PostGISShapeStore.RawShapeFormat() with POINT shapes"""
274     self.assertEquals(self.store.RawShapeFormat(), RAW_WKT)
275    
276     #
277     # Shapetype specific tests
278     #
279    
280     def test_shape_type(self):
281     """Test PostGISShapeStore.ShapeType() with POINT shapes"""
282     self.assertEquals(self.store.ShapeType(), SHAPETYPE_POINT)
283    
284     def test_num_shapes(self):
285     """Test PostGISShapeStore.NumShapes() with POINT shapes"""
286     self.assertEquals(self.store.NumShapes(), 34)
287    
288     def test_bounding_box(self):
289     """Test PostGISShapeStore.BoundingBox() with POINT shapes"""
290     self.assertFloatSeqEqual(self.store.BoundingBox(),
291     [-23.806047439575195, 63.405960083007812,
292     -15.12291431427002, 66.36572265625])
293    
294     def test_shape_shapeid(self):
295     """Test PostGISShapeStore.Shape(i).ShapeID() with POINT shapes"""
296     self.assertEquals(self.store.Shape(5).ShapeID(), 5)
297    
298     def test_shape_points(self):
299     """Test PostGISShapeStore.Shape(i).Points() with POINT shapes"""
300     self.assertPointListEquals(self.store.Shape(0).Points(),
301     [[(-22.711074829101562, 66.36572265625)]])
302    
303     def test_shape_raw_data(self):
304     """Test PostGISShapeStore.Shape(i).RawData() with POINT shapes"""
305     self.assertEquals(self.store.Shape(0).RawData(),
306     'POINT(-22.7110748291016 66.36572265625)')
307    
308     def test_shapes_in_region(self):
309     """Test PostGISShapeStore:ShapesInRegion() with POINT shapes"""
310     shapes = self.store.ShapesInRegion((-20.0, 64.0, -24.0, 67))
311     self.assertEquals([s.ShapeID() for s in shapes],
312     [0, 1, 2, 3, 4, 5, 27])
313    
314    
315     class TestPostGISShapestoreArc(PostGISStaticTests):
316    
317     """Tests for PostGISShapeStore objects with MULTILINESTRING data"""
318    
319     def setUp(self):
320     """Extend inherited method to set self.table to a PostGISShapeStore"""
321     PostGISStaticTests.setUp(self)
322     self.store = PostGISShapeStore(self.db, "roads")
323    
324     def test_shape_type(self):
325     """Test PostGISShapeStore.ShapeType() with ARC shapes"""
326     self.assertEquals(self.store.ShapeType(), SHAPETYPE_ARC)
327    
328     def test_num_shapes(self):
329     """Test PostGISShapeStore.NumShapes() with ARC shapes"""
330     self.assertEquals(self.store.NumShapes(), 839)
331    
332     def test_bounding_box(self):
333     """Test PostGISShapeStore.BoundingBox() with ARC shapes"""
334     self.assertFloatSeqEqual(self.store.BoundingBox(),
335     [-24.450359344482422, 63.426830291748047,
336     -13.55668830871582, 66.520111083984375])
337    
338     def test_shape_shapeid(self):
339     """Test PostGISShapeStore.Shape(i).ShapeID() with ARC shapes"""
340     self.assertEquals(self.store.Shape(5).ShapeID(), 5)
341    
342     def test_shape_points(self):
343     """Test PostGISShapeStore.Shape(i).Points() with ARC shapes"""
344     self.assertPointListEquals(self.store.Shape(32).Points(),
345     [[(-15.0821743011474, 66.2773818969726),
346     (-15.0263500213623, 66.2733917236328)]])
347    
348     def test_shape_raw_data(self):
349     """Test PostGISShapeStore.Shape(i).RawData() with ARC shapes"""
350     self.assertEquals(self.store.Shape(32).RawData(),
351     "MULTILINESTRING((-15.0821743011475 66.2773818969727,"
352     "-15.0263500213623 66.2733917236328))")
353    
354     def test_shapes_in_region(self):
355     """Test PostGISShapeStore.ShapesInRegion() with ARC shapes"""
356     shapes = self.store.ShapesInRegion((-24.0, 64.5, -23.5, 65.0))
357     self.assertEquals([s.ShapeID() for s in shapes], [573, 581, 613])
358    
359    
360     class TestPostGISShapestorePolygon(PostGISStaticTests):
361    
362     """Tests for PostGISShapeStore objects with POLYGON data"""
363    
364     def setUp(self):
365     """Extend inherited method to set self.table to a PostGISShapeStore"""
366     PostGISStaticTests.setUp(self)
367     self.store = PostGISShapeStore(self.db, "political")
368    
369     def test_shape_type(self):
370     """Test PostGISShapeStore.ShapeType() with POLYGON shapes"""
371     self.assertEquals(self.store.ShapeType(), SHAPETYPE_POLYGON)
372    
373     def test_num_shapes(self):
374     """Test PostGISShapeStore.NumShapes() with POLYGON shapes"""
375     self.assertEquals(self.store.NumShapes(), 156)
376    
377     def test_bounding_box(self):
378     """Test PostGISShapeStore.BoundingBox() with POLYGON shapes"""
379     self.assertFloatSeqEqual(self.store.BoundingBox(),
380     [-24.546524047851562, 63.286754608154297,
381     -13.495815277099609, 66.563774108886719])
382    
383     def test_shape_shapeid(self):
384     """Test PostGISShapeStore.Shape(i).ShapeID() with POLYGON shapes"""
385     self.assertEquals(self.store.Shape(5).ShapeID(), 5)
386    
387     def test_shape_points(self):
388     """Test PostGISShapeStore.Shape(i).Points() with POLYGON shapes"""
389     self.assertPointListEquals(self.store.Shape(4).Points(),
390     [[(-22.40639114379882, 64.714111328125),
391     (-22.41621208190918, 64.716003417968),
392     (-22.40605163574218, 64.719200134277),
393     (-22.40639114379882, 64.714111328125)]])
394    
395     def test_shape_raw_data(self):
396     """Test PostGISShapeStore.Shape(i).RawData() with POLYGON shapes"""
397     self.assertEquals(self.store.Shape(4).RawData(),
398     "POLYGON((-22.4063911437988 64.714111328125,"
399     "-22.4162120819092 64.7160034179688,"
400     "-22.4060516357422 64.7192001342773,"
401     "-22.4063911437988 64.714111328125))")
402    
403     def test_shapes_in_region(self):
404     """Test PostGISShapeStore.ShapesInRegion() with POLYGON shapes"""
405     shapes = self.store.ShapesInRegion((-23.0, 65.5, -22.8, 65.25))
406     self.assertEquals([s.ShapeID() for s in shapes],
407     [47, 56, 59, 61, 62, 71, 144])
408    
409    
410     if __name__ == "__main__":
411     support.run_tests()

Properties

Name Value
svn:eol-style native
svn:keywords Author Date Id Revision

[email protected]
ViewVC Help
Powered by ViewVC 1.1.26