import attr from base64 import b64decode import hashlib from io import BytesIO import logging import os import py import re import requests import time from urllib.parse import urlencode from selenium.webdriver.common.by import By from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait from . import wait_assert from .constants import DATA_DIR from .hg import LocalRepo from .group import UserNameSpace from .selenium import could_click_element logger = logging.getLogger(__name__) class ProjectAccess: NO_ACCESS = 0 GUEST = 10 REPORTER = 20 DEVELOPER = 30 MAINTAINER = 40 OWNER = 50 def extract_gitlab_branch_titles(branches): return {name: info['commit']['title'] for name, info in branches.items()} @attr.s class Project(object): heptapod = attr.ib() name = attr.ib() group = attr.ib() # owner=None or id=None means it's not known yet, this should happen only # in the present module functions owner = attr.ib(default=None) id = attr.ib(default=None) # If vcs_type is None, this means Mercurial. # These functional tests should only exceptionally make a difference # between various ways Mercurial is supported ('hg_git', 'hgitaly'), and # preferably for temporary workarounds. vcs_type = attr.ib(default=None) is_legacy = attr.ib(default=False) def owner_get(self, **kwargs): """A shortcut to perform a simple GET, with BasicAuth as the owner. All `kwargs` are passed directly to `requests.get()` """ return requests.get(self.url, auth=self.owner_credentials, **kwargs) def get_session_cookie(self, webdriver): for cookie in webdriver.get_cookies(): if cookie['name'].startswith('_gitlab_session'): return cookie raise LookupError("Could not find GitLab session cookie") def session_api_get(self, webdriver, subpath='', **kwargs): """Perform a simple GET, with the session cookie found in webdriver. The full URL is made of the API URL of the project, together with the given subpath (example '/merge_requests/1') """ cookie = self.get_session_cookie(webdriver) return requests.get( '/'.join((self.api_url, subpath)), cookies={cookie['name']: cookie['value']}) def api_edit(self, **params): """Perform a project API edit.""" return self.owner_api_put(data=params) @property def owner_user(self): return self.heptapod.get_user(self.owner) def owner_api_hgrc_set(self, **values): return self.owner_api_put(subpath='hgrc', data=values) def owner_api_hgrc_get(self): return self.owner_api_get(subpath='hgrc') def owner_api_hgrc_reset(self): return self.owner_api_delete(subpath='hgrc') def owner_api_hg_heptapod_config(self): return self.owner_api_get(subpath='hg_heptapod_config') def api_request(self, method, user, subpath='', **kwargs): """Perform a simple API HTTP request as the given user. `method` is the HTTP method to use, same as in `requests.request`. The full URL is made of the API URL of the project, together with the given subpath (example 'merge_requests/1'). Appropriate authentication headers are added on the fly. All kwargs are passed to `requests.request()` """ headers = kwargs.pop('headers', {}) headers['Private-Token'] = user.token return requests.request(method, '/'.join((self.api_url, subpath)), headers=headers, **kwargs) def owner_api_request(self, method, *args, **kwargs): """Perform a simple API HTTP request as the owner. See :meth:`api_request` for more details. """ return self.api_request(method, self.owner_user, *args, **kwargs) def owner_api_get(self, **kwargs): """Perform a simple HTTP API GET as the owner. See :meth:`api_request` for details. """ return self.owner_api_request('GET', **kwargs) def owner_api_put(self, **kwargs): """Perform a simple HTTP API PUT as the owner. See :meth:`api_request` for details. """ return self.owner_api_request('PUT', **kwargs) def owner_api_post(self, **kwargs): """Perform a simple HTTP API POST as the owner. See :meth:`api_request` for details. """ return self.owner_api_request('POST', **kwargs) def owner_api_delete(self, **kwargs): """Perform a simple HTTP API DELETE as the owner. See :meth:`api_request` for details. """ return self.owner_api_request('DELETE', **kwargs) def api_get_field(self, key): """Return the value of a field by performing an API GET request. The request is made with full owner credentials. """ resp = self.owner_api_get() assert resp.status_code < 400 return resp.json().get(key) @property def owner_credentials(self): """Return (user, password).""" user = self.owner_user return user.name, user.password @property def owner_token(self): return self.owner_user.token @property def owner_webdriver(self): return self.heptapod.get_user_webdriver(self.owner) @property def relative_uri(self): return '/'.join((self.group.full_path, self.name)) @property def url(self): return '/'.join((self.heptapod.url, self.relative_uri)) @property def owner_ssh_params(self): """See `ssh_params()` """ return self.ssh_params(self.owner) def ssh_params(self, user_name): """Provide command and URL to perform SSH operations as `user_name` Example:: ('ssh -i /tmp/id_rsa', 'git@localhost:root/test_project.hg') """ heptapod = self.heptapod ext = '.git' if self.vcs_type == 'git' else '.hg' url = '/'.join((heptapod.ssh_url, self.relative_uri + ext)) return self.heptapod.get_user(user_name).ssh_command, url def git_ssh_params(self, user_name): """Similar to ssh_params, tailored for Git. """ heptapod = self.heptapod cmd = self.ssh_params(user_name)[0] + ' -p %d' % heptapod.ssh_port address = '{heptapod.ssh_user}@{heptapod.host}:{path}'.format( heptapod=heptapod, path=self.relative_uri + '.git', ) return cmd, address def basic_auth_url(self, user_name, pwd=None): """Produce an URL suitable for basic authentication, hence hg CLI. :param pwd: if not supplied, will be read from the permanent users known of :attr:`heptapod` """ heptapod = self.heptapod if pwd is None: pwd = self.heptapod.get_user(user_name).password url = heptapod.parsed_url return "{scheme}://{auth}@{netloc}/{path}".format( scheme=url.scheme, netloc=url.netloc, auth=':'.join((user_name, pwd)), path=self.relative_uri, ) @property def owner_basic_auth_url(self): return self.basic_auth_url(self.owner) def deploy_token_url(self, token): return self.basic_auth_url(token['username'], pwd=token['token']) @property def api_url(self): return '/'.join(( self.heptapod.url, 'api', 'v4', 'projects', '/'.join((self.group.full_path, self.name)).replace('/', '%2F') )) @property def fs_common_path(self): """Common abspath on Heptapod server FS (not ending with .hg nor .git) Meaningful only for those tests that require file system access. Relies on knowledge of internal GitLab details that may well change. (since these are tests, we would notice quickly). """ disk_path = getattr(self, '_disk_path', None) if disk_path is not None: return disk_path if not self.is_legacy: sha2 = hashlib.sha256(b'%d' % self.id).hexdigest() rpath = '@hashed/%s/%s/%s' % (sha2[:2], sha2[2:4], sha2) disk_path = os.path.join(self.heptapod.repositories_root, rpath) else: disk_path = '/'.join((self.heptapod.repositories_root, self.group.full_path, self.name)) self._disk_path = disk_path return disk_path @property def fs_path(self): """Path to the Mercurial repo on Heptapod server file system.""" return self.fs_common_path + '.hg' @property def fs_path_git(self): """Path to the Git repo on Heptapod server file system.""" return self.fs_common_path + '.git' def _change_storage(self, legacy): label = 'legacy' if legacy else 'hashed' if legacy is self.is_legacy: logger.warn("_change_storage: project %d is already on %s storage", self.id, label) rake_cmd = 'rollback_to_legacy' if legacy else 'migrate_to_hashed' self.heptapod.rake('gitlab:storage:' + rake_cmd, 'ID_FROM=%d' % self.id, 'ID_TO=%d' % self.id) self.is_legacy = legacy self._disk_path = None # we're not inconsistent assert ('@hashed' in self.fs_path) is (not legacy) wait_assert( lambda: self.heptapod.execute(('test', '-e', self.fs_path))[0], lambda code: code == 0, timeout=120, retry_wait=2, msg="Repository %r not found in %s storage" % (self.fs_path, label) ) def make_storage_legacy(self): self._change_storage(True) def make_storage_hashed(self): self._change_storage(False) def archive_url(self, gitlab_branch, file_ext): base_file_name = '-'.join([self.name] + gitlab_branch.split('/')) return '/'.join((self.url, '-', 'archive', gitlab_branch, base_file_name + '.' + file_ext )) def get_archive(self, gitlab_branch, fmt='tar'): """Retrieve a repository archive by URL. :param fmt: the wished format. At this point, this is directly mapped as a file extension in the request, and only the `tar` value is tested. :returns: the archive content, as a files-like object """ resp = requests.get(self.archive_url(gitlab_branch, fmt)) assert resp.status_code == 200 return BytesIO(resp.content) def api_branches(self): """Retrieve and pre-sort branches info through the REST API.""" resp = self.owner_api_get(subpath='repository/branches') assert resp.status_code == 200 branches = resp.json() return dict((branch.pop('name'), branch) for branch in branches) def api_default_branch(self): branch = self.api_get_field('default_branch') assert branch is not None return branch def api_branch_titles(self): """Keep only commit titles from `api_branches()` With a test scenario in which those titles are characterizing the commit uniquely, this is what's very often needed for assertions. """ return extract_gitlab_branch_titles(self.api_branches()) def wait_assert_api_branches(self, condition, msg=None, timeout=10, **kw): """Assert some condition to become True on GitLab branches. Since the update of pushed or pruned branches is asynchronous and becomes even more so as GitLab progresses, this provides the means to retry several calls to :meth:`api_branches`. :param condition: any callable returning boolean that can take a single argument, the payload of :meth:`api_branches` :returns: branches after the wait :raises: AssertionError if the condition doesn't become True before timeout """ if msg is None: msg = ("The given condition on GitLab branches was still not " "fulfilled after retrying for %s seconds" % timeout) return wait_assert(lambda: self.api_branches(), condition, msg=msg, timeout=timeout, **kw) def api_tags(self): """Retrieve and pre-sort tags info through the REST API.""" resp = self.owner_api_get(subpath='repository/tags') assert resp.status_code == 200 tags = resp.json() return dict((tag.pop('name'), tag) for tag in tags) def api_protected_branches(self): resp = self.owner_api_get(subpath='protected_branches') assert resp.status_code == 200 return {br['name']: br for br in resp.json()} def api_commit(self, sha, check=True): """Retrieve a commit by its SHA. The SHA is the native one to GitLab, typically obtained through the API. For Mercurial SHAs, it's usually simpler to just perform a pull. """ resp = self.owner_api_get(subpath='repository/commits/' + sha) if not check: return resp assert resp.status_code == 200 return resp.json() def api_file_create(self, path, check=True, **data): """data is transferred directly into the JSON expected by the API.""" data['file_path'] = path resp = self.owner_api_post( subpath='repository/files/' + path, data=data) if not check: return resp assert resp.status_code < 400 return resp.json() def api_file_update(self, path, check=True, **data): """data is transferred directly into the JSON expected by the API.""" data['file_path'] = path resp = self.owner_api_put( subpath='repository/files/' + path, data=data) if not check: return resp assert resp.status_code < 400 return resp.json() def api_wiki_page_create(self, check=True, **data): resp = self.owner_api_post(subpath='wikis', data=data) if not check: return resp assert resp.status_code == 201 return resp.json() def api_wiki_page_get(self, slug, check=True): resp = self.owner_api_get(subpath='wikis/' + slug) if not check: return resp assert resp.status_code == 200 return resp.json() def api_wiki_page_update(self, slug, check=True, **data): resp = self.owner_api_put(subpath='wikis/' + slug, data=data) if not check: return resp assert resp.status_code == 200 return resp.json() def api_wiki_pages_list(self, check=True): resp = self.owner_api_get(subpath='wikis') if not check: return resp assert resp.status_code == 200 return resp.json() def hg_wiki_url(self, user_name=None): """Return an authenticated URL suitage for hg pull/push. :param user_name: any user name known to :attr:`heptapod` """ if user_name is None: user_name = self.owner return self.basic_auth_url(user_name) + '.wiki' def api_file_get(self, path, ref, content_only=True): """Retrieve a repository file through API. :param content_only: if ``True``, the response status code is checked and the content is extracted and returned as bytes. Otherwise the full HTTP response is returned. """ resp = self.owner_api_get(subpath='repository/files/' + path, params=dict(ref=ref)) if not content_only: return resp assert resp.status_code == 200 return b64decode(resp.json()['content']) def api_file_delete(self, path, check=True, **data): data['file_path'] = path resp = self.owner_api_delete(subpath='repository/files/' + path, data=data) if not check: return resp assert resp.status_code < 400 def webdriver_update_merge_request_settings(self, merge_method): driver = self.owner_webdriver driver.get('{url}/{group_path}/{project_name}/edit'.format( group_path=self.group.full_path, url=self.heptapod.url, project_name=self.name, )) elem = driver.find_element_by_xpath( '//section[contains(@class, "merge-requests-feature")]' '//button[contains(text(), "Expand")]' ) elem.click() wait = WebDriverWait(driver, 10) wait.until(could_click_element(lambda d: d.find_element_by_id( 'project_merge_method_' + merge_method))) submit = driver.find_element_by_xpath( '//section[contains(@class, "merge-requests-feature")]' '//input[@type="submit"]') submit.click() def webdriver_create_merge_request(self, source, target): """Create a merge request through the Web UI and return its id. """ heptapod = self.heptapod driver = self.owner_webdriver compare_qs = { 'merge_request[source_project_id]': self.id, 'merge_request[source_branch]': source, 'merge_request[target_project_id]': self.id, 'merge_request[target_branch]': target, } driver.get( '{url}/{group_path}/{project_name}/merge_requests/new?{qs}'.format( group_path=self.group.full_path, url=heptapod.url, project_name=self.name, qs=urlencode(compare_qs), )) assert 'New Merge Request' in driver.title assert self.name in driver.title wait = WebDriverWait(driver, 10) wait.until(could_click_element(lambda d: d.find_element_by_xpath( '//input[@type="submit" ' 'and @value="Submit merge request"]') )) split_url = driver.current_url.rsplit('/', 2) assert split_url[-2] == 'merge_requests' return int(split_url[-1]) def api_create_merge_request(self, source, target='branch/default', title='Created through API'): """As the project owner, create a merge request through API. TODO support creation as another user. :param source: name of the source branch for GitLab, e.g., 'topic/default/foo' :param target: name of the target branch for GitLab :returns: numeric iid of the newly created MR """ resp = self.owner_api_post(subpath='merge_requests', data=dict(id=self.id, source_branch=source, target_branch=target, title=title, )) assert resp.status_code in (200, 201) return resp.json()['iid'] def api_rebase_merge_request(self, mr_iid): resp = self.owner_api_put(subpath='merge_requests/%d/rebase' % mr_iid) assert resp.status_code == 202 # Accepted, this is async mr_info = self.wait_assert_merge_request( mr_iid, lambda info: not info['rebase_in_progress'], with_rebase=True) assert mr_info['merge_error'] is None def api_update_merge_request(self, mr_iid, **data): resp = self.owner_api_put(subpath='merge_requests/%d' % mr_iid, data=data) assert resp.status_code == 200 # this is synchronous return resp.json() def api_get_merge_request(self, mr_id, **params): resp = self.owner_api_get(subpath='merge_requests/%s' % mr_id, params=params) assert resp.status_code == 200 return resp.json() def wait_assert_merge_request(self, mr_id, condition, with_rebase=False, with_diverged_count=False, timeout=10, msg=None, **kw): if msg is None: msg = ( "The given condition on Merge Request %s was still not " "fulfilled after retrying for %s seconds" % (mr_id, timeout)) mr_opts = dict(include_diverged_commits_count=with_diverged_count, include_rebase_in_progress=with_rebase) return wait_assert(lambda: self.api_get_merge_request(mr_id, **mr_opts), condition, msg=msg, timeout=timeout, **kw) def webdriver_get_merge_request_commits(self, mr_id): """Retrieve the commit links from the 'commits' panel of the MR.""" webdriver = self.owner_webdriver # this is the 'commits' pane, loading is dynamic, hence the wait webdriver.get(self.url + '/merge_requests/%d/commits' % mr_id) WebDriverWait(webdriver, 10).until( EC.presence_of_all_elements_located( (By.CSS_SELECTOR, 'a.commit-row-message'))) return webdriver.find_elements_by_css_selector('li.commit') def api_create_deploy_token(self, name, scopes=('read_repository',)): """Will be available with GitLab 12.9.""" resp = self.owner_api_post(subpath='deploy_tokens', data=dict(name=name, scopes=scopes)) assert resp.status_code == 200 return resp.json() def api_delete_deploy_token(self, token): """Will be available with GitLab 12.9.""" self.owner_api_delete(subpath='deploy_tokens', data=dict(token_id=token['id'])) def webdriver_create_deploy_token(self, name): """Create a deploy token with Selenium The API doesn't exist before GitLab 12.9. :param name: required by GitLab, is only a human intended description """ driver = self.owner_webdriver driver.get(self.url + '/-/settings/repository') wait = WebDriverWait(driver, 10) section_xpath = '//section[contains(@class, "deploy-tokens-settings")]' expand_button = driver.find_element_by_xpath( section_xpath + '//button[contains(text(), "Expand")]') expand_button.click() wait.until(lambda d: d.find_element_by_xpath( section_xpath + '//button[contains(text(), "Collapse")]' ).is_displayed) def name_elt(d): return d.find_element_by_id('deploy_token_name') wait.until(lambda d: name_elt(d).is_displayed()) name_elt(driver).send_keys(name) read_repo_scope = driver.find_element_by_id( 'deploy_token_read_repository') read_repo_scope.click() submit = driver.find_element_by_xpath( '//form[@id="new_deploy_token"]' '//input[@type="submit"]') submit.click() def value_for_id(elt_id): return driver.find_element_by_id(elt_id).get_attribute('value') return dict( username=value_for_id('deploy-token-user'), token=value_for_id('deploy-token'), ) def grant_member_access(self, user_name, level): """Grant given user the given access level. It doesn't matter whether the user is already a member or not: this method abstracts over it. This method is idempotent. """ user_id = self.heptapod.get_user(user_name).id resp = self.owner_api_get(subpath='members/%d' % user_id) if resp.status_code == 404: subpath = 'members' meth = self.owner_api_post else: subpath = 'members/%d' % user_id meth = self.owner_api_put resp = meth(subpath=subpath, data=dict(user_id=user_id, access_level=level)) assert resp.status_code < 400 def load_tarball(self, tarball_path): """Replace server-side repository files by the contents of tarball. This should be used right after the project creation. :param tarball_path: path (relative to DATA_DIR) to an uncompressed tar archive, containing `hg` and `git`. These will be renamed to the values of `self.fs_path`` and ``self.fs_path_git``. """ if not self.heptapod.fs_access: raise NotImplementedError( "Can't use load_tarball() without filesystem access") tarball_path = os.path.join(DATA_DIR, tarball_path) # initialize repository # GitLab needs a first clone or push to consider the Git repository to # exist. Otherwise, local pushes such as what the heptapod sync hook # does just fail, with an error about the Git repo existence. tmpdir = py.path.local.mkdtemp() try: LocalRepo.clone(self.owner_basic_auth_url, tmpdir.join('clone')) finally: tmpdir.remove() heptapod = self.heptapod # using a temporary space in same mount point and unique enough srvtmpdir = self.fs_common_path + '.tmp' heptapod.run_shell(['rm', '-rf', self.fs_path, self.fs_path_git, srvtmpdir]) heptapod.run_shell(['mkdir', '-p', srvtmpdir]) heptapod.put_archive(srvtmpdir, tarball_path) heptapod.run_shell(['mv', srvtmpdir + '/hg', self.fs_path]) heptapod.run_shell(['mv', srvtmpdir + '/git', self.fs_path_git]) def get_hgrc(self, managed=False): """Return repo's server-side HGRC, as lines, uid and gid :param managed: if ``True``, the contents returned are those of the file managed by the Rails app, introduced for heptapod#165 """ hgrc_path = '/'.join((self.fs_path, '.hg', 'hgrc.managed' if managed else 'hgrc')) return self.heptapod.get_file_lines(hgrc_path) def put_hgrc(self, lines): """Replace repo's server-side HGRC with given lines. The lines have to include LF, same as with `writelines()`. """ repo_inner_path = '/'.join((self.fs_path, '.hg')) return self.heptapod.put_file_lines( os.path.join(repo_inner_path, 'hgrc'), lines) def extend_hgrc(self, *lines): """Append given lines to repo's server-side HGRC The lines don't have to be newline-terminated. """ hgrc_lines = self.get_hgrc() # just in case original hgrc isn't new-line terminated hgrc_lines.append('\n') hgrc_lines.extend(l + '\n' for l in lines) self.put_hgrc(hgrc_lines) def hg_config(self, section=None): """Return Mercurial configuration item, as really seen by hg process. In other words, this isn't inference on the contents of the various HGRC, it can be used to test propagation of config entries. :return: if ``section`` is passed, a simple ``dict``, otherwise a ``dict`` of ``dicts``. End values are always strings. """ cmd = [self.heptapod.hg_executable, '-R', self.fs_path, '--pager', 'false', 'config'] if section is not None: cmd.append(section) code, out = self.heptapod.execute(cmd, user='git') config = {} if out is None: return config for l in out.splitlines(): print(l) fullkey, val = l.split('=', 1) section, key = fullkey.split('.', 1) if section is not None: config[key] = val else: config.setdefault(section, {})[key] = val return config def api_destroy(self, allow_missing=False, timeout=10): resp = self.owner_api_request('DELETE') if allow_missing and resp.status_code == 404: return print("DELETE request response: %r" % resp.json()) # Even though the deletion meaning that the repos are just # mv'ed on the filesystem, it is still async assert resp.status_code == 202 start = time.time() duration = 0.2 while self.owner_api_get().status_code != 404: time.sleep(duration) assert time.time() - start < timeout return resp def webdriver_destroy(self, skip_missing=False): driver = self.owner_webdriver driver.get(self.url + '/edit') if skip_missing and '404' in driver.title: return assert 'General' in driver.title assert 'Settings' in driver.title assert self.name in driver.title wait = WebDriverWait(driver, 10) elem = driver.find_element_by_xpath( '//section[contains(@class, "advanced-settings")]' '//button[contains(text(), "Expand")]' ) elem.click() wait.until(lambda d: d.find_element_by_xpath( '//section[contains(@class, "advanced-settings")]' '//button[contains(text(), "Collapse")]' ).is_displayed) wait.until(could_click_element(lambda d: d.find_element_by_xpath( # a