# test_client.py -- Compatibilty tests for git client.
# Copyright (C) 2010 Google, Inc.
#
# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
# General Public License as public by the Free Software Foundation; version 2.0
# or (at your option) any later version. You can redistribute it and/or
# modify it under the terms of either of these two licenses.
#
# 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.
#
# You should have received a copy of the licenses; if not, see
# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
# License, Version 2.0.
#

"""Compatibilty tests between the Dulwich client and the cgit server."""

import copy
from io import BytesIO
import os
import select
import signal
import subprocess
import sys
import tarfile
import tempfile
import threading

try:
    from urlparse import unquote
except ImportError:
    from urllib.parse import unquote


try:
    import BaseHTTPServer
    import SimpleHTTPServer
except ImportError:
    import http.server
    BaseHTTPServer = http.server
    SimpleHTTPServer = http.server

from dulwich import (
    client,
    errors,
    file,
    index,
    protocol,
    objects,
    repo,
    )
from dulwich.tests import (
    SkipTest,
    expectedFailure,
    )
from dulwich.tests.compat.utils import (
    CompatTestCase,
    check_for_daemon,
    import_repo_to_dir,
    rmtree_ro,
    run_git_or_fail,
    _DEFAULT_GIT,
    )


if sys.platform == 'win32':
    import ctypes


