3956162881

Build Type

push

github

Committed by Michael Howitz

Commit Message

Update to deprecation warning free releases.

Run Details

4401 of 7036 branches covered (62.55%)

Branch coverage included in aggregate %.

27161 of 31488 relevant lines covered (86.26%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.76

/src/Testing/ZopeTestCase/testBaseTestCase.py

##############################################################################
#
# Copyright (c) 2005 Zope Foundation and Contributors.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Tests the base.TestCase class

NOTE: This is *not* an example TestCase. Do not
use this file as a blueprint for your own tests!

See testPythonScript.py and testShoppingCart.py for
example test cases. See testSkeleton.py for a quick
way of getting started.
"""
import gc
import unittest

import transaction
from AccessControl import getSecurityManager
from AccessControl.SecurityManagement import newSecurityManager
from Acquisition import aq_base
from Testing.ZopeTestCase import base
from Testing.ZopeTestCase import connections
from Testing.ZopeTestCase import sandbox
from Testing.ZopeTestCase import utils


class HookTest(base.TestCase):

    def setUp(self):
        self._called = []
        base.TestCase.setUp(self)

    def beforeSetUp(self):
        self._called.append('beforeSetUp')
        base.TestCase.beforeSetUp(self)

    def _setup(self):
        self._called.append('_setup')
        base.TestCase._setup(self)

    def afterSetUp(self):
        self._called.append('afterSetUp')
        base.TestCase.afterSetUp(self)

    def beforeTearDown(self):
        self._called.append('beforeTearDown')
        base.TestCase.beforeTearDown(self)

    def beforeClose(self):
        self._called.append('beforeClose')
        base.TestCase.beforeClose(self)

    def afterClear(self):
        self._called.append('afterClear')
        base.TestCase.afterClear(self)

    def assertHooks(self, sequence):
        self.assertEqual(self._called, sequence)


class TestTestCase(HookTest):

    def testSetUp(self):
        self.assertHooks(['beforeSetUp', '_setup', 'afterSetUp'])

    def testTearDown(self):
        self._called = []
        self.tearDown()
        self.assertHooks(['beforeTearDown', 'beforeClose', 'afterClear'])

    def testAppOpensConnection(self):
        self.assertEqual(connections.count(), 1)
        self._app()
        self.assertEqual(connections.count(), 2)

    def testClearCallsCloseHook(self):
        self._called = []
        self._clear(1)
        self.assertHooks(['beforeClose', 'afterClear'])

    def testClearSkipsCloseHook(self):
        self._called = []
        self._clear()
        self.assertHooks(['afterClear'])

    def testClearAbortsTransaction(self):
        self.assertEqual(len(self.getObjectsInTransaction()), 0)
        self.app.foo = 1
        self.assertEqual(len(self.getObjectsInTransaction()), 1)
        self._clear()
        self.assertEqual(len(self.getObjectsInTransaction()), 0)

    def testClearClosesConnection(self):
        self.assertEqual(connections.count(), 1)
        self._clear()
        self.assertEqual(connections.count(), 0)

    def testClearClosesAllConnections(self):
        self._app()
        self.assertEqual(connections.count(), 2)
        self._clear()
        self.assertEqual(connections.count(), 0)

    def testClearLogsOut(self):
        uf = self.app.acl_users
        uf.userFolderAddUser('user_1', '', [], [])
        newSecurityManager(None, uf.getUserById('user_1').__of__(uf))
        self.assertEqual(
            getSecurityManager().getUser().getUserName(), 'user_1')
        self._clear()
        self.assertEqual(
            getSecurityManager().getUser().getUserName(), 'Anonymous User')

    def testClearSurvivesDoubleCall(self):
        self._called = []
        self._clear()
        self._clear()
        self.assertHooks(['afterClear', 'afterClear'])

    def testClearSurvivesClosedConnection(self):
        self._called = []
        self._close()
        self._clear()
        self.assertHooks(['afterClear'])

    def testClearSurvivesBrokenApp(self):
        self._called = []
        self.app = None
        self._clear()
        self.assertHooks(['afterClear'])

    def testClearSurvivesMissingApp(self):
        self._called = []
        delattr(self, 'app')
        self._clear()
        self.assertHooks(['afterClear'])

    def testClearSurvivesMissingRequest(self):
        self._called = []
        self.app = aq_base(self.app)
        self._clear()
        self.assertHooks(['afterClear'])

    def testCloseAbortsTransaction(self):
        self.assertEqual(len(self.getObjectsInTransaction()), 0)
        self.app.foo = 1
        self.assertEqual(len(self.getObjectsInTransaction()), 1)
        self._close()
        self.assertEqual(len(self.getObjectsInTransaction()), 0)

    def testCloseClosesConnection(self):
        self.assertEqual(connections.count(), 1)
        self._close()
        self.assertEqual(connections.count(), 0)

    def testCloseClosesAllConnections(self):
        self._app()
        self.assertEqual(connections.count(), 2)
        self._close()
        self.assertEqual(connections.count(), 0)

    def testLogoutLogsOut(self):
        uf = self.app.acl_users
        uf.userFolderAddUser('user_1', '', [], [])
        newSecurityManager(None, uf.getUserById('user_1').__of__(uf))
        self.assertEqual(
            getSecurityManager().getUser().getUserName(), 'user_1')
        self.logout()
        self.assertEqual(
            getSecurityManager().getUser().getUserName(), 'Anonymous User')

    def getObjectsInTransaction(self):
        # Lets us spy into the transaction
        t = transaction.get()
        if hasattr(t, '_objects'):      # Zope < 2.8
            return t._objects
        elif hasattr(t, '_resources'):  # Zope >= 2.8
            return t._resources
        else:
            raise Exception('Unknown version')


class TestSetUpRaises(HookTest):

    class Error(Exception):
        pass

    def setUp(self):
        try:
            HookTest.setUp(self)
        except self.Error:
            self.assertHooks(['beforeSetUp', '_setup', 'afterClear'])
            # Connection has been closed
            self.assertEqual(connections.count(), 0)

    def _setup(self):
        HookTest._setup(self)
        raise self.Error

    def testTrigger(self):
        pass


class TestTearDownRaises(HookTest):

    class Error(Exception):
        pass

    def tearDown(self):
        self._called = []
        try:
            HookTest.tearDown(self)
        except self.Error:
            self.assertHooks(['beforeTearDown', 'beforeClose', 'afterClear'])
            # Connection has been closed
            self.assertEqual(connections.count(), 0)

    def beforeClose(self):
        HookTest.beforeClose(self)
        raise self.Error

    def testTrigger(self):
        pass


class TestConnectionRegistry(base.TestCase):
    '''Test the registry with Connection-like objects'''

    class Conn:
        _closed = 0

        def close(self):
            self._closed = 1

        def closed(self):
            return self._closed

    Klass = Conn

    def afterSetUp(self):
        self.reg = connections.ConnectionRegistry()
        self.conns = [self.Klass(), self.Klass(), self.Klass()]
        for conn in self.conns:
            self.reg.register(conn)

    def testRegister(self):
        # Should be able to register connections
        assert len(self.reg) == 3
        assert self.reg.count() == 3

    def testCloseConnection(self):
        # Should be able to close a single registered connection
        assert len(self.reg) == 3
        self.reg.close(self.conns[0])
        assert len(self.reg) == 2
        assert self.conns[0].closed() == 1
        assert self.conns[1].closed() == 0
        assert self.conns[2].closed() == 0

    def testCloseSeveralConnections(self):
        # Should be able to close all registered connections one-by-one
        assert len(self.reg) == 3
        self.reg.close(self.conns[0])
        assert len(self.reg) == 2
        assert self.conns[0].closed() == 1
        assert self.conns[1].closed() == 0
        assert self.conns[2].closed() == 0
        self.reg.close(self.conns[2])
        assert len(self.reg) == 1
        assert self.conns[0].closed() == 1
        assert self.conns[1].closed() == 0
        assert self.conns[2].closed() == 1
        self.reg.close(self.conns[1])
        assert len(self.reg) == 0
        assert self.conns[0].closed() == 1
        assert self.conns[1].closed() == 1
        assert self.conns[2].closed() == 1

    def testCloseForeignConnection(self):
        # Should be able to close a connection that has not been registered
        assert len(self.reg) == 3
        conn = self.Klass()
        self.reg.close(conn)
        assert len(self.reg) == 3
        assert self.conns[0].closed() == 0
        assert self.conns[1].closed() == 0
        assert self.conns[2].closed() == 0
        assert conn.closed() == 1

    def testCloseAllConnections(self):
        # Should be able to close all registered connections at once
        assert len(self.reg) == 3
        self.reg.closeAll()
        assert len(self.reg) == 0
        assert self.conns[0].closed() == 1
        assert self.conns[1].closed() == 1
        assert self.conns[2].closed() == 1

    def testContains(self):
        # Should be able to check if a connection is registered
        assert len(self.reg) == 3
        assert self.reg.contains(self.conns[0])
        assert self.reg.contains(self.conns[1])
        assert self.reg.contains(self.conns[2])


class TestApplicationRegistry(TestConnectionRegistry):
    '''Test the registry with Application-like objects'''

    class App:
        class Conn:
            _closed = 0

            def close(self):
                self._closed = 1

            def closed(self):
                return self._closed

        def __init__(self):
            self.REQUEST = self.Conn()
            self._p_jar = self.Conn()

        def closed(self):
            if self.REQUEST.closed() and self._p_jar.closed():
                return 1
            return 0

    Klass = App


class TestListConverter(base.TestCase):
    '''Test utils.makelist'''

    def testList0(self):
        self.assertEqual(utils.makelist([]), [])

    def testList1(self):
        self.assertEqual(utils.makelist(['foo']), ['foo'])

    def testList2(self):
        self.assertEqual(utils.makelist(['foo', 'bar']), ['foo', 'bar'])

    def testTuple0(self):
        self.assertEqual(utils.makelist(()), [])

    def testTuple1(self):
        self.assertEqual(utils.makelist(('foo',)), ['foo'])

    def testTuple2(self):
        self.assertEqual(utils.makelist(('foo', 'bar')), ['foo', 'bar'])

    def testString0(self):
        self.assertEqual(utils.makelist(''), [])

    def testString1(self):
        self.assertEqual(utils.makelist('foo'), ['foo'])

    def testString2(self):
        self.assertEqual(utils.makelist('foo, bar'), ['foo, bar'])

    def testInteger(self):
        self.assertRaises(ValueError, utils.makelist, 0)

    def testObject(self):
        class Dummy:
            pass
        self.assertRaises(ValueError, utils.makelist, Dummy())


class TestRequestVariables(base.TestCase):
    '''Makes sure the REQUEST contains required variables'''

    def testRequestVariables(self):
        request = self.app.REQUEST
        self.assertNotEqual(request.get('SERVER_NAME', ''), '')
        self.assertNotEqual(request.get('SERVER_PORT', ''), '')
        self.assertNotEqual(request.get('REQUEST_METHOD', ''), '')
        self.assertNotEqual(request.get('URL', ''), '')
        self.assertNotEqual(request.get('SERVER_URL', ''), '')
        self.assertNotEqual(request.get('URL0', ''), '')
        self.assertNotEqual(request.get('URL1', ''), '')
        self.assertNotEqual(request.get('BASE0', ''), '')
        self.assertNotEqual(request.get('BASE1', ''), '')
        self.assertNotEqual(request.get('BASE2', ''), '')
        self.assertNotEqual(request.get('ACTUAL_URL', ''), '')


_sentinel1 = []
_sentinel2 = []
_sentinel3 = []


class TestRequestGarbage1(base.TestCase):
    '''Make sure base.app + base.close does not leak REQUEST._held'''

    class Held:
        def __del__(self):
            _sentinel1.append('__del__')

    def afterSetUp(self):
        _sentinel1[:] = []
        self.anApp = base.app()
        self.anApp.REQUEST._hold(self.Held())

    def testBaseCloseClosesRequest(self):
        base.close(self.anApp)
        gc.collect()
        self.assertEqual(_sentinel1, ['__del__'])


class TestRequestGarbage2(base.TestCase):
    '''Make sure self._app + self._clear does not leak REQUEST._held'''

    class Held:
        def __del__(self):
            _sentinel2.append('__del__')

    def afterSetUp(self):
        _sentinel2[:] = []
        self.app.REQUEST._hold(self.Held())

    def testClearClosesRequest(self):
        self._clear()
        gc.collect()
        self.assertEqual(_sentinel2, ['__del__'])


class TestRequestGarbage3(sandbox.Sandboxed, base.TestCase):
    '''Make sure self._app + self._clear does not leak REQUEST._held'''

    class Held:
        def __del__(self):
            _sentinel3.append('__del__')

    def afterSetUp(self):
        _sentinel3[:] = []
        self.app.REQUEST._hold(self.Held())

    def testClearClosesRequest(self):
        self._clear()
        gc.collect()
        self.assertEqual(_sentinel3, ['__del__'])


def test_suite():
    return unittest.TestSuite((
        unittest.defaultTestLoader.loadTestsFromTestCase(TestTestCase),
        unittest.defaultTestLoader.loadTestsFromTestCase(TestSetUpRaises),
        unittest.defaultTestLoader.loadTestsFromTestCase(TestTearDownRaises),
        unittest.defaultTestLoader.loadTestsFromTestCase(
            TestConnectionRegistry),
        unittest.defaultTestLoader.loadTestsFromTestCase(
            TestApplicationRegistry),
        unittest.defaultTestLoader.loadTestsFromTestCase(TestListConverter),
        unittest.defaultTestLoader.loadTestsFromTestCase(TestRequestVariables),
        unittest.defaultTestLoader.loadTestsFromTestCase(TestRequestGarbage1),
        unittest.defaultTestLoader.loadTestsFromTestCase(TestRequestGarbage2),
        unittest.defaultTestLoader.loadTestsFromTestCase(TestRequestGarbage3),
    ))

1	##############################################################################
2	#
3	# Copyright (c) 2005 Zope Foundation and Contributors.
4	#
5	# This software is subject to the provisions of the Zope Public License,
6	# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
7	# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
8	# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
9	# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
10	# FOR A PARTICULAR PURPOSE.
11	#
12	##############################################################################
13	"""Tests the base.TestCase class	1✔
14
15	NOTE: This is not an example TestCase. Do not
16	use this file as a blueprint for your own tests!
17
18	See testPythonScript.py and testShoppingCart.py for
19	example test cases. See testSkeleton.py for a quick
20	way of getting started.
21	"""
22	import gc	1✔
23	import unittest	1✔
24
25	import transaction	1✔
26	from AccessControl import getSecurityManager	1✔
27	from AccessControl.SecurityManagement import newSecurityManager	1✔
28	from Acquisition import aq_base	1✔
29	from Testing.ZopeTestCase import base	1✔
30	from Testing.ZopeTestCase import connections	1✔
31	from Testing.ZopeTestCase import sandbox	1✔
32	from Testing.ZopeTestCase import utils	1✔
33
34
35	class HookTest(base.TestCase):	1✔
36
37	def setUp(self):	1✔
38	self._called = []	1✔
39	base.TestCase.setUp(self)	1✔
40
41	def beforeSetUp(self):	1✔
42	self._called.append('beforeSetUp')	1✔
43	base.TestCase.beforeSetUp(self)	1✔
44
45	def _setup(self):	1✔
46	self._called.append('_setup')	1✔
47	base.TestCase._setup(self)	1✔
48
49	def afterSetUp(self):	1✔
50	self._called.append('afterSetUp')	1✔
51	base.TestCase.afterSetUp(self)	1✔
52
53	def beforeTearDown(self):	1✔
54	self._called.append('beforeTearDown')	1✔
55	base.TestCase.beforeTearDown(self)	1✔
56
57	def beforeClose(self):	1✔
58	self._called.append('beforeClose')	1✔
59	base.TestCase.beforeClose(self)	1✔
60
61	def afterClear(self):	1✔
62	self._called.append('afterClear')	1✔
63	base.TestCase.afterClear(self)	1✔
64
65	def assertHooks(self, sequence):	1✔
66	self.assertEqual(self._called, sequence)	1✔
67
68
69	class TestTestCase(HookTest):	1✔
70
71	def testSetUp(self):	1✔
72	self.assertHooks(['beforeSetUp', '_setup', 'afterSetUp'])	1✔
73
74	def testTearDown(self):	1✔
75	self._called = []	1✔
76	self.tearDown()	1✔
77	self.assertHooks(['beforeTearDown', 'beforeClose', 'afterClear'])	1✔
78
79	def testAppOpensConnection(self):	1✔
80	self.assertEqual(connections.count(), 1)	1✔
81	self._app()	1✔
82	self.assertEqual(connections.count(), 2)	1✔
83
84	def testClearCallsCloseHook(self):	1✔
85	self._called = []	1✔
86	self._clear(1)	1✔
87	self.assertHooks(['beforeClose', 'afterClear'])	1✔
88
89	def testClearSkipsCloseHook(self):	1✔
90	self._called = []	1✔
91	self._clear()	1✔
92	self.assertHooks(['afterClear'])	1✔
93
94	def testClearAbortsTransaction(self):	1✔
95	self.assertEqual(len(self.getObjectsInTransaction()), 0)	1✔
96	self.app.foo = 1	1✔
97	self.assertEqual(len(self.getObjectsInTransaction()), 1)	1✔
98	self._clear()	1✔
99	self.assertEqual(len(self.getObjectsInTransaction()), 0)	1✔
100
101	def testClearClosesConnection(self):	1✔
102	self.assertEqual(connections.count(), 1)	1✔
103	self._clear()	1✔
104	self.assertEqual(connections.count(), 0)	1✔
105
106	def testClearClosesAllConnections(self):	1✔
107	self._app()	1✔
108	self.assertEqual(connections.count(), 2)	1✔
109	self._clear()	1✔
110	self.assertEqual(connections.count(), 0)	1✔
111
112	def testClearLogsOut(self):	1✔
113	uf = self.app.acl_users	1✔
114	uf.userFolderAddUser('user_1', '', [], [])	1✔
115	newSecurityManager(None, uf.getUserById('user_1').__of__(uf))	1✔
116	self.assertEqual(	1✔
117	getSecurityManager().getUser().getUserName(), 'user_1')
118	self._clear()	1✔
119	self.assertEqual(	1✔
120	getSecurityManager().getUser().getUserName(), 'Anonymous User')
121
122	def testClearSurvivesDoubleCall(self):	1✔
123	self._called = []	1✔
124	self._clear()	1✔
125	self._clear()	1✔
126	self.assertHooks(['afterClear', 'afterClear'])	1✔
127
128	def testClearSurvivesClosedConnection(self):	1✔
129	self._called = []	1✔
130	self._close()	1✔
131	self._clear()	1✔
132	self.assertHooks(['afterClear'])	1✔
133
134	def testClearSurvivesBrokenApp(self):	1✔
135	self._called = []	1✔
136	self.app = None	1✔
137	self._clear()	1✔
138	self.assertHooks(['afterClear'])	1✔
139
140	def testClearSurvivesMissingApp(self):	1✔
141	self._called = []	1✔
142	delattr(self, 'app')	1✔
143	self._clear()	1✔
144	self.assertHooks(['afterClear'])	1✔
145
146	def testClearSurvivesMissingRequest(self):	1✔
147	self._called = []	1✔
148	self.app = aq_base(self.app)	1✔
149	self._clear()	1✔
150	self.assertHooks(['afterClear'])	1✔
151
152	def testCloseAbortsTransaction(self):	1✔
153	self.assertEqual(len(self.getObjectsInTransaction()), 0)	1✔
154	self.app.foo = 1	1✔
155	self.assertEqual(len(self.getObjectsInTransaction()), 1)	1✔
156	self._close()	1✔
157	self.assertEqual(len(self.getObjectsInTransaction()), 0)	1✔
158
159	def testCloseClosesConnection(self):	1✔
160	self.assertEqual(connections.count(), 1)	1✔
161	self._close()	1✔
162	self.assertEqual(connections.count(), 0)	1✔
163
164	def testCloseClosesAllConnections(self):	1✔
165	self._app()	1✔
166	self.assertEqual(connections.count(), 2)	1✔
167	self._close()	1✔
168	self.assertEqual(connections.count(), 0)	1✔
169
170	def testLogoutLogsOut(self):	1✔
171	uf = self.app.acl_users	1✔
172	uf.userFolderAddUser('user_1', '', [], [])	1✔
173	newSecurityManager(None, uf.getUserById('user_1').__of__(uf))	1✔
174	self.assertEqual(	1✔
175	getSecurityManager().getUser().getUserName(), 'user_1')
176	self.logout()	1✔
177	self.assertEqual(	1✔
178	getSecurityManager().getUser().getUserName(), 'Anonymous User')
179
180	def getObjectsInTransaction(self):	1✔
181	# Lets us spy into the transaction
182	t = transaction.get()	1✔
183	if hasattr(t, '_objects'): # Zope < 2.8	1!
184	return t._objects	×
185	elif hasattr(t, '_resources'): # Zope >= 2.8	1!
186	return t._resources	1✔
187	else:
188	raise Exception('Unknown version')	×
189
190
191	class TestSetUpRaises(HookTest):	1✔
192
193	class Error(Exception):	1✔
194	pass	1✔
195
196	def setUp(self):	1✔
197	try:	1✔
198	HookTest.setUp(self)	1✔
199	except self.Error:	1✔
200	self.assertHooks(['beforeSetUp', '_setup', 'afterClear'])	1✔
201	# Connection has been closed
202	self.assertEqual(connections.count(), 0)	1✔
203
204	def _setup(self):	1✔
205	HookTest._setup(self)	1✔
206	raise self.Error	1✔
207
208	def testTrigger(self):	1✔
209	pass	1✔
210
211
212	class TestTearDownRaises(HookTest):	1✔
213
214	class Error(Exception):	1✔
215	pass	1✔
216
217	def tearDown(self):	1✔
218	self._called = []	1✔
219	try:	1✔
220	HookTest.tearDown(self)	1✔
221	except self.Error:	1✔
222	self.assertHooks(['beforeTearDown', 'beforeClose', 'afterClear'])	1✔
223	# Connection has been closed
224	self.assertEqual(connections.count(), 0)	1✔
225
226	def beforeClose(self):	1✔
227	HookTest.beforeClose(self)	1✔
228	raise self.Error	1✔
229
230	def testTrigger(self):	1✔
231	pass	1✔
232
233
234	class TestConnectionRegistry(base.TestCase):	1✔
235	'''Test the registry with Connection-like objects'''
236
237	class Conn:	1✔
238	_closed = 0	1✔
239
240	def close(self):	1✔
241	self._closed = 1	1✔
242
243	def closed(self):	1✔
244	return self._closed	1✔
245
246	Klass = Conn	1✔
247
248	def afterSetUp(self):	1✔
249	self.reg = connections.ConnectionRegistry()	1✔
250	self.conns = [self.Klass(), self.Klass(), self.Klass()]	1✔
251	for conn in self.conns:	1✔
252	self.reg.register(conn)	1✔
253
254	def testRegister(self):	1✔
255	# Should be able to register connections
256	assert len(self.reg) == 3	1✔
257	assert self.reg.count() == 3	1✔
258
259	def testCloseConnection(self):	1✔
260	# Should be able to close a single registered connection
261	assert len(self.reg) == 3	1✔
262	self.reg.close(self.conns[0])	1✔
263	assert len(self.reg) == 2	1✔
264	assert self.conns[0].closed() == 1	1✔
265	assert self.conns[1].closed() == 0	1✔
266	assert self.conns[2].closed() == 0	1✔
267
268	def testCloseSeveralConnections(self):	1✔
269	# Should be able to close all registered connections one-by-one
270	assert len(self.reg) == 3	1✔
271	self.reg.close(self.conns[0])	1✔
272	assert len(self.reg) == 2	1✔
273	assert self.conns[0].closed() == 1	1✔
274	assert self.conns[1].closed() == 0	1✔
275	assert self.conns[2].closed() == 0	1✔
276	self.reg.close(self.conns[2])	1✔
277	assert len(self.reg) == 1	1✔
278	assert self.conns[0].closed() == 1	1✔
279	assert self.conns[1].closed() == 0	1✔
280	assert self.conns[2].closed() == 1	1✔
281	self.reg.close(self.conns[1])	1✔
282	assert len(self.reg) == 0	1✔
283	assert self.conns[0].closed() == 1	1✔
284	assert self.conns[1].closed() == 1	1✔
285	assert self.conns[2].closed() == 1	1✔
286
287	def testCloseForeignConnection(self):	1✔
288	# Should be able to close a connection that has not been registered
289	assert len(self.reg) == 3	1✔
290	conn = self.Klass()	1✔
291	self.reg.close(conn)	1✔
292	assert len(self.reg) == 3	1✔
293	assert self.conns[0].closed() == 0	1✔
294	assert self.conns[1].closed() == 0	1✔
295	assert self.conns[2].closed() == 0	1✔
296	assert conn.closed() == 1	1✔
297
298	def testCloseAllConnections(self):	1✔
299	# Should be able to close all registered connections at once
300	assert len(self.reg) == 3	1✔
301	self.reg.closeAll()	1✔
302	assert len(self.reg) == 0	1✔
303	assert self.conns[0].closed() == 1	1✔
304	assert self.conns[1].closed() == 1	1✔
305	assert self.conns[2].closed() == 1	1✔
306
307	def testContains(self):	1✔
308	# Should be able to check if a connection is registered
309	assert len(self.reg) == 3	1✔
310	assert self.reg.contains(self.conns[0])	1✔
311	assert self.reg.contains(self.conns[1])	1✔
312	assert self.reg.contains(self.conns[2])	1✔
313
314
315	class TestApplicationRegistry(TestConnectionRegistry):	1✔
316	'''Test the registry with Application-like objects'''
317
318	class App:	1✔
319	class Conn:	1✔
320	_closed = 0	1✔
321
322	def close(self):	1✔
323	self._closed = 1	1✔
324
325	def closed(self):	1✔
326	return self._closed	1✔
327
328	def __init__(self):	1✔
329	self.REQUEST = self.Conn()	1✔
330	self._p_jar = self.Conn()	1✔
331
332	def closed(self):	1✔
333	if self.REQUEST.closed() and self._p_jar.closed():	1✔
334	return 1	1✔
335	return 0	1✔
336
337	Klass = App	1✔
338
339
340	class TestListConverter(base.TestCase):	1✔
341	'''Test utils.makelist'''
342
343	def testList0(self):	1✔
344	self.assertEqual(utils.makelist([]), [])	1✔
345
346	def testList1(self):	1✔
347	self.assertEqual(utils.makelist(['foo']), ['foo'])	1✔
348
349	def testList2(self):	1✔
350	self.assertEqual(utils.makelist(['foo', 'bar']), ['foo', 'bar'])	1✔
351
352	def testTuple0(self):	1✔
353	self.assertEqual(utils.makelist(()), [])	1✔
354
355	def testTuple1(self):	1✔
356	self.assertEqual(utils.makelist(('foo',)), ['foo'])	1✔
357
358	def testTuple2(self):	1✔
359	self.assertEqual(utils.makelist(('foo', 'bar')), ['foo', 'bar'])	1✔
360
361	def testString0(self):	1✔
362	self.assertEqual(utils.makelist(''), [])	1✔
363
364	def testString1(self):	1✔
365	self.assertEqual(utils.makelist('foo'), ['foo'])	1✔
366
367	def testString2(self):	1✔
368	self.assertEqual(utils.makelist('foo, bar'), ['foo, bar'])	1✔
369
370	def testInteger(self):	1✔
371	self.assertRaises(ValueError, utils.makelist, 0)	1✔
372
373	def testObject(self):	1✔
374	class Dummy:	1✔
375	pass	1✔
376	self.assertRaises(ValueError, utils.makelist, Dummy())	1✔
377
378
379	class TestRequestVariables(base.TestCase):	1✔
380	'''Makes sure the REQUEST contains required variables'''
381
382	def testRequestVariables(self):	1✔
383	request = self.app.REQUEST	1✔
384	self.assertNotEqual(request.get('SERVER_NAME', ''), '')	1✔
385	self.assertNotEqual(request.get('SERVER_PORT', ''), '')	1✔
386	self.assertNotEqual(request.get('REQUEST_METHOD', ''), '')	1✔
387	self.assertNotEqual(request.get('URL', ''), '')	1✔
388	self.assertNotEqual(request.get('SERVER_URL', ''), '')	1✔
389	self.assertNotEqual(request.get('URL0', ''), '')	1✔
390	self.assertNotEqual(request.get('URL1', ''), '')	1✔
391	self.assertNotEqual(request.get('BASE0', ''), '')	1✔
392	self.assertNotEqual(request.get('BASE1', ''), '')	1✔
393	self.assertNotEqual(request.get('BASE2', ''), '')	1✔
394	self.assertNotEqual(request.get('ACTUAL_URL', ''), '')	1✔
395
396
397	_sentinel1 = []	1✔
398	_sentinel2 = []	1✔
399	_sentinel3 = []	1✔
400
401
402	class TestRequestGarbage1(base.TestCase):	1✔
403	'''Make sure base.app + base.close does not leak REQUEST._held'''
404
405	class Held:	1✔
406	def __del__(self):	1✔
407	_sentinel1.append('__del__')	1✔
408
409	def afterSetUp(self):	1✔
410	_sentinel1[:] = []	1✔
411	self.anApp = base.app()	1✔
412	self.anApp.REQUEST._hold(self.Held())	1✔
413
414	def testBaseCloseClosesRequest(self):	1✔
415	base.close(self.anApp)	1✔
416	gc.collect()	1✔
417	self.assertEqual(_sentinel1, ['__del__'])	1✔
418
419
420	class TestRequestGarbage2(base.TestCase):	1✔
421	'''Make sure self._app + self._clear does not leak REQUEST._held'''
422
423	class Held:	1✔
424	def __del__(self):	1✔
425	_sentinel2.append('__del__')	1✔
426
427	def afterSetUp(self):	1✔
428	_sentinel2[:] = []	1✔
429	self.app.REQUEST._hold(self.Held())	1✔
430
431	def testClearClosesRequest(self):	1✔
432	self._clear()	1✔
433	gc.collect()	1✔
434	self.assertEqual(_sentinel2, ['__del__'])	1✔
435
436
437	class TestRequestGarbage3(sandbox.Sandboxed, base.TestCase):	1✔
438	'''Make sure self._app + self._clear does not leak REQUEST._held'''
439
440	class Held:	1✔
441	def __del__(self):	1✔
442	_sentinel3.append('__del__')	1✔
443
444	def afterSetUp(self):	1✔
445	_sentinel3[:] = []	1✔
446	self.app.REQUEST._hold(self.Held())	1✔
447
448	def testClearClosesRequest(self):	1✔
449	self._clear()	1✔
450	gc.collect()	1✔
451	self.assertEqual(_sentinel3, ['__del__'])	1✔
452
453
454	def test_suite():	1✔
455	return unittest.TestSuite((	1✔
456	unittest.defaultTestLoader.loadTestsFromTestCase(TestTestCase),
457	unittest.defaultTestLoader.loadTestsFromTestCase(TestSetUpRaises),
458	unittest.defaultTestLoader.loadTestsFromTestCase(TestTearDownRaises),
459	unittest.defaultTestLoader.loadTestsFromTestCase(
460	TestConnectionRegistry),
461	unittest.defaultTestLoader.loadTestsFromTestCase(
462	TestApplicationRegistry),
463	unittest.defaultTestLoader.loadTestsFromTestCase(TestListConverter),
464	unittest.defaultTestLoader.loadTestsFromTestCase(TestRequestVariables),
465	unittest.defaultTestLoader.loadTestsFromTestCase(TestRequestGarbage1),
466	unittest.defaultTestLoader.loadTestsFromTestCase(TestRequestGarbage2),
467	unittest.defaultTestLoader.loadTestsFromTestCase(TestRequestGarbage3),
468	))

zopefoundation / Zope / 3956162881

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous