#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2005, Giovanni Bajo
# All rights reserved.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
import sys, os
import types
import re
import unittest
from cStringIO import StringIO
import shutil
import svnmerge
import stat
import atexit
import getopt

####
# IMPORTANT NOTE TO TEST AUTHORS
#
# Any quoted strings inside the arguments of the parameter "cmd" must
# be enclosed in double-, not single-quotes, so that the command parser
# knows to keep them together. For example, do not write this:
#     launch("svn ci -m 'log comment'") # BAD
# ...but one of these:
#     launch('svn ci -m "log comment"') # GOOD
#     launch("svn ci -m \"log comment\"") # GOOD, but why?
# Otherwise, you get an error saying
#     '<path>/comment' is not under version control
# ...when running the tests on Windows.
####

# True/False constants are Python 2.2+
try:
    True, False
except NameError:
    True, False = 1, 0

class TestCase_kwextract(unittest.TestCase):
    def test_basic(self):
        self.assertEqual(svnmerge.kwextract("$Rev: 134 rasky $"), "134 rasky")
        self.assertEqual(svnmerge.kwextract("$Date: 2005-09-25 13:45 CET+1$"),
                         "2005-09-25 13:45 CET+1")

    def test_failure(self):
        self.assertEqual(svnmerge.kwextract("$Rev: $"), "<unknown>")
        self.assertEqual(svnmerge.kwextract("$Date$"), "<unknown>")

class TestCase_launch(unittest.TestCase):
    if os.name == "nt":
        cmd = "attrib"
    else:
        cmd = "ls"

    def test_basic(self):
        out = svnmerge.launch(self.cmd)
        self.assert_(out)
        for o in out:
            self.assertEqual(o[-1], "\n")

    def test_failure(self):
        self.assertRaises(svnmerge.LaunchError, svnmerge.launch, self.cmd*10)

    def test_failurecode(self):
        try:
            svnmerge.launch(self.cmd*10)
        except svnmerge.LaunchError, (ret, cmd, out):
            self.assertNotEqual(ret, 0)
            self.assertNotEqual(ret, None)
            self.assert_(out)
            self.assertEqual(cmd, self.cmd*10)
        else:
            self.fail("svnmerge.launch did not cause a LaunchError as expected")

class TestCase_PrefixLines(unittest.TestCase):
    def test_basic(self):
        self.assertEqual("zz\n", svnmerge.prefix_lines("zz", "\n"))
        self.assertEqual("zzfoo\n", svnmerge.prefix_lines("zz", "foo\n"))
        self.assertEqual("zzfoo\nzzbar\n", svnmerge.prefix_lines("zz", "foo\nbar\n"))
        self.assertEqual("zz\nzzfoo\n", svnmerge.prefix_lines("zz", "\nfoo\n"))
        self.assertEqual("zz\nzzfoo\nzzbar\n", svnmerge.prefix_lines("zz", "\nfoo\nbar\n"))

class TestCase_RevisionSet(unittest.TestCase):
    def test_constr_string(self):
        rs = svnmerge.RevisionSet("10- 15, 12-48,2 ")
        self.assert_(17 in rs)
        self.assert_(2 in rs)
        self.assert_(9 not in rs)

        rs = svnmerge.RevisionSet("10: 15, 12:48,2 ")
        self.assert_(17 in rs)
        self.assert_(2 in rs)
        self.assert_(9 not in rs)

    def test_constr_dict(self):
        rs = svnmerge.RevisionSet({18:1, 24:1, 25:1, 43:1})
        self.assert_(24 in rs)
        self.assert_(18 in rs)
        self.assert_(44 not in rs)

    def test_constr_error(self):
        self.assertRaises(ValueError, svnmerge.RevisionSet, "10-12-15")
        self.assertRaises(ValueError, svnmerge.RevisionSet, "10;12-15")
        self.assertRaises(ValueError, svnmerge.RevisionSet, "10,foo,3-15")

        self.assertRaises(ValueError, svnmerge.RevisionSet, "10:12:15")
        self.assertRaises(ValueError, svnmerge.RevisionSet, "10;12:15")
        self.assertRaises(ValueError, svnmerge.RevisionSet, "10,foo,3:15")

    def test_normalized(self):
        rs = svnmerge.RevisionSet("8-15,16-18, 4-6, 9, 18, 1-1, 3-3")
        self.assertEqual(rs.normalized(), [(1,1), (3,6), (8,18)])
        self.assertEqual(str(rs), "1,3-6,8-18")

        rs = svnmerge.RevisionSet("8:15,16:18, 4:6, 9, 18, 1:1, 3:3")
        self.assertEqual(rs.normalized(), [(1,1), (3,6), (8,18)])
        self.assertEqual(str(rs), "1,3-6,8-18")

    def test_sorted(self):
        "Test the sorted() function of the RevisionSet class."
        rs = svnmerge.RevisionSet("8-15,16-18, 4-6, 9, 18, 1-1, 3-3")
        self.assertEqual(rs.sorted(), [1, 3, 4, 5, 6, 8, 9, 10, 11,
                                       12, 13, 14, 15, 16, 17, 18])

        rs = svnmerge.RevisionSet("8:15,16:18, 4:6, 9, 18, 1:1, 3:3")
        self.assertEqual(rs.sorted(), [1, 3, 4, 5, 6, 8, 9, 10, 11,
                                       12, 13, 14, 15, 16, 17, 18])

    def test_length(self):
        rs = svnmerge.RevisionSet("3-8")
        self.assertEqual(len(rs), 6)
        rs = svnmerge.RevisionSet("3-8,4-10")
        self.assertEqual(len(rs), 8)
        rs = svnmerge.RevisionSet("1,3,5")
        self.assertEqual(len(rs), 3)

        rs = svnmerge.RevisionSet("3:8")
        self.assertEqual(len(rs), 6)
        rs = svnmerge.RevisionSet("3:8,4:10")
        self.assertEqual(len(rs), 8)
        rs = svnmerge.RevisionSet("1,3,5")
        self.assertEqual(len(rs), 3)

    def test_iter(self):
        try:
            iter
        except NameError:
            pass
        else:
            rs = svnmerge.RevisionSet("4-13,1-5,34,20-22,18-21")
            self.assertEqual(list(iter(rs)), range(1,14)+range(18,23)+[34])

            rs = svnmerge.RevisionSet("4:13,1:5,34,20:22,18:21")
            self.assertEqual(list(iter(rs)), range(1,14)+range(18,23)+[34])

    def test_union(self):
        rs = svnmerge.RevisionSet("3-8,4-10") | svnmerge.RevisionSet("7-14,1")
        self.assertEqual(str(rs), "1,3-14")

        rs = svnmerge.RevisionSet("3:8,4:10") | svnmerge.RevisionSet("7:14,1")
        self.assertEqual(str(rs), "1,3-14")

    def test_subtraction(self):
        rs = svnmerge.RevisionSet("3-8,4-10") - svnmerge.RevisionSet("7-14,1")
        self.assertEqual(str(rs), "3-6")

        rs = svnmerge.RevisionSet("3:8,4:10") - svnmerge.RevisionSet("7:14,1")
        self.assertEqual(str(rs), "3-6")

    def test_constr_empty(self):
        rs = svnmerge.RevisionSet("")
        self.assertEqual(str(rs), "")

class TestCase_MinimalMergeIntervals(unittest.TestCase):
    def test_basic(self):
        rs = svnmerge.RevisionSet("4-8,12,18,24")
        phantom = svnmerge.RevisionSet("8-11,13-16,19-23")
        revs = svnmerge.minimal_merge_intervals(rs, phantom)
        self.assertEqual(revs, [(4,12), (18,24)])

class TestCase_SvnMerge(unittest.TestCase):
    def svnmerge(self, cmds, *args, **kwargs):
        return self.svnmerge2(cmds.split(), *args, **kwargs)

    def svnmerge2(self, args, error=False, match=None, nonmatch=None):
        out = StringIO()
        sys.stdout = sys.stderr = out
        try:
            try:
                # Clear svnmerge's internal cache before running any
                # commands.
                svnmerge._cache_svninfo = {}
                svnmerge._cache_reporoot = {}

                ret = svnmerge.main(args)
            except SystemExit, e:
                ret = e.code
        finally:
            sys.stdout = sys.__stdout__
            sys.stderr = sys.__stderr__

        if ret is None:
            ret = 0

        if error:
            self.assertNotEqual(ret, 0,
                "svnmerge did not fail, with this output:\n%s" % out.getvalue())
        else:
            self.assertEqual(ret, 0,
                "svnmerge failed, with this output:\n%s" % out.getvalue())

        if match is not None:
            self.assert_(re.search(match, out.getvalue()),
                "pattern %r not found in output:\n%s" % (match, out.getvalue()))
        if nonmatch is not None:
            self.assert_(not re.search(nonmatch, out.getvalue()),
                "pattern %r found in output:\n%s" % (nonmatch, out.getvalue()))

        return out.getvalue()

    def _parseoutput(self, ret, out, error=False, match=None, nonmatch=None):
        if error:
            self.assertNotEqual(ret,
                                0,
                                "svnmerge did not fail, with this output:\n%s" % out)
        else:
            self.assertEqual(ret,
                             0,
                             "svnmerge failed, with this output:\n%s" % out)

        if match is not None:
            self.assert_(re.search(match, out),
                         "pattern %r not found in output:\n%s" % (match, out))

        if nonmatch is not None:
            self.assert_(not re.search(nonmatch, out),
                         "pattern %r found in output:\n%s" % (nonmatch, out))

        return out

    def launch(self, cmd, **kwargs):
        try:
            out = svnmerge.launch(cmd, split_lines=False)
        except svnmerge.LaunchError, (ret, cmd, out):
            return self._parseoutput(ret, out, **kwargs)
        return self._parseoutput(0, out, **kwargs)

class TestCase_CommandLineOptions(TestCase_SvnMerge):
    def test_empty(self):
        self.svnmerge("")

    def test_help_commands(self):
        self.svnmerge("help")
        self.svnmerge("--help")
        self.svnmerge("-h")
        for cmd in svnmerge.command_table.keys():
            self.svnmerge("help %s" % cmd)
            self.svnmerge("%s --help" % cmd)
            self.svnmerge("%s -h" % cmd)

    def test_wrong_commands(self):
        self.svnmerge("asijdoiasjd", error=True)
        self.svnmerge("help asijdoiasjd", error=True)

    def test_wrong_option(self):
        self.svnmerge("--asdsad", error=True)
        self.svnmerge("help --asdsad", error=True)
        self.svnmerge("init --asdsad", error=True)
        self.svnmerge("--asdsad init", error=True)

    def test_version(self):
        out = self.svnmerge("--version")
        self.assert_(out.find("Giovanni Bajo") >= 0)
        out = self.svnmerge("-V")
        self.assert_(out.find("Giovanni Bajo") >= 0)
        out = self.svnmerge("init -V")
        self.assert_(out.find("Giovanni Bajo") >= 0)

    def testOptionOrder(self):
        """Make sure you can intermix command name, arguments and
        options in any order."""
        self.svnmerge("--log avail",
                      error=True,
                      match=r"no integration info")  # accepted
        self.svnmerge("-l avail",
                      error=True,
                      match=r"no integration info")  # accepted
        self.svnmerge("-r123 merge",
                      error=True,
                      match=r"no integration info")  # accepted
        self.svnmerge("-s -v -r92481 merge",
                      error=True,
                      match=r"no integration info")  # accepted
        self.svnmerge("--log merge",
                      error=True,
                      match=r"option --log not recognized")
        self.svnmerge("--diff foobar", error=True, match=r"foobar")

        # This requires gnu_getopt support to be parsed
        if hasattr(getopt, "gnu_getopt"):
            self.svnmerge("-r123 merge . --log",
                          error=True,
                          match=r"option --log not recognized")

def temp_path():
    try:
        return os.environ["TEMP"]
    except KeyError:
        pass
    if os.name == "posix":
        return "/tmp"
    return "."

def rmtree(path):
    def onerror(func, path, excinfo):
        if func in [os.remove, os.rmdir]:
            if os.path.exists(path):
                os.chmod(path, stat.S_IWRITE)
                func(path)

    if os.path.isdir(path):
        shutil.rmtree(path, onerror=onerror)

def get_template_path():
    p = os.path.join(temp_path(), "__svnmerge_test_template")
    return os.path.abspath(p)

def get_test_path():
    p = os.path.join(temp_path(), "__svnmerge_test")
    return os.path.abspath(p)

def abspath_to_url(path):
    assert path == os.path.abspath(path)
    path = path.replace("\\", "/")
    if path[0] != '/':
        path = '/' + path
    return "file://" + path

