# -*- coding: utf-8 -*-

"""

    vcs.backends.git.repository

    ~~~~~~~~~~~~~~~~~~~~~~~~~~~

    Git repository implementation.

    :created_on: Apr 8, 2010

    :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.

"""

import errno

import logging

import os

import re

import time

import urllib.error

import urllib.parse

import urllib.request

from collections import OrderedDict

from dulwich.client import SubprocessGitClient

from dulwich.config import ConfigFile

from dulwich.objects import Tag

from dulwich.repo import NotGitRepository, Repo

from dulwich.server import update_server_info

from kallithea.lib.vcs import subprocessio

from kallithea.lib.vcs.backends.base import BaseRepository, CollectionGenerator

from kallithea.lib.vcs.conf import settings

from kallithea.lib.vcs.exceptions import (BranchDoesNotExistError, ChangesetDoesNotExistError, EmptyRepositoryError, RepositoryError, TagAlreadyExistError,

                                          TagDoesNotExistError)

from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, date_fromtimestamp, makedate, safe_bytes, safe_str

from kallithea.lib.vcs.utils.helpers import get_urllib_request_handlers

from kallithea.lib.vcs.utils.lazy import LazyProperty

from kallithea.lib.vcs.utils.paths import abspath, get_user_home

from . import changeset, inmemory, workdir

SHA_PATTERN = re.compile(r'^([0-9a-fA-F]{12}|[0-9a-fA-F]{40})$')

log = logging.getLogger(__name__)

            raise urllib.error.URLError("Error parsing URL: %r" % url)

        # check first if it's not an local url

        if os.path.isabs(url) and os.path.isdir(url):

            return

        unparsed_url = urllib.parse.urlunparse(parsed_url)

        if unparsed_url != url:

            raise urllib.error.URLError("Invalid url: '%s' normalizes to '%s'" % (url, unparsed_url))

        if parsed_url.scheme == 'git':

            # Mitigate problems elsewhere with incorrect handling of encoded paths.

            # Don't trust urllib.parse.unquote but be prepared for more flexible implementations elsewhere.

            # Space is the only allowed whitespace character - directly or % encoded. No other % or \ is allowed.

            for c in parsed_url.path.replace('%20', ' '):

                if c in '%\\':

                    raise urllib.error.URLError("Invalid escape character in path: '%s'" % c)

                if c.isspace() and c != ' ':

                    raise urllib.error.URLError("Invalid whitespace character in path: %r" % c)

            return

        if parsed_url.scheme not in ['http', 'https']:

            raise urllib.error.URLError("Unsupported protocol in URL %r" % url)

        test_uri, handlers = get_urllib_request_handlers(url_obj)

        if not test_uri.endswith(b'info/refs'):

            test_uri = test_uri.rstrip(b'/') + b'/info/refs'

        url_obj.passwd = b'*****'

        cleaned_uri = str(url_obj)

        o = urllib.request.build_opener(*handlers)

        o.addheaders = [('User-Agent', 'git/1.7.8.0')]  # fake some git

        req = urllib.request.Request(

            "%s?%s" % (

                safe_str(test_uri),

                urllib.parse.urlencode({"service": 'git-upload-pack'})

            ))

        try:

            resp = o.open(req)

            if resp.code != 200:

                raise Exception('Return Code is not 200')

        except Exception as e:

            # means it cannot be cloned

            raise urllib.error.URLError("[%s] org_exc: %s" % (cleaned_uri, e))

import logging

import os

import time

import urllib.error

import urllib.parse

import urllib.request

from collections import OrderedDict

import mercurial.commands

import mercurial.error

import mercurial.exchange

import mercurial.hg

import mercurial.hgweb

import mercurial.httppeer

import mercurial.localrepo

import mercurial.match

import mercurial.mdiff

import mercurial.node

import mercurial.patch

import mercurial.scmutil

import mercurial.sshpeer

import mercurial.tags

import mercurial.ui

import mercurial.unionrepo

from kallithea.lib.vcs.backends.base import BaseRepository, CollectionGenerator

from kallithea.lib.vcs.exceptions import (BranchDoesNotExistError, ChangesetDoesNotExistError, EmptyRepositoryError, RepositoryError, TagAlreadyExistError,

                                          TagDoesNotExistError, VCSError)

from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, author_email, author_name, date_fromtimestamp, makedate, safe_bytes, safe_str

from kallithea.lib.vcs.utils.helpers import get_urllib_request_handlers

from kallithea.lib.vcs.utils.lazy import LazyProperty

from kallithea.lib.vcs.utils.paths import abspath

from . import changeset, inmemory, workdir

log = logging.getLogger(__name__)

class MercurialRepository(BaseRepository):

    """

    Mercurial repository backend

    """

    DEFAULT_BRANCH_NAME = 'default'

    scm = 'hg'

    def __init__(self, repo_path, create=False, baseui=None, src_url=None,

                 update_after_clone=False):

            parsed_url = urllib.parse.urlparse(url)

            parsed_url.port  # trigger netloc parsing which might raise ValueError

        except ValueError:

            raise urllib.error.URLError("Error parsing URL: %r" % url)

        # check first if it's not an local url

        if os.path.isabs(url) and os.path.isdir(url) or parsed_url.scheme == 'file':

            # When creating repos, _get_url will use file protocol for local paths

            return

        if parsed_url.scheme not in ['http', 'https', 'ssh', 'git+http', 'git+https']:

            raise urllib.error.URLError("Unsupported protocol in URL %r" % url)

        url = safe_bytes(url)

        if parsed_url.scheme == 'ssh':

            # in case of invalid uri or authentication issues, sshpeer will

            # throw an exception.

            mercurial.sshpeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')

            return

        if '+' in parsed_url.scheme:  # strip 'git+' for hg-git URLs

            url = url.split(b'+', 1)[1]

        test_uri, handlers = get_urllib_request_handlers(url_obj)

        url_obj.passwd = b'*****'

        cleaned_uri = str(url_obj)

        o = urllib.request.build_opener(*handlers)

        o.addheaders = [('Content-Type', 'application/mercurial-0.1'),

                        ('Accept', 'application/mercurial-0.1')]

        req = urllib.request.Request(

            "%s?%s" % (

                safe_str(test_uri),

                urllib.parse.urlencode({

                    'cmd': 'between',

                    'pairs': "%s-%s" % ('0' * 40, '0' * 40),

                })

            ))

        try:

            resp = o.open(req)

            if resp.code != 200:

                raise Exception('Return Code is not 200')

        except Exception as e:

            # means it cannot be cloned

            if groupdict['weeks']:

                days += int(matched.groupdict()['weeks']) * 7

            past = datetime.datetime.now() - datetime.timedelta(days=days)

            return datetime.datetime(*past.timetuple()[:3])

    raise ValueError('Wrong date: "%s"' % text)

def get_dict_for_attrs(obj, attrs):

    """

    Returns dictionary for each attribute from given ``obj``.

    """

    data = {}

    for attr in attrs:

        data[attr] = getattr(obj, attr)

    return data

def get_urllib_request_handlers(url_obj):

    handlers = []

    test_uri, authinfo = url_obj.authinfo()

    if authinfo:

        # authinfo is a tuple (realm, uris, user, password) where 'uris' itself

        # is a tuple of URIs.

        # values will be bytes, e.g.

        #    (None, (b'http://127.0.0.1/repo', b'127.0.0.1'), b'user', b'pass')

        # However, urllib expects strings, not bytes, so we must convert them.

        # create a password manager

        passmgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()

        passmgr.add_password(

            safe_str(authinfo[0]) if authinfo[0] else None, # realm

            tuple(safe_str(x) for x in authinfo[1]),        # uris

            safe_str(authinfo[2]),                          # user

            safe_str(authinfo[3]),                          # password

        )

        handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),

                         mercurial.url.httpdigestauthhandler(passmgr)))

    return test_uri, handlers