subvertpy/tests/test_client.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 client library tests."""
  17
  18 from datetime import datetime, timedelta
  19 import os
  20 from io import BytesIO
  21 import shutil
  22 import tempfile
  23 from unittest import SkipTest
  24
  25 from subvertpy import (
  26     SubversionException,
  27     NODE_DIR, NODE_FILE,
  28     client,
  29     ra,
  30     wc,
  31     )
  32 from subvertpy.tests import (
  33     SubversionTestCase,
  34     TestCase,
  35     )
  36
  37
  38 class VersionTest(TestCase):
  39
  40     def test_version_length(self):
  41         self.assertEqual(4, len(client.version()))
  42
  43     def test_api_version_length(self):
  44         self.assertEqual(4, len(client.api_version()))
  45
  46     def test_api_version_later_same(self):
  47         self.assertTrue(client.api_version() <= client.version())
  48
  49
  50 class TestClient(SubversionTestCase):
  51
  52     def setUp(self):
  53
  54         super(TestClient, self).setUp()
  55         self.repos_url = self.make_client("d", "dc")
  56         self.client = client.Client(auth=ra.Auth([ra.get_username_provider()]))
  57
  58     def tearDown(self):
  59         del self.client
  60         super(TestClient, self).tearDown()
  61
  62     def test_add(self):
  63         self.build_tree({"dc/foo": None})
  64         self.client.add("dc/foo")
  65
  66     def test_commit(self):
  67         self.build_tree({"dc/foo": None})
  68         self.client.add("dc/foo")
  69         self.client.log_msg_func = lambda c: "Amessage"
  70         self.client.commit(["dc"])
  71         r = ra.RemoteAccess(self.repos_url)
  72         revprops = r.rev_proplist(1)
  73         self.assertEqual("Amessage", revprops["svn:log"])
  74
  75     def test_commit_start(self):
  76         self.build_tree({"dc/foo": None})
  77         self.client = client.Client(auth=ra.Auth([ra.get_username_provider()]),
  78                 log_msg_func=lambda c: "Bmessage")
  79         self.client.add("dc/foo")
  80         self.client.commit(["dc"])
  81         r = ra.RemoteAccess(self.repos_url)
  82         revprops = r.rev_proplist(1)
  83         self.assertEqual("Bmessage", revprops["svn:log"])
  84
  85     def test_mkdir(self):
  86         self.client.mkdir(["dc/foo"])
  87         self.client.mkdir("dc/bar")
  88         self.client.mkdir("dc/bla", revprops={"svn:log": "foo"})
  89
  90     def test_export(self):
  91         self.build_tree({"dc/foo": b"bla"})
  92         self.client.add("dc/foo")
  93         self.client.commit(["dc"])
  94         self.client.export(self.repos_url, "de")
  95         self.assertEqual(["foo"], os.listdir("de"))
  96
  97     def test_export_new_option(self):
  98         self.build_tree({"dc/foo": b"bla"})
  99         self.client.add("dc/foo")
 100         self.client.commit(["dc"])
 101         self.client.export(self.repos_url, "de", ignore_externals=True, ignore_keywords=True)
 102         self.assertEqual(["foo"], os.listdir("de"))
 103
 104     def test_add_recursive(self):
 105         self.build_tree({"dc/trunk/foo": b'bla', "dc/trunk": None})
 106         self.client.add("dc/trunk")
 107         if getattr(wc, "WorkingCopy", None) is None:
 108             raise SkipTest(
 109                 "subversion 1.7 API not supported for WorkingCopy yet")
 110         adm = wc.WorkingCopy(None, os.path.join(os.getcwd(), "dc"))
 111         e = adm.entry(os.path.join(os.getcwd(), "dc", "trunk"))
 112         self.assertEqual(e.kind, NODE_DIR)
 113         adm2 = wc.WorkingCopy(None, os.path.join(os.getcwd(), "dc", "trunk"))
 114         e = adm2.entry(os.path.join(os.getcwd(), "dc", "trunk", "foo"))
 115         self.assertEqual(e.kind, NODE_FILE)
 116         self.assertEqual(e.revision, 0)
 117
 118     def test_get_config(self):
 119         self.assertIsInstance(client.get_config(), client.Config)
 120         try:
 121             base_dir = tempfile.mkdtemp()
 122             base_dir_basename = os.path.basename(base_dir)
 123             svn_cfg_dir = os.path.join(base_dir, '.subversion')
 124             os.mkdir(svn_cfg_dir)
 125             with open(os.path.join(svn_cfg_dir, 'config'), 'w') as svn_cfg:
 126                 svn_cfg.write('[miscellany]\n')
 127                 svn_cfg.write('global-ignores = %s' % base_dir_basename)
 128             config = client.get_config(svn_cfg_dir)
 129             self.assertIsInstance(config, client.Config)
 130             ignores = config.get_default_ignores()
 131             self.assertTrue(base_dir_basename in ignores)
 132         finally:
 133             shutil.rmtree(base_dir)
 134
 135     def test_diff(self):
 136         r = ra.RemoteAccess(self.repos_url,
 137                 auth=ra.Auth([ra.get_username_provider()]))
 138         dc = self.get_commit_editor(self.repos_url)
 139         f = dc.add_file("foo")
 140         f.modify(b"foo1")
 141         dc.close()
 142
 143         dc = self.get_commit_editor(self.repos_url)
 144         f = dc.open_file("foo")
 145         f.modify(b"foo2")
 146         dc.close()
 147
 148         if client.api_version() < (1, 5):
 149             self.assertRaises(NotImplementedError, self.client.diff, 1, 2,
 150                 self.repos_url, self.repos_url)
 151             return # Skip test
 152
 153         (outf, errf) = self.client.diff(1, 2, self.repos_url, self.repos_url)
 154         self.addCleanup(outf.close)
 155         self.addCleanup(errf.close)
 156         self.assertEqual(b"""Index: foo
 157 ===================================================================
 158 --- foo\t(revision 1)
 159 +++ foo\t(revision 2)
 160 @@ -1 +1 @@
 161 -foo1
 162 \\ No newline at end of file
 163 +foo2
 164 \\ No newline at end of file
 165 """.splitlines(), outf.read().splitlines())
 166         self.assertEqual(b"", errf.read())
 167
 168     def assertCatEquals(self, value, revision=None):
 169         io = BytesIO()
 170         self.client.cat("dc/foo", io, revision)
 171         self.assertEqual(value, io.getvalue())
 172
 173     def test_cat(self):
 174         self.build_tree({"dc/foo": b"bla"})
 175         self.client.add("dc/foo")
 176         self.client.log_msg_func = lambda c: "Commit"
 177         self.client.commit(["dc"])
 178         self.assertCatEquals(b"bla")
 179         self.build_tree({"dc/foo": b"blabla"})
 180         self.client.commit(["dc"])
 181         self.assertCatEquals(b"blabla")
 182         self.assertCatEquals(b"bla", revision=1)
 183         self.assertCatEquals(b"blabla", revision=2)
 184
 185     def assertLogEntryChangedPathsEquals(self, expected, entry):
 186         changed_paths = entry["changed_paths"]
 187         self.assertIsInstance(changed_paths, dict)
 188         self.assertEqual(sorted(expected), sorted(changed_paths.keys()))
 189
 190     def assertLogEntryMessageEquals(self, expected, entry):
 191         self.assertEqual(expected, entry["revprops"]["svn:log"])
 192
 193     def assertLogEntryDateAlmostEquals(self, expected, entry, delta):
 194         actual = datetime.strptime(entry["revprops"]["svn:date"], "%Y-%m-%dT%H:%M:%S.%fZ")
 195         self.assertTrue((actual - expected) < delta)
 196
 197     def test_log(self):
 198         log_entries = []
 199         commit_msg_1 = "Commit"
 200         commit_msg_2 = "Commit 2"
 201         delta = timedelta(hours=1)
 202         def cb(changed_paths, revision, revprops, has_children=False):
 203             log_entries.append({
 204                 'changed_paths': changed_paths,
 205                 'revision': revision,
 206                 'revprops': revprops,
 207                 'has_children': has_children,
 208             })
 209         self.build_tree({"dc/foo": b"bla"})
 210         self.client.add("dc/foo")
 211         self.client.log_msg_func = lambda c: commit_msg_1
 212         self.client.commit(["dc"])
 213         commit_1_dt = datetime.utcnow()
 214         self.client.log(cb, "dc/foo")
 215         self.assertEqual(1, len(log_entries))
 216         self.assertEqual(None, log_entries[0]["changed_paths"])
 217         self.assertEqual(1, log_entries[0]["revision"])
 218         self.assertLogEntryMessageEquals(commit_msg_1, log_entries[0])
 219         self.assertLogEntryDateAlmostEquals(commit_1_dt, log_entries[0], delta)
 220         self.build_tree({
 221             "dc/foo": b"blabla",
 222             "dc/bar": b"blablabla",
 223         })
 224         self.client.add("dc/bar")
 225         self.client.log_msg_func = lambda c: commit_msg_2
 226         self.client.commit(["dc"])
 227         commit_2_dt = datetime.utcnow()
 228         log_entries = []
 229         self.client.log(cb, "dc/foo", discover_changed_paths=True)
 230         self.assertEqual(2, len(log_entries))
 231         self.assertLogEntryChangedPathsEquals(["/foo", "/bar"], log_entries[0])
 232         self.assertEqual(2, log_entries[0]["revision"])
 233         self.assertLogEntryMessageEquals(commit_msg_2, log_entries[0])
 234         self.assertLogEntryDateAlmostEquals(commit_2_dt, log_entries[0], delta)
 235         self.assertLogEntryChangedPathsEquals(["/foo"], log_entries[1])
 236         self.assertEqual(1, log_entries[1]["revision"])
 237         self.assertLogEntryMessageEquals(commit_msg_1, log_entries[1])
 238         self.assertLogEntryDateAlmostEquals(commit_1_dt, log_entries[1], delta)
 239         log_entries = []
 240         self.client.log(cb, "dc/foo", start_rev=2, end_rev=2, discover_changed_paths=True)
 241         self.assertEqual(1, len(log_entries))
 242         self.assertLogEntryChangedPathsEquals(["/foo", "/bar"], log_entries[0])
 243         self.assertEqual(2, log_entries[0]["revision"])
 244         self.assertLogEntryMessageEquals(commit_msg_2, log_entries[0])
 245         self.assertLogEntryDateAlmostEquals(commit_2_dt, log_entries[0], delta)
 246
 247     def test_info(self):
 248         self.build_tree({"dc/foo": b"bla"})
 249         self.client.add("dc/foo")
 250         self.client.log_msg_func = lambda c: "Commit"
 251         self.client.commit(["dc"])
 252         info = self.client.info("dc/foo")
 253         self.assertEqual(["foo"], list(info.keys()))
 254         self.assertEqual(1, info["foo"].revision)
 255         self.assertEqual(3, info["foo"].size)
 256         self.build_tree({"dc/bar": b"blablabla"})
 257         self.client.add(os.path.abspath("dc/bar"))
 258
 259     def test_info_nonexistant(self):
 260         self.build_tree({"dc/foo": b"bla"})
 261         self.client.add("dc/foo")
 262         self.client.log_msg_func = lambda c: "Commit"
 263         self.client.commit(["dc"])
 264         self.assertRaises(SubversionException, self.client.info, "dc/missing")