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))

        # Read an `interesting' record
        self.assertEquals(table.ReadRowAsDict(144),
                          {'POPYCOUN': 'IC', 'POPYADMIN': '', 'PONET_': 146,
                           'AREA': 19.462,
                           'POPYTYPE': 1, 'PERIMETER': 88.518000000000001,
                           'POPYREG': '1',
                           'PONET_ID': 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())


    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 siply
        # 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])

    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))

        # Read an `interesting' record
        self.assertEquals(table.ReadRowAsDict(22),
                          {'PERIMETER': 0.0, 'CLPOINT_': 23,
                           'AREA': 0.0, 'CLPTLABEL': 'RUINS',
                           'CLPOINT_ID': 38, 'CLPTFLAG': 0,
                           'code': 1, 'type': 'RUINS'})

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


    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	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	jan	805	from Thuban.Model.table import DBFTable, MemoryTable, FIELDTYPE_STRING, \
23	bh	839	FIELDTYPE_INT, FIELDTYPE_DOUBLE
24	bh	765	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	bh	818	self.assertEquals(table.NumRows(), 156)
51			self.assertEquals(table.NumColumns(), 8)
52	bh	765
53			# Check one each of the possible field types. The width and
54			# decimal precision is always 0.
55	bh	818	columns = table.Columns()
56			self.assertEquals(columns[0].name, 'AREA')
57	bh	839	self.assertEquals(columns[0].type, FIELDTYPE_DOUBLE)
58	bh	818	self.assertEquals(columns[3].name, 'PONET_ID')
59	bh	839	self.assertEquals(columns[3].type, FIELDTYPE_INT)
60	bh	818	self.assertEquals(columns[6].name, 'POPYCOUN')
61	bh	839	self.assertEquals(columns[6].type, FIELDTYPE_STRING)
62	bh	765
63	bh	839	# 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	bh	765	# Read an `interesting' record
71	bh	818	self.assertEquals(table.ReadRowAsDict(144),
72	bh	765	{'POPYCOUN': 'IC', 'POPYADMIN': '', 'PONET_': 146,
73			'AREA': 19.462,
74			'POPYTYPE': 1, 'PERIMETER': 88.518000000000001,
75			'POPYREG': '1',
76			'PONET_ID': 145})
77
78	bh	839	# ValueRange may induce a copy to the transient database.
79	bh	765	# Therefore we put it last so that we can execute this method
80			# twice to check whether the other methods still work after the
81			# copy
82	bh	818	self.assertEquals(table.ValueRange("AREA"), (0.0, 19.462))
83	bh	765
84	bh	818	unique = table.UniqueValues("PONET_ID")
85	bh	765	unique.sort()
86			self.assertEquals(unique, range(1, 157))
87
88			def test_transient_table(self):
89			"""Test TransientTable(dbftable)
90
91			The TransientTable should copy the data to the
92			TransientDatabase.
93			"""
94			orig_table = DBFTable(os.path.join("..", "Data", "iceland",
95			"political.dbf"))
96			table = TransientTable(self.transientdb, orig_table)
97			self.run_iceland_political_tests(table)
98
99			# The transient_table method should return the table itself
100			self.assert_(table is table.transient_table())
101
102
103			def test_auto_transient_table(self):
104			"""Test AutoTransientTable(dbftable)
105
106			The AutoTransientTable should copy the data to the
107			TransientDatabase on demand.
108			"""
109			orig_table = DBFTable(os.path.join("..", "Data", "iceland",
110			"political.dbf"))
111			table = AutoTransientTable(self.transientdb, orig_table)
112
113			# Run the tests twice so that we execute them once when the data
114			# has not been copied to the transient db yet and once when it
115			# has. This assumes that run_iceland_political_tests does at
116			# least one call to a method that copies to the transient db at
117			# its end.
118			self.run_iceland_political_tests(table)
119			self.run_iceland_political_tests(table)
120
121	bh	845	def test_auto_transient_table_query(self):
122			"""Test AutoTransientTable.SimpleQuery()"""
123			orig_table = DBFTable(os.path.join("..", "Data", "iceland",
124			"political.dbf"))
125			table = AutoTransientTable(self.transientdb, orig_table)
126			# Only a simple test here. The AutoTransientTable siply
127			# delegates to its transient table so it should be OK that the
128			# real test for it is in test_transient_table_query. However,
129			# it's important to check that the column handling works
130			# correctly because the AutoTransientTable and it's underlying
131			# transient table use different column object types.
132			self.assertEquals(table.SimpleQuery(table.Column("AREA"), ">", 10.0),
133			[144])
134	bh	765
135			def test_transient_joined_table(self):
136			"""Test TransientJoinedTable"""
137	jan	805	simple = MemoryTable([("type", FIELDTYPE_STRING),
138	bh	765	("code", FIELDTYPE_INT)],
139			[("OTHER/UNKNOWN", 0),
140			("RUINS", 1),
141			("FARM", 2),
142			("BUILDING", 3),
143			("HUT", 4),
144			("LIGHTHOUSE", 5)])
145			auto = AutoTransientTable(self.transientdb, simple)
146			filename = os.path.join("..", "Data", "iceland",
147			"cultural_landmark-point.dbf")
148			landmarks = AutoTransientTable(self.transientdb, DBFTable(filename))
149
150			table = TransientJoinedTable(self.transientdb, landmarks, "CLPTLABEL",
151			auto, "type")
152
153	bh	818	self.assertEquals(table.NumRows(), 34)
154			self.assertEquals(table.NumColumns(), 8)
155	bh	839	self.assertEquals(table.Column(0).type, FIELDTYPE_DOUBLE)
156			self.assertEquals(table.Column(0).name, 'AREA')
157			self.assertEquals(table.Column(7).type, FIELDTYPE_INT)
158			self.assertEquals(table.Column(7).name, 'code')
159			self.assertEquals(table.Column(4).type, FIELDTYPE_STRING)
160			self.assertEquals(table.Column(4).name, 'CLPTLABEL')
161			# HasColumn
162			self.failUnless(table.HasColumn("AREA"))
163			self.failUnless(table.HasColumn(1))
164			# HasColumn for non-exisiting columns
165			self.failIf(table.HasColumn("non_existing_name"))
166			self.failIf(table.HasColumn(100))
167	bh	765
168			# Read an `interesting' record
169	bh	839	self.assertEquals(table.ReadRowAsDict(22),
170	bh	765	{'PERIMETER': 0.0, 'CLPOINT_': 23,
171			'AREA': 0.0, 'CLPTLABEL': 'RUINS',
172			'CLPOINT_ID': 38, 'CLPTFLAG': 0,
173			'code': 1, 'type': 'RUINS'})
174
175			# The transient_table method should return the table itself
176			self.assert_(table is table.transient_table())
177
178
179	bh	785	def test_transient_table_read_twice(self):
180	bh	842	"""Test TransientTable.ReadRowAsDict() reading the same record twice"""
181	jan	805	simple = MemoryTable([("type", FIELDTYPE_STRING),
182	bh	785	("code", FIELDTYPE_INT)],
183			[("OTHER/UNKNOWN", 0),
184			("RUINS", 1),
185			("FARM", 2),
186			("BUILDING", 3),
187			("HUT", 4),
188			("LIGHTHOUSE", 5)])
189			table = TransientTable(self.transientdb, simple)
190	bh	765
191	bh	785	# There was a bug where reading the same record twice would
192			# raise an exception in the second call because of an
193			# unitialized local variable, so for passing the test it's
194			# enough if reading simply succeeds. OTOH, while we're at it we
195			# might as well check whether the results are equal anyway :)
196	bh	839	result1 = table.ReadRowAsDict(3)
197			result2 = table.ReadRowAsDict(3)
198	bh	785	self.assertEquals(result1, result2)
199
200	bh	818
201	bh	842	def test_transient_table_query(self):
202			"""Test TransientTable.SimpleQuery()"""
203			simple = MemoryTable([("type", FIELDTYPE_STRING),
204			("value", FIELDTYPE_DOUBLE),
205			("code", FIELDTYPE_INT)],
206			[("OTHER/UNKNOWN", -1.5, 11),
207			("RUINS", 0.0, 1),
208			("FARM", 3.141, 2),
209			("BUILDING", 2.5, 3),
210			("HUT", 1e6, 4),
211			("LIGHTHOUSE", -0.01, 5)])
212			table = TransientTable(self.transientdb, simple)
213
214			# A column and a value
215			self.assertEquals(table.SimpleQuery(table.Column(0), "==", "RUINS"),
216			[1])
217			self.assertEquals(table.SimpleQuery(table.Column(2), "!=", 2),
218			[0, 1, 3, 4, 5])
219			self.assertEquals(table.SimpleQuery(table.Column(1), "<", 1.0),
220			[0, 1, 5])
221			self.assertEquals(table.SimpleQuery(table.Column(1), "<=", -1.5),
222			[0])
223			self.assertEquals(table.SimpleQuery(table.Column(2), ">", 3),
224			[0, 4, 5])
225			self.assertEquals(table.SimpleQuery(table.Column(2), ">=", 3),
226			[0, 3, 4, 5])
227
228			# Two columns as operands
229			self.assertEquals(table.SimpleQuery(table.Column(1),
230			"<=", table.Column(2)),
231			[0, 1, 3, 5])
232
233			# Test whether invalid operators raise a ValueError
234			self.assertRaises(ValueError,
235			table.SimpleQuery,
236			table.Column(1), "<<", table.Column(2))
237
238
239	bh	765	if __name__ == "__main__":
240			support.run_tests()
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision