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,
                                    port = self.server.port,
                                    host = self.server.host)

        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,
                                            port = self.server.port,
                                            host = self.server.host))
        # 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			port = self.server.port,
39			host = self.server.host)
40
41			self.session.Subscribe(DBCONN_ADDED,
42			self.subscribe_with_params, DBCONN_ADDED)
43			self.session.Subscribe(DBCONN_REMOVED,
44			self.subscribe_with_params, DBCONN_REMOVED)
45			self.clear_messages()
46
47			def tearDown(self):
48			self.session.Destroy()
49			self.clear_messages()
50
51			def test_add_dbconn(self):
52			"""Test Session.AddDBConnection()"""
53			# Sanity check. No messages should have been generated so far
54			self.check_messages([])
55
56			self.session.AddDBConnection(self.db)
57
58			# After the connection has been added it should show up in the
59			# list returned by DBConnections and a DBCONN_ADDED message
60			# should have been sent
61			self.assertEquals(self.session.DBConnections(), [self.db])
62			self.check_messages([(DBCONN_ADDED,)])
63
64			def test_remove_dbconn(self):
65			"""Test Session.RemoveDBConnection()"""
66			self.session.AddDBConnection(self.db)
67			self.clear_messages()
68			self.session.RemoveDBConnection(self.db)
69
70			# After the connection has been added it should not show up in
71			# the list returned by DBConnections any more and a
72			# DBCONN_REMOVED message should have been sent
73			self.assertEquals(self.session.DBConnections(), [])
74			self.check_messages([(DBCONN_REMOVED,)])
75
76			def test_remove_dbconn_exception(self):
77			"""Test Session.RemoveDBConnection() with unknown connection"""
78			self.session.AddDBConnection(self.db)
79			self.clear_messages()
80
81			# Trying to remove an unknown connection will raise a
82			# ValueError.
83			self.assertRaises(ValueError, self.session.RemoveDBConnection,
84			PostGISConnection(dbname = self.dbname,
85			port = self.server.port,
86			host = self.server.host))
87			# No message should have been sent when the removal fails
88			self.check_messages([])
89
90			def test_open_db_shapestore(self):
91			"""Test Session.OpenDBShapeStore()"""
92			self.session.AddDBConnection(self.db)
93			store = self.session.OpenDBShapeStore(self.db, "landmarks")
94			self.assertEquals(store.NumShapes(), 34)
95
96			def test_remove_dbconn_still_in_use(self):
97			"""Test Session.RemoveDBConnection() with connectin still in use"""
98			self.session.AddDBConnection(self.db)
99			store = self.session.OpenDBShapeStore(self.db, "landmarks")
100
101			# Removing a db connection that's still in use raises a
102			# ValueError (not sure the choice of ValueError is really good
103			# here).
104			self.assertRaises(ValueError, self.session.RemoveDBConnection, self.db)
105
106			def test_can_remove_db_con(self):
107			"""Test Session.CanRemoveDBConnection()"""
108			self.session.AddDBConnection(self.db)
109			store = self.session.OpenDBShapeStore(self.db, "landmarks")
110
111			# The db connection is in use by store, so CanRemoveDBConnection
112			# should return false
113			self.failIf(self.session.CanRemoveDBConnection(self.db))
114
115			# The only reference to the shapestore is in store, so deleting
116			# the it should remove the weak reference kept by the session so
117			# that afterwards CanRemoveDBConnection should return true
118			del store
119
120			self.failUnless(self.session.CanRemoveDBConnection(self.db))
121
122			def test_has_db_conections(self):
123			"""Test Session.HasDBConnections()"""
124			self.failIf(self.session.HasDBConnections())
125
126			self.session.AddDBConnection(self.db)
127			self.failUnless(self.session.HasDBConnections())
128
129			self.session.RemoveDBConnection(self.db)
130			self.failIf(self.session.HasDBConnections())
131
132
133			if __name__ == "__main__":
134			support.run_tests()
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision