import contextlib import json from io import BytesIO import logging import os from pathlib import Path import requests import selenium.webdriver import shutil import subprocess import sys import tarfile import tempfile import time from urllib.parse import urlparse from tests.utils.user import User from tests.utils import docker from tests.utils import session from .constants import DATA_DIR logger = logging.getLogger(__name__) INITIAL_TIMEOUT = 600 # seconds BRANCH_PROTECTIONS = dict(none=0, dev_can_push=1, full=2, dev_can_merge=3, ) class Heptapod: """Base class and minimum common control of Heptapod server. This is used directly in case `--heptapod-remote` is selected. """ fs_access = False """True if manipulation of files is possible. Implies :attr:`repositories_root` not to be None. """ shell_access = False """True if running arbitrary commands as in a shell is possible.""" repositories_root = None """Path to the repositories, from the Heptapod server point of view. This is potentially not meaningful on the system that runs the test, see :class:`DockerHeptapod` for an example. """ reverse_call_host = None """Network address for the system running these tests, seen from Heptapod. """ webdriver_remote_url = None """URL for a Selenium RemoteWebDriver.""" wait_after_first_response = 0 """Time to wait after we've had a first non error HTTP response. """ hg_executable = 'hg' chrome_driver_args = () def __init__(self, url, ssh_user, ssh_port, hg_native=False, reverse_call_host=None, wait_after_first_response=0, webdriver_remote_url=None): self.parsed_url = urlparse(url) self.url = url self.ssh_port = ssh_port self.ssh_user = ssh_user self.hg_native = hg_native self.users = {} if reverse_call_host is not None: self.reverse_call_host = reverse_call_host self.dead = None self.webdriver_remote_url = webdriver_remote_url self.wait_after_first_response = wait_after_first_response self.settings = {} @property def heptapod(self): return self @property def ssh_url(self): return 'ssh://{user}@{host}:{port}'.format( host=self.host, user=self.ssh_user, port=self.ssh_port, ) @property def host(self): return self.parsed_url.netloc.rsplit(':', 1)[0] @property def api_url(self): return '/'.join((self.url, 'api', 'v4')) @property def root_token_headers(self): return {'Private-Token': self.users['root']['token']} @property def basic_user_token_headers(self): return {'Private-Token': self.users['test_basic']['token']} def run_shell(self, command, **kw): exit_code, output = self.execute(command, **kw) if exit_code != 0: raise RuntimeError( ('Heptapod command {command} returned a non-zero ' 'exit code {exit_code}').format( command=command, exit_code=exit_code, )) return output def get_user(self, name): """Return a :class:`User` instance, or `None`.""" info = self.users.get(name) if info is None: return None user_id = info.get('id') password = info.get('password') if user_id is None: logger.info( "Searching for known user %r because its id is unknown.", name) user = User.search(self, name) user.password = password user.store_in_heptapod() return user return User(heptapod=self, id=user_id, name=name, password=password) def new_webdriver(self): options = selenium.webdriver.ChromeOptions() for arg in self.chrome_driver_args: options.add_argument(arg) options.add_argument('--headless') if self.webdriver_remote_url: return selenium.webdriver.Remote( command_executor=self.webdriver_remote_url, options=options ) else: return selenium.webdriver.Chrome(options=options) def get_user_webdriver(self, user_name): info = self.users[user_name] driver = info.get('webdriver') if driver is not None: return driver logger.info("Initializing a signed-in webdriver for user %r", user_name) if user_name == 'root': driver = self.new_webdriver() # guaranteeing driver to be available for teardown # as soon as created info['webdriver'] = driver session.login_as_root(driver, self, info['password']) else: # TODO should init webdriver from here and store it before login # attempt as well driver = session.make_webdriver(self, user_name, info['password']) info['webdriver'] = driver return driver def wait_startup(self, first_response_timeout=INITIAL_TIMEOUT, wait_after_first_response=None): """Wait for Heptapod to be ready after startup. We have to take into account that the server may have just started (that's frequent in development and it's annoying for a human to wait) or could even be starting from scratch, configuring itself, creating the DB schema etc. (typical of CI). In that latter case, an amount of extra wait after the first successful HTTP response is often needed. """ logger.info("Waiting for Heptapod to answer requests") dead_msg = ("Heptapod server did not " "respond in %s seconds" % first_response_timeout) start = time.time() while time.time() < start + first_response_timeout: try: resp = requests.get(self.url, allow_redirects=False) except IOError: resp = None if resp is None: logger.debug("Couldn't reach Heptapod") elif resp.status_code >= 400: logger.debug("Heptapod response code %r", resp.status_code) else: logger.info("Heptapod is up") self.dead = False if wait_after_first_response: logger.info("Waiting additional %d seconds " "after first successful HTTP call", wait_after_first_response) time.sleep(wait_after_first_response) return duration = 1 logger.debug("Retrying in %.1f seconds", duration) time.sleep(duration) self.dead = True raise AssertionError(dead_msg) def instance_cache_file(self): return os.path.join(DATA_DIR, 'instance.cache') def load_instance_cache(self): path = self.instance_cache_file() try: with open(path) as cachef: cached = json.load(cachef) except Exception: logger.info("Cache file %r not available or not readable. " "Heptapod instance info will be retrieved " "or initialized", path) else: for name, info in cached['users'].items(): self.users[name] = dict(name=name, token=info['token']) def update_instance_cache(self): users = {name: dict(token=info['token']) for name, info in self.users.items()} with open(self.instance_cache_file(), 'w') as cachef: json.dump(dict(users=users), cachef) def prepare(self, root_password): """Make all preparations for the Heptapod instance to be testable. This currently amounts to - defining the root password - activating and retrieving a root API token - creating a persistent `test_basic` user - activating and retrieving an API token for `test_basic` - keeping a logged in webdriver for each persistent user """ assert not self.dead, "Heptapod server marked dead by previous test." if self.dead is None: self.wait_startup( wait_after_first_response=self.wait_after_first_response) self.load_instance_cache() logger.info("Preparing root user.") start = time.time() root = User.init_root(self, root_password) root.ensure_private_token() logger.info("Preparing application settings.") # necessary if we want to listen to web hooks from these tests # in GitLab v12.2.0, this is deprecated for ...from_web_hooks... self.set_application_settings( allow_local_requests_from_hooks_and_services="true") # Mercurial native projects (vcs_types='hg') arent allowed by # default if self.hg_native and 'hg' not in self.settings.get('vcs_types', ()): self.set_vcs_types_settings(['hg', 'git', 'hg_git']) logger.info("Preparing basic user.") session.ensure_user(self, 'test_basic', fullname='Bäsîc Test', password='test_basic') logger.info("Uploading users SSH keys.") self.load_ssh_keys() self.upload_ssh_pub_keys() subprocess.call(( 'ssh-keygen', '-R', '[{host}]:{port}'.format( host=self.host, port=self.ssh_port))) self.update_instance_cache() logger.info("All preparations done in %.2f seconds. " "Proceeding with tests.", time.time() - start) def set_application_settings(self, **settings): """Change GitLab application settings and update :attr:`settings`.""" resp = requests.put( self.api_url + '/application/settings', headers=self.root_token_headers, data=settings, ) assert resp.status_code == 200 self.settings = resp.json() def sync_application_settings(self): """Update :attr:`settings` for current values""" resp = requests.get( self.api_url + '/application/settings', headers=self.root_token_headers, ) assert resp.status_code == 200 self.settings = resp.json() def api_request(self, method, user=None, subpath='', **kwargs): """Perform a simple API HTTP request `method` is the HTTP method to use, same as in `requests.request`. The full URL is made of the API URL of the instance, together with the given subpath (example 'snippets/42'). Appropriate authentication headers are added on the fly. :param user: the :class:`User` to run the request as. If not specified, the request is sent as the root user. All other kwargs are passed to `requests.request()` """ headers = kwargs.pop('headers', {}) token = self.owner_token if user is None else user.token headers['Private-Token'] = token return requests.request(method, '/'.join((self.api_url, subpath)), headers=headers, **kwargs) def api_get(self, **kwargs): return self.api_request('GET', **kwargs) def api_post(self, **kwargs): return self.api_request('POST', **kwargs) def api_put(self, **kwargs): return self.api_request('PUT', **kwargs) def api_delete(self, **kwargs): return self.api_request('DELETE', **kwargs) def load_ssh_keys(self): """Load client-side information to use SSH keys Also makes sure the keys are actually usable (perms) """ ssh_dir = Path(__file__).parent.parent / 'data' / 'ssh' for name, info in self.users.items(): base_fname = 'id_rsa_heptapod_' + name priv = ssh_dir / base_fname pub = ssh_dir / (base_fname + '.pub') # VCSes tend not to preserve non-executable perm bits priv.chmod(0o600) info['ssh'] = dict(priv=str(priv), pub=pub.read_text()) def upload_ssh_pub_keys(self): """Upload SSH public keys for all users to Heptapod.""" # it's really time to put the actual user object in our `self.users` for name, info in self.users.items(): user = User.search(self, name) user.ensure_ssh_pub_key(info['ssh']['pub']) def close_webdrivers(self): for user in self.users.values(): driver = user.pop('webdriver', None) if driver is not None: logger.info("Closing webdriver for user %r", user['name']) driver.close() def close(self): if self.dead is not False: return self.close_webdrivers() def execute(self, command, **kw): raise NotImplementedError('execute') def force_remove_route(self, route_path, source_type='Project'): logger.error("Attempt to force-remove route %r, not implemented " "for %r", route_path, self.__class__) raise NotImplementedError('force_remove_route') def gitlab_ctl(self, command, services=None): """Apply service management command. 'command' would typically be 'start', 'stop', etc. :param services: an iterable of service names (who can themselves be different depending on the concrete subclass). If supplied, the command will apply only to those services. """ raise NotImplementedError('gitlab_ctl') def rake(self, *args): """Call GitLab Rake""" raise NotImplementedError('rake') def remove_all_backups(self): """Remove all existing backups with no impediment for new backups. """ raise NotImplementedError('remove_all_backups') def backup_create(self, clean_previous=True): """Create a new backup :param bool clean_previous: if true, any previously existing backups are removed. This is handy so that the restore rake task knows which one to restore without any need to tell it. """ if clean_previous: self.remove_all_backups() self.rake('gitlab:backup:create') @contextlib.contextmanager def backup_restore(self): """Context manager for backups restoration. This is a context manager as a way to provide resuming of the tests session on errors, in a convenient way for the caller. That means ensuring as much as possible that the server is running, maybe wait again for it, reinitialize passwords and tokens… """ try: self.gitlab_ctl('stop', services=self.RAILS_SERVICES) self.rake('gitlab:backup:restore', 'force=yes') self.gitlab_ctl('start', services=self.RAILS_SERVICES) self.wait_startup() yield except Exception: # these are idempotent self.gitlab_ctl('start', services=self.RAILS_SERVICES) # Worst case scenario, we lost all our data. We need to # reprepare the server for subsequent tests self.prepare(self.users['root']['password']) raise def set_vcs_types_settings(self, vcs_types): self.set_application_settings(vcs_types=','.join(vcs_types)) def apply_hashed_storage_setting(self, hashed_storage): # TODO it would be tempting not to restart if the setting is already # at the wished value, but this currently cannot take into account # rollbacked values that aren't followed by a restart. This will # be more complicated and take more time than we can afford right now # to really make work. self.set_application_settings(hashed_storage_enabled=hashed_storage) # let's be sure that redis is restarted when the Rails services # start self.gitlab_ctl('stop', self.RAILS_SERVICES) self.gitlab_ctl('stop', ['redis']) # we restart everything in case a service would depend on Redis # and would fail to reconnect automatically # closing all webdrivers, because restart of Redis will kill sessions self.close_webdrivers() self.gitlab_ctl('restart') self.wait_startup() # recheck that the setting is applied self.sync_application_settings() assert self.settings['hashed_storage_enabled'] is hashed_storage class OmnibusHeptapod(Heptapod): fs_access = True shell_access = True repositories_root = '/var/opt/gitlab/git-data/repositories' backups_dir = '/var/opt/gitlab/backups' RAILS_SERVICES = ('puma', 'sidekiq') gitlab_ctl_command = ('sudo', 'gitlab-ctl') reverse_call_host = 'localhost' ssh_url = 'ssh://git@localhost' # TODO read from conf for logical dependency loosening. # Not an immediate priority, since we're not concerned about Python 2 # any more (see heptapod#353) hg_executable = '/opt/gitlab/embedded/bin/hg' chrome_driver_args = ('--no-sandbox', ) def execute(self, command, user='git'): if user != 'git': raise NotImplementedError( "On Omnibus Heptapod, only 'git' user is allowed") logger.debug("OmnibusHeptapod: executing command %r", command) process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out, err = [o.decode() for o in process.communicate()] retcode = process.poll() if out: sys.stdout.write(out) if err: sys.stderr.write(err) return retcode, out def put_archive(self, dest, path, owner='git'): # this assumes owner has write access to parent directory, # but that should be enough, and avoids some processing as root self.run_shell(['tar', '-xf', path, '-C', dest], user=owner) def put_archive_bin(self, dest, fobj): with tempfile.NamedTemporaryFile() as tempf: tempf.write(fobj.read()) self.put_archive(dest, tempf.name) def put_file_lines(self, path, lines, uid=998, gid=998): with open(path, 'w') as fobj: fobj.writelines(lines) os.chown(path, uid, gid) def get_file_lines(self, path): with open(path) as fobj: return fobj.readlines() def remove_all_backups(self): # using find in order not to rely on shell expansion for *.tar self.run_shell(('find', '/var/opt/gitlab/backups', '-name', '*.tar', '-delete')) def rake(self, *args): cmd = ['gitlab-rake'] cmd.extend(args) code, out = self.execute(cmd, user='git') return out.encode() # Consistency with HDK (returns bytes) def gitlab_ctl(self, command, services=None): base_cmd = self.gitlab_ctl_command + (command, ) if services is None: self.run_shell(base_cmd) else: for service in services: self.run_shell(base_cmd + (service, )) class DockerHeptapod(OmnibusHeptapod): gitlab_ctl_command = ('gitlab-ctl', ) git_executable = 'git' chrome_driver_args = () def __init__(self, docker_container, **kw): super(DockerHeptapod, self).__init__(**kw) self.docker_container = docker_container if self.reverse_call_host is None: self.reverse_call_host = docker.host_address(docker_container) def execute(self, command, user='root'): return docker.heptapod_exec(self.docker_container, command, user=user) def run_shell(self, command, **kw): return docker.heptapod_run_shell(self.docker_container, command, **kw) @property def ssh_url(self): return super(OmnibusHeptapod, self).ssh_url def put_archive(self, dest, path, owner='git'): res = docker.heptapod_put_archive(self.docker_container, dest, path) self.run_shell(['chown', '-R', 'git:root', dest]) return res def put_archive_bin(self, dest, fobj): return docker.heptapod_put_archive_bin( self.docker_container, dest, fobj) def get_archive(self, path, tarf): return docker.heptapod_get_archive(self.docker_container, path, tarf) def put_file_lines(self, path, lines, uid=998, gid=998): dirpath, filename = path.rsplit('/', 1) tar_buf = BytesIO() tarf = tarfile.open(mode='w:', fileobj=tar_buf) tinfo = tarfile.TarInfo(name='hgrc') contents_buf = BytesIO() contents_buf.writelines(l.encode() for l in lines) tinfo.size = contents_buf.tell() tinfo.uid, tinfo.gid = uid, gid contents_buf.seek(0) tarf.addfile(tinfo, fileobj=contents_buf) tar_buf.seek(0) self.put_archive_bin(dirpath, tar_buf) def get_file_lines(self, path): dirname, filename = path.rsplit('/', 1) buf = BytesIO() self.get_archive(path, buf) buf.seek(0) tarf = tarfile.open(mode='r:', fileobj=buf) return [l.decode() for l in tarf.extractfile(filename).readlines()] def force_remove_route(self, route_path, source_type='Project'): """Delete a route from the database. Sometimes GitLab fails to clean Project routes after failed tests. """ logger.warn("Cleaning up leftover route at %r", route_path) self.run_shell([ 'gitlab-psql', 'gitlabhq_production', '-c', "DELETE FROM routes " "WHERE source_type='%s' " " AND path='%s'" % (source_type, route_path) ]) class SourceHeptapod(Heptapod): """An Heptapod server installed from source on the same system. Same system means without using any container technology (Docker or otherwise) that would insulate the tests from the server. """ fs_access = True shell_access = True reverse_call_host = 'localhost' @property def ssh_url(self): return 'ssh://{host}:{port}'.format( host=self.host, port=self.ssh_port, ) def __init__(self, repositories_root, **kw): super(SourceHeptapod, self).__init__(**kw) self.repositories_root = repositories_root def execute(self, command, user='git'): if user != 'git': raise NotImplementedError( "On source Heptapod, only same user as for Rails and HgServe " "is allowed") logger.debug("SourceHeptapod: executing command %r", command) process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out, err = [o.decode() for o in process.communicate()] retcode = process.poll() if out: sys.stdout.write(out) if err: sys.stderr.write(err) return retcode, out def put_archive(self, dest, path, owner='git'): if owner != 'git': raise NotImplementedError( "On source Heptapod, only same owner as for Rails and HgServe " "is allowed") subprocess.check_call(['tar', 'xf', path], cwd=dest) def get_file_lines(self, path): with open(path, 'r') as fobj: return fobj.readlines() def put_file_lines(self, path, lines): with open(path, 'w') as fobj: fobj.writelines(lines) class GdkHeptapod(SourceHeptapod): """An Heptapod server running with the GDK. """ fs_access = True shell_access = True reverse_call_host = 'localhost' RAILS_SERVICES = ('rails-web', 'rails-background-jobs') def __init__(self, gdk_root, **kw): self.gdk_root = gdk_root self.rails_root = os.path.join(gdk_root, 'gitlab') super(GdkHeptapod, self).__init__( repositories_root=os.path.join(gdk_root, 'repositories'), **kw) @property def backups_dir(self): return os.path.join(self.rails_root, 'tmp', 'backups') def remove_all_backups(self): if os.path.exists(self.backups_dir): shutil.rmtree(self.backups_dir) # as of GitLab 12.10, parent dir is always present os.mkdir(self.backups_dir) def rake(self, *args): cmd = ['bundle', 'exec', 'rake'] cmd.extend(args) logger.debug("GdkHeptapod: calling %r", cmd) return subprocess.check_output(cmd, cwd=self.rails_root) def gitlab_ctl(self, command, services=None): base_cmd = ('gdk', command) def do_command(*opt_args): cmd = base_cmd + opt_args logger.debug("GdkHeptapod: calling %r", cmd) subprocess.check_call(cmd, cwd=self.rails_root) if services is None: do_command() else: for service in services: do_command(service)