Commit 5d7d7012 authored by Georges Racinet's avatar Georges Racinet 🦑
Browse files

Merged stable branch into default

and fixing conflicts in tests for external users due
to different page structure with GitLab 13.11
Pipeline #23190 failed with stage
in 27 minutes and 55 seconds
from io import BytesIO
import json
from .utils import (
unique_name,
)
from .utils.git import LocalRepo as GitRepo
from .utils.project import ProjectAccess
from .utils.project import (
Project,
ProjectAccess
)
from .test_merge_requests import api_accept_merge_request
......@@ -269,3 +276,33 @@ def test_fork_mergerequest_api_explicit_merge_message(
commit_links = proj.webdriver_get_merge_request_commits(mr_id)
assert any(expected in l.text for l in commit_links), (
"Commit summaries don't contain %r" % expected)
def test_export_import(git_project, tmpdir):
proj = git_project
repo_path = tmpdir.join('repo')
repo = GitRepo.init(repo_path, default_url=proj.owner_basic_auth_url)
repo_path.join('foo').write("Hey this is in Git!\n")
repo.git('add', 'foo')
repo.git('commit', '-m', 'Commit 0')
repo.git('push', '--set-upstream', 'origin', 'master')
tarball = BytesIO()
git_project.api_export(tarball, timeout=30)
tarball.seek(0)
with Project.api_import_tarball(
proj.heptapod,
proj.owner,
unique_name('test_import'),
tarball,
) as imported:
assert imported.vcs_type == git_project.vcs_type
assert imported.api_branch_titles() == {
'master': 'Commit 0',
}
clone = GitRepo.clone(imported.owner_basic_auth_url,
tmpdir / 'imported-clone')
assert clone.sha('master') == repo.sha('master')
from io import BytesIO
import pytest
import requests
from selenium.common.exceptions import (
......@@ -330,7 +331,8 @@ def test_housekeeping(test_project):
assert resp.status_code == 201
def test_external_user_can_create_project(test_group, external_user):
@parametrize('how', ['bumped-quota', 'group-member'])
def test_external_user_can_create_project(how, test_group, external_user):
heptapod = test_group.heptapod
webdriver = external_user.webdriver
......@@ -344,8 +346,12 @@ def test_external_user_can_create_project(test_group, external_user):
with pytest.raises(NoSuchElementException):
webdriver.find_element_by_xpath(dashboard_new_proj_link_xpath)
resp = test_group.api_add_member(external_user, ProjectAccess.DEVELOPER)
assert resp.status_code == 201
if how == 'bumped-quota':
external_user.edit(projects_limit=1)
elif how == 'group-member':
resp = test_group.api_add_member(external_user,
ProjectAccess.DEVELOPER)
assert resp.status_code == 201
# Now we have this big link in the middle of the page
webdriver.get(heptapod.url)
......@@ -382,31 +388,41 @@ def test_external_user_can_create_project(test_group, external_user):
lambda d: d.find_element_by_xpath(
'//a[@href = "#blank_project"]')))
# The default selected namespace is test_group
# (hoping its id won't change too much with GitLab versions)
namespace_select_id = 'project_namespace_id'
WebDriverWait(webdriver, 10).until(
lambda d: d.find_element_by_id(namespace_select_id).is_displayed)
namespace_select = webdriver.find_element_by_id(namespace_select_id)
selected = namespace_select.find_element_by_xpath(
'.//option[@selected="selected"]')
assert selected.get_attribute('data-name').strip() == test_group.path
if how == 'group-member':
namespace = test_group
# The default selected namespace is test_group
# (hoping its id won't change too much with GitLab versions)
namespace_select_id = 'project_namespace_id'
WebDriverWait(webdriver, 10).until(
lambda d: d.find_element_by_id(namespace_select_id).is_displayed)
namespace_select = webdriver.find_element_by_id(namespace_select_id)
selected = namespace_select.find_element_by_xpath(
'.//option[@selected="selected"]')
assert selected.get_attribute('data-name').strip() == test_group.path
elif how == 'bumped-quota':
namespace = external_user.personal_namespace
# creation in personal namespace is the only possibility.
# Page structure is different
namespace_elt = webdriver.find_element_by_css_selector(
'.input-group-text')
WebDriverWait(webdriver, 10).until(
lambda _: namespace_elt.is_displayed)
assert namespace_elt.text.strip().rstrip('/') == namespace.url
project = None
try:
Project.webdriver_new_project_submit(webdriver, 'ext_create')
# external_user doesn't have an access token, making one just to
# use it once would be too costly.
project = Project.api_retrieve(heptapod, 'root', test_group,
project = Project.api_retrieve(heptapod, 'root', namespace,
'ext_create')
assert (project.owner == external_user.name
or project.creator_id == external_user.id)
finally:
if project is not None:
# ugly hack until we can just pass a different user
project.owner = 'root'
project.api_destroy()
# in the group case, user doesn't have the right to delete
# the project
project.api_destroy(as_user='root')
def test_archive(public_project, tmpdir):
......@@ -436,3 +452,50 @@ def test_archive(public_project, tmpdir):
'.hg_archival.txt').readlines()
assert 'node: %s\n' % node in metadata_lines
assert extract_dir.join(main_dir, 'foo').read() == 'archived'
def test_export_import(test_project, tmpdir):
proj = test_project
repo_path = tmpdir.join('repo1')
repo = LocalRepo.clone(proj.owner_basic_auth_url, 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.hg('push')
tarball = BytesIO()
test_project.api_export(tarball, timeout=30)
tarball.seek(0)
with Project.api_import_tarball(
proj.heptapod,
proj.owner,
unique_name('test_import'),
tarball,
) as imported:
# in case the default vcs_type would be `hg_git` but the
# test run would be for `hg` (native mode), this would fail:
assert imported.vcs_type == test_project.vcs_type
assert imported.api_branch_titles() == {
'branch/default': 'Commit 0',
'topic/default/zetop': 'Commit 1',
}
# being able to clone is a thing in itself
clone = LocalRepo.clone(imported.owner_basic_auth_url,
tmpdir.join('import_clone'))
# hg exchange commands exit with code 1 when there are no changes
clone.hg('outgoing', repo_path, expected_return_code=1)
clone.hg('incoming', repo_path, expected_return_code=1)
# final check for phases
log = clone.hg('log', '-T', '{desc}:{phase}:{topic}\n')
assert log.splitlines() == ['Commit 1:draft:zetop',
'Commit 0:public:',
]
......@@ -472,7 +472,7 @@ def test_push_tags_branch_heads(test_project, tmpdir):
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
# to reproduce heptapod#96, 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')
......@@ -488,6 +488,30 @@ def test_push_tags_branch_heads(test_project, tmpdir):
assert tags['other-1.0']['commit']['title'] == "Commit 2"
def test_tag_edit_remove(test_project, tmpdir):
repo_path = tmpdir.join('repo')
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.hg('tag', 'v1')
repo.hg('push')
assert list(test_project.api_tags()) == ['v1']
repo_path.join('foo').write('foo1')
repo.hg('commit', '-m', "Commit 1")
repo.hg('tag', '--force', 'v1')
repo.hg('push')
tags = test_project.api_tags()
assert list(tags) == ['v1']
assert tags['v1']['commit']['title'] == 'Commit 1'
repo.hg('tag', '--remove', 'v1')
repo.hg('push')
assert not test_project.api_tags()
@suitable.prod_server
def test_push_tag_ci_job(test_project_with_runner, tmpdir):
proj, runner = test_project_with_runner
......@@ -540,6 +564,39 @@ def test_internal_force_push_default_branch(test_project, tmpdir):
assert log.splitlines() == ['Amended:draft', 'Commit 0:public']
def test_auto_publish_abort(test_project, tmpdir):
"""Using `experimental.auto-publish=abort` on the client-side."""
proj = test_project
repo_path = tmpdir / 'repo'
repo = LocalRepo.init(repo_path, default_url=proj.owner_basic_auth_url)
file_path = repo_path / 'foo'
file_path.write('foo0')
repo.hg('commit', '-Am', "Commit 0")
repo.hg('phase', '-p', ".")
repo.hg('push')
file_path.write('foo1')
repo.hg('commit', '-Am', "Commit 1")
def assert_phase(rev, phase):
assert repo.hg('log', '-r', rev, '-T', '{phase}').strip() == phase
assert_phase('.', 'draft')
# using auto-publish=abort refuses the push of a draft changeset with
# no topic because that's an automatic publication
push_cmd = ['--config', 'experimental.auto-publish=abort',
'push', '-r', '.']
code, _, _ = repo.hg_unchecked(*push_cmd)
assert code not in [0, 1]
assert_phase('.', 'draft')
# yet explicit push --publish works
push_cmd.append('--publish')
repo.hg(*push_cmd)
assert_phase('.', 'public')
def test_push_subrepos(test_project, tmpdir):
"""
The server just ignores subrepos in ordinary pushes.
......
......@@ -2,8 +2,15 @@ import attr
import requests
class NameSpace:
@property
def url(self):
return '/'.join((self.heptapod.url, self.full_path))
@attr.s
class UserNameSpace:
class UserNameSpace(NameSpace):
user_name = attr.ib()
heptapod = attr.ib()
......@@ -16,7 +23,7 @@ class UserNameSpace:
@attr.s
class Group:
class Group(NameSpace):
id = attr.ib()
path = attr.ib()
......
......@@ -811,8 +811,14 @@ class Project(object):
config.setdefault(section, {})[key] = val
return config
def api_destroy(self, allow_missing=False, timeout=10):
resp = self.owner_api_request('DELETE')
def api_destroy(self, allow_missing=False, timeout=10, as_user=None):
if as_user is None:
resp = self.owner_api_request('DELETE')
else:
if isinstance(as_user, str):
as_user = self.heptapod.get_user(as_user)
resp = requests.delete(self.api_url,
headers=as_user.token_headers())
if allow_missing and resp.status_code == 404:
return
print("DELETE request response: %r" % resp.json())
......@@ -1129,6 +1135,53 @@ class Project(object):
print("Project information after import or timeout: \n%r" % proj_info)
return status, err, project
@classmethod
def api_import_tarball(cls, heptapod, user_name,
project_name, tarball_fobj,
group=None, timeout=30):
"""Import project from a tarball, wait for completion and returns it.
:param user_name: name of the user to run the import with
:param group: name of the project group (by default, will be
user personal space.
:param tarball_fobj: open file-like object for the tarball to import
:returns: Project instance
"""
user = heptapod.get_user(user_name)
headers = user.token_headers()
data = dict(path=project_name)
if group is None:
group = UserNameSpace(heptapod=heptapod, user_name=user_name)
else:
data['namespace_id'] = group.id
resp = requests.post(heptapod.url + '/api/v4/projects/import',
headers=headers,
data=data,
files=dict(file=('proj.tar.gz', tarball_fobj)))
assert resp.status_code < 400
proj_info = resp.json()
proj = cls(heptapod=heptapod,
group=group, name=project_name,
id=proj_info['id'],
# proj_info['vcs_type'] is at this point just a default one
# that the import process will correct from information
# in the tarball.
owner=user_name)
resp = wait_assert(
lambda: proj.owner_api_get(subpath='import'),
lambda resp: (resp.status_code < 400
and resp.json()['import_status'] == 'finished'),
timeout=timeout,
retry_wait=2,
msg="Import not finished in %d seconds" % timeout,
)
proj.vcs_type = resp.json()['vcs_type']
return proj
@classmethod
def api_list(cls, heptapod, search_as='root', **params):
"""List the projects owned by the given user
......@@ -1168,6 +1221,30 @@ class Project(object):
self._disk_path = None
return resp
def api_export(self, target_fobj, timeout=30):
"""Export the project in a given file
This is synchronous from caller's perspective.
:target_fobj: a writable, binary, file-like object.
"""
resp = self.owner_api_post(subpath='export')
assert resp.status_code == 202
wait_assert(
lambda: self.owner_api_get(subpath='export'),
lambda resp: (resp.status_code < 400
and resp.json()['export_status'] == 'finished'),
timeout=timeout,
retry_wait=2,
msg="Export not finished in %d seconds" % timeout,
)
resp = self.owner_api_get(subpath='export/download', stream=True)
assert resp.status_code == 200
for chunk in resp.iter_content(chunk_size=8192):
target_fobj.write(chunk)
def only_specific_runners(self):
resp = self.api_edit(shared_runners_enabled=False)
assert resp.status_code < 400
......
......@@ -4,6 +4,7 @@ import requests
import time
from gitlab import Gitlab # python-gitlab
from .group import UserNameSpace
from .selenium import (
webdriver_wait_get,
)
......@@ -125,6 +126,11 @@ class User(object):
def store_in_heptapod(self):
self.heptapod.users[self.name] = self
@property
def personal_namespace(self):
"""Return a UserNameSpace instance for self."""
return UserNameSpace(user_name=self.name, heptapod=self.heptapod)
@property
def webdriver(self):
"""Selenium webdriver, authentified as self.
......@@ -161,7 +167,7 @@ class User(object):
driver.close()
self._webdriver = None
def api_request(self, method, subpath=None, **data):
def root_api_request(self, method, subpath=None, **data):
"""Perform a request on user API as root"""
hepta = self.heptapod
segments = [hepta.api_url, 'users', str(self.id)]
......@@ -169,10 +175,11 @@ class User(object):
segments.append(subpath)
return requests.request(method, '/'.join(segments),
headers=hepta.root_token_headers)
headers=hepta.root_token_headers,
data=data)
def api_post(self, subpath=None, **data):
return self.api_request('POST', subpath=subpath, **data)
def root_api_post(self, subpath=None, **data):
return self.root_api_request('POST', subpath=subpath, **data)
def delete(self, hard=True):
self.close_webdriver()
......@@ -187,23 +194,23 @@ class User(object):
assert resp.status_code == 204, "Failed to delete %r" % self
def edit(self, **data):
resp = self.api_request('PUT', **data)
resp = self.root_api_request('PUT', **data)
assert resp.status_code == 200, "Failed to edit %r" % self
def block(self):
resp = self.api_post(subpath='block')
resp = self.root_api_post(subpath='block')
assert resp.status_code == 201, "Failed to block %r" % self
def unblock(self):
resp = self.api_post(subpath='unblock')
resp = self.root_api_post(subpath='unblock')
assert resp.status_code == 201, "Failed to unblock %r" % self
def deactivate(self):
resp = self.api_post(subpath='deactivate')
resp = self.root_api_post(subpath='deactivate')
assert resp.status_code == 201, "Failed to deactivate %r" % self
def activate(self):
resp = self.api_post(subpath='activate')
resp = self.root_api_post(subpath='activate')
assert resp.status_code == 201, "Failed to activate %r" % self
def ensure_ssh_pub_key(self, pubkey, title='heptapod-tests'):
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment