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

bootstrap: demo CommitService servicer tested with pytest

We use the `pytest-grpc` helpers.

This implements only the `FindCommit` RPC call, but in a complete
manner.

This raises several questions:

- for now, we have a single `servicer` module, but Gitaly has many
  services:  How shall we organize our code?
- the fixtures for gRPC server have the module scope. Shall we have
  such a fixture for each service?
- will we be able to create commits directly with the Gitaly API in
  the future?
- do we want to separate the tests in several layers, ie those that
  don't need a gRPC server (unit-testing the `changectx_to` functions)
  and those that do?
parent 7196784b950a
No related branches found
No related tags found
1 merge request!1Bootstrapping hgitaly development
......@@ -2,6 +2,9 @@
- main
- compat
before_script:
- pip3 install --user -r dev-requirements.txt
variables:
EVOLVE_REPO_URL: https://mirror.octobus.net/evolve
EVOLVE_LOCAL_REPO: /ci/repos/evolve
......
import pytest
@pytest.fixture(scope='module')
def grpc_add_to_server():
from hgitaly.stub.commit_pb2_grpc import add_CommitServiceServicer_to_server
return add_CommitServiceServicer_to_server
@pytest.fixture(scope='module')
def server_repos_root(tmp_path_factory):
return tmp_path_factory.mktemp("server-repos")
@pytest.fixture(scope='module')
def grpc_servicer(server_repos_root):
from hgitaly.servicer import Servicer
return Servicer(str(server_repos_root).encode())
@pytest.fixture(scope='module')
def grpc_stub_cls(grpc_channel):
from hgitaly.stub.commit_pb2_grpc import CommitServiceStub
return CommitServiceStub
import logging
import os
from mercurial import (
error as hg_error,
hg,
node,
ui as uimod,
)
from mercurial.utils import stringutil as hg_stringutil
from google.protobuf.timestamp_pb2 import Timestamp
from .stub.shared_pb2 import (
CommitAuthor,
GitCommit,
Repository,
)
from .stub.commit_pb2 import (
FindCommitRequest,
FindCommitResponse,
)
from .stub.commit_pb2_grpc import CommitServiceServicer
logger = logging.getLogger(__name__)
def changectx_to_git_author(ctx):
auth = ctx.user()
date = Timestamp()
# hg time resolution is the second, see
# https://www.mercurial-scm.org/wiki/ChangeSet
date.FromSeconds(int(ctx.date()[0]))
return CommitAuthor(
email=hg_stringutil.email(auth),
name=hg_stringutil.person(auth),
date=date,
)
def changectx_to_git_commit(ctx):
"""Return :class:`GitCommit` object from Mercurial :class:`changectx`.
subject and body are as in gitaly/internal/git/log/commitmessage.go::
var body string
if split := strings.SplitN(commitString, "\n\n", 2); len(split) == 2 {
body = split[1]
}
subject := strings.TrimRight(strings.SplitN(body, "\n", 2)[0], "\r\n")
See also slightly different stripping gitlab/lib/gitlab/git/commit.rb::
message_split = raw_commit.message.split("\n", 2)
Gitaly::GitCommit.new(
id: raw_commit.oid,
subject: message_split[0] ? message_split[0].chomp.b : "",
body: raw_commit.message.b,
parent_ids: raw_commit.parent_ids,
author: gitaly_commit_author_from_rugged(raw_commit.author),
committer: gitaly_commit_author_from_rugged(raw_commit.committer)
)
"""
descr = ctx.description()
author = changectx_to_git_author(ctx)
return GitCommit(id=ctx.hex(),
subject=descr.split(b'\n', 1)[0].rstrip(b'\r\n'),
body=descr,
parent_ids=[p.hex().decode()
for p in ctx.parents()
if p.rev() != node.nullrev],
author=author,
committer=author,
)
class Servicer(CommitServiceServicer):
"""CommitService implementation.
Attributes:
- `repositories_root`: should be given as bytes, since we'll have
to convert paths to bytes anyway, which is the only promise a filesystem
can make, and what Mercurial expects.
"""
def __init__(self, repositories_root):
self.repos_root = repositories_root
self.ui = uimod.ui.load()
def load_repo(self, repository: Repository):
# shamelessly taken from heptapod.wsgi for the Hgitaly bootstrap
# note that Gitaly Repository has more than just a relative path,
# we'll have to decide what we make of the extra information
rpath = repository.relative_path
if rpath.endswith('.git'):
rpath = rpath[:-4] + '.hg'
# GitLab filesystem paths are always ASCII
repo_path = os.path.join(self.repos_root, rpath.encode('ascii'))
logger.info("loading repo at %r", repo_path)
# ensure caller gets private copy of ui
return hg.repository(self.ui.copy(), repo_path)
def FindCommit(self,
request: FindCommitRequest, context) -> FindCommitResponse:
repo = self.load_repo(request.repository)
logger.debug("FindCommit revision=%r", request.revision)
try:
ctx = repo[request.revision]
except hg_error.RepoLookupError:
commit = None
else:
commit = changectx_to_git_commit(ctx)
return FindCommitResponse(commit=commit)
import time
from ..stub.commit_pb2 import FindCommitRequest
from ..stub.shared_pb2 import Repository
from ..testhelpers import LocalRepoWrapper
def test_find_commit(grpc_stub, server_repos_root):
repo_path = server_repos_root / 'repo1'
wrapper = LocalRepoWrapper.init(repo_path)
now = time.time()
ctx = wrapper.write_commit('foo',
utc_timestamp=now,
user="HGitaly Test <hgitaly@heptapod.test>")
ctx2 = wrapper.write_commit('foo',
parent=ctx,
message="Foo deux\n\nA very interesting bar")
request = FindCommitRequest(
repository=Repository(relative_path='repo1'),
revision=ctx.node())
response = grpc_stub.FindCommit(request)
commit = response.commit
assert commit is not None
assert commit.id == ctx.hex().decode()
assert commit.parent_ids == []
assert commit.author.name == b"HGitaly Test"
assert commit.author.email == b"hgitaly@heptapod.test"
assert commit.author.date.seconds == int(now)
request = FindCommitRequest(
repository=Repository(relative_path='repo1'),
revision=ctx2.node())
response = grpc_stub.FindCommit(request)
commit2 = response.commit
assert commit2 is not None
assert commit2.subject == b'Foo deux'
assert commit2.body == b"Foo deux\n\nA very interesting bar"
assert commit2.parent_ids == [ctx.hex().decode()]
# TODO check with two parents, it'd be nice to have a helper to create
# merge commits very quickly
request = FindCommitRequest(
repository=Repository(relative_path='repo1'),
revision=b'.' * 20)
response = grpc_stub.FindCommit(request)
assert not response.HasField('commit')
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