#!/usr/bin/env python3

#
# This file is a test harness for svn-load
#
# Copyright (c) 2007, Hewlett-Packard Development Company, L.P. <dannf@hp.com>
# Copyright (c) 2011, 2018, dann frazier <dannf@dannf.org>
#
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
#     * Redistributions of source code must retain the above
#       copyright notice, this list of conditions and the following
#       disclaimer.
#
#     * Redistributions in binary form must reproduce the above
#       copyright notice, this list of conditions and the following
#       disclaimer in the documentation and/or other materials
#       provided with the distribution.
#
#     * Neither the name of the Hewlett-Packard Co. nor the names
#       of its contributors may be used to endorse or promote
#       products derived from this software without specific prior
#       written permission.
#
# THIS SOFTWARE IS PROVIDED BY HEWLETT-PACKARD DEVELOPMENT COMPANY
# ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT
# NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
# HEWLETT-PACKARD DEVELOPMENT COMPANY BE LIABLE FOR ANY DIRECT,
# INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
# STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING
# IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#

import unittest
import tempfile
import shutil
import os
import filecmp
import subprocess
import time
import xml.etree.ElementTree as ET


# This case simply imports an empty directory into the toplevel of an
# empty repository. But, it provides the framework for other classes
# to inherit and do more complex things
class BaseLoadCase(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.mkdtemp()
        self.importDirSetUp()
        self.loadDirsSetUp()
        self.exportparent = os.path.join(self.tempdir, 'exportparent')
        os.mkdir(self.exportparent)
        self.repo = self.createRepository()
        self.url = 'file://' + self.repo
        # idir is the relative path in repo where things get imported
        self.idir = ''
        self.username = None
        self.password = None
        self.extra_args = []
        self.extra_svn_args = []

    def importDirSetUp(self):
        self.importdir = os.path.join(self.tempdir, 'importdir')
        os.mkdir(self.importdir)

    def loadDirsSetUp(self):
        self.loaddirs = [os.path.join(self.tempdir, 'loaddir0')]
        os.mkdir(self.loaddirs[0])

    def tearDown(self):
        shutil.rmtree(self.tempdir)

    def createRepository(self):
        repo = os.path.join(self.tempdir, 'repo')
        os.mkdir(repo)
        subprocess.call(['/usr/bin/svnadmin', 'create', repo])
        return repo

    def exportRepository(self):
        args = ['/usr/bin/svn', 'export', '--quiet']
        args = args + self.extra_svn_args
        args = args + [self.url, os.path.join(self.exportparent, 'export')]

        ret = subprocess.call(args)

        if ret == 0:
            return True
        else:
            return False

    def loaddir(self, ldir):
        args = ['./svn-load', self.url, self.idir, ldir, '--no-prompt']
        args = args + self.extra_args

        ret = subprocess.call(args)

        if ret == 0:
            return True
        else:
            return False

    # This should allow us to accept cases that dircmp considers "funny".
    # The only known case so far is where both trees have the same identical
    # symlink
    def funny_check(self, a, b, funny):
        haha = True
        strange = False
        for f in funny:
            funnyA = os.path.join(a, f)
            funnyB = os.path.join(b, f)

            if os.path.islink(funnyA) and os.path.islink(funnyB):
                if os.readlink(funnyA) == os.readlink(funnyB):
                    continue
            else:
                return strange
        return haha

    def compare_dirs(self, a, b):
        if os.path.islink(a) and not os.path.islink(b) or \
           not os.path.islink(a) and os.path.islink(b):
            return False

        c = filecmp.dircmp(a, b)
        funny_ok = self.funny_check(a, b, c.common_funny)
        if c.left_only or c.right_only or c.diff_files or not funny_ok:
            return False
        else:
            return c.common_dirs

    def compare_trees(self, a, b):
        subdirs = self.compare_dirs(a, b)
        if subdirs is False:
            return False
        else:
            for d in subdirs:
                if not self.compare_trees(os.path.join(a, d),
                                          os.path.join(b, d)):
                    return False
            return True

    def svnImport(self, importdir, url):
        subprocess.check_call(['svn', 'import'] + self.extra_svn_args +
                              [importdir, url, '-m', 'Initial import'])

    def testLoad(self):
        self.svnImport(self.importdir, self.url)
        for d in self.loaddirs:
            self.assertTrue(self.loaddir(d))
            self.assertTrue(self.exportRepository())
            self.assertTrue(self.compare_trees(d,
                            os.path.join(self.exportparent, 'export')))


class EmptyToSingleFile(BaseLoadCase):
    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        f = open(os.path.join(self.loaddirs[0], 'foo'), 'w')
        f.write("baz\n")
        f.close()


class EmptyToSingleFile3LevelsDown(EmptyToSingleFile):
    def loadDirsSetUp(self):
        EmptyToSingleFile.loadDirsSetUp(self)
        os.makedirs(os.path.join(self.loaddirs[0], '1/2/3'))
        f = open(os.path.join(self.loaddirs[0], '1/2/3/foo'), 'w')
        f.write("baz\n")
        f.close()


class EmptyToSymLink(BaseLoadCase):
    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        os.symlink('baz', os.path.join(self.loaddirs[0], 'foo'))


class EmptyToSingleFileAndSymLink(EmptyToSingleFile):
    def loadDirsSetUp(self):
        EmptyToSingleFile.loadDirsSetUp(self)
        os.symlink('foo', os.path.join(self.loaddirs[0], 'baz'))


class SingleFileContentChange(BaseLoadCase):
    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        f = open(os.path.join(self.importdir, 'foo'), 'w')
        f.write("bar\n")
        f.close()

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        f = open(os.path.join(self.loaddirs[0], 'foo'), 'w')
        f.write("baz\n")
        f.close()

# We'd need to be able to control the user's svn config file to test this.
#class AutoProps(BaseLoadCase):
#    def loadDirsSetUp(self):
#        BaseLoadCase.loadDirsSetUp(self)
#        f = open(os.path.join(self.loaddirs[0], 'foo.txt'), 'w')
#        f.write("baz\n")
#        f.close()
#
#    def testLoad(self):
#        BaseLoadCase.testLoad(self)
#        client = pysvn.Client()
#        proplist = client.proplist('%s/foo.txt' % (self.url))
#        self.failUnless(<some known property is set>)


class BaseMoveCase(BaseLoadCase):
    # files_old: list of files in "old" side
    # files_new: list of files in "new" side, must be identical in length to
    #            files_old, and files_new[i] is the new path for files_old[i]
    files_old = []
    files_new = []

    def setUp(self):
        BaseLoadCase.setUp(self)
        self.movemap = tempfile.mkstemp()[1]
        self.makeMoveMap()
        self.extra_args.extend(['--move-map', self.movemap])
        self.filesmap_forward = dict(list(zip(self.files_old, self.files_new)))
        self.filesmap_backwards = dict(list(zip(self.files_new, self.files_old)))

    def tearDown(self):
        BaseLoadCase.tearDown(self)
        os.unlink(self.movemap)

    def makeMoveMap(self):
        # Override this method to make the move map
        f = open(self.movemap, 'w')
        f.write('\n')
        f.close()

    def makeFiles(dir, files):
        for filename in files:
            # strip off leading slashes, just in case
            filename = filename.lstrip('/')
            try:
                d = os.path.join(dir, os.path.dirname(filename))
                os.makedirs(d)
            except OSError:
                # already made
                pass
            f = open(os.path.join(dir, filename), 'w')
            f.write('foo\n')
            f.close()
    makeFiles = staticmethod(makeFiles)

    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        self.makeFiles(self.importdir, self.files_old)

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        self.makeFiles(self.loaddirs[0], self.files_new)

    def testLoad(self):
        # superclass's testLoad method does some stuff that would conflict with
        # testFilesMoved
        pass

    def testFilesMoved(self):
        if not (self.files_old and self.files_new):
            # If empty (like on the BaseMoveCase), just ignore
            return

        self.svnImport(self.importdir, self.url)
        self.assertTrue(self.loaddir(self.loaddirs[0]))

        svnxml = subprocess.check_output(['svn', 'log'] + self.extra_svn_args +
                                         ['-l1', '-v', '--xml', self.url])
        root = ET.fromstring(svnxml)
        paths = root.findall("./logentry/paths/path")
        for e in paths:
            # Assume only paths with actions are in the log unless
            # we find a counter-example
            self.assertTrue('action' in list(e.attrib.keys()))
            path = os.path.relpath(e.text, start=os.sep)
            if path in self.files_old:
                self.assertEqual(e.attrib['action'], 'D')
            elif path in self.files_new:
                self.assertEqual(e.attrib['action'], 'A')
                self.assertTrue('copyfrom-path' in list(e.attrib.keys()))
                from_path = os.path.relpath(e.attrib['copyfrom-path'],
                                            start=os.sep)
                self.assertTrue(from_path in self.files_old)


class SingleFileMove(BaseMoveCase):
    files_old = ['foo']
    files_new = ['bar']

    def makeMoveMap(self):
        f = open(self.movemap, 'w')
        f.write("^foo$  'bar'\n")
        f.close()


class GraphicsFilesMove(BaseMoveCase):
    files_old = [
        'src/main.c',
        'src/foo.jpg',
        'src/bar.gif',
        'src/baz.png',
        'src/bang.png',
        'src/blast/boom.jpg',
    ]
    files_new = [
        'src/main.c',
        'graphics/foo.jpg',
        'graphics/bar.gif',
        'graphics/baz.png',
        'graphics/bang.png',
        'graphics/blast/boom.jpg',
    ]

    def makeMoveMap(self):
        f = open(self.movemap, 'w')
        f.write("^src/(?P<filename>.+\\.(gif|jpg|png))$  "
                "lambda m: 'graphics/%s' % m.group('filename')\n")
        f.close()


class MultipleStringMoves(BaseMoveCase):
    files_old = [
        'foo',
        'bar',
        'baz',
    ]

    files_new = [
        'foo',
        'bang',
        'eek',
    ]

    def makeMoveMap(self):
        f = open(self.movemap, 'w')
        f.write("^bar$  'bang'\n")
        f.write("^baz$  'eek'\n")


class MoveMapWithDollarSign(BaseMoveCase):
    files_old = [
        'foo',
        'bar$baz',
    ]

    files_new = [
        'foo',
        'baz',
    ]

    def makeMoveMap(self):
        f = open(self.movemap, 'w')
        f.write(r"^bar\$baz$  'baz'")
        f.write("\n")


class SingleFileToBrokenSymLink(BaseLoadCase):
    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        f = open(os.path.join(self.importdir, 'foo'), 'w')
        f.write("bar\n")
        f.close()

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        os.symlink('broken', os.path.join(self.loaddirs[0], 'foo'))


class SingleFileToSymLink(BaseLoadCase):
    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        f = open(os.path.join(self.importdir, 'foo'), 'w')
        f.close()
        f = open(os.path.join(self.importdir, 'bar'), 'w')
        f.close()

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        f = open(os.path.join(self.loaddirs[0], 'foo'), 'w')
        f.close()
        os.symlink('foo', os.path.join(self.loaddirs[0], 'bar'))


class SingleFileToDirectorySymLink(BaseLoadCase):
    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        os.mkdir(os.path.join(self.importdir, 'foo'))
        open(os.path.join(self.importdir, 'bar'), 'w').close()

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        os.mkdir(os.path.join(self.loaddirs[0], 'foo'))
        os.symlink('foo', os.path.join(self.loaddirs[0], 'bar'))


class BrokenSymLinkToFile(BaseLoadCase):
    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        os.symlink('broken', os.path.join(self.importdir, 'foo'))

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        f = open(os.path.join(self.loaddirs[0], 'foo'), 'w')
        f.write("bar\n")
        f.close()


class SymLinkToFile(BaseLoadCase):
    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        open(os.path.join(self.importdir, 'foo'), 'w').close()
        os.symlink('foo', os.path.join(self.importdir, 'bar'))

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        open(os.path.join(self.loaddirs[0], 'foo'), 'w').close()
        open(os.path.join(self.loaddirs[0], 'bar'), 'w').close()


class DirectorySymLinkToFile(BaseLoadCase):
    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        os.mkdir(os.path.join(self.importdir, 'foo'))
        os.symlink('foo', os.path.join(self.importdir, 'bar'))

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        os.mkdir(os.path.join(self.loaddirs[0], 'foo'))
        open(os.path.join(self.loaddirs[0], 'bar'), 'w').close()


class DirectoryToFile(BaseLoadCase):
    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        open(os.path.join(self.importdir, 'foo'), 'w').close()

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        os.mkdir(os.path.join(self.loaddirs[0], 'foo'))

    def testLoad(self):
        # A failure here is expected - svn does not support changing
        # a file to a directory in a single commit. Perhaps we should
        # support a multiple-commit mode to deal with such things, but
        # until then we'll just test that this fails as expected.
        # https://bugs.debian.org/573090
        try:
            BaseLoadCase.testLoad(self)
        except AssertionError:
            return
        raise AssertionError


class ReimportSymlinkBase(BaseLoadCase):
    def _fillDirectory(self, directory):
        # overwrite in derived class
        pass

    def importDirSetUp(self):
        BaseLoadCase.importDirSetUp(self)
        self._fillDirectory(self.importdir)

    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        self._fillDirectory(self.loaddirs[0])
        # create files one alphabetically before and one after 'bar'
        open(os.path.join(self.loaddirs[0], 'a-file'), 'w').close()
        open(os.path.join(self.loaddirs[0], 'zoo-file'), 'w').close()
        # create a subdirectory hierarchy
        os.mkdir(os.path.join(self.loaddirs[0], 'zoo'))
        os.mkdir(os.path.join(self.loaddirs[0], 'zoo', 'subzoo'))
        os.mkdir(os.path.join(self.loaddirs[0], 'zoo', 'subzoo', 'subsubzoo'))
        open(os.path.join(self.loaddirs[0], 'zoo', 'subzoo', 'subsubzoo',
                          'subsubzoo-file'),
             'w').close()


class ReimportSymlink(ReimportSymlinkBase):
    def _fillDirectory(self, directory):
        open(os.path.join(directory, 'foo'), 'w').close()
        os.symlink('foo', os.path.join(directory, 'bar'))


class ReimportSymlinkToDirectory(ReimportSymlinkBase):
    """svn-load version 0.9 fails on this and _commits_ an
    _inconsistent_ content
    """
    def _fillDirectory(self, directory):
        os.mkdir(os.path.join(directory, 'foo'))
        os.symlink('foo', os.path.join(directory, 'bar'))


class ReimportBrokenSymlink(ReimportSymlinkBase):
    def _fillDirectory(self, directory):
        os.symlink('foo', os.path.join(directory, 'bar'))


class SyblingDirSymlink(BaseLoadCase):
    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        os.mkdir(os.path.join(self.loaddirs[0], 'foo'))
        os.symlink('foo', os.path.join(self.loaddirs[0], 'bar'))


class GlobalIgnoreFile(BaseLoadCase):
    # As of this writing, svn-load assumes that all files in a directory
    # will be added when the subdirectory is added, so it doesn't bother
    # descending down a tree it just added. However, some files are always
    # ignored by svn, even if no svn:ignore property exists. This is
    # controlled by the global-ignores parameter. svn-load blows up
    # on these files during its check_permissions pass, as it assumes it
    # can query their svn:executable property
    def loadDirsSetUp(self):
        BaseLoadCase.loadDirsSetUp(self)
        os.mkdir(os.path.join(self.loaddirs[0], 'foo'))
        f = open(os.path.join(self.loaddirs[0], 'foo', 'bar.o'), 'w')
        f.write("baz\n")
        f.close()


# This test creates two load dirs: foo and foo2, then attempts to load
# 'foo*'. On UNIX this should work because the shell expands the glob.
# On Windows it should work because svn-load expands the glob
class GlobHandling(BaseLoadCase):
    def loadDirsSetUp(self):
        self.loaddirparent = os.path.join(self.tempdir, 'loaddirparent')
        os.mkdir(self.loaddirparent)
        self.loaddirs = [os.path.join(self.loaddirparent, 'foo'),
                         os.path.join(self.loaddirparent, 'foo2')]
        for d in self.loaddirs:
            os.mkdir(d)
            f = open(os.path.join(d, 'test'), 'w')
            f.write(d + '\n')
            f.close()

    def testLoad(self):
        svnload = './svn-load'
        self.svnImport(self.importdir, self.url)
        os.system('%s %s / %s*' % (svnload, self.url,
                                   os.path.join(self.loaddirparent, "foo*")))
        self.assertTrue(self.exportRepository())
        self.assertTrue(
            self.compare_trees(self.loaddirs[-1],
                               os.path.join(self.exportparent, 'export')))


class TempConfigFile:
    def __init__(self):
        self.f = tempfile.NamedTemporaryFile(mode='w', delete=False)
        self.path = self.f.name

    def __del__(self):
        os.unlink(self.path)


class SvnServePasswdFile(TempConfigFile):
    def __init__(self, userDict):
        TempConfigFile.__init__(self)
        self.f.write('[users]\n')
        users = list(userDict.keys())
        for u in users:
            self.f.write('%s = %s\n' % (u, userDict[u]))
        self.f.close()


class SvnServeConf(TempConfigFile):
    def __init__(self, passwd):
        TempConfigFile.__init__(self)
        self.f.write('[general]\n')
        self.f.write('anon-access = none\n')
        self.f.write('auth-access = write\n')
        self.f.write('password-db = %s\n' % (passwd))
        self.f.close()


class AuthenticatedServer(BaseLoadCase):
    def startServer(self, userDict):
        self.passwd = SvnServePasswdFile(userDict)
        self.conf = SvnServeConf(self.passwd.path)
        self.port = 8686
        self.server = subprocess.Popen(['svnserve', '--daemon', '--foreground',
                                        '--root=%s' % (self.tempdir),
                                        '--listen-host=127.0.0.1',
                                        '--listen-port=%d' % (self.port),
                                        '--config-file=%s' % (self.conf.path)])
        # Give server time to listen()
        time.sleep(1)

    def setUp(self, auth=True):
        BaseLoadCase.setUp(self)
        self.username = 'alice'
        self.password = 'secret'
        if auth:
            self.extra_args.extend(['--svn_username=%s' % (self.username)])
            self.extra_args.extend(['--svn_password=%s' % (self.password)])
        self.extra_svn_args.extend(['--username', self.username])
        self.extra_svn_args.extend(['--password', self.password])
        self.startServer({self.username: self.password})
        self.url = 'svn://127.0.0.1:%s/repo' % (self.port)

    def tearDown(self):
        self.server.kill()
        BaseLoadCase.tearDown(self)


class AuthenticatedServerPrompt(AuthenticatedServer):
    def setUp(self):
        AuthenticatedServer.setUp(self, auth=False)

    def loaddir(self, ldir):
        import pexpect
        import sys
        args = ['./svn-load', self.url, os.sep, ldir]
        args = args + self.extra_args
        child = pexpect.spawn(' '.join(args))
        child.logfile = sys.stdout.buffer
        child.expect('Username \(svn\): ')
        child.sendline('alice')
        child.expect('Password for.*: ')
        child.sendline('secret')
        child.expect(pexpect.EOF)
        return True

if __name__ == '__main__':
    unittest.main()