WIP-pyshapelib-bramz/test/test_postgis_session.py

# Copyright (C) 2003 by Intevation GmbH
# Authors:
# Bernhard Herzog <[email protected]>
#
# 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$"
# $Source$
# $Id$

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()
1	bh	1607	# Copyright (C) 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 the software for details.
7
8			"""Test PostGISConnection and Session interaction"""
9
10			__version__ = "$Revision$"
11			# $Source$
12			# $Id$
13
14			import unittest
15
16			import postgissupport
17			import support
18			support.initthuban()
19
20			from Thuban.Model.postgisdb import PostGISConnection
21			from Thuban.Model.session import Session
22			from Thuban.Model.messages import DBCONN_ADDED, DBCONN_REMOVED
23
24			class TestSessionWithPostGIS(unittest.TestCase, support.SubscriberMixin):
25
26			def setUp(self):
27			"""Start the server and create a database.
28
29			The database name will be stored in self.dbname, the server
30			object in self.server and the db object in self.db.
31			"""
32			postgissupport.skip_if_no_postgis()
33			self.server = postgissupport.get_test_server()
34			self.dbref = self.server.get_default_static_data_db()
35			self.dbname = self.dbref.dbname
36			self.session = Session("PostGIS Session")
37			self.db = PostGISConnection(dbname = self.dbname,
38	bh	1634	**self.server.connection_params("user"))
39	bh	1607
40			self.session.Subscribe(DBCONN_ADDED,
41			self.subscribe_with_params, DBCONN_ADDED)
42			self.session.Subscribe(DBCONN_REMOVED,
43			self.subscribe_with_params, DBCONN_REMOVED)
44			self.clear_messages()
45
46			def tearDown(self):
47			self.session.Destroy()
48			self.clear_messages()
49
50			def test_add_dbconn(self):
51			"""Test Session.AddDBConnection()"""
52			# Sanity check. No messages should have been generated so far
53			self.check_messages([])
54
55			self.session.AddDBConnection(self.db)
56
57			# After the connection has been added it should show up in the
58			# list returned by DBConnections and a DBCONN_ADDED message
59			# should have been sent
60			self.assertEquals(self.session.DBConnections(), [self.db])
61			self.check_messages([(DBCONN_ADDED,)])
62
63			def test_remove_dbconn(self):
64			"""Test Session.RemoveDBConnection()"""
65			self.session.AddDBConnection(self.db)
66			self.clear_messages()
67			self.session.RemoveDBConnection(self.db)
68
69			# After the connection has been added it should not show up in
70			# the list returned by DBConnections any more and a
71			# DBCONN_REMOVED message should have been sent
72			self.assertEquals(self.session.DBConnections(), [])
73			self.check_messages([(DBCONN_REMOVED,)])
74
75			def test_remove_dbconn_exception(self):
76			"""Test Session.RemoveDBConnection() with unknown connection"""
77			self.session.AddDBConnection(self.db)
78			self.clear_messages()
79
80			# Trying to remove an unknown connection will raise a
81			# ValueError.
82			self.assertRaises(ValueError, self.session.RemoveDBConnection,
83			PostGISConnection(dbname = self.dbname,
84	bh	1634	**self.server.connection_params("user")))
85	bh	1607	# No message should have been sent when the removal fails
86			self.check_messages([])
87
88			def test_open_db_shapestore(self):
89			"""Test Session.OpenDBShapeStore()"""
90			self.session.AddDBConnection(self.db)
91			store = self.session.OpenDBShapeStore(self.db, "landmarks")
92			self.assertEquals(store.NumShapes(), 34)
93
94			def test_remove_dbconn_still_in_use(self):
95			"""Test Session.RemoveDBConnection() with connectin still in use"""
96			self.session.AddDBConnection(self.db)
97			store = self.session.OpenDBShapeStore(self.db, "landmarks")
98
99			# Removing a db connection that's still in use raises a
100			# ValueError (not sure the choice of ValueError is really good
101			# here).
102			self.assertRaises(ValueError, self.session.RemoveDBConnection, self.db)
103
104			def test_can_remove_db_con(self):
105			"""Test Session.CanRemoveDBConnection()"""
106			self.session.AddDBConnection(self.db)
107			store = self.session.OpenDBShapeStore(self.db, "landmarks")
108
109			# The db connection is in use by store, so CanRemoveDBConnection
110			# should return false
111			self.failIf(self.session.CanRemoveDBConnection(self.db))
112
113			# The only reference to the shapestore is in store, so deleting
114			# the it should remove the weak reference kept by the session so
115			# that afterwards CanRemoveDBConnection should return true
116			del store
117
118			self.failUnless(self.session.CanRemoveDBConnection(self.db))
119
120			def test_has_db_conections(self):
121			"""Test Session.HasDBConnections()"""
122			self.failIf(self.session.HasDBConnections())
123
124			self.session.AddDBConnection(self.db)
125			self.failUnless(self.session.HasDBConnections())
126
127			self.session.RemoveDBConnection(self.db)
128			self.failIf(self.session.HasDBConnections())
129
130
131			if __name__ == "__main__":
132			support.run_tests()
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision