WIP-pyshapelib-bramz/test/test_connector.py

# Copyright (c) 2002, 2003 by Intevation GmbH
# Authors:
# Bernhard Herzog <[email protected]>
#
# This program is free software under the GPL (>=v2)
# Read the file COPYING coming with Thuban for details.

"""
Test the Connector class
"""

__version__ = "$Revision$"
# $Source$
# $Id$

import unittest

import support
support.initthuban()

from Thuban.Lib.connector import Connector, Publisher

# some messages used in the tests
SIMPLE = "SIMPLE"
PARAM = "PARAM"

class SimplePublisher:

    """A version of Publisher that uses a specific connector.

    The Publisher class in Thuban.Lib.connector uses the global
    connector in the same module.
    """

    def __init__(self, connector):
        self.connector = connector

    def __del__(self):
        self.connector.RemovePublisher(self)

    def issue(self):
        """Issue a SIMPLE message without parameters"""
        self.connector.Issue(self, SIMPLE)

    def issue_arg(self):
        """Issue a PARAM message with 42 as parameter"""
        self.connector.Issue(self, PARAM, 42)


class RealPublisher(Publisher):

    """Extended version of Publisher for testing purposes.

    Publisher is not intended to be used directly. It is used as a base
    class for objects that send messages when they change. So we do just
    that here and derive from Publisher to provide some simple methods
    that issue messages.
    """

    def simple_action(self):
        """Issue a SIMPLE message without parameters"""
        self.issue(SIMPLE)

    def param_action(self):
        """Issue a PARAM message with 42 as parameter"""
        self.issue(PARAM, 42)


class Receiver:

    """Class to be used as a generic receiver of messages.

    An instance of this class has some methods that can be used as
    subscribers for messages. These messages put information about the
    messages they received into the public instance variable messages.
    See the method's doc-strings for more information.

    Furthermore, the class is instantiated with a test case object as
    parameter and the instance notifies the test case when it's being
    instantiated and deleted so that the test case can determine which
    objects weren't deleted.
    """

    def __init__(self, testcase):
        """Initialize the object for the given testcase.

        Call the testcase's expect_delete method with self as parameter.
        """
        self.testcase = testcase
        self.testcase.expect_delete(self)
        self.reset()

    def __del__(self):
        """Tell the test case that the object has been deleted"""
        self.testcase.deleted(self)

    def reset(self):
        """Clear the list of received messages"""
        self.messages = []

    def no_params(self):
        """Method for subscriptions without parameters

        Add the tuple ("no_params",) to self.messages
        """
        self.messages.append(("no_params",))

    def with_params(self, *args):
        """Method for subscriptions with parameters

        Add a tuple with the string 'params' followed by the arguments
        of this function (except for the self parameter) to
        self.messages.
        """
        self.messages.append(("params",) + args)


class DeletionTestMixin:

    """Mixin class to check for memory leaks.

    Mixin class for test that want to determine whether certain objects
    have been destroyed.

    This class maintains two lists, deleted_objects and
    expected_deletions to determine whether all objects which are
    expected to be deleted by a test are actually deleted.
    """

    def setUp(self):
        """Initialize self.deleted_objects and self.expected_deletions"""
        self.deleted_objects = []
        self.expected_deletions = []

    def expect_delete(self, obj):
        """Append the id of obj to the self.expected_deletions"""
        self.expected_deletions.append(id(obj))

    def deleted(self, obj):
        """Append the id of obj to the self.deleted_objects"""
        self.deleted_objects.append(id(obj))

    def check_deletetions(self):
        """Assert equality of self.expected_deletions and self.deleted_objects

        This check simply compares the lists for equality and thus
        effectively assumes that the objects are deleted in the same
        order in which they're added to the list which if used only for
        Receiver instances is the order in which they're instantiated.
        """
        self.assertEquals(self.expected_deletions, self.deleted_objects)


