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