class TestCase_TestRepo(TestCase_SvnMerge):
    def setUp(self):
        """Creates a working copy of a branch at r13 with the
        following structure, containing revisions (3-6, 13):

          test-branch/
           test1
           test2
           test3

        ...from a repository with the following structure:

          Path                        Created rev
          ----                        -----------
          /                           0
           trunk/                     3
            test1                     4
            test2                     5
            test3                     6
            test4                     9
            test5                     10
           branches/                  1
            testYYY-branch/           11 (renamed from testXXX-branch in 12)
             test1                    4
             test2                    5
             test3                    6
            test-branch/              13 (copied from trunk@6)
             test1                    4
             test2                    5
             test3                    6
           tags/                      2
        """
        self.cwd = os.getcwd()

        test_path = get_test_path()
        template_path = get_template_path()

        self.template_path = template_path
        self.test_path = test_path

        self.template_repo_path = os.path.join(template_path, "repo")
        self.template_repo_url = abspath_to_url(self.template_repo_path)

        self.test_repo_path = os.path.join(test_path, "repo")
        self.test_repo_url = abspath_to_url(self.test_repo_path)

        if not os.path.isdir(template_path):
            rmtree(template_path)
            os.makedirs(template_path)
            os.chdir(template_path)

            self.multilaunch("""
                svnadmin create --fs-type fsfs %(TEMPLATE_REPO_PATH)s
                svn mkdir -m "create /branches" %(TEMPLATE_REPO_URL)s/branches
                svn mkdir -m "create /tags" %(TEMPLATE_REPO_URL)s/tags
                svn mkdir -m "create /trunk" %(TEMPLATE_REPO_URL)s/trunk
                svn co %(TEMPLATE_REPO_URL)s/trunk trunk
            """)

            os.chdir("trunk")
            open("test1", "w").write("test 1")
            open("test2", "w").write("test 2")
            open("test3", "w").write("test 3")
            open("test4", "w").write("test 4")
            open("test5", "w").write("test 5")

            self.multilaunch("""
                svn add test1
                svn ci -m "add test1"
                svn add test2
                svn ci -m "add test2"
                svn add test3
                svn ci -m "add test3"
                svn mkdir -m "create /foobar" %(TEMPLATE_REPO_URL)s/foobar
                svn rm -m "remove /foobar" %(TEMPLATE_REPO_URL)s/foobar
                svn add test4
                svn ci -m "add test4"
                svn add test5
                svn ci -m "add test5"
                svn cp -r6 -m "create branch" %(TEMPLATE_REPO_URL)s/trunk %(TEMPLATE_REPO_URL)s/branches/testXXX-branch
                svn mv -m "rename branch" %(TEMPLATE_REPO_URL)s/branches/testXXX-branch %(TEMPLATE_REPO_URL)s/branches/testYYY-branch
                svn cp -r6 -m "create branch" %(TEMPLATE_REPO_URL)s/trunk %(TEMPLATE_REPO_URL)s/branches/test-branch
            """)

            os.chdir("..")

            self.launch("svn co %(TEMPLATE_REPO_URL)s/branches/test-branch")

        os.chdir(self.cwd)

        rmtree(self.test_path)
        shutil.copytree(self.template_path, self.test_path)
        os.chdir(self.test_path)

        # Relocate the test working copies from using the template
        # repository to the test repository so the template repository
        # is not affected by commits.
        self.launch("svn switch --relocate %(TEMPLATE_REPO_URL)s %(TEST_REPO_URL)s trunk test-branch")

        os.chdir("test-branch")

        # Always remove the template directory when the tests have
        # completed.
        atexit.register(lambda: rmtree(template_path))

    def tearDown(self):
        os.chdir(self.cwd)
        rmtree(self.test_path)

    def command_dict(self):
        return {
            "TEMPLATE_PATH": self.template_path,
            "TEMPLATE_REPO_PATH": self.template_repo_path,
            "TEMPLATE_REPO_URL": self.template_repo_url,
            "TEST_PATH": self.test_path,
            "TEST_REPO_PATH": self.test_repo_path,
            "TEST_REPO_URL": self.test_repo_url,
        }

    def launch(self, cmd, *args, **kwargs):
        cmd = cmd % self.command_dict()
        return TestCase_SvnMerge.launch(self, cmd, *args, **kwargs)

    def multilaunch(self, cmds):
        for cmd in cmds.split("\n"):
            cmd = cmd.strip()
            if len(cmd) > 0:
                svnmerge.launch(cmd % self.command_dict())

    def revert(self):
        self.multilaunch("svn revert -R .")

    def getproperty(self):
        out = svnmerge.launch("svn pg %s ." % svnmerge.opts["prop"])
        if len(out) == 0:
            return None
        else:
            return out[0].strip()

    def getBlockedProperty(self):
        out = svnmerge.launch("svn pg %s ." % svnmerge.opts["block-prop"])
        if len(out) == 0:
            return None
        else:
            return out[0].strip()

    def testNoWc(self):
        os.mkdir("foo")
        os.chdir("foo")
        self.svnmerge("init", error=True, match=r"working dir")
        self.svnmerge("avail", error=True, match=r"working dir")
        self.svnmerge("integrated", error=True, match=r"working dir")
        self.svnmerge("merge", error=True, match=r"working dir")
        self.svnmerge("block", error=True, match=r"working dir")
        self.svnmerge("unblock", error=True, match=r"working dir")

    def testCheckNoIntegrationInfo(self):
        self.svnmerge("avail", error=True, match=r"no integration")
        self.svnmerge("integrated", error=True, match=r"no integration")
        self.svnmerge("merge", error=True, match=r"no integration")
        self.svnmerge("block", error=True, match=r"no integration")
        self.svnmerge("unblock", error=True, match=r"no integration")

    def testSelfReferentialInit(self):
        self.svnmerge2(["init", self.test_repo_url + "/branches/test-branch"],
                       error=True, match=r"cannot init integration source")

    def testBlocked(self):

        # Initialize svnmerge
        self.svnmerge("init")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision")

        # Block revisions that have already been merged
        self.svnmerge("block -r5", error=True, match=r"no available revisions")

        # Block phantom revisions
        self.svnmerge("block -r8", error=True, match=r"no available revisions")

        # Block available revisions
        self.svnmerge("block -r9", match=r"'svnmerge-blocked' set")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision")

        # Check that the revision is still available
        self.svnmerge("avail", match=r"\A10$")

        # Check that the revision was blocked correctly
        self.svnmerge("avail -B", match=r"\A9$")

        # Check that both revisions are available with avail -A
        self.svnmerge("avail -A", match=r"\A9-10$")

        # Block all remaining revisions
        self.svnmerge("block", match=r"'svnmerge-blocked' set")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision")

        # Check that all revisions were blocked correctly
        self.svnmerge("avail -B", match=r"\A9-10$")

        # Check that all revisions are available using avail -A
        self.svnmerge("avail -A", match=r"\A9-10$")

        # Check that no revisions are available, now that they have
        # been blocked
        self.svnmerge("avail", match=r"\A\Z")

        # Unblock all revisions
        self.svnmerge("unblock", match=r"'svnmerge-blocked' deleted")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision")

        # Check that all revisions are available
        self.svnmerge("avail", match=r"\A9-10$")
        self.svnmerge("avail -A", match=r"\A9-10$")

        # Check that no revisions are blocked
        self.svnmerge("avail -B", match=r"\A$")

    def testTransitiveMerge(self):
        """
        Test a merge of a change from testYYY-branch -> test-branch -> trunk
        """
        os.chdir("..")
        self.launch("svn co %(TEST_REPO_URL)s/branches/testYYY-branch testYYY-branch")

        os.chdir("trunk")
        self.launch("svn up")
        self.svnmerge("init ../test-branch")
        self.launch('svn ci -m "init test-branch -> trunk"',
            match=r"Committed revision 14")

        os.chdir("../test-branch")
        self.svnmerge("init -r 1-6 ../testYYY-branch")
        self.launch('svn ci -m "init testYYY-branch -> test-branch"',
            match=r"Committed revision 15")

        os.chdir("../testYYY-branch")
        open("test6", "w").write("test6")
        self.launch("svn add test6")
        self.launch('svn ci -m "add test6"',
            match=r"Committed revision 16")

        os.chdir("../test-branch")
        self.svnmerge("merge -r 16")
        self.launch('svn ci -m "merge r16"',
            match=r"Committed revision 17")

        #os.chdir("../test-branch")
        open("test5", "w").write("test5")
        self.launch("svn add test5")
        self.launch('svn ci -m "add test5"',
            match=r"Committed revision 18")

        os.chdir("../trunk")
        self.svnmerge("block -r 18")
        self.launch('svn ci -m "block r18"',
            match=r"Committed revision 19")

        #os.chdir("../trunk")
        out = self.svnmerge("merge -r 17")

        self.launch('svn ci -m "merge r17 from test-branch"',
            match=r"Committed revision 20")

        p = self.getproperty()
        self.assertEqual(p, '/branches/test-branch:1-13,17')
        p = self.getBlockedProperty()
        self.assertEqual(p, '/branches/test-branch:18')

    def testTransitiveMergeWithBlock(self):
        """
        Test a merge of a change from testYYY-branch -> test-branch -> trunk
        """
        os.chdir("..")
        self.launch("svn co %(TEST_REPO_URL)s/branches/testYYY-branch testYYY-branch")

        os.chdir("trunk")
        self.launch("svn up")
        self.svnmerge("init ../test-branch")
        self.launch('svn ci -m "init test-branch -> trunk"',
            match=r"Committed revision 14")

        os.chdir("../test-branch")
        self.svnmerge("init -r 1-6 ../testYYY-branch")
        self.launch('svn ci -m "init testYYY-branch -> test-branch"',
            match=r"Committed revision 15")

        os.chdir("../testYYY-branch")
        open("test4", "w").write("test4")
        self.launch("svn add test4")
        self.launch('svn ci -m "add test4"',
            match=r"Committed revision 16")

        os.chdir("../test-branch")
        self.svnmerge("block -r 16")
        self.launch('svn ci -m "block r16"',
            match=r"Committed revision 17")

        #os.chdir("../test-branch")
        open("test5", "w").write("test5")
        self.launch("svn add test5")
        self.launch('svn ci -m "add test5"',
            match=r"Committed revision 18")

        os.chdir("../trunk")
        self.svnmerge("block -r 18")
        self.launch('svn ci -m "block r18"',
            match=r"Committed revision 19")

        #os.chdir("../trunk")
        self.svnmerge("merge -r 17")
        self.launch('svn ci -m "merge r17 from test-branch"',
            match=r"Committed revision 20")

        p = self.getproperty()
        self.assertEqual(p, '/branches/test-branch:1-13,17')
        p = self.getBlockedProperty()
        self.assertEqual(p, '/branches/test-branch:18')

    def testBasic(self):
        self.svnmerge("init")
        p = self.getproperty()
        self.assertEqual("/trunk:1-6", p)

        self.svnmerge("avail", match=r"\A9-10$")
        self.svnmerge("avail -v", match=r"phantom.*7-8")

        self.svnmerge("avail -B", match=r"\A$")
        self.svnmerge("avail -A", match=r"\A9-10$")

        self.svnmerge("avail --log", match=r"| r7.*| r8")
        self.svnmerge("avail --diff -r9", match=r"Index: test4")

        self.svnmerge("avail --log -r5", match=r"\A\Z")
        self.svnmerge("avail --diff -r5", match=r"\A\Z")

        self.svnmerge("integrated", match=r"^3-6$")
        self.svnmerge("integrated --log -r5", match=r"| r5 ")
        self.svnmerge("integrated --diff -r5", match=r"Index: test2")

    def test_log_msg_suggest(self):
        self.svnmerge("init -vf commit-log.txt", match=r"wrote commit message")
        self.assert_(os.path.exists("commit-log.txt"))
        os.remove("commit-log.txt")

    def testInitForce(self):
        open("test1", "a").write("foo")
        self.svnmerge("init", error=True, match=r"clean")
        self.svnmerge("init -F")
        p = self.getproperty()
        self.assertEqual("/trunk:1-6", p)

    def testUninit(self):
        """Test that uninit works, for both merged and blocked revisions."""
        os.chdir("..")
        self.launch("svn co %(TEST_REPO_URL)s/branches/testYYY-branch testYYY-branch")

        os.chdir("trunk")
        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 13")
        self.svnmerge2(["init", "-r1-13",
                        self.test_repo_url + "/branches/test-branch"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 14")

        self.svnmerge2(["init", self.test_repo_url + "/branches/testYYY-branch"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 15")

        # Create changes on test-branch that we can block
        os.chdir("..")
        os.chdir("test-branch")
        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 15")

        open("test1", "w").write("test 1-changed_on_test-branch")

        self.launch("svn commit -m \"Change to test1 on test-branch\"",
                    match=r"Committed revision 16")

        # Create changes on testYYY-branch that we can block
        os.chdir("..")
        os.chdir("testYYY-branch")
        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 16")

        open("test2", "w").write("test 2-changed_on_testYYY-branch")

        self.launch("svn commit -m \"Change to test2 on testYYY-branch\"",
                    match=r"Committed revision 17")

        # Block changes from both branches on the trunk
        os.chdir("..")
        os.chdir("trunk")
        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 17")
        self.svnmerge("block -S testYYY-branch", match=r"'svnmerge-blocked' set")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 18")

        self.svnmerge("block -S test-branch", match=r"'svnmerge-blocked' set")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 19")

        # Do the uninit
        self.svnmerge2(["uninit", "--source", self.test_repo_url + "/branches/testYYY-branch"])

        # Check that the merged property for testYYY-branch was removed, but
        # not for test-branch
        pmerged = self.getproperty()
        self.assertEqual("/branches/test-branch:1-13", pmerged)

        # Check that the blocked property for testYYY-branch was removed, but
        # not for test-branch
        pblocked = self.getBlockedProperty()
        self.assertEqual("/branches/test-branch:16", pblocked)

        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 20")

        self.svnmerge2(["uninit", "--source", self.test_repo_url + "/branches/test-branch"])

        # Check that the merged and blocked properties for test-branch have been removed too
        pmerged = self.getproperty()
        self.assertEqual(None, pmerged)

        pblocked = self.getBlockedProperty()
        self.assertEqual(None, pblocked)

    def testUninitForce(self):
        self.svnmerge2(["init", self.test_repo_url + "/trunk"])

        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision")

        self.svnmerge2(["init", self.test_repo_url + "/branches/testYYY-branch"])

        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision")

        p = self.getproperty()
        self.assert_(re.search("/branches/testYYY-branch:1-\d+? /trunk:1-\d+?", p))

        open("test1", "a").write("foo")

        self.svnmerge("uninit --source " + self.test_repo_url + "/branches/testYYY-branch",
                      error=True, match=r"clean")

        self.svnmerge("uninit -F --source " + self.test_repo_url + "/branches/testYYY-branch")
        p = self.getproperty()
        self.assert_(re.search("^/trunk:1-\d+", p))

    def testCheckNoCopyfrom(self):
        os.chdir("..")
        os.chdir("trunk")
        self.svnmerge("init", error=True, match=r"no copyfrom")

    def testInitScenarios(self):
        """ Run various scenarios w/ svnmerge.py init and verify
        the default values that are set as the integrated
        revisions."""

        # Run init with branch as merge source and trunk as merge target
        os.chdir("..")
        os.chdir("trunk")
        self.svnmerge("init ../test-branch")
        # Verify range ends at rev in which branch was created
        self.launch("svn proplist -v", match=r":1-13")
        self.revert()

        # Run init with TRUNK as merge source and BRANCH as merge target
        os.chdir("..")
        os.chdir("test-branch")
        self.svnmerge("init ../trunk")
        # Verify range ends at rev of trunk which was copied to create branch
        self.launch("svn proplist -v", match=r":1-6")
        self.revert()

        # Same thing, but with no explicit parameter (should work implicitly)
        self.svnmerge("init")
        # Verify range ends at rev of trunk which was copied to create branch
        self.launch("svn proplist -v", match=r":1-6")
        self.revert()

        # Run init with TRUNK as merge src, & any other branch which is not
        # a copy of trunk (or the source from which trunk was copied)
        # as the merge target.
        os.chdir("../trunk")
        os.chdir("..")
        self.launch('svn mkdir -m "create /other" %(TEST_REPO_URL)s/other') # creates r14
        self.launch("svn co %(TEST_REPO_URL)s/other")
        os.chdir("other")
        self.svnmerge("init ../trunk")
        # Verify integrated range ends with merge source's latest rev as of
        # the time of initialization:
        self.launch("svn proplist -v", match=r":1-14")
        self.revert()

        # Run init w/ explicit parms; verify them
        self.svnmerge("init -r 1-999 ../trunk")
        self.launch("svn proplist -v", match=r":1-999")

    def testTrimmedAvailMerge(self):
        """Check that both avail and merge do not search for phantom revs too hard."""
        self.svnmerge("init")
        self.svnmerge("avail -vv -r8-9", match=r"svn log.*-r8:9")
        self.svnmerge("merge -F -vv -r8-9", match=r"svn log.*-r8:9")
        self.svnmerge("avail -vv -r2", nonmatch=r"svn log")
        self.svnmerge("integrated", match=r"^3-6,8-9$")

    def testMergeRecordOnly(self):
        """Check that flagging revisions as manually merged works."""
        self.svnmerge("init")
        self.svnmerge("avail -vv -r9", match=r"svn log.*-r9:9")
        self.svnmerge("merge --record-only -F -vv -r9",
                      nonmatch=r"svn merge -r 8:9")
        self.svnmerge("avail -r9", match=r"\A$")
        self.svnmerge("integrated", match=r"^3-6,9$")
        self.svnmerge("integrated -r9", match=r"^9$")

    def testBidirectionalMerges(self):
        """Check that reflected revisions are recognized properly for bidirectional merges."""

        os.chdir("..")
        os.chdir("test-branch")

        self.svnmerge2(["init", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 14")
        os.remove("svnmerge-commit-message.txt")

        os.chdir("..")
        os.chdir("trunk")

        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 14")
        self.svnmerge2(["init", "-r1-14",
                       self.test_repo_url + "/branches/test-branch"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 15")
        os.remove("svnmerge-commit-message.txt")

        open("test1", "w").write("test 1-changed_on_trunk")

        self.launch("svn commit -m \"Change to test1 on trunk\"",
                    match=r"Committed revision 16")

        self.svnmerge("integrated", match=r"^13-14$")

        os.chdir("..")
        os.chdir("test-branch")

        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 16")

        # test-branch was copied from trunk's r6.  So non-phantom revs
        # since that point should still be available to merge from
        # trunk to test-branch:
        self.svnmerge("avail -vv --bidirectional", match=r"\n9-10,16$")
        self.svnmerge("merge -vv --bidirectional", match=r"svn merge --force -r 15:16")
        p = self.getproperty()
        self.assertEqual("/trunk:1-16", p)
        self.svnmerge("integrated", match=r"^3-16$")

        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 17")
        os.remove("svnmerge-commit-message.txt")

        open("test1", "w").write("test 1-changed_on_branch_after_merge_from_trunk")

        self.launch("svn commit -m \"Change to test1 on branch\"",
                    match=r"Committed revision 18")

        os.chdir("..")
        os.chdir("trunk")

        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 18")

        # Ensure default is not to check for reflected revisions.
        self.svnmerge("avail -vv", match=r"\n17-18$")

        # Now check reflected revision is excluded with --bidirectional flag.
        self.svnmerge("avail -vv --bidirectional", match=r"\n18$")

        self.svnmerge("merge -vv --bidirectional", match=r"svn merge --force -r 17:18")
        p = self.getproperty()
        self.assertEqual("/branches/test-branch:1-18", p)

        self.svnmerge("integrated", match=r"^13-18$")

    def testBidirectionalMergesMultiBranch(self):
        """Check that merges from a second branch are not considered reflected for other branches."""

        os.chdir("..")

        self.multilaunch("""
            svn cp -m "Create test-branch2" %(TEST_REPO_URL)s/trunk %(TEST_REPO_URL)s/branches/test-branch2
            svn co %(TEST_REPO_URL)s/branches/test-branch2 test-branch2
        """)

        os.chdir("test-branch")

        self.svnmerge2(["init", "-r1-13", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 15")
        os.remove("svnmerge-commit-message.txt")

        os.chdir("..")
        os.chdir("test-branch2")

        self.svnmerge2(["init",  "-r1-14", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 16")
        os.remove("svnmerge-commit-message.txt")

        os.chdir("..")
        os.chdir("trunk")

        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 16")

        self.svnmerge2(["init", "-r1-16",
                       self.test_repo_url + "/branches/test-branch"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 17")
        os.remove("svnmerge-commit-message.txt")
        self.svnmerge2(["init", "-r1-17",
                       self.test_repo_url + "/branches/test-branch2"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 18")
        os.remove("svnmerge-commit-message.txt")

        os.chdir("..")
        os.chdir("test-branch2")

        open("test1", "w").write("test 1-changed_on_branch2")

        self.launch("svn commit -m \"Change to test1 on branch2\"",
                    match=r"Committed revision 19")

        os.chdir("..")
        os.chdir("trunk")

        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 19")

        # Merge into trunk
        self.svnmerge("merge -vv -S branch2",
                      match=r"merge --force -r 18:19")
        p = self.getproperty()
        self.assertEqual("/branches/test-branch:1-16 /branches/test-branch2:1-19", p)

        self.svnmerge("integrated -S branch2", match=r"^14-19$")
        self.svnmerge("integrated -S ../test-branch", match=r"^13-16$")

        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 20")
        os.remove("svnmerge-commit-message.txt")

        os.chdir("..")
        os.chdir("test-branch")

        # Not using switch, so must update to get latest repository rev.
        self.launch("svn update", match=r"At revision 20")

        # Initialized revs should not be available for merge
        self.svnmerge("avail -v --bidirectional", match=r"initialized.*17-18")

        # Latest revision on trunk which was merged from test-branch2
        # should be available for test-branch with --bidirectional flag.
        self.svnmerge("avail -vv --bidirectional", match=r"merged are:\n20$")

        self.svnmerge("merge -vv --bidirectional", match=r"merge --force -r 19:20")
        p = self.getproperty()
        self.assertEqual("/trunk:1-20", p)

        self.svnmerge("integrated", match=r"^3-20$")

    def testRollbackWithoutInit(self):
        """Rollback should error out if invoked prior to init"""

        self.svnmerge("rollback -vv -S ../trunk",
                      error = True,
                      match = r"no integration info available for path")

    def testRollbackOutsidePossibleRange(self):
        """`svnmerge rollback' should error out if range contains revisions prior to
        SOURCE creation date."""

        # Initialize svnmerge
        self.svnmerge2(["init", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 14")
        os.remove("svnmerge-commit-message.txt")

        expected_error  = r"""Specified revision range falls out of the rollback range."""
        self.svnmerge("rollback -vv -S ../trunk -r 2-14",
                      error = True,
                      match = expected_error)

    def testRollbackWithoutRevisionOpt(self):
        """`svnmerge rollback' should error out if -r option is not given"""

        # Initialize svnmerge
        self.svnmerge2(["init", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 14")
        os.remove("svnmerge-commit-message.txt")

        self.svnmerge("rollback -vv -S ../trunk",
                      error = True,
                      match = r"The '-r' option is mandatory for rollback")

    def testInitAndRollbackRecordOnly(self):
        """Init svnmerge, modify source head, merge, rollback --record-only."""

        # Initialize svnmerge
        self.svnmerge2(["init", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 14")
        os.remove("svnmerge-commit-message.txt")

        # Rollback record-only
        expected_output = r"property 'svnmerge-integrated' set on '.'"
        detested_output = r"""
D    test2
D    test3"""
        self.svnmerge("rollback -vv --record-only -S ../trunk -r5-7",
                      match = expected_output,
                      nonmatch = detested_output)

    def testInitAndRollback(self):
        """Init svnmerge, modify source head, merge, rollback."""

        # Initialize svnmerge
        self.svnmerge2(["init", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 14")
        os.remove("svnmerge-commit-message.txt")

        # Svnmerge rollback r5-7
        expected_output = "D\s+test2\s+D\s+test3"
        self.svnmerge("rollback -vv -S ../trunk -r5-7",
                      match = expected_output)

    def testMergeAndRollbackEmptyRevisionRange(self):
        """Init svnmerge, modify source head, merge, rollback where no merge
           occured."""

        # Initialize svnmerge
        self.svnmerge2(["init", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 14")
        os.remove("svnmerge-commit-message.txt")

        # Make changes to trunk
        os.chdir("../trunk")
        open("newfile", "w").close()
        self.launch("svn add newfile")
        self.launch('svn commit -m "Adding newfile"', match=r"Committed revision 15")
        open("anothernewfile", "w").close()
        self.launch("svn add anothernewfile")
        self.launch('svn commit -m "Adding anothernewfile"', match=r"Committed revision 16")

        # Svnmerge block r15,16
        os.chdir("../test-branch")
        self.launch("svn up ..",
                    error = False)
        self.svnmerge("block -r 15,16 -S ../trunk")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 17")
        self.svnmerge("merge -S ../trunk")
        self.launch("svn commit -F svnmerge-commit-message.txt")

        # Svnmerge rollback r15-16
        self.svnmerge("rollback -vv -S ../trunk -r15-16",
                      error = False,
                      match = r"Nothing to rollback in revision range r15-16")

    def testMergeAndRollback(self):
        """Init svnmerge, modify source head, merge, rollback."""

        # Initialize svnmerge
        self.svnmerge2(["init", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 14")
        os.remove("svnmerge-commit-message.txt")

        # Make changes to trunk
        os.chdir("../trunk")
        open("newfile", "w").close()
        self.launch("svn add newfile")
        self.launch('svn commit -m "Adding newfile"', match=r"Committed revision 15")

        # Svnmerge merge r15
        os.chdir("../test-branch")
        self.svnmerge("merge -r 15 -S ../trunk")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 16")

        # Svnmerge rollback r15
        self.svnmerge("rollback -vv -S ../trunk -r15",
                      match = r"-r 15:14")

    def testBlockMergeAndRollback(self):
        """Init svnmerge, block, modify head, merge, rollback."""

        # Initialize svnmerge
        self.svnmerge2(["init", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 14")
        os.remove("svnmerge-commit-message.txt")

        # Make changes to trunk
        os.chdir("../trunk")
        open("newfile", "w").close()
        self.launch("svn add newfile")
        self.launch('svn commit -m "Adding newfile"', match=r"Committed revision 15")
        open("anothernewfile", "w").close()
        self.launch("svn add anothernewfile")
        self.launch('svn commit -m "Adding anothernewfile"', match=r"Committed revision 16")

        # Svnmerge block r16, merge r15
        os.chdir("../test-branch")
        self.launch("svn up ..",
                    error = False)
        self.svnmerge("block -r 16 -S ../trunk")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 17")
        self.svnmerge("merge -S ../trunk",
                      nonmatch = r"A    anothernewfile",
                      match = r"A    newfile")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 18")

        # Svnmerge rollback revision range 15-18 (in effect only 15,17)
        self.svnmerge("rollback -vv -S ../trunk -r15-18",
                      nonmatch = r"D    anothernewfile")

    def testMergeWithPotentialPropertyConflict(self):
        """Init branch B, merge changes from branch A to branch B,
        init branch C, and attempt a merge of changes from branch B to
        branch C."""

        # Initialize merge info for test-branch.
        self.svnmerge2(["init", "-r 3-6", self.test_repo_url + "/trunk"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match = r"Committed revision 14")
        os.remove("svnmerge-commit-message.txt")

        # Make a change to trunk.
        os.chdir("../trunk")
        open("newfile", "w").close()
        self.launch("svn add newfile")
        self.launch('svn commit -m "Adding newfile"',
                    match=r"Committed revision 15")

        # Merge a change from trunk to test-branch.
        os.chdir("../test-branch")
        self.svnmerge("merge -r 15 -S ../trunk")
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 16")

        # Get a WC for testYYY-branch.
        os.chdir("..")
        self.launch("svn co %s/branches/testYYY-branch" % self.test_repo_url)
        os.chdir("testYYY-branch")

        # Initialize merge info for testYYY-branch.
        self.svnmerge2(["init", "-r 13",
                        self.test_repo_url + "/branches/test-branch"])
        self.launch("svn commit -F svnmerge-commit-message.txt",
                    match=r"Committed revision 17")
        os.remove("svnmerge-commit-message.txt")

        ### FIXME: Unfortunately, we're getting a conflict for the
        ### merge info property in the following svnmerge invocation
        ### due to Subversion's (reasonable) lack of property value
        ### merging (which isn't possible without knowing a property's
        ### MIME type)

        # Attempt a merge of changes from test-branch to
        # testYYY-branch.
        self.svnmerge("merge -r 16 -S ../test-branch")
        try:
            self.launch("svn commit -F svnmerge-commit-message.txt",
                        match=r"Committed revision 18")
        except AssertionError:
            self.assert_(os.path.isfile("dir_conflicts.prej"))

if __name__ == "__main__":
    # If an existing template repository and working copy for testing
    # exists, then always remove it.  This prevents any problems if
    # this test suite is modified and there exists an older template
    # directory that may be used.  This will also prevent problems if
    # in a previous run of this script, the template was being created
    # when the script was canceled, leaving it in an inconsistent
    # state.
    template_path = get_template_path()
    if os.path.exists(template_path):
        rmtree(template_path)

    unittest.main()