subvertpy/tests/test_wc.py

   1 # Copyright (C) 2005-2007 Jelmer Vernooij <jelmer@jelmer.uk>
   2
   3 # This program is free software; you can redistribute it and/or modify
   4 # it under the terms of the GNU Lesser General Public License as published by
   5 # the Free Software Foundation; either version 2.1 of the License, or
   6 # (at your option) any later version.
   7
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU Lesser General Public License for more details.
  12
  13 # You should have received a copy of the GNU Lesser General Public License
  14 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  15
  16 """Subversion ra library tests."""
  17
  18 import hashlib
  19 from io import BytesIO
  20 import os
  21
  22 import subvertpy
  23 from subvertpy import (
  24     NODE_DIR,
  25     NODE_FILE,
  26     wc,
  27     )
  28 from subvertpy.tests import (
  29     SubversionTestCase,
  30     TestCase,
  31     )
  32
  33
  34 class VersionTest(TestCase):
  35
  36     def test_version_length(self):
  37         self.assertEqual(4, len(wc.version()))
  38
  39     def test_api_version_length(self):
  40         self.assertEqual(4, len(wc.api_version()))
  41
  42     def test_api_version_later_same(self):
  43         self.assertTrue(wc.api_version() <= wc.version())
  44
  45
  46 class AdmTests(TestCase):
  47
  48     def test_get_adm_dir(self):
  49         self.assertEqual(".svn", wc.get_adm_dir())
  50
  51     def test_set_adm_dir(self):
  52         old_dir_name = wc.get_adm_dir()
  53         try:
  54             wc.set_adm_dir(b"_svn")
  55             self.assertEqual("_svn", wc.get_adm_dir())
  56         finally:
  57             wc.set_adm_dir(old_dir_name)
  58
  59     def test_is_normal_prop(self):
  60         self.assertTrue(wc.is_normal_prop("svn:ignore"))
  61
  62     def test_is_entry_prop(self):
  63         self.assertTrue(wc.is_entry_prop("svn:entry:foo"))
  64
  65     def test_is_wc_prop(self):
  66         self.assertTrue(wc.is_wc_prop("svn:wc:foo"))
  67
  68     def test_match_ignore_list(self):
  69         if wc.api_version() < (1, 5):
  70             self.assertRaises(
  71                 NotImplementedError, wc.match_ignore_list, "foo", [])
  72             self.skipTest("match_ignore_list not supported with svn < 1.5")
  73         self.assertTrue(wc.match_ignore_list("foo", ["f*"]))
  74         self.assertTrue(wc.match_ignore_list("foo", ["foo"]))
  75         self.assertFalse(wc.match_ignore_list("foo", []))
  76         self.assertFalse(wc.match_ignore_list("foo", ["bar"]))
  77
  78
  79 class WcTests(SubversionTestCase):
  80
  81     def test_revision_status(self):
  82         self.make_client("repos", "checkout")
  83         ret = wc.revision_status("checkout")
  84         self.assertEqual((0, 0, 0, 0), ret)
  85
  86     def test_revision_status_trailing(self):
  87         self.make_client("repos", "checkout")
  88         ret = wc.revision_status("checkout/")
  89         self.assertEqual((0, 0, 0, 0), ret)
  90
  91
  92 class AdmObjTests(SubversionTestCase):
  93
  94     def test_has_binary_prop(self):
  95         self.make_client("repos", "checkout")
  96         self.build_tree({"checkout/bar": b"\x00 \x01"})
  97         self.client_add('checkout/bar')
  98         self.client_set_prop('checkout/bar', 'svn:mime-type', 'text/bar')
  99         adm = wc.Adm(None, "checkout")
 100         self.assertFalse(adm.has_binary_prop("checkout/bar"))
 101         adm.close()
 102
 103     def test_get_ancestry(self):
 104         repos_url = self.make_client("repos", "checkout")
 105         self.build_tree({"checkout/bar": b"\x00 \x01"})
 106         self.client_add('checkout/bar')
 107         adm = wc.Adm(None, "checkout")
 108         self.assertEqual(("%s/bar" % repos_url, 0),
 109                          adm.get_ancestry("checkout/bar"))
 110         adm.close()
 111
 112     def test_maybe_set_repos_root(self):
 113         repos_url = self.make_client("repos", "checkout")
 114         adm = wc.Adm(None, "checkout")
 115         adm.maybe_set_repos_root(
 116             os.path.join(self.test_dir, "checkout"), repos_url)
 117         adm.close()
 118
 119     def test_add_repos_file(self):
 120         if wc.api_version() < (1, 6):
 121             self.skipTest("doesn't work with svn < 1.6")
 122         self.make_client("repos", "checkout")
 123         adm = wc.Adm(None, "checkout", True)
 124         adm.add_repos_file("checkout/bar", BytesIO(b"basecontents"),
 125                            BytesIO(b"contents"), {}, {})
 126         if wc.api_version() >= (1, 7):
 127             self.skipTest("TODO: doesn't yet work with svn >= 1.7")
 128         self.assertEqual(b"basecontents",
 129                          wc.get_pristine_contents("checkout/bar").read())
 130
 131     def test_mark_missing_deleted(self):
 132         if wc.api_version() >= (1, 7):
 133             self.skipTest("TODO: doesn't yet work with svn >= 1.7")
 134         self.make_client("repos", "checkout")
 135         self.build_tree({"checkout/bar": b"\x00 \x01"})
 136         self.client_add('checkout/bar')
 137         adm = wc.Adm(None, "checkout", True)
 138         os.remove("checkout/bar")
 139         adm.mark_missing_deleted("checkout/bar")
 140         self.assertFalse(os.path.exists("checkout/bar"))
 141
 142     def test_remove_from_revision_control(self):
 143         self.make_client("repos", "checkout")
 144         self.build_tree({"checkout/bar": b"\x00 \x01"})
 145         self.client_add('checkout/bar')
 146         adm = wc.Adm(None, "checkout", True)
 147         adm.remove_from_revision_control("bar")
 148         self.assertTrue(os.path.exists("checkout/bar"))
 149
 150     def test_relocate(self):
 151         self.make_client("repos", "checkout")
 152         adm = wc.Adm(None, "checkout", True)
 153         adm.relocate("checkout", "file://", "http://")
 154
 155     def test_translated_stream(self):
 156         self.skipTest("TODO: doesn't yet work")
 157         self.make_client("repos", "checkout")
 158         self.build_tree({"checkout/bar": b"My id: $Id$"})
 159         self.client_add('checkout/bar')
 160         self.client_set_prop("checkout/bar", "svn:keywords", "Id\n")
 161         self.client_commit("checkout", "foo")
 162         adm = wc.Adm(None, "checkout", True)
 163         stream = adm.translated_stream(
 164                 'checkout/bar', 'checkout/bar', wc.TRANSLATE_TO_NF)
 165         body = stream.read()
 166         self.assertTrue(body.startswith(b"My id: $Id: "), body)
 167
 168     def test_text_modified(self):
 169         self.make_client("repos", "checkout")
 170         self.build_tree({"checkout/bar": b"My id: $Id$"})
 171         self.client_add('checkout/bar')
 172         self.client_set_prop("checkout/bar", "svn:keywords", "Id\n")
 173         self.client_commit("checkout", "foo")
 174         adm = wc.Adm(None, "checkout")
 175         self.assertFalse(adm.text_modified("checkout/bar"))
 176         self.build_tree({"checkout/bar": b"gambon"})
 177         self.assertTrue(adm.text_modified("checkout/bar", True))
 178
 179     def test_props_modified(self):
 180         self.make_client("repos", "checkout")
 181         self.build_tree({"checkout/bar": b"My id: $Id$"})
 182         self.client_add('checkout/bar')
 183         self.client_set_prop("checkout/bar", "svn:keywords", "Id\n")
 184         self.client_commit("checkout", "foo")
 185         adm = wc.Adm(None, "checkout", True)
 186         self.assertFalse(adm.props_modified("checkout/bar"))
 187         adm.prop_set("aprop", "avalue", "checkout/bar")
 188         self.assertTrue(adm.props_modified("checkout/bar"))
 189
 190     def test_prop_set(self):
 191         self.make_client("repos", "checkout")
 192         self.build_tree({"checkout/bar": b"file"})
 193         self.client_add('checkout/bar')
 194         adm = wc.Adm(None, "checkout", True)
 195         adm.prop_set("aprop", "avalue", "checkout/bar")
 196         self.assertEqual(adm.prop_get("aprop", "checkout/bar"), b"avalue")
 197         adm.prop_set("aprop", None, "checkout/bar")
 198         self.assertEqual(adm.prop_get("aprop", "checkout/bar"), None)
 199
 200     def test_committed_queue(self):
 201         cq = wc.CommittedQueue()
 202         self.make_client("repos", "checkout")
 203         adm = wc.Adm(None, "checkout", True)
 204         adm.process_committed_queue(cq, 1, "2010-05-31T08:49:22.430000Z",
 205                                     "jelmer")
 206
 207     def test_entry_not_found(self):
 208         self.make_client("repos", "checkout")
 209         adm = wc.Adm(None, "checkout")
 210         self.assertRaises(KeyError, adm.entry, "bar")
 211
 212     def test_entry(self):
 213         self.make_client("repos", "checkout")
 214         self.build_tree({"checkout/bar": b"\x00 \x01"})
 215         self.client_add('checkout/bar')
 216         adm = wc.Adm(None, "checkout")
 217         entry = adm.entry("checkout/bar")
 218         self.assertEqual("bar", entry.name)
 219         self.assertEqual(NODE_FILE, entry.kind)
 220         self.assertEqual(0, entry.revision)
 221         self.client_commit("checkout", "msg")
 222         adm = wc.Adm(None, "checkout")
 223         entry = adm.entry("checkout/bar")
 224         self.assertEqual("bar", entry.name)
 225         self.assertEqual(NODE_FILE, entry.kind)
 226         self.assertEqual(1, entry.revision)
 227
 228     def test_get_actual_target(self):
 229         self.make_client("repos", ".")
 230         self.assertEqual((self.test_dir, "bla"),
 231                          wc.get_actual_target("%s/bla" % self.test_dir))
 232
 233     def test_is_wc_root(self):
 234         self.make_client("repos", ".")
 235         self.build_tree({"bar": None})
 236         self.client_add('bar')
 237         adm = wc.Adm(None, ".")
 238         self.assertTrue(adm.is_wc_root(self.test_dir))
 239         self.assertFalse(adm.is_wc_root(os.path.join(self.test_dir, "bar")))
 240
 241     def test_status(self):
 242         self.make_client("repos", "checkout")
 243         self.build_tree({"checkout/bar": b"text"})
 244         self.client_add('checkout/bar')
 245         adm = wc.Adm(None, "checkout")
 246         self.assertEqual(
 247                 wc.STATUS_ADDED,
 248                 adm.status('checkout/bar').status)
 249         self.client_commit("checkout", "foo")
 250         adm = wc.Adm(None, "checkout")
 251         self.assertEqual(
 252                 wc.STATUS_NORMAL,
 253                 adm.status('checkout/bar').status)
 254
 255     def test_transmit_text_deltas(self):
 256         if wc.api_version() >= (1, 7):
 257             self.skipTest("TODO: doesn't yet work with svn >= 1.7")
 258         self.make_client("repos", ".")
 259         self.build_tree({"bar": b"blala"})
 260         self.client_add('bar')
 261         adm = wc.Adm(None, ".", True)
 262
 263         class Editor(object):
 264             """Editor"""
 265
 266             def __init__(self):
 267                 self._windows = []
 268                 self._prop = {}
 269
 270             def change_prop(self, name, value):
 271                 self._prop[name] = value
 272
 273             def apply_textdelta(self, checksum):
 274                 def window_handler(window):
 275                     self._windows.append(window)
 276                 return window_handler
 277
 278             def close(self):
 279                 pass
 280         editor = Editor()
 281         (tmpfile, md5_digest) = adm.transmit_text_deltas("bar", True, editor)
 282         self.assertEqual(editor._windows,
 283                          [(0, 0, 5, 0, [(2, 0, 5)], b'blala'), None])
 284         self.assertIsInstance(tmpfile, str)
 285         self.assertEqual(16, len(md5_digest))
 286         self.assertEqual(hashlib.md5(b'blala').digest(), md5_digest)
 287
 288         bar = adm.entry("bar")
 289         self.assertEqual(-1, bar.cmt_rev)
 290         self.assertEqual(0, bar.revision)
 291         self.assertIn(bar.checksum, (None, hashlib.md5(b'blala').hexdigest()))
 292
 293         cq = wc.CommittedQueue()
 294         cq.queue("bar", adm, wcprop_changes=editor._prop,
 295                  md5_digest=md5_digest)
 296         adm.process_committed_queue(cq, 1, "2010-05-31T08:49:22.430000Z",
 297                                     "jelmer")
 298         bar = adm.entry("bar")
 299         self.assertEqual("bar", bar.name)
 300         self.assertEqual(NODE_FILE, bar.kind)
 301         self.assertEqual(wc.SCHEDULE_NORMAL, bar.schedule)
 302         self.assertIn(bar.checksum, (None, hashlib.md5(b'blala').hexdigest()))
 303         self.assertEqual(1, bar.cmt_rev)
 304         self.assertEqual(1, bar.revision)
 305
 306     def test_process_committed_queue(self):
 307         if wc.api_version() >= (1, 7):
 308             self.skipTest("TODO: doesn't yet work with svn >= 1.7")
 309         self.make_client("repos", "checkout")
 310         self.build_tree({"checkout/bar": b"blala"})
 311         self.client_add('checkout/bar')
 312         adm = wc.Adm(None, "checkout", True)
 313
 314         class Editor(object):
 315             """Editor"""
 316
 317             def __init__(self):
 318                 self._windows = []
 319                 self._prop = {}
 320
 321             def change_prop(self, name, value):
 322                 self._prop[name] = value
 323
 324             def apply_textdelta(self, checksum):
 325                 def window_handler(window):
 326                     self._windows.append(window)
 327                 return window_handler
 328
 329             def close(self):
 330                 pass
 331
 332         editor = Editor()
 333         (tmpfile, md5_digest) = adm.transmit_text_deltas(
 334                 "checkout/bar", True, editor)
 335         self.assertEqual(editor._windows,
 336                          [(0, 0, 5, 0, [(2, 0, 5)], b'blala'), None])
 337         self.assertIsInstance(tmpfile, str)
 338         self.assertEqual(16, len(md5_digest))
 339         self.assertEqual(hashlib.md5(b'blala').digest(), md5_digest)
 340
 341         cq = wc.CommittedQueue()
 342         cq.queue("checkout/bar", adm, wcprop_changes=editor._prop,
 343                  md5_digest=md5_digest)
 344         adm.process_committed_queue(cq, 1, "2010-05-31T08:49:22.430000Z",
 345                                     "jelmer")
 346         bar = adm.entry("checkout/bar")
 347         self.assertEqual("bar", bar.name)
 348         self.assertEqual(NODE_FILE, bar.kind)
 349         self.assertEqual(wc.SCHEDULE_NORMAL, bar.schedule)
 350
 351     def test_process_committed(self):
 352         if wc.api_version() >= (1, 7):
 353             self.skipTest("TODO: doesn't yet work with svn >= 1.7")
 354         self.make_client("repos", ".")
 355         self.build_tree({"bar": b"la"})
 356         self.client_add('bar')
 357         adm = wc.Adm(None, ".", True)
 358
 359         class Editor(object):
 360             """Editor"""
 361
 362             def __init__(self):
 363                 self._windows = []
 364
 365             def apply_textdelta(self, checksum):
 366                 def window_handler(window):
 367                     self._windows.append(window)
 368                 return window_handler
 369
 370             def close(self):
 371                 pass
 372         editor = Editor()
 373         (tmpfile, digest) = adm.transmit_text_deltas("bar", True, editor)
 374         self.assertEqual(editor._windows,
 375                          [(0, 0, 2, 0, [(2, 0, 2)], b'la'), None])
 376         self.assertIsInstance(tmpfile, str)
 377         self.assertEqual(16, len(digest))
 378         self.assertEqual(hashlib.md5(b'blala').digest(), digest)
 379         bar = adm.entry("bar")
 380         self.assertEqual(-1, bar.cmt_rev)
 381         self.assertEqual(0, bar.revision)
 382
 383         adm.process_committed(
 384             "bar", False, 1, "2010-05-31T08:49:22.430000Z", "jelmer")
 385         bar = adm.entry("bar")
 386         self.assertEqual("bar", bar.name)
 387         self.assertEqual(NODE_FILE, bar.kind)
 388         self.assertEqual(wc.SCHEDULE_NORMAL, bar.schedule)
 389
 390     def test_probe_try(self):
 391         self.make_client("repos", "checkout")
 392         self.build_tree({"checkout/bar": b"la"})
 393         self.client_add('checkout/bar')
 394         adm = wc.Adm(None, "checkout", True)
 395         try:
 396             self.assertIs(None, adm.probe_try(self.test_dir))
 397         except subvertpy.SubversionException as e:
 398             (msg, num) = e.args
 399             if num != subvertpy.ERR_WC_NOT_WORKING_COPY:
 400                 raise
 401         self.assertEqual(
 402             os.path.abspath("checkout"),
 403             adm.probe_try(os.path.join("checkout", "bar")).access_path())
 404
 405     def test_lock(self):
 406         if wc.api_version() >= (1, 9):
 407             self.skipTest("TODO: doesn't yet work with svn >= 1.9")
 408         self.make_client("repos", "checkout")
 409         self.build_tree({"checkout/bar": b"la"})
 410         self.client_add('checkout/bar')
 411         adm = wc.Adm(None, "checkout", True)
 412         lock = wc.Lock()
 413         lock.token = b"blah"
 414         adm.add_lock("checkout", lock)
 415         adm.remove_lock("checkout")
 416
 417
 418 class ContextTests(SubversionTestCase):
 419
 420     def setUp(self):
 421         super(ContextTests, self).setUp()
 422         if wc.api_version() < (1, 7):
 423             self.skipTest("context API not available on Subversion < 1.7")
 424
 425     def test_create(self):
 426         context = wc.Context()
 427         self.assertIsInstance(context, wc.Context)
 428
 429     def test_locked(self):
 430         context = wc.Context()
 431         self.make_client("repos", "checkout")
 432         self.assertEqual((False, False), context.locked("checkout"))
 433
 434     def test_check_wc(self):
 435         context = wc.Context()
 436         self.make_client("repos", "checkout")
 437         self.assertIsInstance(context.check_wc("checkout"), int)
 438
 439     def test_text_modified(self):
 440         context = wc.Context()
 441         self.make_client("repos", "checkout")
 442         with open('checkout/bla.txt', 'w') as f:
 443             f.write("modified")
 444         self.client_add("checkout/bla.txt")
 445         self.assertTrue(context.text_modified("checkout/bla.txt"))
 446
 447     def test_props_modified(self):
 448         context = wc.Context()
 449         self.make_client("repos", "checkout")
 450         with open('checkout/bla.txt', 'w') as f:
 451             f.write("modified")
 452         self.client_add("checkout/bla.txt")
 453         self.assertFalse(context.props_modified("checkout/bla.txt"))
 454
 455     def test_conflicted(self):
 456         context = wc.Context()
 457         self.make_client("repos", "checkout")
 458         with open('checkout/bla.txt', 'w') as f:
 459             f.write("modified")
 460         self.client_add("checkout/bla.txt")
 461         self.assertEqual(
 462             (False, False, False),
 463             context.conflicted("checkout/bla.txt"))
 464
 465     def test_crawl_revisions(self):
 466         context = wc.Context()
 467         self.make_client("repos", "checkout")
 468         with open('checkout/bla.txt', 'w') as f:
 469             f.write("modified")
 470         self.client_add("checkout/bla.txt")
 471         ret = []
 472
 473         class Reporter(object):
 474             def set_path(self, *args):
 475                 ret.append(args)
 476
 477             def finish(self):
 478                 pass
 479         context.crawl_revisions("checkout", Reporter())
 480
 481         self.assertEqual(ret, [('', 0, 0, None, 3)])
 482
 483     def test_get_update_editor(self):
 484         self.make_client("repos", "checkout")
 485         context = wc.Context()
 486         editor = context.get_update_editor("checkout", "")
 487         editor.close()
 488
 489     def test_status(self):
 490         self.make_client("repos", "checkout")
 491         context = wc.Context()
 492         status = context.status("checkout")
 493         self.assertEqual(NODE_DIR, status.kind)
 494
 495     def test_walk_status(self):
 496         self.make_client("repos", "checkout")
 497         with open('checkout/bla.txt', 'w') as f:
 498             f.write("modified")
 499         self.client_add("checkout/bla.txt")
 500         context = wc.Context()
 501         result = {}
 502         context.walk_status("checkout", result.__setitem__)
 503         self.assertEqual(
 504                 set(result.keys()),
 505                 {os.path.abspath("checkout"),
 506                  os.path.abspath("checkout/bla.txt")})
 507
 508     def test_locking(self):
 509         if wc.api_version() >= (1, 7):
 510             self.skipTest("TODO: doesn't yet work with svn >= 1.7")
 511         self.make_client("repos", "checkout")
 512         with open('checkout/bla.txt', 'w') as f:
 513             f.write("modified")
 514         self.client_add("checkout/bla.txt")
 515         context = wc.Context()
 516         lock = wc.Lock(token=b'foo')
 517         self.assertEqual((False, False), context.locked("checkout"))
 518         context.add_lock("checkout/", lock)
 519         self.assertEqual((True, True), context.locked("checkout"))
 520         context.remove_lock("checkout/", lock)
 521
 522     def test_add_from_disk(self):
 523         if wc.api_version() >= (1, 7):
 524             self.skipTest("TODO: doesn't yet work with svn >= 1.7")
 525         self.make_client("repos", "checkout")
 526         with open('checkout/bla.txt', 'w') as f:
 527             f.write("modified")
 528         context = wc.Context()
 529         lock = wc.Lock(token=b'foo')
 530         lock.path = os.path.abspath('checkout')+"/"
 531         context.add_lock("checkout", lock)
 532         context.add_from_disk('checkout/bla.txt')
 533         context.remove_lock("checkout", lock)
 534
 535     def test_get_prop_diffs(self):
 536         self.make_client("repos", "checkout")
 537         context = wc.Context()
 538         (orig_props, propdelta) = context.get_prop_diffs("checkout")
 539         self.assertEqual({}, orig_props)
 540         self.assertEqual([], propdelta)
 541
 542     def test_process_committed_queue(self):
 543         if wc.api_version() >= (1, 7):
 544             self.skipTest("TODO: doesn't yet work with svn >= 1.7")
 545         self.make_client("repos", "checkout")
 546         adm = wc.Context()
 547         self.build_tree({"checkout/bar": b"blala"})
 548         self.client_add('checkout/bar')
 549         adm.add_lock("checkout", wc.Lock(token=b'foo'))
 550         cq = wc.CommittedQueue()
 551         cq.queue("checkout/bar", adm)
 552         adm.process_committed_queue(cq, 1, "2010-05-31T08:49:22.430000Z",
 553                                     "jelmer")