WIP-pyshapelib-bramz/test/test_connector.py

# Copyright (c) 2002 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()


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	329	# Copyright (c) 2002 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 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
235			class TestPublisher(unittest.TestCase, DeletionTestMixin):
236
237			"""Tests for the Publisher class"""
238
239			def setUp(self):
240			DeletionTestMixin.setUp(self)
241
242			def test_issue_simple(self):
243			"""Test Publisher message without parameters"""
244			# Make a publisher and a subscriber and connect the two
245			pub = RealPublisher()
246			rec = Receiver(self)
247			pub.Subscribe(SIMPLE, rec.no_params)
248
249			# Issue a message and check whether the receiver got it
250			pub.simple_action()
251			self.assertEquals(rec.messages, [("no_params",)])
252			rec.reset()
253
254			# disconnect and check that the message doesn't get send anymore
255			pub.Unsubscribe(SIMPLE, rec.no_params)
256			pub.simple_action()
257			self.assertEquals(rec.messages, [])
258
259			# make sure that all references have been deleted
260			del rec
261			self.check_deletetions()
262
263			def test_issue_param(self):
264			"""Test Publisher message with parameters"""
265			pub = RealPublisher()
266			rec = Receiver(self)
267			# Three cases: 1. The parameter supplied by pub.issue_arg, 2.
268			# only the parameter given when connecting, 3. both
269			pub.Subscribe(PARAM, rec.with_params)
270			pub.Subscribe(SIMPLE, rec.with_params, "deliverator")
271			pub.Subscribe(PARAM, rec.with_params, "loglo")
272
273			pub.param_action()
274			pub.simple_action()
275			self.assertEquals(rec.messages, [("params", 42),
276			("params", 42, "loglo"),
277			("params", "deliverator")])
278
279			# make sure that all references have been deleted
280			pub.Destroy()
281			del rec
282			self.check_deletetions()
283
284			def test_cyclic_references(self):
285			"""Test whether Publisher avoids cyclic references"""
286			pub = RealPublisher()
287			rec = Receiver(self)
288			pub.Subscribe(SIMPLE, rec.no_params, ())
289
290			# deleting pub and rec should be enough that the last reference
291			# to rec has been dropped because the connector doesn't keep
292			# references to the publishers and SimplePublisher's __del__
293			# method removes all subscriptions
294			del pub
295			del rec
296			self.check_deletetions()
297
298
299
300			if __name__ == "__main__":
301			unittest.main()
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision