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