class ConnectorTest(unittest.TestCase, DeletionTestMixin):

    """Test cases for the Connector class.

    These tests use the SimplePublisher class instead of the Publisher
    class in Thuban.Lib.connector because we only want to test the
    connector here.
    """

    def setUp(self):
        """Extend the inherited method to create a Connector instance.

        Bind the Connector to self.connector.
        """
        self.connector = Connector()
        DeletionTestMixin.setUp(self)

    def test_issue_simple(self):
        """Test connector issue without parameters"""
        # Make a publisher and a subscriber and connect the two
        pub = SimplePublisher(self.connector)
        rec = Receiver(self)
        self.connector.Connect(pub, SIMPLE, rec.no_params, ())

        # now the publisher should have subscribers
        self.assert_(self.connector.HasSubscribers(pub))

        # Issue a message and check whether the receiver got it
        pub.issue()
        self.assertEquals(rec.messages, [("no_params",)])
        rec.reset()

        # disconnect and check that the message doesn't get send anymore
        self.connector.Disconnect(pub, SIMPLE, rec.no_params, ())
        pub.issue()
        self.assertEquals(rec.messages, [])

        # now the publisher should have no subscribers
        self.failIf(self.connector.HasSubscribers(pub))

        # make sure that all references have been deleted
        del rec
        self.check_deletetions()

    def test_issue_param(self):
        """Test connector issue with parameters"""
        pub = SimplePublisher(self.connector)
        rec = Receiver(self)
        # Three cases: 1. The parameter supplied by pub.issue_arg, 2.
        # only the parameter given when connecting, 3. both
        self.connector.Connect(pub, PARAM, rec.with_params, ())
        self.connector.Connect(pub, SIMPLE, rec.with_params, ("deliverator",))
        self.connector.Connect(pub, PARAM, rec.with_params, ("loglo",))

        pub.issue_arg()
        pub.issue()
        self.assertEquals(rec.messages, [("params", 42),
                                         ("params", 42, "loglo"),
                                         ("params", "deliverator")])

        # make sure that all references have been deleted
        self.connector.RemovePublisher(pub)
        del rec
        self.check_deletetions()

    def test_cyclic_references(self):
        """Test whether connector avoids cyclic references"""
        pub = SimplePublisher(self.connector)
        rec = Receiver(self)
        self.connector.Connect(pub, SIMPLE, rec.no_params, ())

        # deleting pub and rec should be enough that the last reference
        # to rec has been dropped because the connector doesn't keep
        # references to the publishers and SimplePublisher's __del__
        # method removes all subscriptions
        del pub
        del rec
        self.check_deletetions()

    def test_disconnect_in_receiver(self):
        """Test unsubscribing from a channel while receiving a message

        There was a bug in the connector implementation in the following
        situation:

         - 2 receivers for the same channel

         - the reiver called first unsubscribes itself from that channel
           in response to a message on that channel

        Now the second receiver is never called because the list of
        receivers was modified by Disconnect while the connecter was
        iterating over it.
        """
        messages = []
        def rec1(*args):
            try:
                messages.append("rec1")
                self.connector.Disconnect(None, SIMPLE, rec1, ())
            except:
                self.fail("Exception in rec1")
        def rec2(*args):
            try:
                messages.append("rec2")
                self.connector.Disconnect(None, SIMPLE, rec2, ())
            except:
                self.fail("Exception in rec1")

        self.connector.Connect(None, SIMPLE, rec1, ())
        self.connector.Connect(None, SIMPLE, rec2, ())

        self.connector.Issue(None, SIMPLE)

        self.assertEquals(messages, [("rec1"), ("rec2")])