class DulwichClientTestBase(object):
    """Tests for client/server compatibility."""

    def setUp(self):
        self.gitroot = os.path.dirname(
                import_repo_to_dir('server_new.export').rstrip(os.sep))
        self.dest = os.path.join(self.gitroot, 'dest')
        file.ensure_dir_exists(self.dest)
        run_git_or_fail(['init', '--quiet', '--bare'], cwd=self.dest)

    def tearDown(self):
        rmtree_ro(self.gitroot)

    def assertDestEqualsSrc(self):
        repo_dir = os.path.join(self.gitroot, 'server_new.export')
        dest_repo_dir = os.path.join(self.gitroot, 'dest')
        with repo.Repo(repo_dir) as src:
            with repo.Repo(dest_repo_dir) as dest:
                self.assertReposEqual(src, dest)

    def _client(self):
        raise NotImplementedError()

    def _build_path(self):
        raise NotImplementedError()

    def _do_send_pack(self):
        c = self._client()
        srcpath = os.path.join(self.gitroot, 'server_new.export')
        with repo.Repo(srcpath) as src:
            sendrefs = dict(src.get_refs())
            del sendrefs[b'HEAD']
            c.send_pack(self._build_path('/dest'), lambda _: sendrefs,
                        src.object_store.generate_pack_data)

    def test_send_pack(self):
        self._do_send_pack()
        self.assertDestEqualsSrc()

    def test_send_pack_nothing_to_send(self):
        self._do_send_pack()
        self.assertDestEqualsSrc()
        # nothing to send, but shouldn't raise either.
        self._do_send_pack()

    def test_send_without_report_status(self):
        c = self._client()
        c._send_capabilities.remove(b'report-status')
        srcpath = os.path.join(self.gitroot, 'server_new.export')
        with repo.Repo(srcpath) as src:
            sendrefs = dict(src.get_refs())
            del sendrefs[b'HEAD']
            c.send_pack(self._build_path('/dest'), lambda _: sendrefs,
                        src.object_store.generate_pack_data)
            self.assertDestEqualsSrc()

    def make_dummy_commit(self, dest):
        b = objects.Blob.from_string(b'hi')
        dest.object_store.add_object(b)
        t = index.commit_tree(dest.object_store, [(b'hi', b.id, 0o100644)])
        c = objects.Commit()
        c.author = c.committer = b'Foo Bar <foo@example.com>'
        c.author_time = c.commit_time = 0
        c.author_timezone = c.commit_timezone = 0
        c.message = b'hi'
        c.tree = t
        dest.object_store.add_object(c)
        return c.id

    def disable_ff_and_make_dummy_commit(self):
        # disable non-fast-forward pushes to the server
        dest = repo.Repo(os.path.join(self.gitroot, 'dest'))
        run_git_or_fail(['config', 'receive.denyNonFastForwards', 'true'],
                        cwd=dest.path)
        commit_id = self.make_dummy_commit(dest)
        return dest, commit_id

    def compute_send(self, src):
        sendrefs = dict(src.get_refs())
        del sendrefs[b'HEAD']
        return sendrefs, src.object_store.generate_pack_data

    def test_send_pack_one_error(self):
        dest, dummy_commit = self.disable_ff_and_make_dummy_commit()
        dest.refs[b'refs/heads/master'] = dummy_commit
        repo_dir = os.path.join(self.gitroot, 'server_new.export')
        with repo.Repo(repo_dir) as src:
            sendrefs, gen_pack = self.compute_send(src)
            c = self._client()
            try:
                c.send_pack(self._build_path('/dest'),
                            lambda _: sendrefs, gen_pack)
            except errors.UpdateRefsError as e:
                self.assertEqual('refs/heads/master failed to update',
                                 e.args[0])
                self.assertEqual({b'refs/heads/branch': b'ok',
                                  b'refs/heads/master': b'non-fast-forward'},
                                 e.ref_status)

    def test_send_pack_multiple_errors(self):
        dest, dummy = self.disable_ff_and_make_dummy_commit()
        # set up for two non-ff errors
        branch, master = b'refs/heads/branch', b'refs/heads/master'
        dest.refs[branch] = dest.refs[master] = dummy
        repo_dir = os.path.join(self.gitroot, 'server_new.export')
        with repo.Repo(repo_dir) as src:
            sendrefs, gen_pack = self.compute_send(src)
            c = self._client()
            try:
                c.send_pack(self._build_path('/dest'), lambda _: sendrefs,
                            gen_pack)
            except errors.UpdateRefsError as e:
                self.assertIn(
                        str(e),
                        ['{0}, {1} failed to update'.format(
                            branch.decode('ascii'), master.decode('ascii')),
                         '{1}, {0} failed to update'.format(
                             branch.decode('ascii'), master.decode('ascii'))])
                self.assertEqual({branch: b'non-fast-forward',
                                  master: b'non-fast-forward'},
                                 e.ref_status)

    def test_archive(self):
        c = self._client()
        f = BytesIO()
        c.archive(self._build_path('/server_new.export'), b'HEAD', f.write)
        f.seek(0)
        tf = tarfile.open(fileobj=f)
        self.assertEqual(['baz', 'foo'], tf.getnames())

    def test_fetch_pack(self):
        c = self._client()
        with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest:
            result = c.fetch(self._build_path('/server_new.export'), dest)
            for r in result.refs.items():
                dest.refs.set_if_equals(r[0], None, r[1])
            self.assertDestEqualsSrc()

    def test_fetch_pack_depth(self):
        c = self._client()
        with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest:
            result = c.fetch(self._build_path('/server_new.export'), dest,
                             depth=1)
            for r in result.refs.items():
                dest.refs.set_if_equals(r[0], None, r[1])
            self.assertEqual(
                    dest.get_shallow(),
                    set([b'35e0b59e187dd72a0af294aedffc213eaa4d03ff',
                         b'514dc6d3fbfe77361bcaef320c4d21b72bc10be9']))

    def test_repeat(self):
        c = self._client()
        with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest:
            result = c.fetch(self._build_path('/server_new.export'), dest)
            for r in result.refs.items():
                dest.refs.set_if_equals(r[0], None, r[1])
            self.assertDestEqualsSrc()
            result = c.fetch(self._build_path('/server_new.export'), dest)
            for r in result.refs.items():
                dest.refs.set_if_equals(r[0], None, r[1])
            self.assertDestEqualsSrc()

    def test_incremental_fetch_pack(self):
        self.test_fetch_pack()
        dest, dummy = self.disable_ff_and_make_dummy_commit()
        dest.refs[b'refs/heads/master'] = dummy
        c = self._client()
        repo_dir = os.path.join(self.gitroot, 'server_new.export')
        with repo.Repo(repo_dir) as dest:
            result = c.fetch(self._build_path('/dest'), dest)
            for r in result.refs.items():
                dest.refs.set_if_equals(r[0], None, r[1])
            self.assertDestEqualsSrc()

    def test_fetch_pack_no_side_band_64k(self):
        c = self._client()
        c._fetch_capabilities.remove(b'side-band-64k')
        with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest:
            result = c.fetch(self._build_path('/server_new.export'), dest)
            for r in result.refs.items():
                dest.refs.set_if_equals(r[0], None, r[1])
            self.assertDestEqualsSrc()

    def test_fetch_pack_zero_sha(self):
        # zero sha1s are already present on the client, and should
        # be ignored
        c = self._client()
        with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest:
            result = c.fetch(
                self._build_path('/server_new.export'), dest,
                lambda refs: [protocol.ZERO_SHA])
            for r in result.refs.items():
                dest.refs.set_if_equals(r[0], None, r[1])

    def test_send_remove_branch(self):
        with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest:
            dummy_commit = self.make_dummy_commit(dest)
            dest.refs[b'refs/heads/master'] = dummy_commit
            dest.refs[b'refs/heads/abranch'] = dummy_commit
            sendrefs = dict(dest.refs)
            sendrefs[b'refs/heads/abranch'] = b"00" * 20
            del sendrefs[b'HEAD']

            def gen_pack(have, want, ofs_delta=False):
                return 0, []
            c = self._client()
            self.assertEqual(dest.refs[b"refs/heads/abranch"], dummy_commit)
            c.send_pack(
                self._build_path('/dest'), lambda _: sendrefs, gen_pack)
            self.assertFalse(b"refs/heads/abranch" in dest.refs)

    def test_get_refs(self):
        c = self._client()
        refs = c.get_refs(self._build_path('/server_new.export'))

        repo_dir = os.path.join(self.gitroot, 'server_new.export')
        with repo.Repo(repo_dir) as dest:
            self.assertDictEqual(dest.refs.as_dict(), refs)


class DulwichTCPClientTest(CompatTestCase, DulwichClientTestBase):

    def setUp(self):
        CompatTestCase.setUp(self)
        DulwichClientTestBase.setUp(self)
        if check_for_daemon(limit=1):
            raise SkipTest('git-daemon was already running on port %s' %
                           protocol.TCP_GIT_PORT)
        fd, self.pidfile = tempfile.mkstemp(prefix='dulwich-test-git-client',
                                            suffix=".pid")
        os.fdopen(fd).close()
        args = [_DEFAULT_GIT, 'daemon', '--verbose', '--export-all',
                '--pid-file=%s' % self.pidfile,
                '--base-path=%s' % self.gitroot,
                '--enable=receive-pack', '--enable=upload-archive',
                '--listen=localhost', '--reuseaddr',
                self.gitroot]
        self.process = subprocess.Popen(
            args, cwd=self.gitroot,
            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        if not check_for_daemon():
            raise SkipTest('git-daemon failed to start')

    def tearDown(self):
        with open(self.pidfile) as f:
            pid = int(f.read().strip())
        if sys.platform == 'win32':
            PROCESS_TERMINATE = 1
            handle = ctypes.windll.kernel32.OpenProcess(
                PROCESS_TERMINATE, False, pid)
            ctypes.windll.kernel32.TerminateProcess(handle, -1)
            ctypes.windll.kernel32.CloseHandle(handle)
        else:
            try:
                os.kill(pid, signal.SIGKILL)
                os.unlink(self.pidfile)
            except (OSError, IOError):
                pass
        self.process.wait()
        self.process.stdout.close()
        self.process.stderr.close()
        DulwichClientTestBase.tearDown(self)
        CompatTestCase.tearDown(self)

    def _client(self):
        return client.TCPGitClient('localhost')

    def _build_path(self, path):
        return path

    if sys.platform == 'win32':