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, FIELDTYPE_STRING, FIELDTYPE_INT
from Thuban.Model.transientdb import TransientDatabase, TransientTable, \
     TransientJoinedTable, AutoTransientTable


class SimpleTable:

    """Very simple table implementation that operates on a list of tuples"""

    def __init__(self, fields, data):
        """Initialize the SimpleTable

        Parameters:
        fields -- List of (name, field_type) pairs
        data -- List of tuples, one for each row of data
        """
        self.fields = fields
        self.data = data

    def field_count(self):
        return len(self.fields)

    def field_info(self, index):
        name, type = self.fields[index]
        return (type, name, 0, 0)

    def record_count(self):
        return len(self.data)

    def read_record(self, index):
        return dict([(self.fields[i][0], self.data[index][i])
                      for i in range(len(self.fields))])


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.record_count(), 156)
        self.assertEquals(table.field_count(), 8)

        # Check one each of the possible field types. The width and
        # decimal precision is always 0.
        self.assertEquals(table.field_info(0), ('double', 'AREA', 0, 0))
        self.assertEquals(table.field_info(3), ('int', 'PONET_ID', 0, 0))
        self.assertEquals(table.field_info(6), ('string', 'POPYCOUN', 0, 0))

        # Read an `interesting' record
        self.assertEquals(table.read_record(144),
                          {'POPYCOUN': 'IC', 'POPYADMIN': '', 'PONET_': 146,
                           'AREA': 19.462,
                           'POPYTYPE': 1, 'PERIMETER': 88.518000000000001,
                           'POPYREG': '1',
                           'PONET_ID': 145})

        # field_range 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.field_range("AREA"),
                          ((0.0, None), (19.462, None)))

        unique = table.GetUniqueValues("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_transient_joined_table(self):
        """Test TransientJoinedTable"""
        simple = SimpleTable([("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.record_count(), 34)
        self.assertEquals(table.field_count(), 8)
        self.assertEquals(table.field_info(0), ('double', 'AREA', 0, 0))
        self.assertEquals(table.field_info(7), ('int', 'code', 0, 0))
        self.assertEquals(table.field_info(4), ('string', 'CLPTLABEL', 0, 0))

        # Read an `interesting' record
        self.assertEquals(table.read_record(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.read_record() reading the same record twice"""
        simple = SimpleTable([("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.read_record(3)
        result2 = table.read_record(3)
        self.assertEquals(result1, result2)

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, 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	def test_transient_table_read_twice(self):
176	"""Test TransientTable.read_record() reading the same record twice"""
177	simple = SimpleTable([("type", FIELDTYPE_STRING),
178	("code", FIELDTYPE_INT)],
179	[("OTHER/UNKNOWN", 0),
180	("RUINS", 1),
181	("FARM", 2),
182	("BUILDING", 3),
183	("HUT", 4),
184	("LIGHTHOUSE", 5)])
185	table = TransientTable(self.transientdb, simple)
186
187	# There was a bug where reading the same record twice would
188	# raise an exception in the second call because of an
189	# unitialized local variable, so for passing the test it's
190	# enough if reading simply succeeds. OTOH, while we're at it we
191	# might as well check whether the results are equal anyway :)
192	result1 = table.read_record(3)
193	result2 = table.read_record(3)
194	self.assertEquals(result1, result2)
195
196	if __name__ == "__main__":
197	support.run_tests()
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision