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

   1 import unittest, os, weakref, tempfile, types, setup_path
   2
   3 from svn import core, repos, fs, delta, client, wc
   4 from svn.core import SubversionException
   5
   6 from trac.versioncontrol.tests.svn_fs import SubversionRepositoryTestSetup, \
   7   REPOS_PATH, REPOS_URL
   8 from urlparse import urljoin
   9
  10 class SubversionClientTestCase(unittest.TestCase):
  11   """Test cases for the basic SWIG Subversion client layer"""
  12
  13   def log_message_func(self, items, pool):
  14     """ Simple log message provider for unit tests. """
  15     self.log_message_func_calls += 1
  16     return "Test log message"
  17
  18   def log_receiver(self, changed_paths, revision, author, date, message, pool):
  19     """ Function to recieve log messages retrieved by client.log3(). """
  20     self.log_message = message
  21     self.change_author = author
  22     self.changed_paths = changed_paths
  23
  24   def setUp(self):
  25     """Set up authentication and client context"""
  26     self.client_ctx = client.svn_client_create_context()
  27     self.assertEquals(self.client_ctx.log_msg_baton2, None)
  28     self.assertEquals(self.client_ctx.log_msg_func2, None)
  29     self.client_ctx.log_msg_func2 = client.svn_swig_py_get_commit_log_func
  30     self.client_ctx.log_msg_baton2 = self.log_message_func
  31     self.log_message_func_calls = 0
  32     self.log_message = None
  33     self.changed_paths = None
  34     self.change_author = None
  35
  36     providers = [
  37        client.svn_client_get_simple_provider(),
  38        client.svn_client_get_username_provider(),
  39     ]
  40
  41     self.client_ctx.auth_baton = core.svn_auth_open(providers)
  42
  43   def testBatonPlay(self):
  44     """Test playing with C batons"""
  45     self.client_ctx.log_msg_baton2 = self.client_ctx.auth_baton
  46     self.assertEquals(str(self.client_ctx.log_msg_baton2),
  47                       str(self.client_ctx.auth_baton))
  48     self.client_ctx.log_msg_baton2 = self.client_ctx.log_msg_baton2
  49     self.assertEquals(str(self.client_ctx.log_msg_baton2),
  50                       str(self.client_ctx.auth_baton))
  51     self.client_ctx.log_msg_baton2 = None
  52     self.assertEquals(self.client_ctx.log_msg_baton2, None)
  53
  54     # External objects should retain their current parent pool
  55     self.assertNotEquals(self.client_ctx._parent_pool,
  56                          self.client_ctx.auth_baton._parent_pool)
  57
  58     # notify_func2 and notify_baton2 were generated by
  59     # svn_client_create_context, so they should have
  60     # the same pool as the context
  61     self.assertEquals(self.client_ctx._parent_pool,
  62                       self.client_ctx.notify_func2._parent_pool)
  63     self.assertEquals(self.client_ctx._parent_pool,
  64                       self.client_ctx.notify_baton2._parent_pool)
  65
  66   def testMethodCalls(self):
  67     """Test direct method calls to callbacks"""
  68
  69     # Directly invoking the msg_baton should work
  70     self.client_ctx.log_msg_baton2(None, None)
  71     b = self.client_ctx.log_msg_baton2
  72     b(None, None)
  73     self.assertEqual(self.log_message_func_calls, 2)
  74
  75     # You can also invoke the log_msg_func2. It'd be
  76     # nice if we could get log_msg_func2 function
  77     # to invoke the baton function, but, in order to do that,
  78     # we'd need to supply a value for the first parameter.
  79     self.client_ctx.log_msg_func2(None, self.client_ctx.log_msg_baton2)
  80
  81   def info_receiver(self, path, info, pool):
  82     """Squirrel away the output from 'svn info' so that the unit tests
  83        can get at them."""
  84     self.path = path
  85     self.info = info
  86
  87   def test_client_ctx_baton_lifetime(self):
  88     pool = core.Pool()
  89     temp_client_ctx = client.svn_client_create_context(pool)
  90
  91     # We keep track of these objects in separate variables here
  92     # because you can't get a PyObject back out of a PY_AS_VOID field
  93     test_object1 = lambda *args: "message 1"
  94     test_object2 = lambda *args: "message 2"
  95
  96     # Verify that the refcount of a Python object is incremented when
  97     # you insert it into a PY_AS_VOID field.
  98     temp_client_ctx.log_msg_baton2 = test_object1
  99     test_object1 = weakref.ref(test_object1)
 100     self.assertNotEqual(test_object1(), None)
 101
 102     # Verify that the refcount of the previous Python object is decremented
 103     # when a PY_AS_VOID field is replaced.
 104     temp_client_ctx.log_msg_baton2 = test_object2
 105     self.assertEqual(test_object1(), None)
 106
 107     # Verify that the reference count of the new Python object (which
 108     # replaced test_object1) was incremented.
 109     test_object2 = weakref.ref(test_object2)
 110     self.assertNotEqual(test_object2(), None)
 111
 112     # Verify that the reference count of test_object2 is decremented when
 113     # test_client_ctx is destroyed.
 114     temp_client_ctx = None
 115     self.assertEqual(test_object2(), None)
 116
 117   def test_checkout(self):
 118     """Test svn_client_checkout2."""
 119
 120     rev = core.svn_opt_revision_t()
 121     rev.kind = core.svn_opt_revision_head
 122
 123     path = tempfile.mktemp('-checkout')
 124
 125     self.assertRaises(ValueError, client.checkout2,
 126                       REPOS_URL, path, None, None, True, True,
 127                       self.client_ctx)
 128
 129     client.checkout2(REPOS_URL, path, rev, rev, True, True,
 130             self.client_ctx)
 131
 132   def test_info(self):
 133     """Test svn_client_info on an empty repository"""
 134
 135     # Run info
 136     revt = core.svn_opt_revision_t()
 137     revt.kind = core.svn_opt_revision_head
 138     client.info(REPOS_URL, revt, revt, self.info_receiver,
 139                 False, self.client_ctx)
 140
 141     # Check output from running info. This also serves to verify that
 142     # the internal 'info' object is still valid
 143     self.assertEqual(self.path, os.path.basename(REPOS_PATH))
 144     self.info.assert_valid()
 145     self.assertEqual(self.info.URL, REPOS_URL)
 146     self.assertEqual(self.info.repos_root_URL, REPOS_URL)
 147
 148   def test_mkdir_url(self):
 149     """Test svn_client_mkdir2 on a file:// URL"""
 150     dir = urljoin(REPOS_URL+"/", "dir1")
 151
 152     commit_info = client.mkdir2((dir,), self.client_ctx)
 153     self.assertEqual(commit_info.revision, 13)
 154     self.assertEqual(self.log_message_func_calls, 1)
 155
 156   def test_log3_url(self):
 157     """Test svn_client_log3 on a file:// URL"""
 158     dir = urljoin(REPOS_URL+"/", "trunk/dir1")
 159
 160     start = core.svn_opt_revision_t()
 161     end = core.svn_opt_revision_t()
 162     core.svn_opt_parse_revision(start, end, "4:0")
 163     client.log3((dir,), start, start, end, 1, True, False, self.log_receiver,
 164         self.client_ctx)
 165     self.assertEqual(self.change_author, "john")
 166     self.assertEqual(self.log_message, "More directories.")
 167     self.assertEqual(len(self.changed_paths), 3)
 168     for dir in ('/trunk/dir1', '/trunk/dir2', '/trunk/dir3'):
 169       self.assert_(self.changed_paths.has_key(dir))
 170       self.assertEqual(self.changed_paths[dir].action, 'A')
 171
 172   def test_uuid_from_url(self):
 173     """Test svn_client_uuid_from_url on a file:// URL"""
 174     self.assert_(isinstance(
 175                  client.uuid_from_url(REPOS_URL, self.client_ctx),
 176                  types.StringTypes))
 177
 178   def test_url_from_path(self):
 179     """Test svn_client_url_from_path for a file:// URL"""
 180     self.assertEquals(client.url_from_path(REPOS_URL), REPOS_URL)
 181
 182     rev = core.svn_opt_revision_t()
 183     rev.kind = core.svn_opt_revision_head
 184
 185     path = tempfile.mktemp('-url_from_path')
 186
 187     client.checkout2(REPOS_URL, path, rev, rev, True, True,
 188                      self.client_ctx)
 189
 190     self.assertEquals(client.url_from_path(path), REPOS_URL)
 191
 192   def test_uuid_from_path(self):
 193     """Test svn_client_uuid_from_path."""
 194     rev = core.svn_opt_revision_t()
 195     rev.kind = core.svn_opt_revision_head
 196
 197     path = tempfile.mktemp('uuid_from_path')
 198
 199     client.checkout2(REPOS_URL, path, rev, rev, True, True,
 200                      self.client_ctx)
 201
 202     wc_adm = wc.adm_open3(None, path, False, 0, None)
 203
 204     self.assertEquals(client.uuid_from_path(path, wc_adm, self.client_ctx),
 205                       client.uuid_from_url(REPOS_URL, self.client_ctx))
 206
 207     self.assert_(isinstance(client.uuid_from_path(path, wc_adm,
 208                             self.client_ctx), types.StringTypes))
 209
 210   def test_open_ra_session(self):
 211       """Test svn_client_open_ra_session()."""
 212       client.open_ra_session(REPOS_URL, self.client_ctx)
 213
 214
 215   def test_info_file(self):
 216     """Test svn_client_info on working copy file and remote files."""
 217
 218     # This test requires a file /trunk/README.txt of size 8 bytes
 219     # in the repository.
 220     rev = core.svn_opt_revision_t()
 221     rev.kind = core.svn_opt_revision_head
 222     wc_path = core.svn_path_canonicalize(tempfile.mktemp())
 223
 224     client.checkout2(REPOS_URL, wc_path, rev, rev, True, True,
 225                      self.client_ctx)
 226     adm_access = wc.adm_open3(None, wc_path, True, -1, None)
 227
 228     try:
 229       # Test 1: Run info -r BASE. We expect the size value to be filled in.
 230       rev.kind = core.svn_opt_revision_base
 231       readme_path = '%s/trunk/README.txt' % wc_path
 232       readme_url = '%s/trunk/README.txt' % REPOS_URL
 233       client.info(readme_path, rev, rev, self.info_receiver,
 234                   False, self.client_ctx)
 235
 236       self.assertEqual(self.path, os.path.basename(readme_path))
 237       self.info.assert_valid()
 238       self.assertEqual(self.info.working_size, client.SWIG_SVN_INFO_SIZE_UNKNOWN)
 239       self.assertEqual(self.info.size, 8)
 240
 241       # Test 2: Run info (revision unspecified). We expect the working_size value
 242       # to be filled in.
 243       rev.kind = core.svn_opt_revision_unspecified
 244       client.info(readme_path, rev, rev, self.info_receiver,
 245                   False, self.client_ctx)
 246
 247       self.assertEqual(self.path, readme_path)
 248       self.info.assert_valid()
 249       self.assertEqual(self.info.size, client.SWIG_SVN_INFO_SIZE_UNKNOWN)
 250       # README.txt contains one EOL char, so on Windows it will be expanded from
 251       # LF to CRLF hence the working_size will be 9 instead of 8.
 252       if os.name == 'nt':
 253         self.assertEqual(self.info.working_size, 9)
 254       else:
 255         self.assertEqual(self.info.working_size, 8)
 256
 257       # Test 3: Run info on the repository URL of README.txt. We expect the size
 258       # value to be filled in.
 259       rev.kind = core.svn_opt_revision_head
 260       client.info(readme_url, rev, rev, self.info_receiver,
 261                   False, self.client_ctx)
 262       self.info.assert_valid()
 263       self.assertEqual(self.info.working_size, client.SWIG_SVN_INFO_SIZE_UNKNOWN)
 264       self.assertEqual(self.info.size, 8)
 265     finally:
 266       wc.adm_close(adm_access)
 267       core.svn_io_remove_dir(wc_path)
 268
 269 def suite():
 270     return unittest.makeSuite(SubversionClientTestCase, 'test',
 271                               suiteClass=SubversionRepositoryTestSetup)
 272
 273 if __name__ == '__main__':
 274     runner = unittest.TextTestRunner()
 275     runner.run(suite())
 276