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 from io import BytesIO
  19 import os
  20 from unittest import SkipTest
  21
  22 import subvertpy
  23 from subvertpy import (
  24     NODE_FILE,
  25     wc,
  26     )
  27 from subvertpy.tests import (
  28     SubversionTestCase,
  29     TestCase,
  30     )
  31
  32
  33 class VersionTest(TestCase):
  34
  35     def test_version_length(self):
  36         self.assertEqual(4, len(wc.version()))
  37
  38     def test_api_version_length(self):
  39         self.assertEqual(4, len(wc.api_version()))
  40
  41     def test_api_version_later_same(self):
  42         self.assertTrue(wc.api_version() <= wc.version())
  43
  44
  45 class AdmTests(TestCase):
  46
  47     def test_get_adm_dir(self):
  48         self.assertEqual(b".svn", wc.get_adm_dir())
  49
  50     def test_set_adm_dir(self):
  51         old_dir_name = wc.get_adm_dir()
  52         try:
  53             wc.set_adm_dir(b"_svn")
  54             self.assertEqual(b"_svn", wc.get_adm_dir())
  55         finally:
  56             wc.set_adm_dir(old_dir_name)
  57
  58     def test_is_normal_prop(self):
  59         self.assertTrue(wc.is_normal_prop("svn:ignore"))
  60
  61     def test_is_entry_prop(self):
  62         self.assertTrue(wc.is_entry_prop("svn:entry:foo"))
  63
  64     def test_is_wc_prop(self):
  65         self.assertTrue(wc.is_wc_prop("svn:wc:foo"))
  66
  67     def test_match_ignore_list(self):
  68         if wc.api_version() < (1, 5):
  69             self.assertRaises(
  70                 NotImplementedError, wc.match_ignore_list, "foo", [])
  71             return  # Skip test
  72         self.assertTrue(wc.match_ignore_list("foo", ["f*"]))
  73         self.assertTrue(wc.match_ignore_list("foo", ["foo"]))
  74         self.assertFalse(wc.match_ignore_list("foo", []))
  75         self.assertFalse(wc.match_ignore_list("foo", ["bar"]))
  76
  77
  78 class WcTests(SubversionTestCase):
  79
  80     def test_revision_status(self):
  81         self.make_client("repos", "checkout")
  82         ret = wc.revision_status("checkout")
  83         self.assertEqual((0, 0, 0, 0), ret)
  84
  85     def test_revision_status_trailing(self):
  86         self.make_client("repos", "checkout")
  87         ret = wc.revision_status("checkout/")
  88         self.assertEqual((0, 0, 0, 0), ret)
  89
  90
  91 class AdmTests(SubversionTestCase):
  92
  93     def setUp(self):
  94         super(AdmTests, self).setUp()
  95         if getattr(wc, "WorkingCopy", None) is None:
  96             raise SkipTest(
  97                 "Subversion 1.7 API for WorkingCopy not yet supported")
  98
  99     def test_has_binary_prop(self):
 100         self.make_client("repos", "checkout")
 101         self.build_tree({"checkout/bar": b"\x00 \x01"})
 102         self.client_add('checkout/bar')
 103         adm = wc.Adm(None, "checkout")
 104         path = os.path.join(self.test_dir, "checkout/bar")
 105         self.assertFalse(adm.has_binary_prop(path))
 106         adm.close()
 107
 108     def test_get_ancestry(self):
 109         repos_url = self.make_client("repos", "checkout")
 110         self.build_tree({"checkout/bar": b"\x00 \x01"})
 111         self.client_add('checkout/bar')
 112         adm = wc.Adm(None, "checkout")
 113         self.assertEqual(("%s/bar" % repos_url, 0),
 114                          adm.get_ancestry("checkout/bar"))
 115         adm.close()
 116
 117     def test_maybe_set_repos_root(self):
 118         repos_url = self.make_client("repos", "checkout")
 119         adm = wc.Adm(None, "checkout")
 120         adm.maybe_set_repos_root(
 121             os.path.join(self.test_dir, "checkout"), repos_url)
 122         adm.close()
 123
 124     def test_add_repos_file(self):
 125         self.make_client("repos", "checkout")
 126         adm = wc.Adm(None, "checkout", True)
 127         adm.add_repos_file("checkout/bar", BytesIO(b"basecontents"),
 128                            BytesIO(b"contents"), {}, {})
 129         self.assertEqual(b"basecontents",
 130                          wc.get_pristine_contents("checkout/bar").read())
 131
 132     def test_mark_missing_deleted(self):
 133         self.make_client("repos", "checkout")
 134         self.build_tree({"checkout/bar": b"\x00 \x01"})
 135         self.client_add('checkout/bar')
 136         adm = wc.Adm(None, "checkout", True)
 137         os.remove("checkout/bar")
 138         adm.mark_missing_deleted("checkout/bar")
 139         self.assertFalse(os.path.exists("checkout/bar"))
 140
 141     def test_remove_from_revision_control(self):
 142         self.make_client("repos", "checkout")
 143         self.build_tree({"checkout/bar": b"\x00 \x01"})
 144         self.client_add('checkout/bar')
 145         adm = wc.Adm(None, "checkout", True)
 146         adm.remove_from_revision_control("bar")
 147         self.assertTrue(os.path.exists("checkout/bar"))
 148
 149     def test_relocate(self):
 150         self.make_client("repos", "checkout")
 151         adm = wc.Adm(None, "checkout", True)
 152         adm.relocate("checkout", "file://", "http://")
 153
 154     def test_translated_stream(self):
 155         self.make_client("repos", "checkout")
 156         self.build_tree({"checkout/bar": b"My id: $Id$"})
 157         self.client_add('checkout/bar')
 158         self.client_set_prop("checkout/bar", "svn:keywords", "Id\n")
 159         self.client_commit("checkout", "foo")
 160         adm = wc.Adm(None, "checkout", True)
 161         path = os.path.join(self.test_dir, "checkout/bar")
 162         stream = adm.translated_stream(path, path, wc.TRANSLATE_TO_NF)
 163         self.assertTrue(stream.read().startswith(b"My id: $Id: "))
 164
 165     def test_text_modified(self):
 166         self.make_client("repos", "checkout")
 167         self.build_tree({"checkout/bar": b"My id: $Id$"})
 168         self.client_add('checkout/bar')
 169         self.client_set_prop("checkout/bar", "svn:keywords", "Id\n")
 170         self.client_commit("checkout", "foo")
 171         adm = wc.Adm(None, "checkout")
 172         self.assertFalse(adm.text_modified("checkout/bar"))
 173         self.build_tree({"checkout/bar": b"gambon"})
 174         self.assertTrue(adm.text_modified("checkout/bar", True))
 175
 176     def test_props_modified(self):
 177         self.make_client("repos", "checkout")
 178         self.build_tree({"checkout/bar": b"My id: $Id$"})
 179         self.client_add('checkout/bar')
 180         self.client_set_prop("checkout/bar", "svn:keywords", "Id\n")
 181         self.client_commit("checkout", "foo")
 182         adm = wc.Adm(None, "checkout", True)
 183         self.assertFalse(adm.props_modified("checkout/bar"))
 184         adm.prop_set("aprop", "avalue", "checkout/bar")
 185         self.assertTrue(adm.props_modified("checkout/bar"))
 186
 187     def test_prop_set(self):
 188         self.make_client("repos", "checkout")
 189         self.build_tree({"checkout/bar": b"file"})
 190         self.client_add('checkout/bar')
 191         adm = wc.Adm(None, "checkout", True)
 192         adm.prop_set("aprop", "avalue", "checkout/bar")
 193         self.assertEqual(adm.prop_get("aprop", "checkout/bar"), "avalue")
 194         adm.prop_set("aprop", None, "checkout/bar")
 195         self.assertEqual(adm.prop_get("aprop", "checkout/bar"), None)
 196
 197     def test_committed_queue(self):
 198         if getattr(wc, "CommittedQueue", None) is None:
 199             raise SkipTest("CommittedQueue not available")
 200         cq = wc.CommittedQueue()
 201         self.make_client("repos", "checkout")
 202         adm = wc.Adm(None, "checkout", True)
 203         adm.process_committed_queue(cq, 1, "2010-05-31T08:49:22.430000Z",
 204                                     "jelmer")
 205
 206     def test_entry_not_found(self):
 207         self.make_client("repos", "checkout")
 208         adm = wc.Adm(None, "checkout")
 209         self.assertRaises(KeyError, adm.entry, "bar")
 210
 211     def test_entry(self):
 212         self.make_client("repos", "checkout")
 213         self.build_tree({"checkout/bar": b"\x00 \x01"})
 214         self.client_add('checkout/bar')
 215         adm = wc.Adm(None, "checkout")
 216         entry = adm.entry("checkout/bar")
 217         self.assertEqual("bar", entry.name)
 218         self.assertEqual(NODE_FILE, entry.kind)
 219         self.assertEqual(0, entry.revision)
 220         self.client_commit("checkout", "msg")
 221         adm = wc.Adm(None, "checkout")
 222         entry = adm.entry("checkout/bar")
 223         self.assertEqual("bar", entry.name)
 224         self.assertEqual(NODE_FILE, entry.kind)
 225         self.assertEqual(1, entry.revision)
 226
 227     def test_get_actual_target(self):
 228         self.make_client("repos", ".")
 229         self.assertEqual((self.test_dir, "bla"),
 230                          wc.get_actual_target("%s/bla" % self.test_dir))
 231
 232     def test_is_wc_root(self):
 233         self.make_client("repos", ".")
 234         self.build_tree({"bar": None})
 235         self.client_add('bar')
 236         adm = wc.Adm(None, ".")
 237         self.assertTrue(adm.is_wc_root(self.test_dir))
 238         self.assertFalse(adm.is_wc_root(os.path.join(self.test_dir, "bar")))
 239
 240     def test_status(self):
 241         self.make_client("repos", "checkout")
 242         self.build_tree({"checkout/bar": b"text"})
 243         self.client_add('checkout/bar')
 244         adm = wc.Adm(None, "checkout")
 245         self.assertEqual(wc.STATUS_ADDED, adm.status('bar').status)
 246         self.client_commit("checkout", "foo")
 247         adm = wc.Adm(None, "checkout")
 248         self.assertEqual(wc.STATUS_NORMAL, adm.status('bar').status)
 249
 250     def test_transmit_text_deltas(self):
 251         self.make_client("repos", ".")
 252         self.build_tree({"bar": b"blala"})
 253         self.client_add('bar')
 254         adm = wc.Adm(None, ".", True)
 255
 256         class Editor(object):
 257             """Editor"""
 258
 259             def __init__(self):
 260                 self._windows = []
 261
 262             def apply_textdelta(self, checksum):
 263                 def window_handler(window):
 264                     self._windows.append(window)
 265                 return window_handler
 266
 267             def close(self):
 268                 pass
 269         editor = Editor()
 270         (tmpfile, digest) = adm.transmit_text_deltas("bar", True, editor)
 271         self.assertEqual(editor._windows,
 272                          [(0, 0, 5, 0, [(2, 0, 5)], b'blala'), None])
 273         self.assertIsInstance(tmpfile, str)
 274         self.assertEqual(16, len(digest))
 275
 276         bar = adm.entry("bar")
 277         self.assertEqual(-1, bar.cmt_rev)
 278         self.assertEqual(0, bar.revision)
 279
 280         cq = wc.CommittedQueue()
 281         cq.queue("bar", adm)
 282         adm.process_committed_queue(cq, 1, "2010-05-31T08:49:22.430000Z",
 283                                     "jelmer")
 284         bar = adm.entry("bar")
 285         self.assertEqual("bar", bar.name)
 286         self.assertEqual(NODE_FILE, bar.kind)
 287         self.assertEqual(wc.SCHEDULE_NORMAL, bar.schedule)
 288         self.assertIs(None, bar.checksum)
 289         self.assertEqual(1, bar.cmt_rev)
 290         self.assertEqual(1, bar.revision)
 291
 292     def test_process_committed_queue(self):
 293         self.make_client("repos", "checkout")
 294         self.build_tree({"checkout/bar": b"la"})
 295         self.client_add('checkout/bar')
 296         adm = wc.Adm(None, "checkout", True)
 297         cq = wc.CommittedQueue()
 298         cq.queue(os.path.join(self.test_dir, "checkout/bar"), adm)
 299         adm.process_committed_queue(cq, 1, "2010-05-31T08:49:22.430000Z",
 300                                     "jelmer")
 301         bar = adm.entry("checkout/bar")
 302         self.assertEqual("bar", bar.name)
 303         self.assertEqual(NODE_FILE, bar.kind)
 304         self.assertEqual(wc.SCHEDULE_ADD, bar.schedule)
 305
 306     def test_probe_try(self):
 307         self.make_client("repos", "checkout")
 308         self.build_tree({"checkout/bar": b"la"})
 309         self.client_add('checkout/bar')
 310         adm = wc.Adm(None, "checkout", True)
 311         try:
 312             self.assertIs(None, adm.probe_try(self.test_dir))
 313         except subvertpy.SubversionException as e:
 314             (msg, num) = e.args
 315             if num != subvertpy.ERR_WC_NOT_WORKING_COPY:
 316                 raise
 317         self.assertEqual(
 318             "checkout",
 319             adm.probe_try(os.path.join("checkout", "bar")).access_path())
 320
 321
 322 class ContextTests(SubversionTestCase):
 323
 324     def test_create(self):
 325         context = wc.Context()
 326         self.assertIsInstance(context, wc.Context)
 327
 328     def test_locked(self):
 329         context = wc.Context()
 330         self.make_client("repos", "checkout")
 331         self.assertEqual((False, False), context.locked("checkout"))