subversion/bindings/swig/python/tests/pool.py

   1 import unittest, weakref, setup_path
   2 import svn.core, svn.client, libsvn.core
   3 from svn.core import *
   4 from libsvn.core import application_pool, GenericSWIGWrapper
   5
   6 # Test case for the new automatic pool management infrastructure
   7
   8 class PoolTestCase(unittest.TestCase):
   9
  10   def assertNotNone(self, value):
  11     """Assert that the specified value is not None"""
  12     return self.assertNotEqual(value, None);
  13
  14   def assertNone(self, value):
  15     """Assert that the specified value is None"""
  16     return self.assertEqual(value, None);
  17
  18   def test_object_struct_members(self):
  19     """Check that object struct members work correctly"""
  20
  21     # Test good object assignment operations
  22     client_ctx = svn.client.svn_client_create_context()
  23     config = svn.core.svn_config_get_config(None)
  24     client_ctx.config = config
  25
  26     # Check that parent pools are set correctly on struct accesses
  27     self.assertEqual(client_ctx.config._parent_pool, config._parent_pool)
  28
  29     # Test bad object assignment operations
  30     def test_bad_assignment(self):
  31       head_revision = svn.core.svn_opt_revision_t()
  32       head_revision.kind = config
  33     self.assertRaises(TypeError, test_bad_assignment)
  34
  35   def test_assert_valid(self):
  36     """Test assert_valid method on proxy objects"""
  37
  38     # Test assert_valid with destroy()
  39     client_ctx = svn.client.svn_client_create_context()
  40     config = svn.core.svn_config_get_config(None)
  41     wrapped_config = GenericSWIGWrapper(config, config._parent_pool)
  42     client_ctx.config = config
  43     config.assert_valid()
  44     wrapped_config.assert_valid()
  45     client_ctx.config.assert_valid()
  46     config._parent_pool.destroy()
  47     self.assertRaises(AssertionError, lambda: config.assert_valid())
  48     self.assertRaises(AssertionError, lambda: wrapped_config.assert_valid())
  49     self.assertRaises(AssertionError, lambda: client_ctx.config)
  50
  51     # Test assert_valid with clear()
  52     client_ctx = svn.client.svn_client_create_context()
  53     config = svn.core.svn_config_get_config(None)
  54     wrapped_config = GenericSWIGWrapper(config, config._parent_pool)
  55     client_ctx.config = config
  56     config.assert_valid()
  57     client_ctx.config.assert_valid()
  58     wrapped_config.assert_valid()
  59     config._parent_pool.clear()
  60     self.assertRaises(AssertionError, lambda: config.assert_valid())
  61     self.assertRaises(AssertionError, lambda: wrapped_config.assert_valid())
  62     self.assertRaises(AssertionError, lambda: client_ctx.config)
  63
  64   def test_integer_struct_members(self):
  65     """Check that integer struct members work correctly"""
  66
  67     # Test good integer assignment operations
  68     rev = svn.core.svn_opt_revision_t()
  69     rev.kind = svn.core.svn_opt_revision_number
  70     rev.value.number = 10
  71     self.assertEqual(rev.kind, svn.core.svn_opt_revision_number)
  72     self.assertEqual(rev.value.number, 10)
  73
  74     # Test bad integer assignment operations
  75     def test_bad_assignment(self):
  76       client_ctx = svn.client.svn_client_create_context()
  77       client_ctx.config = 2
  78     self.assertRaises(TypeError, test_bad_assignment)
  79
  80   def test_pool(self):
  81     # Create pools
  82     parent_pool = Pool()
  83     parent_pool_ref = weakref.ref(parent_pool)
  84     pool = Pool(Pool(parent_pool))
  85     pool = Pool(pool)
  86
  87     # Make sure proper exceptions are raised with incorrect input
  88     self.assertRaises(TypeError, lambda: Pool("abcd"));
  89
  90     # Check that garbage collection is working OK
  91     self.assertNotNone(parent_pool_ref())
  92     top_pool_ref = weakref.ref(parent_pool._parent_pool)
  93     del parent_pool
  94     self.assertNotNone(parent_pool_ref())
  95     self.assertNotNone(top_pool_ref())
  96     pool.clear()
  97     newpool = libsvn.core.svn_pool_create(pool)
  98     libsvn.core.apr_pool_destroy(newpool)
  99     self.assertNotNone(newpool)
 100     pool.clear()
 101     self.assertNotNone(parent_pool_ref())
 102     del pool
 103     self.assertNotNone(parent_pool_ref())
 104     del newpool
 105     self.assertNone(parent_pool_ref())
 106     self.assertNone(top_pool_ref())
 107
 108     # Make sure anonymous pools are destroyed properly
 109     anonymous_pool_ref = weakref.ref(Pool())
 110     self.assertNone(anonymous_pool_ref())
 111
 112   def test_compatibility_layer(self):
 113     # Create a new pool
 114     pool = Pool()
 115     parent_pool_ref = weakref.ref(pool)
 116     pool = svn_pool_create(Pool(pool))
 117     pool_ref = weakref.ref(pool)
 118
 119     # Make sure proper exceptions are raised with incorrect input
 120     self.assertRaises(TypeError, lambda: svn_pool_create("abcd"));
 121
 122     # Test whether pools are destroyed properly
 123     pool = svn_pool_create(pool)
 124     self.assertNotNone(pool_ref())
 125     self.assertNotNone(parent_pool_ref())
 126     del pool
 127     self.assertNone(pool_ref())
 128     self.assertNone(parent_pool_ref())
 129
 130     # Ensure that AssertionErrors are raised when a pool is deleted twice
 131     newpool = Pool()
 132     newpool2 = Pool(newpool)
 133     svn_pool_clear(newpool)
 134     self.assertRaises(AssertionError, lambda: libsvn.core.apr_pool_destroy(newpool2))
 135     self.assertRaises(AssertionError, lambda: svn_pool_destroy(newpool2));
 136     svn_pool_destroy(newpool)
 137     self.assertRaises(AssertionError, lambda: svn_pool_destroy(newpool))
 138
 139     # Try to allocate memory from a destroyed pool
 140     self.assertRaises(AssertionError, lambda: svn_pool_create(newpool))
 141
 142     # Create and destroy a pool
 143     svn_pool_destroy(svn_pool_create())
 144
 145     # Make sure anonymous pools are destroyed properly
 146     anonymous_pool_ref = weakref.ref(svn_pool_create())
 147     self.assertNone(anonymous_pool_ref())
 148
 149     # Try to cause a segfault using apr_terminate
 150     apr_terminate()
 151     apr_initialize()
 152     apr_terminate()
 153     apr_terminate()
 154
 155     # Destroy the application pool
 156     svn_pool_destroy(libsvn.core.application_pool)
 157
 158     # Double check that the application pool has been deleted
 159     self.assertNone(libsvn.core.application_pool)
 160
 161     # Try to allocate memory from the old application pool
 162     self.assertRaises(AssertionError, lambda: svn_pool_create(application_pool));
 163
 164     # Bring the application pool back to life
 165     svn_pool_create()
 166
 167     # Double check that the application pool has been created
 168     self.assertNotNone(libsvn.core.application_pool)
 169
 170     # We can still destroy and create pools at will
 171     svn_pool_destroy(svn_pool_create())
 172
 173 def suite():
 174   return unittest.makeSuite(PoolTestCase, 'test')
 175
 176 if __name__ == '__main__':
 177   runner = unittest.TextTestRunner()
 178   runner.run(suite());