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

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

Parent Directory | Revision Log | View Patch Patch

-revision 998 by bh,
Thu May 22 19:29:39 2003 UTC
+revision 1381 by bh,
Tue Jul  8 16:37:46 2003 UTC
 Line 19 
 import unittest
  import support
  support.initthuban()
+ import dbflib
  from Thuban.Model.table import DBFTable, MemoryTable, FIELDTYPE_STRING, \
-                                FIELDTYPE_INT, FIELDTYPE_DOUBLE
+                                FIELDTYPE_INT, FIELDTYPE_DOUBLE, table_to_dbf
  from Thuban.Model.transientdb import TransientDatabase, TransientTable, \
       TransientJoinedTable, AutoTransientTable
-Line 50 
 class TestTransientTable(unittest.TestCa
+Line 51 
 class TestTransientTable(unittest.TestCa
          self.assertEquals(table.NumRows(), 156)
          self.assertEquals(table.NumColumns(), 8)
-         # Check one each of the possible field types. The width and
+         # Check one each of the possible field types.
-         # decimal precision is always 0.
          columns = table.Columns()
          self.assertEquals(columns[0].name, 'AREA')
          self.assertEquals(columns[0].type, FIELDTYPE_DOUBLE)
 Line 215 
 class TestTransientTable(unittest.TestCa
                                                   auto.Title()))
+     def test_transient_joined_table_same_column_name(self):
+         """Test TransientJoinedTable join on columns with same name
+         The transient DB maps the column names used by the tables to
+         another set of names used only inside the SQLite database. There
+         was a bug in the way this mapping was used when joining on
+         fields with the same names in both tables so that the joined
+         table ended up joining on the same column in the same table.
+         """
+         mem_stretches = MemoryTable([("stretch_id", FIELDTYPE_INT)],
+                                     [(i,) for i in range(4)])
+         stretches = AutoTransientTable(self.transientdb, mem_stretches)
+         mem_discharges = MemoryTable([("disch_id", FIELDTYPE_INT),
+                                       ("stretch_id", FIELDTYPE_INT)],
+                                      [(1, 0), (2, 3)])
+         discharges = AutoTransientTable(self.transientdb, mem_discharges)
+         table = TransientJoinedTable(self.transientdb, stretches, "stretch_id",
+                                      discharges, "stretch_id",
+                                      outer_join = True)
+         self.assertEquals(table.NumRows(), 4)
+         self.assertEquals(table.NumColumns(), 3)
+         # HasColumn
+         self.failUnless(table.HasColumn("stretch_id"))
+         self.failUnless(table.HasColumn("disch_id"))
+     def test_transient_joined_table_with_equal_column_names(self):
+         """Test TransientJoinedTable join on tables with equal column names
+         If a name collision occurs for the field names, underscores are
+         appended as long as any collision is resolved.
+         """
+         mem_stretches = MemoryTable([("stretch_id", FIELDTYPE_INT),
+                                      ("name", FIELDTYPE_INT)],
+                                     [(0, 10), (1, 11), (2, 12), (3, 13) ])
+         stretches = AutoTransientTable(self.transientdb, mem_stretches)
+         mem_discharges = MemoryTable([("disch_id", FIELDTYPE_INT),
+                                       ("stretch_id", FIELDTYPE_INT),
+                                       ("name", FIELDTYPE_INT)],
+                                      [(1, 0, 1), (2, 3, 2)])
+         discharges = AutoTransientTable(self.transientdb, mem_discharges)
+         table = TransientJoinedTable(self.transientdb, stretches, "stretch_id",
+                                      discharges, "stretch_id",
+                                      outer_join = True)
+         self.assertEquals(table.NumRows(), 4)
+         self.assertEquals(table.NumColumns(), 5)
+         # HasColumn
+         self.assertEquals([c.name for c in table.Columns()],
+                           ["stretch_id", "name", "disch_id", "stretch_id_",
+                            "name_"])
+     def test_transient_joined_table_name_collisions_dont_modify_in_place(self):
+         """Test TransientJoinedTable name-collisions do not modifying in place
+         The name collision work-around by appending underscores
+         accidentally modified the column objects in place. We do two
+         joins therefore in reverse order to detect this: The first join
+         will lead to a modified name in the column object of the right
+         table which is then used as the left table in the second join so
+         the underscored name will appear before the non-underscored one
+         in the list of column names after the second join.
+         """
+         mem1 = MemoryTable([("stretch_id", FIELDTYPE_INT),
+                             ("name", FIELDTYPE_INT)],
+                            [(0, 10), (1, 11), (2, 12), (3, 13) ])
+         table1 = AutoTransientTable(self.transientdb, mem1)
+         mem2 = MemoryTable([("stretch_id", FIELDTYPE_INT),
+                             ("name", FIELDTYPE_INT)],
+                            [(0, 10), (1, 11), (2, 12), (3, 13) ])
+         table2 = AutoTransientTable(self.transientdb, mem2)
+         table = TransientJoinedTable(self.transientdb, table1, "stretch_id",
+                                      table2, "stretch_id", outer_join = True)
+         table = TransientJoinedTable(self.transientdb, table2, "stretch_id",
+                                      table1, "stretch_id", outer_join = True)
+         self.assertEquals([c.name for c in table.Columns()],
+                           ["stretch_id", "name", "stretch_id_", "name_"])
      def test_transient_table_read_twice(self):
          """Test TransientTable.ReadRowAsDict() reading the same record twice"""
          simple = MemoryTable([("type", FIELDTYPE_STRING),
-Line 274 
 class TestTransientTable(unittest.TestCa
+Line 362 
 class TestTransientTable(unittest.TestCa
                            table.SimpleQuery,
                            table.Column(1), "<<", table.Column(2))
+     def test_transienttable_to_dbf(self):
+         memtable = MemoryTable([("type", FIELDTYPE_STRING),
+                                 ("value", FIELDTYPE_DOUBLE),
+                                 ("code", FIELDTYPE_INT)],
+                                [("UNKNOWN", 0.0, 0),
+                                 ("Foo", 0.5, -1),
+                                 ("Foo", 1.0/256, 100),
+                                 ("bar", 1e10, 17)])
+         table = TransientTable(self.transientdb, memtable)
+         filename = self.temp_file_name("test_transienttable_to_dbf.dbf")
+         table_to_dbf(table, filename)
+         dbf = dbflib.DBFFile(filename)
+         self.assertEquals(dbf.read_record(2),
+                           {'code': 100, 'type': 'Foo', 'value': 0.00390625})
+         self.assertEquals(dbf.field_count(), 3)
+         self.assertEquals(dbf.record_count(), 4)
+         self.assertEquals(dbf.field_info(0),
+                           (dbflib.FTString, "type", 7, 0))
+         self.assertEquals(dbf.field_info(1),
+                           (dbflib.FTDouble, "value", 24, 12))
+         self.assertEquals(dbf.field_info(2),
+                           (dbflib.FTInteger, "code", 3, 0))
  if __name__ == "__main__":
      support.run_tests()

 Legend:



Removed from v.998
 


changed lines


 
Added in v.1381
 Legend:



Removed from v.998
 


changed lines


 
Added in v.1381
-Removed from v.998
+Added in v.1381

[email protected]	ViewVC Help
Powered by ViewVC 1.1.26