1#!/usr/bin/env python 2# Licensed to the Apache Software Foundation (ASF) under one 3# or more contributor license agreements. See the NOTICE file 4# distributed with this work for additional information 5# regarding copyright ownership. The ASF licenses this file 6# to you under the Apache License, Version 2.0 (the 7# "License"); you may not use this file except in compliance 8# with the License. You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, 13# software distributed under the License is distributed on an 14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 15# KIND, either express or implied. See the License for the 16# specific language governing permissions and limitations 17# under the License. 18 19import setup_path 20import unittest 21import locale 22import os 23import shutil 24import tempfile 25from sys import version_info # For Python version check 26if version_info[0] >= 3: 27 # Python >=3.0 28 from io import StringIO 29else: 30 # Python <3.0 31 from StringIO import StringIO 32from csvn.core import * 33from urllib import pathname2url 34from csvn.wc import WC 35from csvn.repos import LocalRepository 36 37locale.setlocale(locale.LC_ALL, "C") 38 39repos_location = os.path.join(tempfile.gettempdir(), "svn_test_repos") 40wc_location = os.path.join(tempfile.gettempdir(), "svn_test_wc") 41repo_url = pathname2url(repos_location) 42if repo_url.startswith("///"): 43 # Don't add extra slashes if they're already present. 44 # (This is important for Windows compatibility). 45 repo_url = "file:" + repo_url 46else: 47 # If the URL simply starts with '/', we need to add two 48 # extra slashes to make it a valid 'file://' URL 49 repo_url = "file://" + repo_url 50 51if os.sep != "/": 52 repos_location = repos_location.replace(os.sep, "/") 53 wc_location = wc_location.replace(os.sep, "/") 54 55class WCTestCase(unittest.TestCase): 56 """Test case for Subversion WC layer.""" 57 58 def setUp(self): 59 dumpfile = open(os.path.join(os.path.split(__file__)[0], 60 'test.dumpfile')) 61 62 # Just in case a previous test instance was not properly cleaned up 63 self.remove_from_disk() 64 65 self.repos = LocalRepository(repos_location, create=True) 66 self.repos.load(dumpfile) 67 68 self.wc = WC(wc_location) 69 self.wc.checkout(repo_url) 70 71 def tearDown(self): 72 self.repos.close() 73 self.wc.close() 74 self.remove_from_disk() 75 self.wc = None 76 77 def remove_from_disk(self): 78 """Remove anything left on disk""" 79 pool = Pool() 80 if os.path.exists(wc_location): 81 svn_io_remove_dir(wc_location, pool) 82 if os.path.exists(repos_location): 83 svn_repos_delete(repos_location, pool) 84 85 def _info_receiver(self, path, info): 86 self.last_info = info 87 88 def test_info(self): 89 self.wc.info(path="trunk/README.txt",info_func=self._info_receiver) 90 self.assertEqual(9, self.last_info.rev) 91 self.assertEqual(svn_node_file, self.last_info.kind) 92 self.assertEqual(repo_url, self.last_info.repos_root_URL) 93 self.assertEqual("890f2569-e600-4cfc-842a-f574dec58d87", 94 self.last_info.repos_UUID) 95 self.assertEqual(9, self.last_info.last_changed_rev) 96 self.assertEqual("bruce", self.last_info.last_changed_author) 97 self.assertEqual(-1, self.last_info.copyfrom_rev) 98 99 def test_copy(self): 100 self.wc.copy("trunk/README.txt", "trunk/DONTREADME.txt") 101 self.wc.info(path="trunk/DONTREADME.txt", 102 info_func=self._info_receiver) 103 self.assertEqual(svn_wc_schedule_add, self.last_info.schedule) 104 self.wc.info(path="trunk/README.txt") 105 self.assertEqual(svn_wc_schedule_normal, self.last_info.schedule) 106 107 def test_move(self): 108 self.wc.move("trunk/README.txt", "trunk/DONTREADMEEITHER.txt") 109 self.wc.info(path="trunk/DONTREADMEEITHER.txt", 110 info_func=self._info_receiver) 111 self.assertEqual(svn_wc_schedule_add, self.last_info.schedule) 112 self.wc.info(path="trunk/README.txt") 113 self.assertEqual(svn_wc_schedule_delete, self.last_info.schedule) 114 115 def test_delete(self): 116 self.wc.delete(["trunk/README.txt"]) 117 self.wc.info(path="trunk/README.txt", 118 info_func=self._info_receiver) 119 self.assertEqual(svn_wc_schedule_delete, self.last_info.schedule) 120 121 def test_mkdir(self): 122 self.wc.mkdir(["trunk/plank"]) 123 self.wc.info(path="trunk/plank", 124 info_func=self._info_receiver) 125 self.assertEqual(svn_wc_schedule_add, self.last_info.schedule) 126 127 def test_add(self): 128 f = open("%s/trunk/ADDED.txt" % wc_location, "w") 129 f.write("Something") 130 f.close() 131 132 self.wc.add("trunk/ADDED.txt") 133 self.wc.info(path="trunk/ADDED.txt", 134 info_func=self._info_receiver) 135 self.assertEqual(svn_wc_schedule_add, self.last_info.schedule) 136 137 def test_revert(self): 138 self.wc.revert([""],True) 139 self.wc.info(path="trunk/README.txt", 140 info_func=self._info_receiver) 141 self.assertEqual(svn_wc_schedule_normal, self.last_info.schedule) 142 143 def test_diff(self): 144 path = "%s/trunk/README.txt" % wc_location 145 146 diffstring="""Index: """+path+""" 147=================================================================== 148--- """+path+"""\t(revision 9) 149+++ """+path+"""\t(working copy) 150@@ -1,7 +0,0 @@ 151-This repository is for test purposes only. Any resemblance to any other 152-repository, real or imagined, is purely coincidental. 153- 154-Contributors: 155-Clark 156-Bruce 157-Henry 158""" 159 f = open(path, "w") 160 f.truncate(0) 161 f.close() 162 difffile = StringIO() 163 self.wc.diff("trunk", outfile=difffile) 164 difffile.seek(0) 165 diffresult = difffile.read().replace("\r","") 166 self.assertEqual(diffstring, diffresult) 167 168 path = "%s/branches/0.x/README.txt" % wc_location 169 diffstring="""Index: """+path+""" 170=================================================================== 171--- """+path+"""\t(nonexistent) 172+++ """+path+"""\t(revision 5) 173@@ -0,0 +1,9 @@ 174+This repository is for test purposes only. Any resemblance to any other 175+repository, real or imagined, is purely coincidental. 176+ 177+This branch preserves and refines the code of the excellent pre-1.0 days. 178+ 179+Contributors: 180+Clark 181+Bruce 182+Henry 183""" 184 difffile.seek(0) 185 self.wc.diff(revnum1=4, revnum2=5, outfile=difffile) 186 difffile.seek(0) 187 diffresult = difffile.read().replace("\r","") 188 self.assertEqual(diffstring, diffresult) 189 190 191 def test_export(self): 192 export_location = os.path.join(tempfile.gettempdir(), "svn_export") 193 self.wc.export("", export_location) 194 if not os.path.exists(export_location): 195 self.fail("Export directory does not exist") 196 else: 197 shutil.rmtree(export_location) 198 199 def test_propget(self): 200 props = self.wc.propget("Awesome") 201 path = "%s/trunk/README.txt" % wc_location 202 if not path in props.keys(): 203 self.fail("File missing in propget") 204 205 def test_propset(self): 206 self.wc.propset("testprop", "testval", "branches/0.x/README.txt") 207 props = self.wc.propget("testprop", "branches/0.x/README.txt") 208 if not "%s/branches/0.x/README.txt" % wc_location in \ 209 props.keys(): 210 211 self.fail("Property not set") 212 213 def test_update(self): 214 path = "trunk/README.txt" 215 results = self.wc.update([path], revnum=7) 216 self.assertEqual(results[0], 7) 217 props = self.wc.propget("Awesome") 218 if "%s/%s" % (wc_location, path) in \ 219 props.keys(): 220 self.fail("File not updated to old revision") 221 results = self.wc.update([path]) 222 self.assertEqual(results[0], 9) 223 self.test_propget() 224 225 def test_switch(self): 226 self.wc.switch("trunk", "%s/tags" % repo_url) 227 if os.path.exists("%s/trunk/README.txt" % wc_location): 228 self.fail("Switch did not happen") 229 230 def test_lock(self): 231 self.wc.lock(["%s/trunk/README.txt" % wc_location], 232 "Test lock") 233 self.wc.info(path="trunk/README.txt", 234 info_func=self._info_receiver) 235 if not self.last_info.lock: 236 self.fail("Lock not acquired") 237 238 def test_unlock(self): 239 path = "%s/trunk/README.txt" % wc_location 240 self.wc.lock([path], "Test lock") 241 self.wc.info(path=path, 242 info_func=self._info_receiver) 243 if not self.last_info.lock: 244 self.fail("Lock not acquired") 245 self.wc.unlock([path]) 246 247 self.wc.info(path="trunk/README.txt", 248 info_func=self._info_receiver) 249 if self.last_info.lock: 250 self.fail("Lock not released") 251 252 253def suite(): 254 return unittest.makeSuite(WCTestCase, 'test') 255 256if __name__ == '__main__': 257 runner = unittest.TextTestRunner() 258 runner.run(suite()) 259