Skip to content
Snippets Groups Projects
Commit a03eb27e authored by Georges Racinet's avatar Georges Racinet
Browse files

tests: using heptapod.testhelpers

With py-heptapod supporting both Python 2 and 3, not so much
was left to make HGitaly tests use its test helpers directly,
ending this de facto fork.

This is also a first use case for the CI running against
the head of py-heptapod default branch.
parent 653a45dd
No related branches found
No related tags found
No related merge requests found
Pipeline #7020 passed with warnings
# Copyright 2019 Georges Racinet <georges.racinet@octobus.net>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
"""Helpers for automatic tests.
These allow both high level operation on testing repos, and lower level
calls and introspections, making it possible to test more exhaustively inner
code paths that with `.t` tests, which are really functional tests.
"""
import os
from mercurial import (
cmdutil,
commands,
context,
hg,
node,
phases,
ui as uimod,
)
import random
import time
# re-exports for stability
NULL_REVISION = node.nullrev
NULL_ID = node.nullid
def as_bytes(s):
"""whether s is a path, an str, or bytes, return a bytes representation."""
if isinstance(s, bytes):
return s
return str(s).encode('utf-8')
def make_ui(base_ui, config=None):
# let's make sure we aren't polluted by surrounding settings
os.environ['HGRCPATH'] = ''
if base_ui is None:
ui = uimod.ui.load()
else:
ui = base_ui.copy()
if config is not None:
for section_name, section in config.items():
for item_name, item_value in section.items():
ui.setconfig(as_bytes(section_name),
as_bytes(item_name),
as_bytes(item_value),
source='tests')
return ui
class LocalRepoWrapper(object):
def __init__(self, repo):
self.repo = repo
@classmethod
def init(cls, path, base_ui=None, config=None):
path = str(path)
bytes_path = as_bytes(path)
init = cmdutil.findcmd(b'init', commands.table)[1][0]
ui = make_ui(base_ui, config)
init(ui, dest=bytes_path)
return cls(hg.repository(ui, bytes_path))
@classmethod
def load(cls, path, base_ui=None, config=None):
ui = make_ui(base_ui, config=config)
return cls(hg.repository(ui, as_bytes(path)))
def command(self, name, *args, **kwargs):
cmd = cmdutil.findcmd(as_bytes(name), commands.table)[1][0]
repo = self.repo
return cmd(repo.ui, repo, *args, **kwargs)
def write_commit(self, rpath,
content=None, message=None,
parent=None, branch=None,
user=None,
utc_timestamp=None):
"""Write content at rpath and commit in one call.
This is meant to allow fast and efficient preparation of
testing repositories. To do so, it goes a bit lower level
than the actual commit command, so is not suitable to test specific
commit options, especially if through extensions.
This leaves the working directoy updated at the new commit.
:param rpath: relative path from repository root. If existing,
will be overwritten by `content`
:param content: what's to be written in ``rpath``.
If not specified, will be replaced by random content.
:param message: message commit. If not specified, defaults to
``content``
:param parent: changectx value. If specified, the repository is
updated to it first. Useful to produce branching
histories. This is single valued, because the purpose
of this method is not to produce merge commits.
:param user: full user name and email, as in ``ui.username`` config
option. Can be :class:`str` or :class:`bytes`
:param utc_timestamp: seconds since Epoch UTC. Good enough for
tests without ambiguity. Can be float (only
seconds will be kept). Defaults to
``time.time()``
:returns: changectx for the resulting commit.
"""
repo = self.repo
path = os.path.join(repo.root, as_bytes(rpath))
if parent is not None:
if isinstance(parent, context.changectx):
parent = parent.node()
self.update_bin(parent)
if content is None:
content = "random: {}\n\nparent: {}\n".format(
random.random(),
node.hex(repo.dirstate.p1()))
if message is None:
message = content
if branch is not None:
self.repo.dirstate.setbranch(as_bytes(branch))
flags = 'wb' if isinstance(content, bytes) else 'w'
with open(path, flags) as fobj:
fobj.write(content)
if isinstance(user, str):
user = user.encode('utf-8')
if utc_timestamp is None:
utc_timestamp = time.time()
def commitfun(ui, repo, message, match, opts):
return self.repo.commit(message,
user,
(int(utc_timestamp), 0),
match=match,
editor=False,
extra=None)
new_node = cmdutil.commit(repo.ui, repo, commitfun, (path, ),
{b'addremove': True,
b'message': as_bytes(message),
})
return repo[new_node]
def update_bin(self, bin_node):
"""Update to a revision specified by its node in binary form.
This is separated in order to avoid ambiguities
"""
# maybe we'll do something lower level later
self.update(node.hex(bin_node))
def update(self, rev):
repo = self.repo
cmdutil.findcmd(b'update', commands.table)[1][0](repo.ui, repo,
as_bytes(rev))
def set_phase(self, phase_name, revs, force=True):
repo = self.repo
if isinstance(revs, str):
revs = [as_bytes(revs)]
elif isinstance(revs, bytes):
revs = [revs]
else:
revs = [as_bytes(revset) for revset in revs]
opts = dict(force=force, rev=revs)
opts.update((phn.decode(), phn == as_bytes(phase_name))
for phn in phases.cmdphasenames)
cmdutil.findcmd(b'phase', commands.table)[1][0](repo.ui, repo, **opts)
import time
from heptapod.testhelpers import LocalRepoWrapper
from ..stub.commit_pb2 import FindCommitRequest
from ..stub.commit_pb2_grpc import CommitServiceStub
from ..stub.shared_pb2 import Repository
......@@ -2,8 +3,7 @@
from ..stub.commit_pb2 import FindCommitRequest
from ..stub.commit_pb2_grpc import CommitServiceStub
from ..stub.shared_pb2 import Repository
from ..testhelpers import LocalRepoWrapper
def test_default_branch(grpc_channel, server_repos_root):
......
# Copyright 2019-2020 Georges Racinet <georges.racinet@octobus.net>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
from __future__ import absolute_import
from mercurial import (
extensions,
phases,
ui as uimod,
)
from ..testhelpers import (
LocalRepoWrapper,
NULL_ID,
NULL_REVISION,
)
import hgext.strip # used as an example
HGEXT_STRIP_SOURCE = hgext.strip.__file__.replace('.pyc', '.py')
def test_init_write_commit(tmpdir):
wrapper = LocalRepoWrapper.init(tmpdir)
node = wrapper.write_commit('foo', content='Foo', message='Foo committed')
ctx = wrapper.repo[node]
assert ctx.description() == b'Foo committed'
parents = ctx.parents()
assert len(parents) == 1
assert parents[0].rev() == NULL_REVISION
assert parents[0].node() == NULL_ID
del wrapper, ctx
reloaded = LocalRepoWrapper.load(tmpdir)
rl_ctx = reloaded.repo[node]
assert rl_ctx.description() == b'Foo committed'
def assert_is_strip_ext(strip_ext):
assert strip_ext is not None
# it's imported with a different name, hence can't be directly compared
# let's also avoid flakiness due to __file__ behaviour depending on
# installation context
assert strip_ext.__doc__ == hgext.strip.__doc__
def test_load_hgrc_extension(tmpdir):
LocalRepoWrapper.init(tmpdir)
tmpdir.join('.hg', 'hgrc').write('\n'.join((
"[extensions]",
"strip="
)))
wrapper = LocalRepoWrapper.load(tmpdir, config=dict(foo=dict(bar='17')))
exts = dict(extensions.extensions(wrapper.repo.ui))
assert_is_strip_ext(exts.get(b'strip'))
assert wrapper.repo.ui.configint(b'foo', b'bar') == 17
def test_init_baseui_config_extension(tmpdir):
ui = uimod.ui.load()
ui.setconfig(b'foo', b'bar', b'yes', source='tests')
ui.setconfig(b'extensions', b'strip', b'', source='tests')
wrapper = LocalRepoWrapper.init(tmpdir, base_ui=ui)
assert wrapper.repo.ui.configbool(b'foo', b'bar')
exts = dict(extensions.extensions(wrapper.repo.ui))
assert_is_strip_ext(exts.get(b'strip'))
def test_init_config_extension(tmpdir):
ui = uimod.ui.load()
ui.setconfig(b'foo', b'bar', b'yes', source='tests')
ui.setconfig(b'extensions', b'strip', b'', source='tests')
wrapper = LocalRepoWrapper.init(
tmpdir,
config=dict(foo=dict(bar='yes'),
extensions=dict(strip=''),
))
assert wrapper.repo.ui.configbool(b'foo', b'bar')
exts = dict(extensions.extensions(wrapper.repo.ui))
assert_is_strip_ext(exts.get(b'strip'))
def test_update(tmpdir):
wrapper = LocalRepoWrapper.init(tmpdir)
wrapper.write_commit('foo', content='Foo 0')
ctx1 = wrapper.write_commit('foo', content='Foo 1')
foo = tmpdir.join('foo')
assert foo.read() == 'Foo 1'
wrapper.update('0')
assert foo.read() == 'Foo 0'
wrapper.update_bin(NULL_ID)
assert not foo.isfile()
wrapper.update_bin(ctx1.node())
assert foo.read() == 'Foo 1'
def test_write_commit_named_branch(tmpdir):
"""Demonstrate the use of write_commit with parent."""
wrapper = LocalRepoWrapper.init(tmpdir)
ctx0 = wrapper.write_commit('foo', content='Foo 0')
wrapper.write_commit('foo', content='Foo 1')
nodebr = wrapper.write_commit('foo', content='Foo branch',
parent=ctx0.node(), branch='other')
ctxbr = wrapper.repo[nodebr]
assert ctxbr.branch() == b'other'
assert all(c == ctx0 for c in ctxbr.parents())
def test_write_commit_wild_branch(tmpdir):
"""Demonstrate the use of write_commit with parent."""
wrapper = LocalRepoWrapper.init(tmpdir)
ctx0 = wrapper.write_commit('foo', content='Foo 0')
wrapper.write_commit('foo', content='Foo 1')
nodebr = wrapper.write_commit('foo', content='Foo branch',
parent=ctx0)
ctxbr = wrapper.repo[nodebr]
assert ctxbr.branch() == b'default'
assert all(c == ctx0 for c in ctxbr.parents())
def test_write_commit_random(tmpdir):
"""Demonstrate how random content is generated."""
wrapper = LocalRepoWrapper.init(tmpdir)
ctx0 = wrapper.write_commit('foo')
ctx1 = wrapper.write_commit('foo', parent=ctx0)
ctx2 = wrapper.write_commit('foo', parent=ctx0)
assert ctx1.p1() == ctx2.p1()
assert ctx1 != ctx2
def test_write_commit_user_date(tmpdir):
"""Test timestamp and username options."""
wrapper = LocalRepoWrapper.init(tmpdir)
ctx = wrapper.write_commit('foo',
utc_timestamp=1578604387,
user="Test GR <testgr@hg.test>")
assert ctx.date() == (1578604387, 0)
assert ctx.user() == b"Test GR <testgr@hg.test>"
def test_phase(tmpdir):
wrapper = LocalRepoWrapper.init(tmpdir)
node = wrapper.write_commit('foo', content='Foo 0')
ctx = wrapper.repo[node]
assert ctx.phase() == phases.draft
wrapper.set_phase('public', ['.'], force=False)
assert ctx.phase() == phases.public
wrapper.set_phase('draft', ['.'], force=True)
assert ctx.phase() == phases.draft
# also works directly with a single revspec, either as str
wrapper.set_phase('draft', '.', force=False)
assert ctx.phase() == phases.draft
# or as bytes
wrapper.set_phase('draft', b'.', force=True)
assert ctx.phase() == phases.draft
def test_command(tmpdir):
"""Test the generic command wrapper."""
wrapper = LocalRepoWrapper.init(tmpdir)
wrapper.write_commit('foo', content='Foo 0')
path = tmpdir.join('foo')
wrapper.command('rm', str(path).encode())
assert not(path.exists())
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment