#!/usr/bin/env python
#    Licensed to the Apache Software Foundation (ASF) under one
#    or more contributor license agreements.  See the NOTICE file
#    distributed with this work for additional information
#    regarding copyright ownership.  The ASF licenses this file
#    to you under the Apache License, Version 2.0 (the
#    "License"); you may not use this file except in compliance
#    with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing,
#    software distributed under the License is distributed on an
#    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#    KIND, either express or implied.  See the License for the
#    specific language governing permissions and limitations
#    under the License.

import setup_path
import unittest
import locale
import os
import shutil
import tempfile
from sys import version_info # For Python version check
if version_info[0] >= 3:
  # Python >=3.0
  from io import StringIO
else:
  # Python <3.0
  from StringIO import StringIO
from csvn.core import *
from urllib import pathname2url
from csvn.wc import WC
from csvn.repos import LocalRepository

locale.setlocale(locale.LC_ALL, "C")

repos_location = os.path.join(tempfile.gettempdir(), "svn_test_repos")
wc_location = os.path.join(tempfile.gettempdir(), "svn_test_wc")
repo_url = pathname2url(repos_location)
if repo_url.startswith("///"):
  # Don't add extra slashes if they're already present.
  # (This is important for Windows compatibility).
  repo_url = "file:" + repo_url
else:
  # If the URL simply starts with '/', we need to add two
  # extra slashes to make it a valid 'file://' URL
  repo_url = "file://" + repo_url

if os.sep != "/":
    repos_location = repos_location.replace(os.sep, "/")
    wc_location = wc_location.replace(os.sep, "/")

class WCTestCase(unittest.TestCase):
    """Test case for Subversion WC layer."""

    def setUp(self):
        dumpfile = open(os.path.join(os.path.split(__file__)[0],
                        'test.dumpfile'))

        # Just in case a previous test instance was not properly cleaned up
        self.remove_from_disk()

        self.repos = LocalRepository(repos_location, create=True)
        self.repos.load(dumpfile)

        self.wc = WC(wc_location)
        self.wc.checkout(repo_url)

    def tearDown(self):
        self.repos.close()
        self.wc.close()
        self.remove_from_disk()
        self.wc = None

    def remove_from_disk(self):
        """Remove anything left on disk"""
        pool = Pool()
        if os.path.exists(wc_location):
            svn_io_remove_dir(wc_location, pool)
        if os.path.exists(repos_location):
            svn_repos_delete(repos_location, pool)

    def _info_receiver(self, path, info):
        self.last_info = info

    def test_info(self):
        self.wc.info(path="trunk/README.txt",info_func=self._info_receiver)
        self.assertEqual(9, self.last_info.rev)
        self.assertEqual(svn_node_file, self.last_info.kind)
        self.assertEqual(repo_url, self.last_info.repos_root_URL)
        self.assertEqual("890f2569-e600-4cfc-842a-f574dec58d87",
            self.last_info.repos_UUID)
        self.assertEqual(9, self.last_info.last_changed_rev)
        self.assertEqual("bruce", self.last_info.last_changed_author)
        self.assertEqual(-1, self.last_info.copyfrom_rev)

    def test_copy(self):
        self.wc.copy("trunk/README.txt", "trunk/DONTREADME.txt")
        self.wc.info(path="trunk/DONTREADME.txt",
            info_func=self._info_receiver)
        self.assertEqual(svn_wc_schedule_add, self.last_info.schedule)
        self.wc.info(path="trunk/README.txt")
        self.assertEqual(svn_wc_schedule_normal, self.last_info.schedule)

    def test_move(self):
        self.wc.move("trunk/README.txt", "trunk/DONTREADMEEITHER.txt")
        self.wc.info(path="trunk/DONTREADMEEITHER.txt",
            info_func=self._info_receiver)
        self.assertEqual(svn_wc_schedule_add, self.last_info.schedule)
        self.wc.info(path="trunk/README.txt")
        self.assertEqual(svn_wc_schedule_delete, self.last_info.schedule)

    def test_delete(self):
        self.wc.delete(["trunk/README.txt"])
        self.wc.info(path="trunk/README.txt",
            info_func=self._info_receiver)
        self.assertEqual(svn_wc_schedule_delete, self.last_info.schedule)

    def test_mkdir(self):
        self.wc.mkdir(["trunk/plank"])
        self.wc.info(path="trunk/plank",
            info_func=self._info_receiver)
        self.assertEqual(svn_wc_schedule_add, self.last_info.schedule)

    def test_add(self):
        f = open("%s/trunk/ADDED.txt" % wc_location, "w")
        f.write("Something")
        f.close()

        self.wc.add("trunk/ADDED.txt")
        self.wc.info(path="trunk/ADDED.txt",
            info_func=self._info_receiver)
        self.assertEqual(svn_wc_schedule_add, self.last_info.schedule)

    def test_revert(self):
        self.wc.revert([""],True)
        self.wc.info(path="trunk/README.txt",
            info_func=self._info_receiver)
        self.assertEqual(svn_wc_schedule_normal, self.last_info.schedule)

    def test_diff(self):
        path = "%s/trunk/README.txt" % wc_location

        diffstring="""Index: """+path+"""
===================================================================
--- """+path+"""\t(revision 9)
+++ """+path+"""\t(working copy)
@@ -1,7 +0,0 @@
-This repository is for test purposes only. Any resemblance to any other
-repository, real or imagined, is purely coincidental.
-
-Contributors:
-Clark
-Bruce
-Henry
"""
        f = open(path, "w")
        f.truncate(0)
        f.close()
        difffile = StringIO()
        self.wc.diff("trunk", outfile=difffile)
        difffile.seek(0)
        diffresult = difffile.read().replace("\r","")
        self.assertEqual(diffstring, diffresult)

        path = "%s/branches/0.x/README.txt" % wc_location
        diffstring="""Index: """+path+"""
===================================================================
--- """+path+"""\t(nonexistent)
+++ """+path+"""\t(revision 5)
@@ -0,0 +1,9 @@
+This repository is for test purposes only. Any resemblance to any other
+repository, real or imagined, is purely coincidental.
+
+This branch preserves and refines the code of the excellent pre-1.0 days.
+
+Contributors:
+Clark
+Bruce
+Henry
"""
        difffile.seek(0)
        self.wc.diff(revnum1=4, revnum2=5, outfile=difffile)
        difffile.seek(0)
        diffresult = difffile.read().replace("\r","")
        self.assertEqual(diffstring, diffresult)


    def test_export(self):
        export_location = os.path.join(tempfile.gettempdir(), "svn_export")
        self.wc.export("", export_location)
        if not os.path.exists(export_location):
            self.fail("Export directory does not exist")
        else:
            shutil.rmtree(export_location)

    def test_propget(self):
        props = self.wc.propget("Awesome")
        path = "%s/trunk/README.txt" % wc_location
        if not path in props.keys():
            self.fail("File missing in propget")

    def test_propset(self):
        self.wc.propset("testprop", "testval", "branches/0.x/README.txt")
        props = self.wc.propget("testprop", "branches/0.x/README.txt")
        if not "%s/branches/0.x/README.txt" % wc_location in \
                props.keys():

            self.fail("Property not set")

    def test_update(self):
        path = "trunk/README.txt"
        results = self.wc.update([path], revnum=7)
        self.assertEqual(results[0], 7)
        props = self.wc.propget("Awesome")
        if "%s/%s" % (wc_location, path) in \
                props.keys():
            self.fail("File not updated to old revision")
        results = self.wc.update([path])
        self.assertEqual(results[0], 9)
        self.test_propget()

    def test_switch(self):
        self.wc.switch("trunk", "%s/tags" % repo_url)
        if os.path.exists("%s/trunk/README.txt" % wc_location):
            self.fail("Switch did not happen")

    def test_lock(self):
        self.wc.lock(["%s/trunk/README.txt" % wc_location],
                        "Test lock")
        self.wc.info(path="trunk/README.txt",
            info_func=self._info_receiver)
        if not self.last_info.lock:
            self.fail("Lock not acquired")

    def test_unlock(self):
        path = "%s/trunk/README.txt" % wc_location
        self.wc.lock([path], "Test lock")
        self.wc.info(path=path,
            info_func=self._info_receiver)
        if not self.last_info.lock:
            self.fail("Lock not acquired")
        self.wc.unlock([path])

        self.wc.info(path="trunk/README.txt",
            info_func=self._info_receiver)
        if self.last_info.lock:
            self.fail("Lock not released")


def suite():
    return unittest.makeSuite(WCTestCase, 'test')

if __name__ == '__main__':
    runner = unittest.TextTestRunner()
    runner.run(suite())