class TestPublisher(unittest.TestCase, DeletionTestMixin):

    """Tests for the Publisher class"""

    def setUp(self):
        DeletionTestMixin.setUp(self)

    def test_issue_simple(self):
        """Test Publisher message without parameters"""
        # Make a publisher and a subscriber and connect the two
        pub = RealPublisher()
        rec = Receiver(self)
        pub.Subscribe(SIMPLE, rec.no_params)

        # Issue a message and check whether the receiver got it
        pub.simple_action()
        self.assertEquals(rec.messages, [("no_params",)])
        rec.reset()

        # disconnect and check that the message doesn't get send anymore
        pub.Unsubscribe(SIMPLE, rec.no_params)
        pub.simple_action()
        self.assertEquals(rec.messages, [])

        # make sure that all references have been deleted
        del rec
        self.check_deletetions()

    def test_issue_param(self):
        """Test Publisher message with parameters"""
        pub = RealPublisher()
        rec = Receiver(self)
        # Three cases: 1. The parameter supplied by pub.issue_arg, 2.
        # only the parameter given when connecting, 3. both
        pub.Subscribe(PARAM, rec.with_params)
        pub.Subscribe(SIMPLE, rec.with_params, "deliverator")
        pub.Subscribe(PARAM, rec.with_params, "loglo")

        pub.param_action()
        pub.simple_action()
        self.assertEquals(rec.messages, [("params", 42),
                                         ("params", 42, "loglo"),
                                         ("params", "deliverator")])

        # make sure that all references have been deleted
        pub.Destroy()
        del rec
        self.check_deletetions()

    def test_cyclic_references(self):
        """Test whether Publisher avoids cyclic references"""
        pub = RealPublisher()
        rec = Receiver(self)
        pub.Subscribe(SIMPLE, rec.no_params, ())

        # deleting pub and rec should be enough that the last reference
        # to rec has been dropped because the connector doesn't keep
        # references to the publishers and SimplePublisher's __del__
        # method removes all subscriptions
        del pub
        del rec
        self.check_deletetions()


if __name__ == "__main__":
    unittest.main()
