WIP-pyshapelib-bramz/test/test_transientdb.py

# Copyright (c) 2002, 2003 by Intevation GmbH
# Authors:
# Bernhard Herzog <[email protected]>
#
# This program is free software under the GPL (>=v2)
# Read the file COPYING coming with Thuban for details.

"""
Test the Transient DB classes
"""

__version__ = "$Revision$"
# $Source$
# $Id$

import os
import unittest

import support
support.initthuban()

from Thuban.Model.table import DBFTable, MemoryTable, FIELDTYPE_STRING, \
                               FIELDTYPE_INT, FIELDTYPE_DOUBLE
from Thuban.Model.transientdb import TransientDatabase, TransientTable, \
     TransientJoinedTable, AutoTransientTable


class TestTransientTable(unittest.TestCase, support.FileTestMixin):

    def setUp(self):
        """Create a transient database as self.transientdb"""
        filename = self.temp_file_name("transient_table.sqlite")
        if os.path.exists(filename):
            os.remove(filename)
        journal = filename + "-journal"
        if os.path.exists(journal):
            print "removing journal", journal
            os.remove(journal)
        self.transientdb = TransientDatabase(filename)

    def tearDown(self):
        self.transientdb.close()

    def run_iceland_political_tests(self, table):
        """Run some tests on tablte

        Assume that table holds the data of the file
        ../Data/iceland/political.dbf sample file.
        """
        self.assertEquals(table.NumRows(), 156)
        self.assertEquals(table.NumColumns(), 8)

        # Check one each of the possible field types. The width and
        # decimal precision is always 0.
        columns = table.Columns()
        self.assertEquals(columns[0].name, 'AREA')
        self.assertEquals(columns[0].type, FIELDTYPE_DOUBLE)
        self.assertEquals(columns[3].name, 'PONET_ID')
        self.assertEquals(columns[3].type, FIELDTYPE_INT)
        self.assertEquals(columns[6].name, 'POPYCOUN')
        self.assertEquals(columns[6].type, FIELDTYPE_STRING)

        # HasColumn
        self.failUnless(table.HasColumn("AREA"))
        self.failUnless(table.HasColumn(1))
        # HasColumn for non-exisiting columns
        self.failIf(table.HasColumn("non_existing_name"))
        self.failIf(table.HasColumn(100))

        # Reading rows and values.
        self.assertEquals(table.ReadRowAsDict(144),
                          {'POPYCOUN': 'IC', 'POPYADMIN': '', 'PONET_': 146,
                           'AREA': 19.462,
                           'POPYTYPE': 1, 'PERIMETER': 88.518000000000001,
                           'POPYREG': '1',
                           'PONET_ID': 145})
        self.assertEquals(table.ReadValue(144, "AREA"), 19.462)
        self.assertEquals(table.ReadValue(144, 3), 145)

        # ValueRange may induce a copy to the transient database.
        # Therefore we put it last so that we can execute this method
        # twice to check whether the other methods still work after the
        # copy
        self.assertEquals(table.ValueRange("AREA"), (0.0, 19.462))

        unique = table.UniqueValues("PONET_ID")
        unique.sort()
        self.assertEquals(unique, range(1, 157))

    def test_transient_table(self):
        """Test TransientTable(dbftable)

        The TransientTable should copy the data to the
        TransientDatabase.
        """
        orig_table = DBFTable(os.path.join("..", "Data", "iceland",
                                           "political.dbf"))
        table = TransientTable(self.transientdb, orig_table)
        self.run_iceland_political_tests(table)

        # The transient_table method should return the table itself
        self.assert_(table is table.transient_table())

        # The title is simply copied over from the original table
        self.assertEquals(table.Title(), orig_table.Title())

        # The TransientTable class itself doesn't implement the
        # Dependencies method, so we don't test it.


    def test_auto_transient_table(self):
        """Test AutoTransientTable(dbftable)

        The AutoTransientTable should copy the data to the
        TransientDatabase on demand.
        """
        orig_table = DBFTable(os.path.join("..", "Data", "iceland",
                                           "political.dbf"))
        table = AutoTransientTable(self.transientdb, orig_table)

        # Run the tests twice so that we execute them once when the data
        # has not been copied to the transient db yet and once when it
        # has. This assumes that run_iceland_political_tests does at
        # least one call to a method that copies to the transient db at
        # its end.
        self.run_iceland_political_tests(table)
        self.run_iceland_political_tests(table)

    def test_auto_transient_table_query(self):
        """Test AutoTransientTable.SimpleQuery()"""
        orig_table = DBFTable(os.path.join("..", "Data", "iceland",
                                           "political.dbf"))
        table = AutoTransientTable(self.transientdb, orig_table)
        # Only a simple test here. The AutoTransientTable simply
        # delegates to its transient table so it should be OK that the
        # real test for it is in test_transient_table_query. However,
        # it's important to check that the column handling works
        # correctly because the AutoTransientTable and it's underlying
        # transient table use different column object types.
        self.assertEquals(table.SimpleQuery(table.Column("AREA"), ">", 10.0),
                          [144])

        # test using a Column object as the right parameter
        self.assertEquals(table.SimpleQuery(table.Column("POPYTYPE"),
                                            "==",
                                            table.Column("POPYREG")),
                          range(156))

    def test_auto_transient_table_dependencies(self):
        """Test AutoTransientTable.Dependencies()"""
        orig_table = DBFTable(os.path.join("..", "Data", "iceland",
                                           "political.dbf"))
        table = AutoTransientTable(self.transientdb, orig_table)
        self.assertEquals(table.Dependencies(), (orig_table,))

    def test_auto_transient_table_title(self):
        """Test AutoTransientTable.Title()"""
        orig_table = DBFTable(os.path.join("..", "Data", "iceland",
                                           "political.dbf"))
        table = AutoTransientTable(self.transientdb, orig_table)
        # The title is of course the same as that of the original table
        self.assertEquals(table.Title(), orig_table.Title())

    def test_transient_joined_table(self):
        """Test TransientJoinedTable"""
        simple = MemoryTable([("type", FIELDTYPE_STRING),
                              ("code", FIELDTYPE_INT)],
                             [("OTHER/UNKNOWN", 0),
                              ("RUINS", 1),
                              ("FARM", 2),
                              ("BUILDING", 3),
                              ("HUT", 4),
                              ("LIGHTHOUSE", 5)])
        auto = AutoTransientTable(self.transientdb, simple)
        filename = os.path.join("..", "Data", "iceland",
                                "cultural_landmark-point.dbf")
        landmarks = AutoTransientTable(self.transientdb, DBFTable(filename))

        table = TransientJoinedTable(self.transientdb, landmarks, "CLPTLABEL",
                                     auto, "type")

        self.assertEquals(table.NumRows(), 34)
        self.assertEquals(table.NumColumns(), 8)
        self.assertEquals(table.Column(0).type, FIELDTYPE_DOUBLE)
        self.assertEquals(table.Column(0).name, 'AREA')
        self.assertEquals(table.Column(7).type, FIELDTYPE_INT)
        self.assertEquals(table.Column(7).name, 'code')
        self.assertEquals(table.Column(4).type, FIELDTYPE_STRING)
        self.assertEquals(table.Column(4).name, 'CLPTLABEL')
        # HasColumn
        self.failUnless(table.HasColumn("AREA"))
        self.failUnless(table.HasColumn(1))
        # HasColumn for non-exisiting columns
        self.failIf(table.HasColumn("non_existing_name"))
        self.failIf(table.HasColumn(100))

        # Reading rows and values
        self.assertEquals(table.ReadRowAsDict(22),
                          {'PERIMETER': 0.0, 'CLPOINT_': 23,
                           'AREA': 0.0, 'CLPTLABEL': 'RUINS',
                           'CLPOINT_ID': 38, 'CLPTFLAG': 0,
                           'code': 1, 'type': 'RUINS'})
        self.assertEquals(table.ReadValue(22, "type"), 'RUINS')
        self.assertEquals(table.ReadValue(22, 7), 1)

        # The transient_table method should return the table itself
        self.assert_(table is table.transient_table())

        # The TransientJoinedTable depends on both input tables
        self.assertEquals(table.Dependencies(), (landmarks, auto))

        # The title is constructed from the titles of the input tables.
        self.assertEquals(table.Title(),
                          "Join of %s and %s" % (landmarks.Title(),
                                                 auto.Title()))


    def test_transient_table_read_twice(self):
        """Test TransientTable.ReadRowAsDict() reading the same record twice"""
        simple = MemoryTable([("type", FIELDTYPE_STRING),
                              ("code", FIELDTYPE_INT)],
                             [("OTHER/UNKNOWN", 0),
                              ("RUINS", 1),
                              ("FARM", 2),
                              ("BUILDING", 3),
                              ("HUT", 4),
                              ("LIGHTHOUSE", 5)])
        table = TransientTable(self.transientdb, simple)

        # There was a bug where reading the same record twice would
        # raise an exception in the second call because of an
        # unitialized local variable, so for passing the test it's
        # enough if reading simply succeeds. OTOH, while we're at it we
        # might as well check whether the results are equal anyway :)
        result1 = table.ReadRowAsDict(3)
        result2 = table.ReadRowAsDict(3)
        self.assertEquals(result1, result2)


    def test_transient_table_query(self):
        """Test TransientTable.SimpleQuery()"""
        simple = MemoryTable([("type", FIELDTYPE_STRING),
                              ("value", FIELDTYPE_DOUBLE),
                              ("code", FIELDTYPE_INT)],
                             [("OTHER/UNKNOWN", -1.5, 11),
                              ("RUINS", 0.0, 1),
                              ("FARM", 3.141, 2),
                              ("BUILDING", 2.5, 3),
                              ("HUT", 1e6, 4),
                              ("LIGHTHOUSE", -0.01, 5)])
        table = TransientTable(self.transientdb, simple)

        # A column and a value
        self.assertEquals(table.SimpleQuery(table.Column(0), "==", "RUINS"),
                          [1])
        self.assertEquals(table.SimpleQuery(table.Column(2), "!=", 2),
                          [0, 1, 3, 4, 5])
        self.assertEquals(table.SimpleQuery(table.Column(1), "<", 1.0),
                          [0, 1, 5])
        self.assertEquals(table.SimpleQuery(table.Column(1), "<=", -1.5),
                          [0])
        self.assertEquals(table.SimpleQuery(table.Column(2), ">", 3),
                          [0, 4, 5])
        self.assertEquals(table.SimpleQuery(table.Column(2), ">=", 3),
                          [0, 3, 4, 5])

        # Two columns as operands
        self.assertEquals(table.SimpleQuery(table.Column(1),
                                            "<=", table.Column(2)),
                          [0, 1, 3, 5])

        # Test whether invalid operators raise a ValueError
        self.assertRaises(ValueError,
                          table.SimpleQuery,
                          table.Column(1), "<<", table.Column(2))


if __name__ == "__main__":
    support.run_tests()
1	# 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, MemoryTable, FIELDTYPE_STRING, \
23	FIELDTYPE_INT, FIELDTYPE_DOUBLE
24	from Thuban.Model.transientdb import TransientDatabase, TransientTable, \
25	TransientJoinedTable, AutoTransientTable
26
27
28	class TestTransientTable(unittest.TestCase, support.FileTestMixin):
29
30	def setUp(self):
31	"""Create a transient database as self.transientdb"""
32	filename = self.temp_file_name("transient_table.sqlite")
33	if os.path.exists(filename):
34	os.remove(filename)
35	journal = filename + "-journal"
36	if os.path.exists(journal):
37	print "removing journal", journal
38	os.remove(journal)
39	self.transientdb = TransientDatabase(filename)
40
41	def tearDown(self):
42	self.transientdb.close()
43
44	def run_iceland_political_tests(self, table):
45	"""Run some tests on tablte
46
47	Assume that table holds the data of the file
48	../Data/iceland/political.dbf sample file.
49	"""
50	self.assertEquals(table.NumRows(), 156)
51	self.assertEquals(table.NumColumns(), 8)
52
53	# Check one each of the possible field types. The width and
54	# decimal precision is always 0.
55	columns = table.Columns()
56	self.assertEquals(columns[0].name, 'AREA')
57	self.assertEquals(columns[0].type, FIELDTYPE_DOUBLE)
58	self.assertEquals(columns[3].name, 'PONET_ID')
59	self.assertEquals(columns[3].type, FIELDTYPE_INT)
60	self.assertEquals(columns[6].name, 'POPYCOUN')
61	self.assertEquals(columns[6].type, FIELDTYPE_STRING)
62
63	# HasColumn
64	self.failUnless(table.HasColumn("AREA"))
65	self.failUnless(table.HasColumn(1))
66	# HasColumn for non-exisiting columns
67	self.failIf(table.HasColumn("non_existing_name"))
68	self.failIf(table.HasColumn(100))
69
70	# Reading rows and values.
71	self.assertEquals(table.ReadRowAsDict(144),
72	{'POPYCOUN': 'IC', 'POPYADMIN': '', 'PONET_': 146,
73	'AREA': 19.462,
74	'POPYTYPE': 1, 'PERIMETER': 88.518000000000001,
75	'POPYREG': '1',
76	'PONET_ID': 145})
77	self.assertEquals(table.ReadValue(144, "AREA"), 19.462)
78	self.assertEquals(table.ReadValue(144, 3), 145)
79
80	# ValueRange may induce a copy to the transient database.
81	# Therefore we put it last so that we can execute this method
82	# twice to check whether the other methods still work after the
83	# copy
84	self.assertEquals(table.ValueRange("AREA"), (0.0, 19.462))
85
86	unique = table.UniqueValues("PONET_ID")
87	unique.sort()
88	self.assertEquals(unique, range(1, 157))
89
90	def test_transient_table(self):
91	"""Test TransientTable(dbftable)
92
93	The TransientTable should copy the data to the
94	TransientDatabase.
95	"""
96	orig_table = DBFTable(os.path.join("..", "Data", "iceland",
97	"political.dbf"))
98	table = TransientTable(self.transientdb, orig_table)
99	self.run_iceland_political_tests(table)
100
101	# The transient_table method should return the table itself
102	self.assert_(table is table.transient_table())
103
104	# The title is simply copied over from the original table
105	self.assertEquals(table.Title(), orig_table.Title())
106
107	# The TransientTable class itself doesn't implement the
108	# Dependencies method, so we don't test it.
109
110
111	def test_auto_transient_table(self):
112	"""Test AutoTransientTable(dbftable)
113
114	The AutoTransientTable should copy the data to the
115	TransientDatabase on demand.
116	"""
117	orig_table = DBFTable(os.path.join("..", "Data", "iceland",
118	"political.dbf"))
119	table = AutoTransientTable(self.transientdb, orig_table)
120
121	# Run the tests twice so that we execute them once when the data
122	# has not been copied to the transient db yet and once when it
123	# has. This assumes that run_iceland_political_tests does at
124	# least one call to a method that copies to the transient db at
125	# its end.
126	self.run_iceland_political_tests(table)
127	self.run_iceland_political_tests(table)
128
129	def test_auto_transient_table_query(self):
130	"""Test AutoTransientTable.SimpleQuery()"""
131	orig_table = DBFTable(os.path.join("..", "Data", "iceland",
132	"political.dbf"))
133	table = AutoTransientTable(self.transientdb, orig_table)
134	# Only a simple test here. The AutoTransientTable simply
135	# delegates to its transient table so it should be OK that the
136	# real test for it is in test_transient_table_query. However,
137	# it's important to check that the column handling works
138	# correctly because the AutoTransientTable and it's underlying
139	# transient table use different column object types.
140	self.assertEquals(table.SimpleQuery(table.Column("AREA"), ">", 10.0),
141	[144])
142
143	# test using a Column object as the right parameter
144	self.assertEquals(table.SimpleQuery(table.Column("POPYTYPE"),
145	"==",
146	table.Column("POPYREG")),
147	range(156))
148
149	def test_auto_transient_table_dependencies(self):
150	"""Test AutoTransientTable.Dependencies()"""
151	orig_table = DBFTable(os.path.join("..", "Data", "iceland",
152	"political.dbf"))
153	table = AutoTransientTable(self.transientdb, orig_table)
154	self.assertEquals(table.Dependencies(), (orig_table,))
155
156	def test_auto_transient_table_title(self):
157	"""Test AutoTransientTable.Title()"""
158	orig_table = DBFTable(os.path.join("..", "Data", "iceland",
159	"political.dbf"))
160	table = AutoTransientTable(self.transientdb, orig_table)
161	# The title is of course the same as that of the original table
162	self.assertEquals(table.Title(), orig_table.Title())
163
164	def test_transient_joined_table(self):
165	"""Test TransientJoinedTable"""
166	simple = MemoryTable([("type", FIELDTYPE_STRING),
167	("code", FIELDTYPE_INT)],
168	[("OTHER/UNKNOWN", 0),
169	("RUINS", 1),
170	("FARM", 2),
171	("BUILDING", 3),
172	("HUT", 4),
173	("LIGHTHOUSE", 5)])
174	auto = AutoTransientTable(self.transientdb, simple)
175	filename = os.path.join("..", "Data", "iceland",
176	"cultural_landmark-point.dbf")
177	landmarks = AutoTransientTable(self.transientdb, DBFTable(filename))
178
179	table = TransientJoinedTable(self.transientdb, landmarks, "CLPTLABEL",
180	auto, "type")
181
182	self.assertEquals(table.NumRows(), 34)
183	self.assertEquals(table.NumColumns(), 8)
184	self.assertEquals(table.Column(0).type, FIELDTYPE_DOUBLE)
185	self.assertEquals(table.Column(0).name, 'AREA')
186	self.assertEquals(table.Column(7).type, FIELDTYPE_INT)
187	self.assertEquals(table.Column(7).name, 'code')
188	self.assertEquals(table.Column(4).type, FIELDTYPE_STRING)
189	self.assertEquals(table.Column(4).name, 'CLPTLABEL')
190	# HasColumn
191	self.failUnless(table.HasColumn("AREA"))
192	self.failUnless(table.HasColumn(1))
193	# HasColumn for non-exisiting columns
194	self.failIf(table.HasColumn("non_existing_name"))
195	self.failIf(table.HasColumn(100))
196
197	# Reading rows and values
198	self.assertEquals(table.ReadRowAsDict(22),
199	{'PERIMETER': 0.0, 'CLPOINT_': 23,
200	'AREA': 0.0, 'CLPTLABEL': 'RUINS',
201	'CLPOINT_ID': 38, 'CLPTFLAG': 0,
202	'code': 1, 'type': 'RUINS'})
203	self.assertEquals(table.ReadValue(22, "type"), 'RUINS')
204	self.assertEquals(table.ReadValue(22, 7), 1)
205
206	# The transient_table method should return the table itself
207	self.assert_(table is table.transient_table())
208
209	# The TransientJoinedTable depends on both input tables
210	self.assertEquals(table.Dependencies(), (landmarks, auto))
211
212	# The title is constructed from the titles of the input tables.
213	self.assertEquals(table.Title(),
214	"Join of %s and %s" % (landmarks.Title(),
215	auto.Title()))
216
217
218	def test_transient_table_read_twice(self):
219	"""Test TransientTable.ReadRowAsDict() reading the same record twice"""
220	simple = MemoryTable([("type", FIELDTYPE_STRING),
221	("code", FIELDTYPE_INT)],
222	[("OTHER/UNKNOWN", 0),
223	("RUINS", 1),
224	("FARM", 2),
225	("BUILDING", 3),
226	("HUT", 4),
227	("LIGHTHOUSE", 5)])
228	table = TransientTable(self.transientdb, simple)
229
230	# There was a bug where reading the same record twice would
231	# raise an exception in the second call because of an
232	# unitialized local variable, so for passing the test it's
233	# enough if reading simply succeeds. OTOH, while we're at it we
234	# might as well check whether the results are equal anyway :)
235	result1 = table.ReadRowAsDict(3)
236	result2 = table.ReadRowAsDict(3)
237	self.assertEquals(result1, result2)
238
239
240	def test_transient_table_query(self):
241	"""Test TransientTable.SimpleQuery()"""
242	simple = MemoryTable([("type", FIELDTYPE_STRING),
243	("value", FIELDTYPE_DOUBLE),
244	("code", FIELDTYPE_INT)],
245	[("OTHER/UNKNOWN", -1.5, 11),
246	("RUINS", 0.0, 1),
247	("FARM", 3.141, 2),
248	("BUILDING", 2.5, 3),
249	("HUT", 1e6, 4),
250	("LIGHTHOUSE", -0.01, 5)])
251	table = TransientTable(self.transientdb, simple)
252
253	# A column and a value
254	self.assertEquals(table.SimpleQuery(table.Column(0), "==", "RUINS"),
255	[1])
256	self.assertEquals(table.SimpleQuery(table.Column(2), "!=", 2),
257	[0, 1, 3, 4, 5])
258	self.assertEquals(table.SimpleQuery(table.Column(1), "<", 1.0),
259	[0, 1, 5])
260	self.assertEquals(table.SimpleQuery(table.Column(1), "<=", -1.5),
261	[0])
262	self.assertEquals(table.SimpleQuery(table.Column(2), ">", 3),
263	[0, 4, 5])
264	self.assertEquals(table.SimpleQuery(table.Column(2), ">=", 3),
265	[0, 3, 4, 5])
266
267	# Two columns as operands
268	self.assertEquals(table.SimpleQuery(table.Column(1),
269	"<=", table.Column(2)),
270	[0, 1, 3, 5])
271
272	# Test whether invalid operators raise a ValueError
273	self.assertRaises(ValueError,
274	table.SimpleQuery,
275	table.Column(1), "<<", table.Column(2))
276
277
278	if __name__ == "__main__":
279	support.run_tests()
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision