kallithea Changeset - f713a37564c0

Changeset - f713a37564c0

Parent rev.

Child rev.

[Not reviewed]

default

0 14 0

Mads Kiilerich - 6 years ago 2020-01-02 20:39:13
mads@kiilerich.com

vcs: drop the superfluous and leaky hgcompat "layer"

Explicit is better. And gives less pyflakes noise.

14 files changed with 103 insertions and 90 deletions:

kallithea/controllers/compare.py

kallithea/controllers/pullrequests.py

kallithea/lib/hooks.py

kallithea/lib/middleware/simplehg.py

kallithea/lib/utils.py

kallithea/lib/vcs/backends/git/repository.py

kallithea/lib/vcs/backends/hg/__init__.py

kallithea/lib/vcs/backends/hg/changeset.py

kallithea/lib/vcs/backends/hg/inmemory.py

kallithea/lib/vcs/backends/hg/repository.py

kallithea/lib/vcs/backends/hg/ssh.py

kallithea/lib/vcs/backends/hg/workdir.py

kallithea/lib/vcs/utils/hgcompat.py

kallithea/tests/vcs/test_hg.py

0 comments (0 inline, 0 general)

kallithea/controllers/compare.py

➞

Show inline comments

@@ @@ -21,38 +21,38 @@ repos, branches, bookmarks or tips @@
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 6, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import re
 import mercurial.unionrepo
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPBadRequest, HTTPFound, HTTPNotFound
 from kallithea.config.routing import url
 from kallithea.controllers.changeset import _context_url, _ignorews_url
 from kallithea.lib import diffs
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, render
 from kallithea.lib.graphmod import graph_data
 from kallithea.lib.utils2 import ascii_bytes, ascii_str, safe_int, safe_str
 from kallithea.lib.vcs.utils.hgcompat import unionrepo
 from kallithea.model.db import Repository
 log = logging.getLogger(__name__)
 class CompareController(BaseRepoController):
     def _before(self, *args, **kwargs):
         super(CompareController, self)._before(*args, **kwargs)
         # The base repository has already been retrieved.
@@ @@ -88,25 +88,25 @@ class CompareController(BaseRepoControll @@
         :param other_repo: repo object, most likely the fork of org_repo. It has
             all changesets that we need to obtain
         :param other_rev: revision we want out compare to be made on other_repo
         """
         ancestors = None
         if org_rev == other_rev:
             org_changesets = []
             other_changesets = []
         elif alias == 'hg':
             # case two independent repos
             if org_repo != other_repo:
                 hgrepo = unionrepo.makeunionrepository(other_repo.baseui,
+                hgrepo = mercurial.unionrepo.makeunionrepository(other_repo.baseui,
                                                        other_repo.path,
                                                        org_repo.path)
                 # all ancestors of other_rev will be in other_repo and
                 # rev numbers from hgrepo can be used in other_repo - org_rev ancestors cannot
             # no remote compare do it on the same repository
             else:
                 hgrepo = other_repo._repo
             ancestors = [ascii_str(hgrepo[ancestor].hex()) for ancestor in
                          hgrepo.revs(b"id(%s) & ::id(%s)", ascii_bytes(other_rev), ascii_bytes(org_rev))]
             if ancestors:

kallithea/controllers/pullrequests.py

➞

Show inline comments

@@ @@ -20,40 +20,40 @@ pull requests controller for Kallithea f @@
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 7, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 import mercurial.unionrepo
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPBadRequest, HTTPForbidden, HTTPFound, HTTPNotFound
 from kallithea.config.routing import url
 from kallithea.controllers.changeset import _context_url, _ignorews_url, create_cs_pr_comment, delete_cs_pr_comment
 from kallithea.lib import diffs
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, jsonify, render
 from kallithea.lib.graphmod import graph_data
 from kallithea.lib.page import Page
 from kallithea.lib.utils2 import ascii_bytes, safe_bytes, safe_int, safe_str
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, EmptyRepositoryError
 from kallithea.lib.vcs.utils.hgcompat import unionrepo
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.comment import ChangesetCommentsModel
 from kallithea.model.db import ChangesetStatus, PullRequest, PullRequestReviewer, Repository, User
 from kallithea.model.forms import PullRequestForm, PullRequestPostForm
 from kallithea.model.meta import Session
 from kallithea.model.pull_request import CreatePullRequestAction, CreatePullRequestIterationAction, PullRequestModel
 log = logging.getLogger(__name__)
 def _get_reviewer(user_id):
@@ @@ -524,25 +524,25 @@ class PullrequestsController(BaseRepoCon @@
                 elif c.pull_request.is_closed():
                     c.update_msg = _('This pull request has been closed and can not be updated.')
                 else: # look for descendants of PR head on source branch in org repo
                     avail_revs = org_scm_instance._repo.revs('%s:: & branch(%s)',
                                                              revs[0], c.cs_branch_name)
                     if len(avail_revs) > 1: # more than just revs[0]
                         # also show changesets that not are descendants but would be merged in
                         targethead = other_scm_instance.get_changeset(c.a_branch_name).raw_id
                         if org_scm_instance.path != other_scm_instance.path:
                             # Note: org_scm_instance.path must come first so all
                             # valid revision numbers are 100% org_scm compatible
                             # - both for avail_revs and for revset results
                             hgrepo = unionrepo.makeunionrepository(org_scm_instance.baseui,
+                            hgrepo = mercurial.unionrepo.makeunionrepository(org_scm_instance.baseui,
                                                                    org_scm_instance.path,
                                                                    other_scm_instance.path)
                         else:
                             hgrepo = org_scm_instance._repo
                         show = set(hgrepo.revs('::%ld & !::parents(%s) & !::%s',
                                                avail_revs, revs[0], targethead))
                         if show:
                             c.update_msg = _('The following additional changes are available on %s:') % c.cs_branch_name
                         else:
                             c.update_msg = _('No additional changesets found for iterating on this pull request.')
                     else:
                         show = set()

kallithea/lib/hooks.py

➞

Show inline comments

@@ @@ -19,30 +19,31 @@ Hooks run by Kallithea @@
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Aug 6, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import time
 import mercurial.scmutil
 from kallithea.lib import helpers as h
 from kallithea.lib.exceptions import UserCreationError
 from kallithea.lib.utils import action_logger, make_ui, setup_cache_regions
 from kallithea.lib.utils2 import ascii_str, get_hook_environment, safe_str, safe_unicode
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.utils.hgcompat import revrange
 from kallithea.model.db import Repository, User
 def _get_scm_size(alias, root_path):
     if not alias.startswith('.'):
         alias += '.'
     size_scm, size_root = 0, 0
     for path, dirs, files in os.walk(safe_str(root_path)):
         if path.find(alias) != -1:
             for f in files:
                 try:
@@ @@ -99,25 +100,25 @@ def log_pull_action(ui, repo, **kwargs): @@
     return 0
 def log_push_action(ui, repo, node, node_last, **kwargs):
     """
     Entry point for Mercurial hook changegroup.push_logger.
     The pushed changesets is given by the revset 'node:node_last'.
     Note: This hook is not only logging, but also the side effect invalidating
     cahes! The function should perhaps be renamed.
     """
     revs = [ascii_str(repo[r].hex()) for r in revrange(repo, [b'%s:%s' % (node, node_last)])]
+    revs = [ascii_str(repo[r].hex()) for r in mercurial.scmutil.revrange(repo, [b'%s:%s' % (node, node_last)])]
     process_pushed_raw_ids(revs)
     return 0
 def process_pushed_raw_ids(revs):
     """
     Register that changes have been added to the repo - log the action *and* invalidate caches.
     Called from  Mercurial changegroup.push_logger calling hook log_push_action,
     or from the Git post-receive hook calling handle_git_post_receive ...
     or from scm _handle_push.
     """

kallithea/lib/middleware/simplehg.py

➞

Show inline comments

@@ @@ -23,28 +23,29 @@ Original author and date, and relevant c @@
 :created_on: Apr 28, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import urllib
 import mercurial.hgweb
 from kallithea.lib.base import BaseVCSController, get_path_info
 from kallithea.lib.utils import make_ui
 from kallithea.lib.utils2 import safe_str, safe_unicode
 from kallithea.lib.vcs.utils.hgcompat import hgweb_mod
 log = logging.getLogger(__name__)
 def get_header_hgarg(environ):
     """Decode the special Mercurial encoding of big requests over multiple headers.
     >>> get_header_hgarg({})
     ''
     >>> get_header_hgarg({'HTTP_X_HGARG_0': ' ', 'HTTP_X_HGARG_1': 'a','HTTP_X_HGARG_2': '','HTTP_X_HGARG_3': 'b+c %20'})
     'ab+c %20'
     """
@@ @@ -130,19 +131,19 @@ class SimpleHg(BaseVCSController): @@
                         action = cmd_mapping.get(cmd, 'push')
                     break # only process one cmd
         return parsed_request
     def _make_app(self, parsed_request):
         """
         Make an hgweb wsgi application.
         """
         str_repo_name = safe_str(parsed_request.repo_name)
         repo_path = os.path.join(safe_str(self.basepath), str_repo_name)
         baseui = make_ui(repo_path=repo_path)
-        hgweb_app = hgweb_mod.hgweb(repo_path, name=str_repo_name, baseui=baseui)
+        hgweb_app = mercurial.hgweb.hgweb(repo_path, name=str_repo_name, baseui=baseui)
         def wrapper_app(environ, start_response):
-            environ['REPO_NAME'] = str_repo_name # used by hgweb_mod.hgweb
+            environ['REPO_NAME'] = str_repo_name # used by mercurial.hgweb.hgweb
             return hgweb_app(environ, start_response)
         return wrapper_app

kallithea/lib/utils.py

➞

Show inline comments

@@ @@ -25,37 +25,38 @@ Original author and date, and relevant c @@
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import logging
 import os
 import re
 import sys
 import traceback
 from distutils.version import StrictVersion
 import beaker
 import mercurial.config
 import mercurial.ui
 from beaker.cache import _cache_decorate
 from tg.i18n import ugettext as _
 import kallithea.config.conf
 from kallithea.lib.exceptions import HgsubversionImportError
 from kallithea.lib.utils2 import ascii_bytes, aslist, get_current_authuser, safe_bytes, safe_str, safe_unicode
 from kallithea.lib.vcs.backends.git.repository import GitRepository
 from kallithea.lib.vcs.backends.hg.repository import MercurialRepository
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import RepositoryError, VCSError
 from kallithea.lib.vcs.utils.fakemod import create_module
 from kallithea.lib.vcs.utils.helpers import get_scm
 from kallithea.lib.vcs.utils.hgcompat import config, ui
 from kallithea.model import meta
 from kallithea.model.db import RepoGroup, Repository, Setting, Ui, User, UserGroup, UserLog
 log = logging.getLogger(__name__)
 REMOVED_REPO_PAT = re.compile(r'rm__\d{8}_\d{6}_\d{6}_.*')
 #==============================================================================
 # PERM DECORATOR HELPERS FOR EXTRACTING NAMES FOR PERM CHECKS
 #==============================================================================
@@ @@ -318,54 +319,54 @@ ui_sections = ['alias', 'auth', @@
                 'hooks', 'http_proxy',
                 'smtp', 'patch',
                 'paths', 'profiling',
                 'server', 'trusted',
                 'ui', 'web', ]
 def make_ui(repo_path=None):
     """
     Create an Mercurial 'ui' object based on database Ui settings, possibly
     augmenting with content from a hgrc file.
     """
     baseui = ui.ui()
+    baseui = mercurial.ui.ui()
     # clean the baseui object
     baseui._ocfg = config.config()
     baseui._ucfg = config.config()
     baseui._tcfg = config.config()
     baseui._ocfg = mercurial.config.config()
     baseui._ucfg = mercurial.config.config()
     baseui._tcfg = mercurial.config.config()
     sa = meta.Session()
     for ui_ in sa.query(Ui).all():
         if ui_.ui_active:
             log.debug('config from db: [%s] %s=%r', ui_.ui_section,
                       ui_.ui_key, ui_.ui_value)
             baseui.setconfig(ascii_bytes(ui_.ui_section), ascii_bytes(ui_.ui_key),
                              b'' if ui_.ui_value is None else safe_bytes(ui_.ui_value))
     # force set push_ssl requirement to False, Kallithea handles that
     baseui.setconfig(b'web', b'push_ssl', False)
     baseui.setconfig(b'web', b'allow_push', b'*')
     # prevent interactive questions for ssh password / passphrase
     ssh = baseui.config(b'ui', b'ssh', default=b'ssh')
     baseui.setconfig(b'ui', b'ssh', b'%s -oBatchMode=yes -oIdentitiesOnly=yes' % ssh)
     # push / pull hooks
     baseui.setconfig(b'hooks', b'changegroup.kallithea_log_push_action', b'python:kallithea.lib.hooks.log_push_action')
     baseui.setconfig(b'hooks', b'outgoing.kallithea_log_pull_action', b'python:kallithea.lib.hooks.log_pull_action')
     if repo_path is not None:
         hgrc_path = os.path.join(repo_path, '.hg', 'hgrc')
         if os.path.isfile(hgrc_path):
             log.debug('reading hgrc from %s', hgrc_path)
             cfg = config.config()
+            cfg = mercurial.config.config()
             cfg.read(hgrc_path)
             for section in ui_sections:
                 for k, v in cfg.items(section):
                     log.debug('config from file: [%s] %s=%s', section, k, v)
                     baseui.setconfig(ascii_bytes(section), ascii_bytes(k), safe_bytes(v))
         else:
             log.debug('hgrc file is not present at %s, skipping...', hgrc_path)
     return baseui
 def set_app_settings(config):

kallithea/lib/vcs/backends/git/repository.py

➞

Show inline comments

@@ @@ -9,35 +9,36 @@ @@
     :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.
 """
 import errno
 import logging
 import os
 import re
 import time
 import urllib
 import urllib2
 from collections import OrderedDict
 import mercurial.url  # import httpbasicauthhandler, httpdigestauthhandler
 import mercurial.util  # import url as hg_url
 from dulwich.config import ConfigFile
 from dulwich.objects import Tag
 from dulwich.repo import NotGitRepository, Repo
 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_str, date_fromtimestamp, makedate, safe_str, safe_unicode
 from kallithea.lib.vcs.utils.hgcompat import hg_url, httpbasicauthhandler, httpdigestauthhandler
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.lib.vcs.utils.paths import abspath, get_user_home
 from .changeset import GitChangeset
 from .inmemory import GitInMemoryChangeset
 from .workdir import GitWorkdir
 SHA_PATTERN = re.compile(r'^([0-9a-fA-F]{12}|[0-9a-fA-F]{40})$')
 log = logging.getLogger(__name__)
@@ @@ -159,39 +160,39 @@ class GitRepository(BaseRepository): @@
         # check first if it's not an local url
         if os.path.isdir(url) or url.startswith('file:'):
             return True
         if url.startswith('git://'):
             return True
         if '+' in url[:url.find('://')]:
             url = url[url.find('+') + 1:]
         handlers = []
-        url_obj = hg_url(url)
+        url_obj = mercurial.util.url(url)
         test_uri, authinfo = url_obj.authinfo()
         if not test_uri.endswith('info/refs'):
             test_uri = test_uri.rstrip('/') + '/info/refs'
         url_obj.passwd = b'*****'
         cleaned_uri = str(url_obj)
         if authinfo:
             # create a password manager
             passmgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
             passmgr.add_password(*authinfo)
             handlers.extend((httpbasicauthhandler(passmgr),
                              httpdigestauthhandler(passmgr)))
             handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),
                              mercurial.url.httpdigestauthhandler(passmgr)))
         o = urllib2.build_opener(*handlers)
         o.addheaders = [('User-Agent', 'git/1.7.8.0')]  # fake some git
         q = {"service": 'git-upload-pack'}
         qs = '?%s' % urllib.urlencode(q)
         cu = "%s%s" % (test_uri, qs)
         req = urllib2.Request(cu, None, {})
         try:
             resp = o.open(req)
             if resp.code != 200:

kallithea/lib/vcs/backends/hg/__init__.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     vcs.backends.hg
     ~~~~~~~~~~~~~~~
     Mercurial backend implementation.
     :created_on: Apr 8, 2010
     :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.
 """
 from kallithea.lib.vcs.utils import hgcompat
 from .changeset import MercurialChangeset
 from .inmemory import MercurialInMemoryChangeset
 from .repository import MercurialRepository
 from .workdir import MercurialWorkdir
 __all__ = [
     'MercurialRepository', 'MercurialChangeset',
     'MercurialInMemoryChangeset', 'MercurialWorkdir',
+]
 hgcompat.monkey_do()

kallithea/lib/vcs/backends/hg/changeset.py

➞

Show inline comments

 import os
 import posixpath
 import mercurial.archival
 import mercurial.node
 import mercurial.obsutil
 from kallithea.lib.vcs.backends.base import BaseChangeset
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, ChangesetError, ImproperArchiveTypeError, NodeDoesNotExistError, VCSError
 from kallithea.lib.vcs.nodes import (
     AddedFileNodesGenerator, ChangedFileNodesGenerator, DirNode, FileNode, NodeKind, RemovedFileNodesGenerator, RootNode, SubModuleNode)
 from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, date_fromtimestamp, safe_str, safe_unicode
 from kallithea.lib.vcs.utils.hgcompat import archival, hex, obsutil
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.lib.vcs.utils.paths import get_dirs_for_path
 class MercurialChangeset(BaseChangeset):
     """
     Represents state of the repository at a revision.
     """
     def __init__(self, repository, revision):
         self.repository = repository
         assert isinstance(revision, basestring), repr(revision)
@@ @@ -63,35 +66,35 @@ class MercurialChangeset(BaseChangeset): @@
     @LazyProperty
     def phase(self):
         if(self._ctx.phase() == 1):
             return 'Draft'
         elif(self._ctx.phase() == 2):
             return 'Secret'
         else:
             return ''
     @LazyProperty
     def successors(self):
         successors = obsutil.successorssets(self._ctx._repo, self._ctx.node(), closest=True)
+        successors = mercurial.obsutil.successorssets(self._ctx._repo, self._ctx.node(), closest=True)
         if successors:
             # flatten the list here handles both divergent (len > 1)
             # and the usual case (len = 1)
             successors = [hex(n)[:12] for sub in successors for n in sub if n != self._ctx.node()]
+            successors = [mercurial.node.hex(n)[:12] for sub in successors for n in sub if n != self._ctx.node()]
         return successors
     @LazyProperty
     def predecessors(self):
         return [hex(n)[:12] for n in obsutil.closestpredecessors(self._ctx._repo, self._ctx.node())]
+        return [mercurial.node.hex(n)[:12] for n in mercurial.obsutil.closestpredecessors(self._ctx._repo, self._ctx.node())]
     @LazyProperty
     def bookmarks(self):
         return [safe_unicode(bookmark) for bookmark in self._ctx.bookmarks()]
     @LazyProperty
     def message(self):
         return safe_unicode(self._ctx.description())
     @LazyProperty
     def committer(self):
         return safe_unicode(self.author)
@@ @@ -262,25 +265,25 @@ class MercurialChangeset(BaseChangeset): @@
         return self.get_file_history(path, limit=1)[0]
     def get_file_history(self, path, limit=None):
         """
         Returns history of file as reversed list of ``Changeset`` objects for
         which file at given ``path`` has been modified.
         """
         fctx = self._get_filectx(path)
         hist = []
         cnt = 0
         for cs in reversed([x for x in fctx.filelog()]):
             cnt += 1
             hist.append(hex(fctx.filectx(cs).node()))
+            hist.append(mercurial.node.hex(fctx.filectx(cs).node()))
             if limit is not None and cnt == limit:
                 break
         return [self.repository.get_changeset(node) for node in hist]
     def get_file_annotate(self, path):
         """
         Returns a generator of four element tuples with
             lineno, sha, changeset lazy loader and line
         """
         annotations = self._get_filectx(path).annotate()
         annotation_lines = [(annotateline.fctx, annotateline.text) for annotateline in annotations]
@@ @@ -311,25 +314,25 @@ class MercurialChangeset(BaseChangeset): @@
         if stream is None:
             raise VCSError('You need to pass in a valid stream for filling'
                            ' with archival data')
         if prefix is None:
             prefix = '%s-%s' % (self.repository.name, self.short_id)
         elif prefix.startswith('/'):
             raise VCSError("Prefix cannot start with leading slash")
         elif prefix.strip() == '':
             raise VCSError("Prefix cannot be empty")
         archival.archive(self.repository._repo, stream, ascii_bytes(self.raw_id),
+        mercurial.archival.archive(self.repository._repo, stream, ascii_bytes(self.raw_id),
                          kind, prefix=prefix, subrepos=subrepos)
     def get_nodes(self, path):
         """
         Returns combined ``DirNode`` and ``FileNode`` objects list representing
         state of changeset at the given ``path``. If node at the given ``path``
         is not instance of ``DirNode``, ChangesetError would be raised.
         """
         if self._get_kind(path) != NodeKind.DIR:
             raise ChangesetError("Directory does not exist for revision %s at "
                 " '%s'" % (self.revision, path))

kallithea/lib/vcs/backends/hg/inmemory.py

➞

Show inline comments

 import datetime
 import mercurial.context
 import mercurial.node
 from kallithea.lib.vcs.backends.base import BaseInMemoryChangeset
 from kallithea.lib.vcs.exceptions import RepositoryError
 from kallithea.lib.vcs.utils import ascii_str, safe_bytes
 from kallithea.lib.vcs.utils.hgcompat import hex, memctx, memfilectx
 class MercurialInMemoryChangeset(BaseInMemoryChangeset):
     def commit(self, message, author, parents=None, branch=None, date=None,
             **kwargs):
         """
         Performs in-memory commit (doesn't check workdir in any way) and
         returns newly created ``Changeset``. Updates repository's
         ``revisions``.
         :param message: message of the commit
@@ @@ -42,65 +44,66 @@ class MercurialInMemoryChangeset(BaseInM @@
             """
             Marks given path as added/changed/removed in a given _repo. This is
             for internal mercurial commit function.
             """
             # check if this path is removed
             if path in (node.path for node in self.removed):
                 return None
             # check if this path is added
             for node in self.added:
                 if node.path == path:
                     return memfilectx(_repo, memctx, path=node.path,
+                    return mercurial.context.memfilectx(_repo, memctx, path=node.path,
                         data=node.content,
                         islink=False,
                         isexec=node.is_executable,
                         copysource=False)
             # or changed
             for node in self.changed:
                 if node.path == path:
                     return memfilectx(_repo, memctx, path=node.path,
+                    return mercurial.context.memfilectx(_repo, memctx, path=node.path,
                         data=node.content,
                         islink=False,
                         isexec=node.is_executable,
                         copysource=False)
             raise RepositoryError("Given path haven't been marked as added,"
                                   "changed or removed (%s)" % path)
         parents = [None, None]
         for i, parent in enumerate(self.parents):
             if parent is not None:
                 parents[i] = parent._ctx.node()
         if date and isinstance(date, datetime.datetime):
             date = date.strftime('%a, %d %b %Y %H:%M:%S')
         commit_ctx = memctx(repo=self.repository._repo,
         commit_ctx = mercurial.context.memctx(
             repo=self.repository._repo,
             parents=parents,
             text=b'',
             files=self.get_paths(),
             filectxfn=filectxfn,
             user=author,
             date=date,
             extra=kwargs)
         # injecting given _repo params
         commit_ctx._text = safe_bytes(message)
         commit_ctx._user = safe_bytes(author)
         commit_ctx._date = date
         # TODO: Catch exceptions!
         n = self.repository._repo.commitctx(commit_ctx)
         # Returns mercurial node
         self._commit_ctx = commit_ctx  # For reference
         # Update vcs repository object & recreate mercurial _repo
         # new_ctx = self.repository._repo[node]
         # new_tip = ascii_str(self.repository.get_changeset(new_ctx.hex()))
         self.repository.revisions.append(ascii_str(hex(n)))
+        self.repository.revisions.append(ascii_str(mercurial.node.hex(n)))
         self._repo = self.repository._get_repo(create=False)
         self.repository.branches = self.repository._get_branches()
         tip = self.repository.get_changeset()
         self.reset()
         return tip

kallithea/lib/vcs/backends/hg/repository.py

➞

Show inline comments

@@ @@ -8,30 +8,45 @@ @@
     :created_on: Apr 8, 2010
     :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.
 """
 import datetime
 import logging
 import os
 import time
 import urllib
 import urllib2
 from collections import OrderedDict
 import mercurial.commands
 import mercurial.error
 import mercurial.exchange
 import mercurial.hg
 import mercurial.hgweb
 import mercurial.httppeer
 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.url
 import mercurial.util
 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_str, author_email, author_name, date_fromtimestamp, makedate, safe_bytes, safe_str, safe_unicode
 from kallithea.lib.vcs.utils.hgcompat import (
     Abort, RepoError, RepoLookupError, clone, diffopts, get_contact, hex, hg_url, httpbasicauthhandler, httpdigestauthhandler, httppeer, localrepo, match_exact, nullid, patch, peer, scmutil, sshpeer, tag, ui)
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.lib.vcs.utils.paths import abspath
 from .changeset import MercurialChangeset
 from .inmemory import MercurialInMemoryChangeset
 from .workdir import MercurialWorkdir
 log = logging.getLogger(__name__)
 class MercurialRepository(BaseRepository):
@@ @@ -53,25 +68,25 @@ class MercurialRepository(BaseRepository @@
         :param baseui=None: user data
         :param src_url=None: would try to clone repository from given location
         :param update_after_clone=False: sets update of working copy after
           making a clone
         """
         if not isinstance(repo_path, str):
             raise VCSError('Mercurial backend requires repository path to '
                            'be instance of <str> got %s instead' %
                            type(repo_path))
         self.path = abspath(repo_path)
         self.baseui = baseui or ui.ui()
+        self.baseui = baseui or mercurial.ui.ui()
         # We've set path and ui, now we can set _repo itself
         self._repo = self._get_repo(create, src_url, update_after_clone)
     @property
     def _empty(self):
         """
         Checks if repository is empty ie. without any changesets
         """
         # TODO: Following raises errors when using InMemoryChangeset...
         # return len(self._repo.changelog) == 0
         return len(self.revisions) == 0
@@ @@ -109,43 +124,43 @@ class MercurialRepository(BaseRepository @@
         :param closed: return also closed branches for mercurial
         :param normal: return also normal branches
         """
         if self._empty:
             return {}
         bt = OrderedDict()
         for bn, _heads, node, isclosed in sorted(self._repo.branchmap().iterbranches()):
             if isclosed:
                 if closed:
                     bt[safe_unicode(bn)] = ascii_str(hex(node))
+                    bt[safe_unicode(bn)] = ascii_str(mercurial.node.hex(node))
             else:
                 if normal:
                     bt[safe_unicode(bn)] = ascii_str(hex(node))
+                    bt[safe_unicode(bn)] = ascii_str(mercurial.node.hex(node))
         return bt
     @LazyProperty
     def tags(self):
         """
         Gets tags for this repository
         """
         return self._get_tags()
     def _get_tags(self):
         if self._empty:
             return {}
         return OrderedDict(sorted(
             ((safe_unicode(n), ascii_str(hex(h))) for n, h in self._repo.tags().items()),
+            ((safe_unicode(n), ascii_str(mercurial.node.hex(h))) for n, h in self._repo.tags().items()),
             reverse=True,
             key=lambda x: x[0],  # sort by name
         ))
     def tag(self, name, user, revision=None, message=None, date=None,
             **kwargs):
         """
         Creates and returns a tag for the given ``revision``.
         :param name: name for new tag
         :param user: full username, i.e.: "Joe Doe <joe.doe@example.com>"
         :param revision: changeset id for which new tag would be created
@@ @@ -158,26 +173,26 @@ class MercurialRepository(BaseRepository @@
             raise TagAlreadyExistError("Tag %s already exists" % name)
         changeset = self.get_changeset(revision)
         local = kwargs.setdefault('local', False)
         if message is None:
             message = "Added tag %s for changeset %s" % (name,
                 changeset.short_id)
         if date is None:
             date = datetime.datetime.now().strftime('%a, %d %b %Y %H:%M:%S')
         try:
             tag(self._repo, name, changeset._ctx.node(), message, local, user, date)
         except Abort as e:
             mercurial.tags.tag(self._repo, name, changeset._ctx.node(), message, local, user, date)
         except mercurial.error.Abort as e:
             raise RepositoryError(e.message)
         # Reinitialize tags
         self.tags = self._get_tags()
         tag_id = self.tags[name]
         return self.get_changeset(revision=tag_id)
     def remove_tag(self, name, user, message=None, date=None):
         """
         Removes tag with the given ``name``.
@@ @@ -188,27 +203,27 @@ class MercurialRepository(BaseRepository @@
         :raises TagDoesNotExistError: if tag with given name does not exists
         """
         if name not in self.tags:
             raise TagDoesNotExistError("Tag %s does not exist" % name)
         if message is None:
             message = "Removed tag %s" % name
         if date is None:
             date = datetime.datetime.now().strftime('%a, %d %b %Y %H:%M:%S')
         local = False
         try:
             tag(self._repo, name, nullid, message, local, user, date)
+            mercurial.tags.tag(self._repo, name, mercurial.commands.nullid, message, local, user, date)
             self.tags = self._get_tags()
         except Abort as e:
+        except mercurial.error.Abort as e:
             raise RepositoryError(e.message)
     @LazyProperty
     def bookmarks(self):
         """
         Gets bookmarks for this repository
         """
         return self._get_bookmarks()
     def _get_bookmarks(self):
         if self._empty:
             return {}
@@ @@ -247,95 +262,95 @@ class MercurialRepository(BaseRepository @@
         if hasattr(rev1, 'raw_id'):
             rev1 = getattr(rev1, 'raw_id')
         if hasattr(rev2, 'raw_id'):
             rev2 = getattr(rev2, 'raw_id')
         # Check if given revisions are present at repository (may raise
         # ChangesetDoesNotExistError)
         if rev1 != self.EMPTY_CHANGESET:
             self.get_changeset(rev1)
         self.get_changeset(rev2)
         if path:
-            file_filter = match_exact(path)
+            file_filter = mercurial.match.exact(path)
         else:
             file_filter = None
         return b''.join(patch.diff(self._repo, rev1, rev2, match=file_filter,
                           opts=diffopts(git=True,
         return b''.join(mercurial.patch.diff(self._repo, rev1, rev2, match=file_filter,
                           opts=mercurial.mdiff.diffopts(git=True,
                                         showfunc=True,
                                         ignorews=ignore_whitespace,
                                         context=context)))
     @classmethod
     def _check_url(cls, url, repoui=None):
         """
         Function will check given url and try to verify if it's a valid
         link. Sometimes it may happened that mercurial will issue basic
         auth request that can cause whole API to hang when used from python
         or other external calls.
         On failures it'll raise urllib2.HTTPError, exception is also thrown
         when the return code is non 200
         """
         # check first if it's not an local url
         if os.path.isdir(url) or url.startswith(b'file:'):
             return True
         if url.startswith(b'ssh:'):
             # in case of invalid uri or authentication issues, sshpeer will
             # throw an exception.
             sshpeer.instance(repoui or ui.ui(), url, False).lookup(b'tip')
+            mercurial.sshpeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')
             return True
         url_prefix = None
         if b'+' in url[:url.find(b'://')]:
             url_prefix, url = url.split(b'+', 1)
         handlers = []
-        url_obj = hg_url(url)
+        url_obj = mercurial.util.url(url)
         test_uri, authinfo = url_obj.authinfo()
         url_obj.passwd = b'*****'
         cleaned_uri = str(url_obj)
         if authinfo:
             # create a password manager
             passmgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
             passmgr.add_password(*authinfo)
             handlers.extend((httpbasicauthhandler(passmgr),
                              httpdigestauthhandler(passmgr)))
             handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),
                              mercurial.url.httpdigestauthhandler(passmgr)))
         o = urllib2.build_opener(*handlers)
         o.addheaders = [('Content-Type', 'application/mercurial-0.1'),
                         ('Accept', 'application/mercurial-0.1')]
         q = {"cmd": 'between'}
         q.update({'pairs': "%s-%s" % ('0' * 40, '0' * 40)})
         qs = '?%s' % urllib.urlencode(q)
         cu = "%s%s" % (test_uri, qs)
         req = urllib2.Request(cu, None, {})
         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 urllib2.URLError("[%s] org_exc: %s" % (cleaned_uri, e))
         if not url_prefix: # skip svn+http://... (and git+... too)
             # now check if it's a proper hg repo
             try:
                 httppeer.instance(repoui or ui.ui(), url, False).lookup(b'tip')
+                mercurial.httppeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')
             except Exception as e:
                 raise urllib2.URLError(
                     "url [%s] does not look like an hg repo org_exc: %s"
                     % (cleaned_uri, e))
         return True
     def _get_repo(self, create, src_url=None, update_after_clone=False):
         """
         Function will check for mercurial repository in given path and return
         a localrepo object. If there is no repository in that path it will
         raise an exception unless ``create`` parameter is set to True - in
@@ @@ -343,50 +358,50 @@ class MercurialRepository(BaseRepository @@
         If ``src_url`` is given, would try to clone repository from the
         location at given clone_point. Additionally it'll make update to
         working copy accordingly to ``update_after_clone`` flag
         """
         try:
             if src_url:
                 url = safe_bytes(self._get_url(src_url))
                 opts = {}
                 if not update_after_clone:
                     opts.update({'noupdate': True})
                 MercurialRepository._check_url(url, self.baseui)
                 clone(self.baseui, url, self.path, **opts)
+                mercurial.commands.clone(self.baseui, url, self.path, **opts)
                 # Don't try to create if we've already cloned repo
                 create = False
             return localrepo.instance(self.baseui, self.path, create=create)
         except (Abort, RepoError) as err:
             return mercurial.localrepo.instance(self.baseui, self.path, create=create)
         except (mercurial.error.Abort, mercurial.error.RepoError) as err:
             if create:
                 msg = "Cannot create repository at %s. Original error was %s" \
                     % (self.path, err)
             else:
                 msg = "Not valid repository at %s. Original error was %s" \
                     % (self.path, err)
             raise RepositoryError(msg)
     @LazyProperty
     def in_memory_changeset(self):
         return MercurialInMemoryChangeset(self)
     @LazyProperty
     def description(self):
         _desc = self._repo.ui.config(b'web', b'description', None, untrusted=True)
         return safe_unicode(_desc or b'unknown')
     @LazyProperty
     def contact(self):
         return safe_unicode(get_contact(self._repo.ui.config)
+        return safe_unicode(mercurial.hgweb.common.get_contact(self._repo.ui.config)
                             or b'Unknown')
     @LazyProperty
     def last_change(self):
         """
         Returns last change made on this repository as datetime object
         """
         return date_fromtimestamp(self._get_mtime(), makedate()[1])
     def _get_mtime(self):
         try:
             return time.mktime(self.get_changeset().date.timetuple())
@@ @@ -407,26 +422,26 @@ class MercurialRepository(BaseRepository @@
         """
         if self._empty:
             raise EmptyRepositoryError("There are no changesets yet")
         if revision in [-1, None]:
             revision = b'tip'
         elif isinstance(revision, unicode):
             revision = safe_bytes(revision)
         try:
             if isinstance(revision, int):
                 return ascii_str(self._repo[revision].hex())
             return ascii_str(scmutil.revsymbol(self._repo, revision).hex())
         except (IndexError, ValueError, RepoLookupError, TypeError):
             return ascii_str(mercurial.scmutil.revsymbol(self._repo, revision).hex())
         except (IndexError, ValueError, mercurial.error.RepoLookupError, TypeError):
             msg = ("Revision %s does not exist for %s" % (revision, self))
             raise ChangesetDoesNotExistError(msg)
         except (LookupError, ):
             msg = ("Ambiguous identifier `%s` for %s" % (revision, self))
             raise ChangesetDoesNotExistError(msg)
     def get_ref_revision(self, ref_type, ref_name):
         """
         Returns revision number for the given reference.
         """
         ref_name = safe_str(ref_name)
         if ref_type == 'rev' and not ref_name.strip('0'):
@@ @@ -436,25 +451,25 @@ class MercurialRepository(BaseRepository @@
                 'branch': 'branch',
                 'book': 'bookmark',
                 'tag': 'tag',
                 'rev': 'id',
+            }
         # avoid expensive branch(x) iteration over whole repo
         rev_spec = "%%s & %s(%%s)" % _revset_predicates[ref_type]
         try:
             revs = self._repo.revs(rev_spec, ref_name, ref_name)
         except LookupError:
             msg = ("Ambiguous identifier %s:%s for %s" % (ref_type, ref_name, self.name))
             raise ChangesetDoesNotExistError(msg)
         except RepoLookupError:
+        except mercurial.error.RepoLookupError:
             msg = ("Revision %s:%s does not exist for %s" % (ref_type, ref_name, self.name))
             raise ChangesetDoesNotExistError(msg)
         if revs:
             revision = revs.last()
         else:
             # TODO: just report 'not found'?
             revision = ref_name
         return self._get_revision(revision)
     def _get_archives(self, archive_name='tip'):
         allowed = self.baseui.configlist(b"web", b"allow_archive",
@@ @@ -524,46 +539,45 @@ class MercurialRepository(BaseRepository @@
             filter_.append(b'branch("%s")' % safe_bytes(branch_name))
         if start_date:
             filter_.append(b'date(">%s")' % start_date)
         if end_date:
             filter_.append(b'date("<%s")' % end_date)
         if filter_ or max_revisions:
             if filter_:
                 revspec = b' and '.join(filter_)
             else:
                 revspec = b'all()'
             if max_revisions:
                 revspec = b'limit(%s, %d)' % (revspec, max_revisions)
             revisions = scmutil.revrange(self._repo, [revspec])
+            revisions = mercurial.scmutil.revrange(self._repo, [revspec])
         else:
             revisions = self.revisions
         # this is very much a hack to turn this into a list; a better solution
         # would be to get rid of this function entirely and use revsets
         revs = list(revisions)[start_pos:end_pos]
         if reverse:
             revs.reverse()
         return CollectionGenerator(self, revs)
     def pull(self, url):
         """
         Tries to pull changes from external location.
         """
         url = self._get_url(url)
         other = peer(self._repo, {}, url)
+        other = mercurial.hg.peer(self._repo, {}, url)
         try:
             from mercurial import exchange
             exchange.pull(self._repo, other, heads=None, force=None)
         except Abort as err:
             mercurial.exchange.pull(self._repo, other, heads=None, force=None)
         except mercurial.error.Abort as err:
             # Propagate error but with vcs's type
             raise RepositoryError(str(err))
     @LazyProperty
     def workdir(self):
         """
         Returns ``Workdir`` instance for this repository.
         """
         return MercurialWorkdir(self)
     def get_config_value(self, section, name=None, config_file=None):
         """
@@ @@ -572,25 +586,25 @@ class MercurialRepository(BaseRepository @@
         :param section: Section we want to retrieve value from
         :param name: Name of configuration we want to retrieve
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         if config_file is None:
             config_file = []
         elif isinstance(config_file, basestring):
             config_file = [config_file]
         config = self._repo.ui
         if config_file:
             config = ui.ui()
+            config = mercurial.ui.ui()
             for path in config_file:
                 config.readconfig(path)
         return config.config(section, name)
     def get_user_name(self, config_file=None):
         """
         Returns user's name from global configuration file.
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         username = self.get_config_value('ui', 'username', config_file=config_file)

kallithea/lib/vcs/backends/hg/ssh.py

➞

Show inline comments

@@ @@ -5,26 +5,26 @@ @@
 # (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, see <http://www.gnu.org/licenses/>.
 import logging
 from mercurial import hg
 from mercurial.wireprotoserver import sshserver
 import mercurial.hg
 import mercurial.wireprotoserver
 from kallithea.lib.utils import make_ui
 from kallithea.lib.vcs.backends.ssh import BaseSshHandler
 from kallithea.lib.vcs.utils import safe_bytes, safe_unicode
 log = logging.getLogger(__name__)
 class MercurialSshHandler(BaseSshHandler):
     vcs_type = 'hg'
@@ @@ -52,15 +52,15 @@ class MercurialSshHandler(BaseSshHandler @@
         return None
     def __init__(self, repo_name):
         self.repo_name = repo_name
     def _serve(self):
         # Note: we want a repo with config based on .hg/hgrc and can thus not use self.db_repo.scm_instance._repo.ui
         baseui = make_ui(repo_path=self.db_repo.repo_full_path)
         if not self.allow_push:
             baseui.setconfig(b'hooks', b'pretxnopen._ssh_reject', b'python:kallithea.lib.hooks.rejectpush')
             baseui.setconfig(b'hooks', b'prepushkey._ssh_reject', b'python:kallithea.lib.hooks.rejectpush')
         repo = hg.repository(baseui, safe_bytes(self.db_repo.repo_full_path))
+        repo = mercurial.hg.repository(baseui, safe_bytes(self.db_repo.repo_full_path))
         log.debug("Starting Mercurial sshserver for %s", self.db_repo.repo_full_path)
         sshserver(baseui, repo).serve_forever()
+        mercurial.wireprotoserver.sshserver(baseui, repo).serve_forever()

kallithea/lib/vcs/backends/hg/workdir.py

➞

Show inline comments

 import mercurial.merge
 from kallithea.lib.vcs.backends.base import BaseWorkdir
 from kallithea.lib.vcs.exceptions import BranchDoesNotExistError
 from kallithea.lib.vcs.utils import ascii_bytes, ascii_str
 from kallithea.lib.vcs.utils.hgcompat import hg_merge
 class MercurialWorkdir(BaseWorkdir):
     def get_branch(self):
         return self.repository._repo.dirstate.branch()
     def get_changeset(self):
         wk_dir_id = ascii_str(self.repository._repo[None].parents()[0].hex())
         return self.repository.get_changeset(wk_dir_id)
     def checkout_branch(self, branch=None):
         if branch is None:
             branch = self.repository.DEFAULT_BRANCH_NAME
         if branch not in self.repository.branches:
             raise BranchDoesNotExistError
         raw_id = self.repository.branches[branch]
-        hg_merge.update(self.repository._repo, ascii_bytes(raw_id), False, False, None)
+        mercurial.merge.update(self.repository._repo, ascii_bytes(raw_id), False, False, None)

kallithea/lib/vcs/utils/hgcompat.py

➞

Show inline comments

 """
 Mercurial libs compatibility
 """
 import mercurial
 from mercurial import archival, config, demandimport, discovery, httppeer, localrepo
 from mercurial import merge as hg_merge
 from mercurial import obsutil, patch, scmutil, sshpeer, ui, unionrepo
 from mercurial.commands import clone, nullid, pull
 from mercurial.context import memctx, memfilectx
 from mercurial.discovery import findcommonoutgoing
 from mercurial.error import Abort, RepoError, RepoLookupError
 from mercurial.hg import peer
 from mercurial.hgweb import hgweb_mod
 from mercurial.hgweb.common import get_contact
 from mercurial.match import exact as match_exact
 from mercurial.match import match
 from mercurial.mdiff import diffopts
 from mercurial.node import hex, nullrev
 from mercurial.scmutil import revrange
 from mercurial.tags import tag
 from mercurial.url import httpbasicauthhandler, httpdigestauthhandler
 from mercurial.util import url as hg_url
 import mercurial.localrepo
 def monkey_do():
     """Apply some Mercurial monkey patching"""
 # workaround for 3.3 94ac64bcf6fe and not calling largefiles reposetup correctly, and test_archival failing
 localrepo.localrepository._lfstatuswriters = [lambda *msg, **opts: None]
+    mercurial.localrepo.localrepository._lfstatuswriters = [lambda *msg, **opts: None]
 # 3.5 7699d3212994 added the invariant that repo.lfstatus must exist before hitting overridearchive
 localrepo.localrepository.lfstatus = False
+    mercurial.localrepo.localrepository.lfstatus = False

kallithea/tests/vcs/test_hg.py

➞

Show inline comments

@@ @@ -226,33 +226,33 @@ VCS @@
 Various Version Control System management abstraction layer for Python.
 Introduction
 ------------
 TODO: To be written...
 """
         node = chset10.get_node('README.rst')
         assert node.kind == NodeKind.FILE
         assert node.content == readme
-    @mock.patch('kallithea.lib.vcs.backends.hg.repository.diffopts')
+    @mock.patch('mercurial.mdiff.diffopts')
     def test_get_diff_does_not_sanitize_zero_context(self, mock_diffopts):
         zero_context = 0
         self.repo.get_diff(0, 1, 'foo', context=zero_context)
         mock_diffopts.assert_called_once_with(git=True, showfunc=True, ignorews=False, context=zero_context)
-    @mock.patch('kallithea.lib.vcs.backends.hg.repository.diffopts')
+    @mock.patch('mercurial.mdiff.diffopts')
     def test_get_diff_sanitizes_negative_context(self, mock_diffopts):
         negative_context = -10
         zero_context = 0
         self.repo.get_diff(0, 1, 'foo', context=negative_context)
         mock_diffopts.assert_called_once_with(git=True, showfunc=True, ignorews=False, context=zero_context)
 class TestMercurialChangeset(object):
     def setup_method(self):

0 comments (0 inline, 0 general)