1	bh	1144	# Copyright (c) 2002, 2003 by Intevation GmbH
2	bh	329	# Authors:
3			# Bernhard Herzog <[email protected]>
4			#
5			# This program is free software under the GPL (>=v2)
6			# Read the file COPYING coming with Thuban for details.
7
8			"""
9			Test the Connector class
10			"""
11
12			__version__ = "$Revision$"
13			# $Source$
14			# $Id$
15
16			import unittest
17
18			import support
19			support.initthuban()
20
21			from Thuban.Lib.connector import Connector, Publisher
22
23			# some messages used in the tests
24			SIMPLE = "SIMPLE"
25			PARAM = "PARAM"
26
27			class SimplePublisher:
28
29			"""A version of Publisher that uses a specific connector.
30
31			The Publisher class in Thuban.Lib.connector uses the global
32			connector in the same module.
33			"""
34
35			def __init__(self, connector):
36			self.connector = connector
37
38			def __del__(self):
39			self.connector.RemovePublisher(self)
40
41			def issue(self):
42			"""Issue a SIMPLE message without parameters"""
43			self.connector.Issue(self, SIMPLE)
44
45			def issue_arg(self):
46			"""Issue a PARAM message with 42 as parameter"""
47			self.connector.Issue(self, PARAM, 42)
48
49
50			class RealPublisher(Publisher):
51
52			"""Extended version of Publisher for testing purposes.
53
54			Publisher is not intended to be used directly. It is used as a base
55			class for objects that send messages when they change. So we do just
56			that here and derive from Publisher to provide some simple methods
57			that issue messages.
58			"""
59
60			def simple_action(self):
61			"""Issue a SIMPLE message without parameters"""
62			self.issue(SIMPLE)
63
64			def param_action(self):
65			"""Issue a PARAM message with 42 as parameter"""
66			self.issue(PARAM, 42)
67
68
69			class Receiver:
70
71			"""Class to be used as a generic receiver of messages.
72
73			An instance of this class has some methods that can be used as
74			subscribers for messages. These messages put information about the
75			messages they received into the public instance variable messages.
76			See the method's doc-strings for more information.
77
78			Furthermore, the class is instantiated with a test case object as
79			parameter and the instance notifies the test case when it's being
80			instantiated and deleted so that the test case can determine which
81			objects weren't deleted.
82			"""
83
84			def __init__(self, testcase):
85			"""Initialize the object for the given testcase.
86
87			Call the testcase's expect_delete method with self as parameter.
88			"""
89			self.testcase = testcase
90			self.testcase.expect_delete(self)
91			self.reset()
92
93			def __del__(self):
94			"""Tell the test case that the object has been deleted"""
95			self.testcase.deleted(self)
96
97			def reset(self):
98			"""Clear the list of received messages"""
99			self.messages = []
100
101			def no_params(self):
102			"""Method for subscriptions without parameters
103
104			Add the tuple ("no_params",) to self.messages
105			"""
106			self.messages.append(("no_params",))
107
108			def with_params(self, *args):
109			"""Method for subscriptions with parameters
110
111			Add a tuple with the string 'params' followed by the arguments
112			of this function (except for the self parameter) to
113			self.messages.
114			"""
115			self.messages.append(("params",) + args)
116
117
118
119			class DeletionTestMixin:
120
121			"""Mixin class to check for memory leaks.
122
123			Mixin class for test that want to determine whether certain objects
124			have been destroyed.
125
126			This class maintains two lists, deleted_objects and
127			expected_deletions to determine whether all objects which are
128			expected to be deleted by a test are actually deleted.
129			"""
130
131			def setUp(self):
132			"""Initialize self.deleted_objects and self.expected_deletions"""
133			self.deleted_objects = []
134			self.expected_deletions = []
135
136			def expect_delete(self, obj):
137			"""Append the id of obj to the self.expected_deletions"""
138			self.expected_deletions.append(id(obj))
139
140			def deleted(self, obj):
141			"""Append the id of obj to the self.deleted_objects"""
142			self.deleted_objects.append(id(obj))
143
144			def check_deletetions(self):
145			"""Assert equality of self.expected_deletions and self.deleted_objects
146
147			This check simply compares the lists for equality and thus
148			effectively assumes that the objects are deleted in the same
149			order in which they're added to the list which if used only for
150			Receiver instances is the order in which they're instantiated.
151			"""
152			self.assertEquals(self.expected_deletions, self.deleted_objects)
153
154
155			class ConnectorTest(unittest.TestCase, DeletionTestMixin):
156
157			"""Test cases for the Connector class.
158
159			These tests use the SimplePublisher class instead of the Publisher
160			class in Thuban.Lib.connector because we only want to test the
161			connector here.
162			"""
163
164			def setUp(self):
165			"""Extend the inherited method to create a Connector instance.
166
167			Bind the Connector to self.connector.
168			"""
169			self.connector = Connector()
170			DeletionTestMixin.setUp(self)
171
172			def test_issue_simple(self):
173			"""Test connector issue without parameters"""
174			# Make a publisher and a subscriber and connect the two
175			pub = SimplePublisher(self.connector)
176			rec = Receiver(self)
177			self.connector.Connect(pub, SIMPLE, rec.no_params, ())
178
179			# now the publisher should have subscribers
180			self.assert_(self.connector.HasSubscribers(pub))
181
182			# Issue a message and check whether the receiver got it
183			pub.issue()
184			self.assertEquals(rec.messages, [("no_params",)])
185			rec.reset()
186
187			# disconnect and check that the message doesn't get send anymore
188			self.connector.Disconnect(pub, SIMPLE, rec.no_params, ())
189			pub.issue()
190			self.assertEquals(rec.messages, [])
191
192			# now the publisher should have no subscribers
193			self.failIf(self.connector.HasSubscribers(pub))
194
195			# make sure that all references have been deleted
196			del rec
197			self.check_deletetions()
198
199			def test_issue_param(self):
200			"""Test connector issue with parameters"""
201			pub = SimplePublisher(self.connector)
202			rec = Receiver(self)
203			# Three cases: 1. The parameter supplied by pub.issue_arg, 2.
204			# only the parameter given when connecting, 3. both
205			self.connector.Connect(pub, PARAM, rec.with_params, ())
206			self.connector.Connect(pub, SIMPLE, rec.with_params, ("deliverator",))
207			self.connector.Connect(pub, PARAM, rec.with_params, ("loglo",))
208
209			pub.issue_arg()
210			pub.issue()
211			self.assertEquals(rec.messages, [("params", 42),
212			("params", 42, "loglo"),
213			("params", "deliverator")])
214
215			# make sure that all references have been deleted
216			self.connector.RemovePublisher(pub)
217			del rec
218			self.check_deletetions()
219
220			def test_cyclic_references(self):
221			"""Test whether connector avoids cyclic references"""
222			pub = SimplePublisher(self.connector)
223			rec = Receiver(self)
224			self.connector.Connect(pub, SIMPLE, rec.no_params, ())
225
226			# deleting pub and rec should be enough that the last reference
227			# to rec has been dropped because the connector doesn't keep
228			# references to the publishers and SimplePublisher's __del__
229			# method removes all subscriptions
230			del pub
231			del rec
232			self.check_deletetions()
233
234	bh	1144	def test_disconnect_in_receiver(self):
235			"""Test unsubscribing from a channel while receiving a message
236	bh	329
237	bh	1144	There was a bug in the connector implementation in the following
238			situation:
239
240			- 2 receivers for the same channel
241
242			- the reiver called first unsubscribes itself from that channel
243			in response to a message on that channel
244
245			Now the second receiver is never called because the list of
246			receivers was modified by Disconnect while the connecter was
247			iterating over it.
248			"""
249			messages = []
250			def rec1(*args):
251			try:
252			messages.append("rec1")
253			self.connector.Disconnect(None, SIMPLE, rec1, ())
254			except:
255			self.fail("Exception in rec1")
256			def rec2(*args):
257			try:
258			messages.append("rec2")
259			self.connector.Disconnect(None, SIMPLE, rec2, ())
260			except:
261			self.fail("Exception in rec1")
262
263			self.connector.Connect(None, SIMPLE, rec1, ())
264			self.connector.Connect(None, SIMPLE, rec2, ())
265
266			self.connector.Issue(None, SIMPLE)
267
268			self.assertEquals(messages, [("rec1"), ("rec2")])
269
270
271	bh	329	class TestPublisher(unittest.TestCase, DeletionTestMixin):
272
273			"""Tests for the Publisher class"""
274
275			def setUp(self):
276			DeletionTestMixin.setUp(self)
277
278			def test_issue_simple(self):
279			"""Test Publisher message without parameters"""
280			# Make a publisher and a subscriber and connect the two
281			pub = RealPublisher()
282			rec = Receiver(self)
283			pub.Subscribe(SIMPLE, rec.no_params)
284
285			# Issue a message and check whether the receiver got it
286			pub.simple_action()
287			self.assertEquals(rec.messages, [("no_params",)])
288			rec.reset()
289
290			# disconnect and check that the message doesn't get send anymore
291			pub.Unsubscribe(SIMPLE, rec.no_params)
292			pub.simple_action()
293			self.assertEquals(rec.messages, [])
294
295			# make sure that all references have been deleted
296			del rec
297			self.check_deletetions()
298
299			def test_issue_param(self):
300			"""Test Publisher message with parameters"""
301			pub = RealPublisher()
302			rec = Receiver(self)
303			# Three cases: 1. The parameter supplied by pub.issue_arg, 2.
304			# only the parameter given when connecting, 3. both
305			pub.Subscribe(PARAM, rec.with_params)
306			pub.Subscribe(SIMPLE, rec.with_params, "deliverator")
307			pub.Subscribe(PARAM, rec.with_params, "loglo")
308
309			pub.param_action()
310			pub.simple_action()
311			self.assertEquals(rec.messages, [("params", 42),
312			("params", 42, "loglo"),
313			("params", "deliverator")])
314
315			# make sure that all references have been deleted
316			pub.Destroy()
317			del rec
318			self.check_deletetions()
319
320			def test_cyclic_references(self):
321			"""Test whether Publisher avoids cyclic references"""
322			pub = RealPublisher()
323			rec = Receiver(self)
324			pub.Subscribe(SIMPLE, rec.no_params, ())
325
326			# deleting pub and rec should be enough that the last reference
327			# to rec has been dropped because the connector doesn't keep
328			# references to the publishers and SimplePublisher's __del__
329			# method removes all subscriptions
330			del pub
331			del rec
332			self.check_deletetions()
333
334
335
336			if __name__ == "__main__":
337			unittest.main()
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision