# Copyright (c) 2013 The Chromium OS Authors. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """Test framework for code that interacts with gerrit. class GerritTestCase -------------------------------------------------------------------------------- This class initializes and runs an a gerrit instance on localhost. To use the framework, define a class that extends GerritTestCase, and then do standard python unittest development as described here: http://docs.python.org/2.7/library/unittest.html#basic-example When your test code runs, the framework will: - Download the latest stable(-ish) binary release of the gerrit code. - Start up a live gerrit instance running in a temp directory on the localhost. - Set up a single gerrit user account with admin priveleges. - Supply credential helpers for interacting with the gerrit instance via http or ssh. Refer to depot_tools/testing_support/gerrit-init.sh for details about how the gerrit instance is set up, and refer to helper methods defined below (createProject, cloneProject, uploadChange, etc.) for ways to interact with the gerrit instance from your test methods. class RepoTestCase -------------------------------------------------------------------------------- This class extends GerritTestCase, and creates a set of project repositories and a manifest repository that can be used in conjunction with the 'repo' tool. Each test method will initialize and sync a brand-new repo working directory. The 'repo' command may be invoked in a subprocess as part of your tests. One gotcha: 'repo upload' will always attempt to use the ssh interface to talk to gerrit. """ import collections import errno import netrc import os import re import shutil import signal import socket import stat import subprocess import sys import tempfile import unittest import urllib import gerrit_util DEPOT_TOOLS_DIR = os.path.normpath(os.path.join( os.path.realpath(__file__), '..', '..')) # When debugging test code, it's sometimes helpful to leave the test gerrit # instance intact and running after the test code exits. Setting TEARDOWN # to False will do that. TEARDOWN = True class GerritTestCase(unittest.TestCase): """Test class for tests that interact with a gerrit server. The class setup creates and launches a stand-alone gerrit instance running on localhost, for test methods to interact with. Class teardown stops and deletes the gerrit instance. Note that there is a single gerrit instance for ALL test methods in a GerritTestCase sub-class. """ COMMIT_RE = re.compile(r'^commit ([0-9a-fA-F]{40})$') CHANGEID_RE = re.compile('^\s+Change-Id:\s*(\S+)$') DEVNULL = open(os.devnull, 'w') TEST_USERNAME = 'test-username' TEST_EMAIL = 'test-username@test.org' GerritInstance = collections.namedtuple('GerritInstance', [ 'credential_file', 'gerrit_dir', 'gerrit_exe', 'gerrit_host', 'gerrit_pid', 'gerrit_url', 'git_dir', 'git_host', 'git_url', 'http_port', 'netrc_file', 'ssh_ident', 'ssh_port', ]) @classmethod def check_call(cls, *args, **kwargs): kwargs.setdefault('stdout', cls.DEVNULL) kwargs.setdefault('stderr', cls.DEVNULL) subprocess.check_call(*args, **kwargs) @classmethod def check_output(cls, *args, **kwargs): kwargs.setdefault('stderr', cls.DEVNULL) return subprocess.check_output(*args, **kwargs) @classmethod def _create_gerrit_instance(cls, gerrit_dir): gerrit_init_script = os.path.join( DEPOT_TOOLS_DIR, 'testing_support', 'gerrit-init.sh') http_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) http_sock.bind(('', 0)) http_port = str(http_sock.getsockname()[1]) ssh_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) ssh_sock.bind(('', 0)) ssh_port = str(ssh_sock.getsockname()[1]) # NOTE: this is not completely safe. These port numbers could be # re-assigned by the OS between the calls to socket.close() and gerrit # starting up. The only safe way to do this would be to pass file # descriptors down to the gerrit process, which is not even remotely # supported. Alas. http_sock.close() ssh_sock.close() cls.check_call(['bash', gerrit_init_script, '--http-port', http_port, '--ssh-port', ssh_port, gerrit_dir]) gerrit_exe = os.path.join(gerrit_dir, 'bin', 'gerrit.sh') cls.check_call(['bash', gerrit_exe, 'start']) with open(os.path.join(gerrit_dir, 'logs', 'gerrit.pid')) as fh: gerrit_pid = int(fh.read().rstrip()) return cls.GerritInstance( credential_file=os.path.join(gerrit_dir, 'tmp', '.git-credentials'), gerrit_dir=gerrit_dir, gerrit_exe=gerrit_exe, gerrit_host='localhost:%s' % http_port, gerrit_pid=gerrit_pid, gerrit_url='http://localhost:%s' % http_port, git_dir=os.path.join(gerrit_dir, 'git'), git_host='%s/git' % gerrit_dir, git_url='file://%s/git' % gerrit_dir, http_port=http_port, netrc_file=os.path.join(gerrit_dir, 'tmp', '.netrc'), ssh_ident=os.path.join(gerrit_dir, 'tmp', 'id_rsa'), ssh_port=ssh_port,) @classmethod def setUpClass(cls): """Sets up the gerrit instances in a class-specific temp dir.""" # Create gerrit instance. gerrit_dir = tempfile.mkdtemp() os.chmod(gerrit_dir, 0o700) gi = cls.gerrit_instance = cls._create_gerrit_instance(gerrit_dir) # Set netrc file for http authentication. cls.gerrit_util_netrc_orig = gerrit_util.NETRC gerrit_util.NETRC = netrc.netrc(gi.netrc_file) # gerrit_util.py defaults to using https, but for testing, it's much # simpler to use http connections. cls.gerrit_util_protocol_orig = gerrit_util.GERRIT_PROTOCOL gerrit_util.GERRIT_PROTOCOL = 'http' # Because we communicate with the test server via http, rather than https, # libcurl won't add authentication headers to raw git requests unless the # gerrit server returns 401. That works for pushes, but for read operations # (like git-ls-remote), gerrit will simply omit any ref that requires # authentication. By default gerrit doesn't permit anonymous read access to # refs/meta/config. Override that behavior so tests can access # refs/meta/config if necessary. clone_path = os.path.join(gi.gerrit_dir, 'tmp', 'All-Projects') cls._CloneProject('All-Projects', clone_path) project_config = os.path.join(clone_path, 'project.config') cls.check_call(['git', 'config', '--file', project_config, '--add', 'access.refs/meta/config.read', 'group Anonymous Users']) cls.check_call(['git', 'add', project_config], cwd=clone_path) cls.check_call( ['git', 'commit', '-m', 'Anonyous read for refs/meta/config'], cwd=clone_path) cls.check_call(['git', 'push', 'origin', 'HEAD:refs/meta/config'], cwd=clone_path) def setUp(self): self.tempdir = tempfile.mkdtemp() os.chmod(self.tempdir, 0o700) def tearDown(self): if TEARDOWN: shutil.rmtree(self.tempdir) @classmethod def createProject(cls, name, description='Test project', owners=None, submit_type='CHERRY_PICK'): """Create a project on the test gerrit server.""" if owners is None: owners = ['Administrators'] body = { 'description': description, 'submit_type': submit_type, 'owners': owners, } path = 'projects/%s' % urllib.quote(name, '') conn = gerrit_util.CreateHttpConn( cls.gerrit_instance.gerrit_host, path, reqtype='PUT', body=body) jmsg = gerrit_util.ReadHttpJsonResponse(conn, accept_statuses=[200, 201]) assert jmsg['name'] == name @classmethod def _post_clone_bookkeeping(cls, clone_path): config_path = os.path.join(clone_path, '.git', 'config') cls.check_call( ['git', 'config', '--file', config_path, 'user.email', cls.TEST_EMAIL]) cls.check_call( ['git', 'config', '--file', config_path, 'credential.helper', 'store --file=%s' % cls.gerrit_instance.credential_file]) @classmethod def _CloneProject(cls, name, path): """Clone a project from the test gerrit server.""" gi = cls.gerrit_instance parent_dir = os.path.dirname(path) if not os.path.exists(parent_dir): os.makedirs(parent_dir) url = '/'.join((gi.gerrit_url, name)) cls.check_call(['git', 'clone', url, path]) cls._post_clone_bookkeeping(path) # Install commit-msg hook to add Change-Id lines. hook_path = os.path.join(path, '.git', 'hooks', 'commit-msg') cls.check_call(['curl', '-o', hook_path, '/'.join((gi.gerrit_url, 'tools/hooks/commit-msg'))]) os.chmod(hook_path, stat.S_IRWXU) return path def cloneProject(self, name, path=None): """Clone a project from the test gerrit server.""" if path is None: path = os.path.basename(name) if path.endswith('.git'): path = path[:-4] path = os.path.join(self.tempdir, path) return self._CloneProject(name, path) @classmethod def _CreateCommit(cls, clone_path, fn=None, msg=None, text=None): """Create a commit in the given git checkout.""" if not fn: fn = 'test-file.txt' if not msg: msg = 'Test Message' if not text: text = 'Another day, another dollar.' fpath = os.path.join(clone_path, fn) with open(fpath, 'a') as fh: fh.write('%s\n' % text) cls.check_call(['git', 'add', fn], cwd=clone_path) cls.check_call(['git', 'commit', '-m', msg], cwd=clone_path) return cls._GetCommit(clone_path) def createCommit(self, clone_path, fn=None, msg=None, text=None): """Create a commit in the given git checkout.""" clone_path = os.path.join(self.tempdir, clone_path) return self._CreateCommit(clone_path, fn, msg, text) @classmethod def _GetCommit(cls, clone_path, ref='HEAD'): """Get the sha1 and change-id for a ref in the git checkout.""" log_proc = cls.check_output(['git', 'log', '-n', '1', ref], cwd=clone_path) sha1 = None change_id = None for line in log_proc.splitlines(): match = cls.COMMIT_RE.match(line) if match: sha1 = match.group(1) continue match = cls.CHANGEID_RE.match(line) if match: change_id = match.group(1) continue assert sha1 assert change_id return (sha1, change_id) def getCommit(self, clone_path, ref='HEAD'): """Get the sha1 and change-id for a ref in the git checkout.""" clone_path = os.path.join(self.tempdir, clone_path) return self._GetCommit(clone_path, ref) @classmethod def _UploadChange(cls, clone_path, branch='master', remote='origin'): """Create a gerrit CL from the HEAD of a git checkout.""" cls.check_call( ['git', 'push', remote, 'HEAD:refs/for/%s' % branch], cwd=clone_path) def uploadChange(self, clone_path, branch='master', remote='origin'): """Create a gerrit CL from the HEAD of a git checkout.""" clone_path = os.path.join(self.tempdir, clone_path) self._UploadChange(clone_path, branch, remote) @classmethod def _PushBranch(cls, clone_path, branch='master'): """Push a branch directly to gerrit, bypassing code review.""" cls.check_call( ['git', 'push', 'origin', 'HEAD:refs/heads/%s' % branch], cwd=clone_path) def pushBranch(self, clone_path, branch='master'): """Push a branch directly to gerrit, bypassing code review.""" clone_path = os.path.join(self.tempdir, clone_path) self._PushBranch(clone_path, branch) @classmethod def createAccount(cls, name='Test User', email='test-user@test.org', password=None, groups=None): """Create a new user account on gerrit.""" username = email.partition('@')[0] gerrit_cmd = 'gerrit create-account %s --full-name "%s" --email %s' % ( username, name, email) if password: gerrit_cmd += ' --http-password "%s"' % password if groups: gerrit_cmd += ' '.join(['--group %s' % x for x in groups]) ssh_cmd = ['ssh', '-p', cls.gerrit_instance.ssh_port, '-i', cls.gerrit_instance.ssh_ident, '-o', 'NoHostAuthenticationForLocalhost=yes', '-o', 'StrictHostKeyChecking=no', '%s@localhost' % cls.TEST_USERNAME, gerrit_cmd] cls.check_call(ssh_cmd) @classmethod def _stop_gerrit(cls, gerrit_instance): """Stops the running gerrit instance and deletes it.""" try: # This should terminate the gerrit process. cls.check_call(['bash', gerrit_instance.gerrit_exe, 'stop']) finally: try: # cls.gerrit_pid should have already terminated. If it did, then # os.waitpid will raise OSError. os.waitpid(gerrit_instance.gerrit_pid, os.WNOHANG) except OSError as e: if e.errno == errno.ECHILD: # If gerrit shut down cleanly, os.waitpid will land here. # pylint: disable=lost-exception return # If we get here, the gerrit process is still alive. Send the process # SIGKILL for good measure. try: os.kill(gerrit_instance.gerrit_pid, signal.SIGKILL) except OSError: if e.errno == errno.ESRCH: # os.kill raised an error because the process doesn't exist. Maybe # gerrit shut down cleanly after all. # pylint: disable=lost-exception return # Announce that gerrit didn't shut down cleanly. msg = 'Test gerrit server (pid=%d) did not shut down cleanly.' % ( gerrit_instance.gerrit_pid) print >> sys.stderr, msg @classmethod def tearDownClass(cls): gerrit_util.NETRC = cls.gerrit_util_netrc_orig gerrit_util.GERRIT_PROTOCOL = cls.gerrit_util_protocol_orig if TEARDOWN: cls._stop_gerrit(cls.gerrit_instance) shutil.rmtree(cls.gerrit_instance.gerrit_dir) class RepoTestCase(GerritTestCase): """Test class which runs in a repo checkout.""" REPO_URL = 'https://chromium.googlesource.com/external/repo' MANIFEST_PROJECT = 'remotepath/manifest' MANIFEST_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?> <manifest> <remote name="remote1" fetch="%(gerrit_url)s" review="%(gerrit_host)s" /> <remote name="remote2" fetch="%(gerrit_url)s" review="%(gerrit_host)s" /> <default revision="refs/heads/master" remote="remote1" sync-j="1" /> <project remote="remote1" path="localpath/testproj1" name="remotepath/testproj1" /> <project remote="remote1" path="localpath/testproj2" name="remotepath/testproj2" /> <project remote="remote2" path="localpath/testproj3" name="remotepath/testproj3" /> <project remote="remote2" path="localpath/testproj4" name="remotepath/testproj4" /> </manifest> """ @classmethod def setUpClass(cls): GerritTestCase.setUpClass() gi = cls.gerrit_instance # Create local mirror of repo tool repository. repo_mirror_path = os.path.join(gi.git_dir, 'repo.git') cls.check_call( ['git', 'clone', '--mirror', cls.REPO_URL, repo_mirror_path]) # Check out the top-level repo script; it will be used for invocation. repo_clone_path = os.path.join(gi.gerrit_dir, 'tmp', 'repo') cls.check_call(['git', 'clone', '-n', repo_mirror_path, repo_clone_path]) cls.check_call( ['git', 'checkout', 'origin/stable', 'repo'], cwd=repo_clone_path) shutil.rmtree(os.path.join(repo_clone_path, '.git')) cls.repo_exe = os.path.join(repo_clone_path, 'repo') # Create manifest repository. cls.createProject(cls.MANIFEST_PROJECT) clone_path = os.path.join(gi.gerrit_dir, 'tmp', 'manifest') cls._CloneProject(cls.MANIFEST_PROJECT, clone_path) manifest_path = os.path.join(clone_path, 'default.xml') with open(manifest_path, 'w') as fh: fh.write(cls.MANIFEST_TEMPLATE % gi.__dict__) cls.check_call(['git', 'add', 'default.xml'], cwd=clone_path) cls.check_call(['git', 'commit', '-m', 'Test manifest.'], cwd=clone_path) cls._PushBranch(clone_path) # Create project repositories. for i in xrange(1, 5): proj = 'testproj%d' % i cls.createProject('remotepath/%s' % proj) clone_path = os.path.join(gi.gerrit_dir, 'tmp', proj) cls._CloneProject('remotepath/%s' % proj, clone_path) cls._CreateCommit(clone_path) cls._PushBranch(clone_path, 'master') def setUp(self): super(RepoTestCase, self).setUp() manifest_url = '/'.join((self.gerrit_instance.gerrit_url, self.MANIFEST_PROJECT)) repo_url = '/'.join((self.gerrit_instance.gerrit_url, 'repo')) self.check_call( [self.repo_exe, 'init', '-u', manifest_url, '--repo-url', repo_url, '--no-repo-verify'], cwd=self.tempdir) self.check_call([self.repo_exe, 'sync'], cwd=self.tempdir) for i in xrange(1, 5): clone_path = os.path.join(self.tempdir, 'localpath', 'testproj%d' % i) self._post_clone_bookkeeping(clone_path) # Tell 'repo upload' to upload this project without prompting. config_path = os.path.join(clone_path, '.git', 'config') self.check_call( ['git', 'config', '--file', config_path, 'review.%s.upload' % self.gerrit_instance.gerrit_host, 'true']) @classmethod def runRepo(cls, *args, **kwargs): # Unfortunately, munging $HOME appears to be the only way to control the # netrc file used by repo. munged_home = os.path.join(cls.gerrit_instance.gerrit_dir, 'tmp') if 'env' not in kwargs: env = kwargs['env'] = os.environ.copy() env['HOME'] = munged_home else: env.setdefault('HOME', munged_home) args[0].insert(0, cls.repo_exe) cls.check_call(*args, **kwargs) def uploadChange(self, clone_path, branch='master', remote='origin'): review_host = self.check_output( ['git', 'config', 'remote.%s.review' % remote], cwd=clone_path).strip() assert(review_host) projectname = self.check_output( ['git', 'config', 'remote.%s.projectname' % remote], cwd=clone_path).strip() assert(projectname) GerritTestCase._UploadChange( clone_path, branch=branch, remote='%s://%s/%s' % ( gerrit_util.GERRIT_PROTOCOL, review_host, projectname))