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

   1 import unittest, os, setup_path, StringIO
   2 from svn import core, repos, fs, delta
   3 from svn.core import SubversionException
   4
   5 from trac.versioncontrol.tests.svn_fs import SubversionRepositoryTestSetup, \
   6   REPOS_PATH
   7
   8 class ChangeReceiver(delta.Editor):
   9   """A delta editor which saves textdeltas for later use"""
  10
  11   def __init__(self, src_root, tgt_root):
  12     self.src_root = src_root
  13     self.tgt_root = tgt_root
  14     self.textdeltas = []
  15
  16   def apply_textdelta(self, file_baton, base_checksum):
  17     def textdelta_handler(textdelta):
  18       if textdelta is not None:
  19         self.textdeltas.append(textdelta)
  20     return textdelta_handler
  21
  22 def _authz_callback(root, path, pool):
  23   "A dummy authz callback which always returns success."
  24   return 1
  25
  26 class SubversionRepositoryTestCase(unittest.TestCase):
  27   """Test cases for the Subversion repository layer"""
  28
  29   def setUp(self):
  30     """Load a Subversion repository"""
  31     self.repos = repos.open(REPOS_PATH)
  32     self.fs = repos.fs(self.repos)
  33     self.rev = fs.youngest_rev(self.fs)
  34
  35   def test_create(self):
  36     """Make sure that repos.create doesn't segfault when we set fs-type
  37        using a config hash"""
  38     fs_config = { "fs-type": "fsfs" }
  39     for i in range(5):
  40       path = os.path.join(REPOS_PATH, "test" + str(i))
  41       repos.create(path, "", "", None, fs_config)
  42       repos.delete(path)
  43
  44   def test_dump_fs2(self):
  45     """Test the dump_fs2 function"""
  46
  47     self.callback_calls = 0
  48
  49     def is_cancelled():
  50       self.callback_calls += 1
  51       return None
  52
  53     dumpstream = StringIO.StringIO()
  54     feedbackstream = StringIO.StringIO()
  55     repos.dump_fs2(self.repos, dumpstream, feedbackstream, 0, self.rev, 0, 0,
  56                    is_cancelled)
  57
  58     # Check that we can dump stuff
  59     dump = dumpstream.getvalue()
  60     feedback = feedbackstream.getvalue()
  61     expected_feedback = "* Dumped revision " + str(self.rev)
  62     self.assertEquals(dump.count("Node-path: trunk/README.txt"), 2)
  63     self.assertEquals(feedback.count(expected_feedback), 1)
  64     self.assertEquals(self.callback_calls, 13)
  65
  66     # Check that the dump can be cancelled
  67     self.assertRaises(SubversionException, repos.dump_fs2,
  68       self.repos, dumpstream, feedbackstream, 0, self.rev, 0, 0, lambda: 1)
  69
  70     dumpstream.close()
  71     feedbackstream.close()
  72
  73     # Check that the dump fails when the dumpstream is closed
  74     self.assertRaises(ValueError, repos.dump_fs2,
  75       self.repos, dumpstream, feedbackstream, 0, self.rev, 0, 0, None)
  76
  77     dumpstream = StringIO.StringIO()
  78     feedbackstream = StringIO.StringIO()
  79
  80     # Check that we can grab the feedback stream, but not the dumpstream
  81     repos.dump_fs2(self.repos, None, feedbackstream, 0, self.rev, 0, 0, None)
  82     feedback = feedbackstream.getvalue()
  83     self.assertEquals(feedback.count(expected_feedback), 1)
  84
  85     # Check that we can grab the dumpstream, but not the feedbackstream
  86     repos.dump_fs2(self.repos, dumpstream, None, 0, self.rev, 0, 0, None)
  87     dump = dumpstream.getvalue()
  88     self.assertEquals(dump.count("Node-path: trunk/README.txt"), 2)
  89
  90     # Check that we can ignore both the dumpstream and the feedbackstream
  91     repos.dump_fs2(self.repos, dumpstream, None, 0, self.rev, 0, 0, None)
  92     self.assertEquals(feedback.count(expected_feedback), 1)
  93
  94     # FIXME: The Python bindings don't check for 'NULL' values for
  95     #        svn_repos_t objects, so the following call segfaults
  96     #repos.dump_fs2(None, None, None, 0, self.rev, 0, 0, None)
  97
  98   def test_get_logs(self):
  99     """Test scope of get_logs callbacks"""
 100     logs = []
 101     def addLog(paths, revision, author, date, message, pool):
 102       if paths is not None:
 103         logs.append(paths)
 104
 105     # Run get_logs
 106     repos.get_logs(self.repos, ['/'], self.rev, 0, True, 0, addLog)
 107
 108     # Count and verify changes
 109     change_count = 0
 110     for log in logs:
 111       for path_changed in log.values():
 112         change_count += 1
 113         path_changed.assert_valid()
 114     self.assertEqual(logs[2]["/tags/v1.1"].action, "A")
 115     self.assertEqual(logs[2]["/tags/v1.1"].copyfrom_path, "/branches/v1x")
 116     self.assertEqual(len(logs), 12)
 117     self.assertEqual(change_count, 19)
 118
 119   def test_dir_delta(self):
 120     """Test scope of dir_delta callbacks"""
 121     # Run dir_delta
 122     this_root = fs.revision_root(self.fs, self.rev)
 123     prev_root = fs.revision_root(self.fs, self.rev-1)
 124     editor = ChangeReceiver(this_root, prev_root)
 125     e_ptr, e_baton = delta.make_editor(editor)
 126     repos.dir_delta(prev_root, '', '', this_root, '', e_ptr, e_baton,
 127                     _authz_callback, 1, 1, 0, 0)
 128
 129     # Check results
 130     self.assertEqual(editor.textdeltas[0].new_data, "This is a test.\n")
 131     self.assertEqual(editor.textdeltas[1].new_data, "A test.\n")
 132     self.assertEqual(len(editor.textdeltas),2)
 133
 134   def test_retrieve_and_change_rev_prop(self):
 135     """Test playing with revprops"""
 136     self.assertEqual(repos.fs_revision_prop(self.repos, self.rev, "svn:log",
 137                                             _authz_callback),
 138                      "''(a few years later)'' Argh... v1.1 was buggy, "
 139                      "after all")
 140
 141     # We expect this to complain because we have no pre-revprop-change
 142     # hook script for the repository.
 143     self.assertRaises(SubversionException, repos.fs_change_rev_prop3,
 144                       self.repos, self.rev, "jrandom", "svn:log",
 145                       "Youngest revision", True, True, _authz_callback)
 146
 147     repos.fs_change_rev_prop3(self.repos, self.rev, "jrandom", "svn:log",
 148                               "Youngest revision", False, False,
 149                               _authz_callback)
 150
 151     self.assertEqual(repos.fs_revision_prop(self.repos, self.rev, "svn:log",
 152                                             _authz_callback),
 153                      "Youngest revision")
 154
 155 def suite():
 156     return unittest.makeSuite(SubversionRepositoryTestCase, 'test',
 157                               suiteClass=SubversionRepositoryTestSetup)
 158
 159 if __name__ == '__main__':
 160     runner = unittest.TextTestRunner()
 161     runner.run(suite())