# Copyright (C) 2003 by Intevation GmbH
# Authors:
# Bernhard Herzog <bh@intevation.de>
#
# This program is free software under the GPL (>=v2)
# Read the file COPYING coming with the software for details.

"""Test PostGISConnection and Session interaction"""

__version__ = "$Revision: 1.2 $"
# $Source: /thubanrepository/thuban/test/test_postgis_session.py,v $
# $Id: test_postgis_session.py,v 1.2 2003/08/22 16:55:19 bh Exp $

import unittest

import postgissupport
import support
support.initthuban()

from Thuban.Model.postgisdb import PostGISConnection
from Thuban.Model.session import Session
from Thuban.Model.messages import DBCONN_ADDED, DBCONN_REMOVED

class TestSessionWithPostGIS(unittest.TestCase, support.SubscriberMixin):

    def setUp(self):
        """Start the server and create a database.

        The database name will be stored in self.dbname, the server
        object in self.server and the db object in self.db.
        """
        postgissupport.skip_if_no_postgis()
        self.server = postgissupport.get_test_server()
        self.dbref = self.server.get_default_static_data_db()
        self.dbname = self.dbref.dbname
        self.session = Session("PostGIS Session")
        self.db = PostGISConnection(dbname = self.dbname,
                                    **self.server.connection_params("user"))

        self.session.Subscribe(DBCONN_ADDED,
                               self.subscribe_with_params, DBCONN_ADDED)
        self.session.Subscribe(DBCONN_REMOVED,
                               self.subscribe_with_params, DBCONN_REMOVED)
        self.clear_messages()

    def tearDown(self):
        self.session.Destroy()
        self.clear_messages()

    def test_add_dbconn(self):
        """Test Session.AddDBConnection()"""
        # Sanity check. No messages should have been generated so far
        self.check_messages([])

        self.session.AddDBConnection(self.db)

        # After the connection has been added it should show up in the
        # list returned by DBConnections and a DBCONN_ADDED message
        # should have been sent
        self.assertEquals(self.session.DBConnections(), [self.db])
        self.check_messages([(DBCONN_ADDED,)])

    def test_remove_dbconn(self):
        """Test Session.RemoveDBConnection()"""
        self.session.AddDBConnection(self.db)
        self.clear_messages()
        self.session.RemoveDBConnection(self.db)

        # After the connection has been added it should not show up in
        # the list returned by DBConnections any more and a
        # DBCONN_REMOVED message should have been sent
        self.assertEquals(self.session.DBConnections(), [])
        self.check_messages([(DBCONN_REMOVED,)])

    def test_remove_dbconn_exception(self):
        """Test Session.RemoveDBConnection() with unknown connection"""
        self.session.AddDBConnection(self.db)
        self.clear_messages()

        # Trying to remove an unknown connection will raise a
        # ValueError.
        self.assertRaises(ValueError, self.session.RemoveDBConnection,
                          PostGISConnection(dbname = self.dbname,
                                   **self.server.connection_params("user")))
        # No message should have been sent when the removal fails
        self.check_messages([])

    def test_open_db_shapestore(self):
        """Test Session.OpenDBShapeStore()"""
        self.session.AddDBConnection(self.db)
        store = self.session.OpenDBShapeStore(self.db, "landmarks")
        self.assertEquals(store.NumShapes(), 34)

    def test_remove_dbconn_still_in_use(self):
        """Test Session.RemoveDBConnection() with connectin still in use"""
        self.session.AddDBConnection(self.db)
        store = self.session.OpenDBShapeStore(self.db, "landmarks")

        # Removing a db connection that's still in use raises a
        # ValueError (not sure the choice of ValueError is really good
        # here).
        self.assertRaises(ValueError, self.session.RemoveDBConnection, self.db)

    def test_can_remove_db_con(self):
        """Test Session.CanRemoveDBConnection()"""
        self.session.AddDBConnection(self.db)
        store = self.session.OpenDBShapeStore(self.db, "landmarks")

        # The db connection is in use by store, so CanRemoveDBConnection
        # should return false
        self.failIf(self.session.CanRemoveDBConnection(self.db))

        # The only reference to the shapestore is in store, so deleting
        # the it should remove the weak reference kept by the session so
        # that afterwards CanRemoveDBConnection should return true
        del store

        self.failUnless(self.session.CanRemoveDBConnection(self.db))

    def test_has_db_conections(self):
        """Test Session.HasDBConnections()"""
        self.failIf(self.session.HasDBConnections())

        self.session.AddDBConnection(self.db)
        self.failUnless(self.session.HasDBConnections())

        self.session.RemoveDBConnection(self.db)
        self.failIf(self.session.HasDBConnections())


if __name__ == "__main__":
    support.run_tests()
