import test_util

import os, sys, cStringIO, difflib
import unittest

from mercurial import commands
from mercurial import hg
from mercurial import node
from mercurial import ui

from hgsubversion import svncommands
from hgsubversion import svnrepo

class TestTags(test_util.TestBase):
    def _load_fixture_and_fetch(self, fixture_name, stupid=False):
        return test_util.load_fixture_and_fetch(fixture_name, self.repo_path,
                                                self.wc_path, stupid=stupid)

    def test_tags(self, stupid=False):
        repo = self._load_fixture_and_fetch('basic_tag_tests.svndump',
                                            stupid=stupid)
        self.assertEqual(sorted(repo.tags()), ['copied_tag', 'tag_r3', 'tip'])
        self.assertEqual(repo['tag_r3'], repo['copied_tag'])
        self.assertEqual(repo['tag_r3'].rev(), 1)

    def test_tags_stupid(self):
        self.test_tags(stupid=True)

    def test_remove_tag(self, stupid=False):
        repo = self._load_fixture_and_fetch('remove_tag_test.svndump',
                                            stupid=stupid)
        self.assertEqual(repo['tag_r3'].rev(), 1)
        self.assert_('copied_tag' not in repo.tags())

    def test_remove_tag_stupid(self):
        self.test_remove_tag(stupid=True)

    def test_rename_tag(self, stupid=False):
        expected = """\
node: hg=default@2:svn=trunk@4
tagging r3
  tag_r3: hg=default@1:svn=trunk@3

node: hg=default@3:svn=trunk@5
tag from a tag
  tag_r3: hg=default@1:svn=trunk@3
  copied_tag: hg=default@1:svn=trunk@3

node: hg=default@4:svn=trunk@6
rename a tag
  tag_r3: hg=default@1:svn=trunk@3
  copied_tag: hg=default@1:svn=trunk@3
  copied_tag: hg=default@-1:svn=unk@unk
  other_tag_r3: hg=default@1:svn=trunk@3
"""
        self._test_tags('rename_tag_test.svndump', expected, stupid)

    def test_rename_tag_stupid(self):
        self.test_rename_tag(stupid=True)

    def test_branch_from_tag(self, stupid=False):
        repo = self._load_fixture_and_fetch('branch_from_tag.svndump',
                                            stupid=stupid)
        self.assert_('branch_from_tag' in repo.branchtags())
        self.assertEqual(repo[1], repo['tag_r3'])
        self.assertEqual(repo['branch_from_tag'].parents()[0], repo['copied_tag'])

    def test_branch_from_tag_stupid(self):
        self.test_branch_from_tag(stupid=True)

    def test_tag_by_renaming_branch(self, stupid=False):
        repo = self._load_fixture_and_fetch('tag_by_rename_branch.svndump',
                                            stupid=stupid)
        branches = set(repo[h] for h in repo.heads())
        self.assert_('dummy' not in branches)
        self.assertEqual(repo['dummy'], repo['tip'].parents()[0],
                         '%r != %r[0]' % (repo['dummy'],
                                              repo['tip'].parents()))
        extra = repo['tip'].extra().copy()
        extra.pop('convert_revision', None)
        self.assertEqual(extra, {'branch': 'dummy', 'close': '1'})

    def test_tag_by_renaming_branch_stupid(self):
        self.test_tag_by_renaming_branch(stupid=True)

    def test_deletion_of_tag_on_trunk_after_branching(self):
        repo = self._load_fixture_and_fetch('tag_deletion_tag_branch.svndump')
        branches = set(repo[h].extra()['branch'] for h in repo.heads())
        self.assertEqual(branches, set(['default', 'from_2', ]))
        self.assertEqual(
            repo.tags(),
            {'tip': 'g\xdd\xcd\x93\x03g\x1e\x7f\xa6-[V%\x99\x07\xd3\x9d>(\x94',
             'new_tag': '=\xb8^\xb5\x18\xa9M\xdb\xf9\xb62Z\xa0\xb5R6+\xfe6.'})

    def test_tags_in_unusual_location(self):
        repo = self._load_fixture_and_fetch('unusual_tags.svndump')
        branches = set(repo[h].extra()['branch']
                       for h in repo.heads())
        self.assertEqual(branches, set(['default', 'dev_branch']))
        tags = repo.tags()
        del tags['tip']
        self.assertEqual(
            tags,
            {'blah/trunktag': '\xd3$@\xd7\xd8Nu\xce\xa6%\xf1u\xd9b\x1a\xb2\x81\xc2\xb0\xb1',
             'versions/branch_version': 'I\x89\x1c>z#\xfc._K#@:\xd6\x1f\x96\xd6\x83\x1b|',
             })

    def test_most_recent_is_edited_stupid(self):
        self.test_most_recent_is_edited(True)

    def test_most_recent_is_edited(self, stupid=False):
        repo = self._load_fixture_and_fetch('most-recent-is-edit-tag.svndump',
                                            stupid=stupid)
        self.repo.ui.status(
            "Note: this test failing may be because of a rebuildmeta failure.\n"
            "You should check that before assuming issues with this test.\n")
        wc2_path = self.wc_path + '2'
        src, dest = test_util.hgclone(repo.ui, self.wc_path, wc2_path, update=False)
        svncommands.rebuildmeta(repo.ui,
                               dest,
                               args=[test_util.fileurl(self.repo_path), ])
        commands.pull(self.repo.ui, self.repo, stupid=stupid)
        dtags, srctags = dest.tags(), self.repo.tags()
        dtags.pop('tip')
        srctags.pop('tip')
        self.assertEqual(dtags, srctags)
        self.assertEqual(dest.heads(), self.repo.heads())

    def test_edited_tag_stupid(self):
        self.test_edited_tag(True)

    def test_edited_tag(self, stupid=False):
       repo = self._load_fixture_and_fetch('commit-to-tag.svndump',
                                           stupid=stupid)
       headcount = 6
       self.assertEqual(len(repo.heads()), headcount)
       heads = repo.heads()
       openheads = [h for h in heads if not repo[h].extra().get('close', False)]
       closedheads = set(heads) - set(openheads)
       self.assertEqual(len(openheads), 1)
       self.assertEqual(len(closedheads), headcount - 1)
       closedheads = sorted(list(closedheads),
                            cmp=lambda x, y: cmp(repo[x].rev(), repo[y].rev()))

       # closeme has no open heads
       for h in openheads:
           self.assertNotEqual('closeme', repo[openheads[0]].branch())

       self.assertEqual(1, len(self.repo.branchheads('magic')))

       alsoedit, editlater, closeme, willedit, editcreate, = closedheads
       self.assertEqual(
           repo[willedit].extra(),
           {'close': '1',
            'branch': 'magic',
            'convert_revision': 'svn:af82cc90-c2d2-43cd-b1aa-c8a78449440a/tags/will-edit@19'})
       self.assertEqual(willedit, repo.tags()['will-edit'])
       self.assertEqual(repo['will-edit'].manifest().keys(), ['alpha',
                                                              'beta',
                                                              'gamma',
                                                              ])
       self.assertEqual(
           repo[alsoedit].extra(),
           {'close': '1',
            'branch': 'magic',
            'convert_revision': 'svn:af82cc90-c2d2-43cd-b1aa-c8a78449440a/tags/also-edit@14'})
       self.assertEqual(repo[alsoedit].parents()[0].node(), repo.tags()['also-edit'])
       self.assertEqual(repo['also-edit'].manifest().keys(),
                        ['beta',
                         '.hgtags',
                         'delta',
                         'alpha',
                         'omega',
                         'iota',
                         'gamma',
                         'lambda',
                         ])

       self.assertEqual(editlater, repo['edit-later'].node())
       self.assertEqual(
           repo[closeme].extra(),
           {'close': '1',
            'branch': 'closeme',
            'convert_revision': 'svn:af82cc90-c2d2-43cd-b1aa-c8a78449440a/branches/closeme@17'})
       self.assertEqual('alpha\nalpha\n', repo['edit-at-create']['alpha'].data())

    def test_tags_in_unusual_location(self):
        repo = self._load_fixture_and_fetch('tag_name_same_as_branch.svndump')
        self.assertEqual(len(repo.heads()), 1)
        branches = set(repo[h].extra()['branch']
                       for h in repo.heads())
        self.assertEqual(branches, set(['magic', ]))
        tags = repo.tags()
        del tags['tip']
        self.assertEqual(
            tags,
            {'magic': '\xa2b\xb9\x03\xc6\xbd\x903\x95\xf5\x0f\x94\xcey\xc4E\xfaE6\xaa',
             'magic2': '\xa3\xa2D\x86aM\xc0v\xb9\xb0\x18\x14\xad\xacwBUi}\xe2',
             })

    def test_old_tag_map_rebuilds(self):
        repo = self._load_fixture_and_fetch('tag_name_same_as_branch.svndump')
        tm = os.path.join(repo.path, 'svn', 'tagmap')
        open(tm, 'w').write('1\n')
        commands.pull(repo.ui, repo)
        self.assertEqual(open(tm).read().splitlines()[0], '2')

    def _debug_print_tags(self, repo, ctx, fp):
        def formatnode(ctx):
            crev = ctx.extra().get('convert_revision', 'unk/unk@unk')
            path, rev = crev.rsplit('@', 1)
            path = path.split('/', 1)[-1]
            branch = ctx.branch() or 'default'
            return 'hg=%s@%d:svn=%s@%s' % (branch, ctx.rev(), path, rev)

        w = fp.write
        desc = ctx.description().splitlines()[0].strip()
        if '.hgtags' not in ctx or not ctx['.hgtags'].data().strip():
            return
        w('node: %s\n' % formatnode(ctx))
        w('%s\n' % desc)
        for line in ctx['.hgtags'].data().splitlines(False):
            node, name = line.split(None, 1)
            w('  %s: %s\n' % (name, formatnode(repo[node])))
        w('\n')

    def _test_tags(self, testpath, expected, stupid=False):
        repo = self._load_fixture_and_fetch(testpath, stupid=stupid)
        fp = cStringIO.StringIO()
        for r in repo:
            self._debug_print_tags(repo, repo[r], fp=fp)
        output = fp.getvalue().strip()
        expected = expected.strip()
        if expected == output:
            return
        expected = expected.splitlines()
        output = output.splitlines()
        diff = difflib.unified_diff(expected, output, 'expected', 'output')
        self.assert_(False, '\n' + '\n'.join(diff))

    def test_tagging_into_tag(self, stupid=False):
        expected = """\
node: hg=test@2:svn=branches/test@4
First tag.
  test-0.1: hg=test@1:svn=branches/test@3

node: hg=test@3:svn=branches/test@5
Weird tag.
  test-0.1: hg=test@1:svn=branches/test@3
  test-0.1/test: hg=test@1:svn=branches/test@3

node: hg=test@4:svn=branches/test@6
Fix tag pt 1.
  test-0.1: hg=test@1:svn=branches/test@3
  test-0.1/test: hg=test@1:svn=branches/test@3
  test-0.1/test: hg=default@-1:svn=unk@unk
  test-0.1-real: hg=test@1:svn=branches/test@3

node: hg=test@5:svn=branches/test@7
Remove weird.
  test-0.1: hg=test@1:svn=branches/test@3
  test-0.1/test: hg=test@1:svn=branches/test@3
  test-0.1/test: hg=default@-1:svn=unk@unk
  test-0.1-real: hg=test@1:svn=branches/test@3
  test-0.1: hg=default@-1:svn=unk@unk

node: hg=test@6:svn=branches/test@8
Fix tag pt 2.
  test-0.1: hg=test@1:svn=branches/test@3
  test-0.1/test: hg=test@1:svn=branches/test@3
  test-0.1/test: hg=default@-1:svn=unk@unk
  test-0.1-real: hg=test@1:svn=branches/test@3
  test-0.1: hg=default@-1:svn=unk@unk
  test-0.1-real: hg=default@-1:svn=unk@unk
  test-0.1: hg=test@1:svn=branches/test@3
"""
        self._test_tags('renametagdir.svndump', expected, stupid=stupid)

    def test_tagging_into_tag_stupid(self):
        self.test_tagging_into_tag(True)


def suite():
    return unittest.TestLoader().loadTestsFromTestCase(TestTags)