import json from urllib.parse import parse_qs import pytest import re import requests from .utils import needs from .utils.reverse_calls import HttpListener from .utils.hg import LocalRepo from .utils.project import ( extract_gitlab_branch_titles, ProjectAccess, ) parametrize = pytest.mark.parametrize def prepare_simple_repo(proj, repo_path): repo = LocalRepo.init(repo_path, default_url=proj.owner_basic_auth_url) repo_path.join('foo').write('foo0') repo.hg('commit', '-Am', "Commit 0") repo.hg('phase', '-p', ".") repo.hg('topic', 'zetop') repo_path.join('foo').write('foo1') repo.hg('commit', '-Am', "Commit 1") repo.hg('push') return repo def test_push_basic(test_project, tmpdir): """ Push two changesets, one being a draft, confirm their arrival and phases """ repo = prepare_simple_repo(test_project, tmpdir.join('repo1')) # let's control what GitLab really sees assert test_project.api_branch_titles() == { 'branch/default': 'Commit 0', 'topic/default/zetop': 'Commit 1', } url = test_project.owner_basic_auth_url clone = LocalRepo.clone(url, tmpdir.join('repo2')) log = clone.hg('log', '-T', '{desc}:{phase}:{topic}\n') assert log.splitlines() == ['Commit 1:draft:zetop', 'Commit 0:public:'] repo.path.join('foo').write('foo2') repo.hg('commit', '-Am', "Commit 2") repo.hg('push', url) clone.hg('pull', url) log = clone.hg('log', '-T', '{desc}:{phase}:{topic}\n') assert log.splitlines() == ['Commit 2:draft:zetop', 'Commit 1:draft:zetop', 'Commit 0:public:'] # now GitLab side: branches = test_project.api_branches() assert extract_gitlab_branch_titles(branches) == { 'branch/default': 'Commit 0', 'topic/default/zetop': 'Commit 2', } webdriver = test_project.owner_webdriver webdriver.get('/'.join((test_project.url, 'commit', branches['branch/default']['commit']['id']))) # we don't have a reliable way to detect an error # the title would be just 500 in production and the error class name in # development mode. assert "Error" not in webdriver.title assert "(500)" not in webdriver.title assert 'Commit 0' in webdriver.title hgsha = repo.hg('log', '-T', '{node}', '-r', '0') assert hgsha[:11] in webdriver.title @needs.reverse_call def test_web_hook(tmpdir, test_project): """Setup a webhook, listen to it, push and record the resulting POST """ listener = HttpListener(test_project.heptapod) repo_path = tmpdir.join('repo1') url = test_project.owner_basic_auth_url test_project.owner_api_post( subpath='hooks', data=dict( id=test_project.id, url=listener.url(test_project.heptapod), push_events=True, enable_ssl_verification=False, ) ) repo = LocalRepo.init(repo_path) repo_path.join('foo').write('foo0') repo.hg('commit', '-Am', "Commit 0") repo.hg('push', '--publish', url) timeout = 30 exit_code = listener.join(timeout) assert exit_code is not None, "No request received in %d seconds" % timeout posted = listener.queue.get() environ = posted['environ'] # stuff that we want to be altered about if it changes assert environ['CONTENT_TYPE'] == 'application/json' assert environ['HTTP_X_GITLAB_EVENT'] == 'Push Hook' hook_data = json.loads(posted['body']) assert hook_data['ref'] == 'refs/heads/branch/default' hgsha = repo.node('.') commits = hook_data.get('commits') assert commits is not None assert len(commits) == 1 commit = commits[0] assert commit['message'].strip() == "Commit 0" assert hook_data['checkout_hgsha'] == hgsha assert hook_data['hg_before'] == '0' * 40 assert hook_data['hg_after'] == hgsha assert commit['hgid'] == hgsha @needs.reverse_call def test_readthedocs_hook(tmpdir, test_project): """Setup the readthedocs integration, push and assert what was posted. There's no point pushing on several named branches: each will be considered a new event on the GitLab side, giving rise to as many POST requests to Read the Docs. """ listener = HttpListener(test_project.heptapod) repo_path = tmpdir.join('repo1') url = test_project.owner_basic_auth_url token = "rtdtkxxx" resp = test_project.owner_api_put( subpath='services/readthedocs', data=dict( project_url=listener.url(test_project.heptapod), token=token, # TLS verification seems inert over plain HTTP active=True, ) ) assert resp.status_code < 400 repo = LocalRepo.init(repo_path) repo_path.join('foo').write('foo0') repo.hg('commit', '-Am', "Commit 0") repo.hg('branch', 'other') repo.hg('push', '--publish', url) timeout = 30 exit_code = listener.join(timeout) assert exit_code is not None, "No request received in %d seconds" % timeout posted = listener.queue.get() environ = posted['environ'] # otherwise this test would be in trouble: assert environ['CONTENT_TYPE'] == 'application/x-www-form-urlencoded' hook_data = parse_qs(posted['body'].decode()) assert hook_data['branches'] == ['default'] assert hook_data['token'] == [token] @needs.fs_access @parametrize('push_proto', ['ssh', 'http']) def test_push_hook_env(test_project, tmpdir, push_proto): repo_path = tmpdir.join('repo1') repo = LocalRepo.init(repo_path) repo_path.join('foo').write('foo0') repo.hg('commit', '-Am', "Commit 0") test_project.extend_hgrc( "[hooks]", "pretxnchangegroup.heptapod_env=python:" "heptapod.hooks.dev_util.print_heptapod_env") if push_proto == 'http': out = repo.hg('push', test_project.owner_basic_auth_url) elif push_proto == 'ssh': out = repo.hg('push', '--ssh', *test_project.owner_ssh_params) expected_env = [('HEPTAPOD_PROJECT_ID', str(test_project.id)), ('HEPTAPOD_PROJECT_NAMESPACE_FULL_PATH', 'root'), ('HEPTAPOD_PROJECT_PATH', test_project.name), ('HEPTAPOD_REPOSITORY_USAGE', 'project'), ('HEPTAPOD_USERINFO_EMAIL', 'admin@example.com'), # root id, good enough for now: ('HEPTAPOD_USERINFO_ID', '1'), ('HEPTAPOD_USERINFO_NAME', 'Administrator'), ('HEPTAPOD_USERINFO_USERNAME', 'root'), ] assert 'remote: %r' % expected_env in out @needs.fs_access def test_push_corrupted(test_project, tmpdir): """If Git repo has more commits (issue hg-git#3).""" url = test_project.owner_basic_auth_url test_project.load_tarball('issue-3.tar') # at the time this tarball has been recorded, we were using inner SSH # push which today would not work test_project.extend_hgrc('[heptapod]', 'mirror-path=') repo = LocalRepo.clone(url, tmpdir.join('repo')) # verifying the tarball has been loaded ok assert repo.branches() == {'default', 'default:topic1'} repo.hg('up', 'default') # just to be explicit repo.path.join('foo').write('fromtest') repo.hg('commit', '-Am', "fromtest") repo.hg('push', test_project.owner_basic_auth_url) def test_protected_branch_refusal(test_project, tmpdir): """Refusals due to branch protection are explained to the end user.""" repo_path = tmpdir.join('repo1') url = test_project.owner_basic_auth_url repo = LocalRepo.init(repo_path) # let's have two branches repo_path.join('foo').write('foo0') repo.hg('commit', '-Am', "Commit 0") repo_path.join('foo').write('foo1') repo.hg('commit', '-Am', "Commit 1") repo.hg('push', url) # make sure 'default' is the GitLab default branch repo.hg('up', '0') repo.hg('branch', 'a_branch') repo_path.join('foo').write('in branch') repo.hg('commit', '-Am', "Commit in branch") repo.hg('phase', '-pr', "default+a_branch") print(repo.graphlog()) repo.hg('push', '--new-branch', url) # now let's protect the branch resp = test_project.owner_api_post(subpath='protected_branches', data=dict(name='branch/a_branch')) assert resp.status_code in (200, 201, 202) # Later on, we'll have a 'publisher' role. Meanwhile, let's # avoid the check_publish hook to prevent developer pushing. test_project.owner_api_hgrc_set(inherit=True, auto_publish='nothing') test_project.grant_member_access(user_name='test_basic', level=ProjectAccess.DEVELOPER) basic_user_url = test_project.basic_auth_url(user_name='test_basic') basic_user_ssh_params = test_project.ssh_params('test_basic') repo_path.join('foo').write('still in branch') repo.hg('commit', '-m', 'New commit that only maintainer can push') code, out, err = repo.hg_unchecked('push', '--ssh', *basic_user_ssh_params) assert code != 0 assert re.search("push.*protected branches", out) code, out, err = repo.hg_unchecked('push', basic_user_url) assert code != 0 assert re.search("push.*protected branches", out) # it is also forbidden to remove protected branches repo.hg('commit', '--close-branch', '-m', "closing") code, out, err = repo.hg_unchecked('push', basic_user_url) assert code != 0 # let's not depend too much on exact message wording assert "delete protected branches" in out # the failed pushes had indeed no effect on the hg repo clone = LocalRepo.clone(url, tmpdir.join('repo2')) log = clone.hg('log', '-T', '{desc}:{branch}\n').splitlines() assert set(l.strip() for l in log) == { 'Commit 0:default', 'Commit 1:default', 'Commit in branch:a_branch', } def test_protected_branch_ultimate(test_project, tmpdir): """Mode in which even maintainers can't push.""" repo_path = tmpdir.join('repo1') url = test_project.owner_basic_auth_url repo = LocalRepo.init(repo_path) # let's have two branches repo_path.join('foo').write('foo0') repo.hg('commit', '-Am', "Commit 0") repo_path.join('foo').write('foo1') repo.hg('commit', '-Am', "Commit 1") repo.hg('push', url) # make sure 'default' is the GitLab default branch repo.hg('up', '0') repo.hg('branch', 'a_branch') repo_path.join('foo').write('in branch') repo.hg('commit', '-Am', "Commit in branch") repo.hg('phase', '-pr', "default+a_branch") print(repo.graphlog()) repo.hg('push', '--new-branch', url) resp = test_project.owner_api_post(subpath='protected_branches', data=dict(name='branch/a_branch', push_access_level=0)) assert resp.status_code in (200, 201, 202) repo_path.join('foo').write('still in branch') repo.hg('commit', '-m', 'New commit that even Owner cannot push') ssh_params = test_project.owner_ssh_params code, out, err = repo.hg_unchecked('push', '--ssh', *ssh_params) assert code != 0 assert re.search("push.*protected branches", out) code, out, err = repo.hg_unchecked('push', url) assert code != 0 assert re.search("push.*protected branches", out) # regular owner is also instance-wide administrator, and should # arguably still be allowed to remove the branch, let's make a Maintainer test_project.grant_member_access(user_name='test_basic', level=ProjectAccess.MAINTAINER) maintainer_url = test_project.basic_auth_url(user_name='test_basic') repo.hg('commit', '--close-branch', '-m', "closing") code, out, err = repo.hg_unchecked('push', maintainer_url) assert code != 0 # let's not depend too much on exact message wording assert "delete protected branches" in out # the failed pushes had indeed no effect on the hg repo clone = LocalRepo.clone(url, tmpdir.join('repo2')) log = clone.hg('log', '-T', '{desc}:{branch}\n').splitlines() assert set(l.strip() for l in log) == { 'Commit 0:default', 'Commit 1:default', 'Commit in branch:a_branch', } # far from being completely blocked the newly appointed Maintainer # can still accept MRs repo.hg('topic', '-r', '.~1', 'zetop') repo.hg('push', '-r', 'zetop', maintainer_url) mr_id = test_project.api_create_merge_request( source='topic/a_branch/zetop', target='branch/a_branch', ) # TODO looks like it's really time to put this kind of MR handling # down to the project level or on a separate MergeRequest object test_project.wait_assert_merge_request( mr_id, lambda mr: mr.get('merge_status') == 'can_be_merged', msg="Mergeability wrong or still unknown") # we have nicer helpers for API requests from other users than owner # in the default branch but can't use them right away (stable branch) headers = { 'Private-Token': test_project.heptapod.get_user('test_basic').token, } resp = requests.put('%s/merge_requests/%d/merge' % (test_project.api_url, mr_id), headers=headers, data=dict(merge_request_iid=mr_id, id=test_project.id)) assert resp.status_code < 300 assert resp.json()['state'] == 'merged' clone.hg('pull') log = clone.hg('log', '-T', '{desc}:{branch}:{phase}\n').splitlines() assert set(l.strip() for l in log) == { 'Commit 0:default:public', 'Commit 1:default:public', 'Commit in branch:a_branch:public', 'New commit that even Owner cannot push:a_branch:public', } @needs.docker def test_push_corrupted_2(test_project, tmpdir): """If Git repo has more commits in a topic (issue heptapod#119).""" repo_path = tmpdir.join('repo1') url = test_project.owner_basic_auth_url repo = LocalRepo.init(repo_path) repo_path.join('foo').write('foo0') repo.hg('commit', '-Am', "Commit 0") repo.hg('phase', '-p', ".") repo.hg('topic', 'zetop') repo_path.join('foo').write('foo1') repo.hg('commit', '-Am', "Commit 1") repo_path.join('foo').write('foo2') repo.hg('commit', '-Am', "Commit 2") repo.hg('push', url) sha = repo.hg('log', '-T', '{node}', '-r', '.').strip() # now, to reproduce, we'll strip Commit 2 directly from the remote. # Not setting HGRCPATH in Heptapod >= 0.8 implies that the inner Git repo # is untouched. heptapod = test_project.heptapod # TODO make a server-side hg method on Heptapod heptapod.run_shell((heptapod.hg_executable, '--config', 'extensions.strip=', '-R', test_project.fs_path, 'strip', sha)) # let's be sure that GitLab still has Commit 2 branches = test_project.api_branches() assert set(branches) == {'branch/default', 'topic/default/zetop'} assert branches['topic/default/zetop']['commit']['title'] == 'Commit 2' # but cloning the repo shows that Commit 2 isn't there clone = LocalRepo.clone(url, tmpdir.join('repo2')) log = clone.hg('log', '-T', '{desc}:{phase}:{topic}\n') assert log.splitlines() == ['Commit 1:draft:zetop', 'Commit 0:public:'] # now, pushing again we should gracefully override the Git remote repo.hg('push', url) branches = test_project.api_branches() assert branches['topic/default/zetop']['commit']['title'] == 'Commit 2' # and the Mercurial repo now has commit 2 clone.hg('pull', url) log = clone.hg('log', '-T', '{desc}:{phase}:{topic}', '-l', 1) assert log == 'Commit 2:draft:zetop' def test_push_tags_branch_heads(test_project, tmpdir): repo_path = tmpdir.join('repo1') # Minimal repo with two branches and one root repo = LocalRepo.init(repo_path, default_url=test_project.owner_basic_auth_url) repo_path.join('foo').write('foo0') repo.hg('commit', '-Am', "Commit 0") repo_path.join('foo').write('foo1') repo.hg('commit', '-Am', "Commit 1") repo.hg('push') repo.hg('up', '0') repo.hg('branch', 'other') repo_path.join('foo').write("other") repo.hg('commit', '-m', "Commit 2") # to reproduce #97, we need the tag commit not to # be the head of the 'other' branch repo.hg('up', '1') repo.hg('tag', 'other-1.0', '-r', '2') print(repo.graphlog()) repo.hg('push', '--new-branch') branches = test_project.api_branches() assert set(branches) == {'branch/default', 'branch/other'} assert branches['branch/other']['commit']['title'] == 'Commit 2' tags = test_project.api_tags() assert set(tags) == {'other-1.0'} assert tags['other-1.0']['commit']['title'] == "Commit 2" def test_push_tag_ci_job(test_project_with_runner, tmpdir): proj, runner = test_project_with_runner repo_path = tmpdir.join('repo1') repo = LocalRepo.init(repo_path, default_url=proj.owner_basic_auth_url) repo.init_gitlab_ci() repo.hg('tag', '1.0') repo.hg('push') jobs = runner.wait_assert_jobs(2) tag_jobs = [j for j in jobs if j['git_info']['ref_type'] == 'tag'] assert len(tag_jobs) == 1 job_vars = {details['key']: details for details in tag_jobs[0]['variables']} assert job_vars['CI_COMMIT_TAG']['value'] == '1.0' assert job_vars['CI_BUILD_TAG']['value'] == '1.0' def test_internal_force_push_default_branch(test_project, tmpdir): """ Although protected, but still accept internal Git force-pushes. See heptapod#129 """ repo_path = tmpdir.join('repo1') url = test_project.owner_basic_auth_url # internal force pushes occur upon amending of drafts without topics. test_project.owner_api_hgrc_set(inherit=True, auto_publish='nothing') repo = LocalRepo.init(repo_path) in_repo = repo_path.join('foo') in_repo.write('foo0') repo.hg('commit', '-Am', "Commit 0") repo.hg('phase', '-p', ".") in_repo.write('foo1') repo.hg('commit', '-Am', "Commit 1") repo.hg('push', url) assert test_project.api_default_branch() == 'branch/default' assert test_project.api_branch_titles() == {'branch/default': 'Commit 1'} in_repo.write('amended') repo.hg('amend', '-m', "Amended") repo.hg('push', url) assert test_project.api_branch_titles() == {'branch/default': 'Amended'} clone = LocalRepo.clone(url, tmpdir.join('repo2')) log = clone.hg('log', '-T', '{desc}:{phase}\n') assert log.splitlines() == ['Amended:draft', 'Commit 0:public'] def test_push_subrepos(test_project, tmpdir): """ The server just ignores subrepos in ordinary pushes. """ repo_path = tmpdir.join('repo1') other_path = tmpdir.join('repo2') nested_path = repo_path.join('nested') url = test_project.owner_basic_auth_url repo = LocalRepo.init(repo_path) other = LocalRepo.init(other_path) nested = LocalRepo.init(nested_path) repo_path.join('foo').write('foo0') repo.hg('commit', '-Am', "Commit 0") repo.hg('phase', '-p', ".") nested_path.join('bar').write('nestedbar') nested.hg('commit', '-Am', "Nested Bar 0") nested_sha = nested.hg('log', '-T', '{node}', '-r', '.') repo_path.join('.hgsub').write("nested = file://%s\n" % other_path) repo.hg('add', '-S') repo.hg('commit', '-Am', "Subrepos config") # subrepositories system did its job substate = repo_path.join('.hgsubstate') assert substate.exists() assert substate.read().split() == [nested_sha, 'nested'] repo.hg('push', url) # no problem with subsequent pushes repo_path.join('foo').write('foo1') repo.hg('commit', '-Am', "Commit 1") repo.hg('push', url) # let's control what GitLab really sees assert test_project.api_branch_titles() == { 'branch/default': 'Commit 1', } # the subrepo also got pushed successfully assert other.hg('log', '-T', '{node}') == nested_sha