# Copyright (c) 2002, 2003, 2006 by Intevation GmbH
# Authors:
# Bernhard Herzog <bh@intevation.de>
#
# 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: 2705 $"
# $Source$
# $Id: test_transientdb.py 2705 2006-09-24 18:55:30Z bernhard $

import os
import unittest

import support
support.initthuban()

import dbflib
from Thuban.Model.table import DBFTable, MemoryTable, FIELDTYPE_STRING, \
                               FIELDTYPE_INT, FIELDTYPE_DOUBLE, table_to_dbf
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 table

        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.
        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.ReadRowAsDict(144, row_is_ordinal = 1),
                          {'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)
        self.assertEquals(table.ReadValue(144, "AREA", row_is_ordinal = 1),
                          19.462)
        self.assertEquals(table.ReadValue(144, 3, row_is_ordinal = 1), 145)

        self.assertEquals(table.RowIdToOrdinal(23), 23)
        self.assertEquals(table.RowOrdinalToId(23), 23)

        # 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)
        # At this point the data has probably not been copied to the
        # transient DB yet, so we force it by calling the
        # transient_table method.
        table.transient_table()

        # Run the tests again.
        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)
        self.assertEquals(table.ReadValue(22, "type", row_is_ordinal = 1),
                          "RUINS")
        self.assertEquals(table.ReadValue(22, 7, row_is_ordinal = 1), 1)
        self.assertEquals(table.RowIdToOrdinal(23), 23)
        self.assertEquals(table.RowOrdinalToId(23), 23)

        # 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_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),
                              ("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))

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