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

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

Parent Directory Parent Directory | Revision Log Revision Log


Revision 765 - (hide annotations)
Tue Apr 29 12:42:14 2003 UTC (21 years, 10 months ago) by bh
Original Path: trunk/thuban/test/test_transientdb.py
File MIME type: text/x-python
File size: 6348 byte(s)
Next step of table implementation. Introduce a transient database
using SQLite that some of the data is copied to on demand. This
allows us to do joins and other operations that require an index
for good performance with reasonable efficiency. Thuban now needs
SQLite 2.8.0 and pysqlite 0.4.1. Older versions may work but I
haven't tested that.

* Thuban/Model/transientdb.py: New. Transient database
implementation.

* test/test_transientdb.py: New. Tests for the transient DB
classes.

* Thuban/Model/session.py (AutoRemoveFile, AutoRemoveDir): New
classes to help automatically remove temporary files and
directories.
(Session.__init__): New instance variables temp_dir for the
temporary directory and transient_db for the SQLite database
(Session.temp_directory): New. Create a temporary directory if not
yet done and return its name. Use AutoRemoveDir to have it
automatically deleted
(Session.TransientDB): Instantiate the transient database if not
done yet and return it.

* Thuban/Model/data.py (ShapefileStore.__init__): Use an
AutoTransientTable so that data is copied to the transient DB on
demand.
(SimpleStore): New class that simply combines a table and a
shapefile

* Thuban/Model/table.py (Table, DBFTable): Rename Table into
DBFTable and update its doc-string to reflect the fact that this
is only the table interface to a DBF file. Table is now an alias
for DBFTable for temporary backwards compatibility.

* Thuban/UI/application.py (ThubanApplication.OnExit): Make sure
the last reference to the session goes away so that the temporary
files are removed properly.

* test/test_load.py (LoadSessionTest.tearDown): Remove the
reference to the session to make sure the temporary files are
removed.

1 bh 765 # Copyright (c) 2002, 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 Thuban for details.
7    
8     """
9     Test the Transient DB classes
10     """
11    
12     __version__ = "$Revision$"
13     # $Source$
14     # $Id$
15    
16     import os
17     import unittest
18    
19     import support
20     support.initthuban()
21    
22     from Thuban.Model.table import DBFTable, FIELDTYPE_STRING, FIELDTYPE_INT
23     from Thuban.Model.transientdb import TransientDatabase, TransientTable, \
24     TransientJoinedTable, AutoTransientTable
25    
26    
27     class SimpleTable:
28    
29     """Very simple table implementation that operates on a list of tuples"""
30    
31     def __init__(self, fields, data):
32     """Initialize the SimpleTable
33    
34     Parameters:
35     fields -- List of (name, field_type) pairs
36     data -- List of tuples, one for each row of data
37     """
38     self.fields = fields
39     self.data = data
40    
41     def field_count(self):
42     return len(self.fields)
43    
44     def field_info(self, index):
45     name, type = self.fields[index]
46     return (type, name, 0, 0)
47    
48     def record_count(self):
49     return len(self.data)
50    
51     def read_record(self, index):
52     return dict([(self.fields[i][0], self.data[index][i])
53     for i in range(len(self.fields))])
54    
55    
56     class TestTransientTable(unittest.TestCase, support.FileTestMixin):
57    
58     def setUp(self):
59     """Create a transient database as self.transientdb"""
60     filename = self.temp_file_name("transient_table.sqlite")
61     if os.path.exists(filename):
62     os.remove(filename)
63     journal = filename + "-journal"
64     if os.path.exists(journal):
65     print "removing journal", journal
66     os.remove(journal)
67     self.transientdb = TransientDatabase(filename)
68    
69     def tearDown(self):
70     self.transientdb.close()
71    
72     def run_iceland_political_tests(self, table):
73     """Run some tests on tablte
74    
75     Assume that table holds the data of the file
76     ../Data/iceland/political.dbf sample file.
77     """
78     self.assertEquals(table.record_count(), 156)
79     self.assertEquals(table.field_count(), 8)
80    
81     # Check one each of the possible field types. The width and
82     # decimal precision is always 0.
83     self.assertEquals(table.field_info(0), ('double', 'AREA', 0, 0))
84     self.assertEquals(table.field_info(3), ('int', 'PONET_ID', 0, 0))
85     self.assertEquals(table.field_info(6), ('string', 'POPYCOUN', 0, 0))
86    
87     # Read an `interesting' record
88     self.assertEquals(table.read_record(144),
89     {'POPYCOUN': 'IC', 'POPYADMIN': '', 'PONET_': 146,
90     'AREA': 19.462,
91     'POPYTYPE': 1, 'PERIMETER': 88.518000000000001,
92     'POPYREG': '1',
93     'PONET_ID': 145})
94    
95     # field_range may induce a copy to the transient database.
96     # Therefore we put it last so that we can execute this method
97     # twice to check whether the other methods still work after the
98     # copy
99     self.assertEquals(table.field_range("AREA"),
100     ((0.0, None), (19.462, None)))
101    
102     unique = table.GetUniqueValues("PONET_ID")
103     unique.sort()
104     self.assertEquals(unique, range(1, 157))
105    
106     def test_transient_table(self):
107     """Test TransientTable(dbftable)
108    
109     The TransientTable should copy the data to the
110     TransientDatabase.
111     """
112     orig_table = DBFTable(os.path.join("..", "Data", "iceland",
113     "political.dbf"))
114     table = TransientTable(self.transientdb, orig_table)
115     self.run_iceland_political_tests(table)
116    
117     # The transient_table method should return the table itself
118     self.assert_(table is table.transient_table())
119    
120    
121     def test_auto_transient_table(self):
122     """Test AutoTransientTable(dbftable)
123    
124     The AutoTransientTable should copy the data to the
125     TransientDatabase on demand.
126     """
127     orig_table = DBFTable(os.path.join("..", "Data", "iceland",
128     "political.dbf"))
129     table = AutoTransientTable(self.transientdb, orig_table)
130    
131     # Run the tests twice so that we execute them once when the data
132     # has not been copied to the transient db yet and once when it
133     # has. This assumes that run_iceland_political_tests does at
134     # least one call to a method that copies to the transient db at
135     # its end.
136     self.run_iceland_political_tests(table)
137     self.run_iceland_political_tests(table)
138    
139    
140     def test_transient_joined_table(self):
141     """Test TransientJoinedTable"""
142     simple = SimpleTable([("type", FIELDTYPE_STRING),
143     ("code", FIELDTYPE_INT)],
144     [("OTHER/UNKNOWN", 0),
145     ("RUINS", 1),
146     ("FARM", 2),
147     ("BUILDING", 3),
148     ("HUT", 4),
149     ("LIGHTHOUSE", 5)])
150     auto = AutoTransientTable(self.transientdb, simple)
151     filename = os.path.join("..", "Data", "iceland",
152     "cultural_landmark-point.dbf")
153     landmarks = AutoTransientTable(self.transientdb, DBFTable(filename))
154    
155     table = TransientJoinedTable(self.transientdb, landmarks, "CLPTLABEL",
156     auto, "type")
157    
158     self.assertEquals(table.record_count(), 34)
159     self.assertEquals(table.field_count(), 8)
160     self.assertEquals(table.field_info(0), ('double', 'AREA', 0, 0))
161     self.assertEquals(table.field_info(7), ('int', 'code', 0, 0))
162     self.assertEquals(table.field_info(4), ('string', 'CLPTLABEL', 0, 0))
163    
164     # Read an `interesting' record
165     self.assertEquals(table.read_record(22),
166     {'PERIMETER': 0.0, 'CLPOINT_': 23,
167     'AREA': 0.0, 'CLPTLABEL': 'RUINS',
168     'CLPOINT_ID': 38, 'CLPTFLAG': 0,
169     'code': 1, 'type': 'RUINS'})
170    
171     # The transient_table method should return the table itself
172     self.assert_(table is table.transient_table())
173    
174    
175    
176     if __name__ == "__main__":
177     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