From 7bb7aed4d00d72dfdb47b3a9e97fb0c76679c97f 2012-05-05 10:56:44
From: Thomas Kluyver <takowl@gmail.com>
Date: 2012-05-05 10:56:44
Subject: [PATCH] Merge pull request #1685 from takluyver/test-pr

Add script to test pull request
---

diff --git a/tools/gh_api.py b/tools/gh_api.py
new file mode 100644
index 0000000..6b042ed
--- /dev/null
+++ b/tools/gh_api.py
@@ -0,0 +1,80 @@
+"""Functions for Github authorisation."""
+from __future__ import print_function
+
+try:
+    input = raw_input
+except NameError:
+    pass
+
+import requests
+import getpass
+import json
+
+# Keyring stores passwords by a 'username', but we're not storing a username and
+# password
+fake_username = 'ipython_tools'
+
+token = None
+def get_auth_token():
+    global token
+    
+    if token is not None:
+        return token
+    
+    import keyring
+    token = keyring.get_password('github', fake_username)
+    if token is not None:
+        return token
+    
+    print("Please enter your github username and password. These are not "
+           "stored, only used to get an oAuth token. You can revoke this at "
+           "any time on Github.")
+    user = input("Username: ")
+    pw = getpass.getpass("Password: ")
+    
+    auth_request = {
+      "scopes": [
+        "public_repo",
+        "gist"
+      ],
+      "note": "IPython tools",
+      "note_url": "https://github.com/ipython/ipython/tree/master/tools",
+    }
+    response = requests.post('https://api.github.com/authorizations',
+                            auth=(user, pw), data=json.dumps(auth_request))
+    response.raise_for_status()
+    token = json.loads(response.text)['token']
+    keyring.set_password('github', fake_username, token)
+    return token
+
+def make_auth_header():
+    return {'Authorization': 'token ' + get_auth_token()}
+
+def post_issue_comment(project, num, body):
+    url = 'https://api.github.com/repos/{project}/issues/{num}/comments'.format(project=project, num=num)
+    payload = json.dumps({'body': body})
+    r = requests.post(url, data=payload, headers=make_auth_header())
+
+def post_gist(content, description='', filename='file', auth=False):
+    """Post some text to a Gist, and return the URL."""
+    post_data = json.dumps({
+      "description": description,
+      "public": True,
+      "files": {
+        filename: {
+          "content": content
+        }
+      }
+    }).encode('utf-8')
+    
+    headers = make_auth_header() if auth else {}
+    response = requests.post("https://api.github.com/gists", data=post_data, headers=headers)
+    response.raise_for_status()
+    response_data = json.loads(response.text)
+    return response_data['html_url']
+    
+def get_pull_request(project, num):
+    url = "https://api.github.com/repos/{project}/pulls/{num}".format(project=project, num=num)
+    response = requests.get(url)
+    response.raise_for_status()
+    return json.loads(response.text)
diff --git a/tools/post_pr_test.py b/tools/post_pr_test.py
new file mode 100755
index 0000000..c5cdcd1
--- /dev/null
+++ b/tools/post_pr_test.py
@@ -0,0 +1,10 @@
+from test_pr import load_results, post_logs, post_results_comment, print_results
+
+num, results, pr = load_results()
+results_urls = post_logs(results)
+print_results(pr, results_urls)
+post_results_comment(pr, results_urls, num)
+
+print()
+print("Posted test results to pull request")
+print("  " + pr['html_url'])
diff --git a/tools/test_pr.py b/tools/test_pr.py
new file mode 100755
index 0000000..338370e
--- /dev/null
+++ b/tools/test_pr.py
@@ -0,0 +1,235 @@
+"""
+This is a script for testing pull requests for IPython. It merges the pull
+request with current master, installs and tests on all available versions of
+Python, and posts the results to Gist if any tests fail.
+
+Usage:
+    python test_pr.py 1657
+"""
+from __future__ import print_function
+
+import errno
+from glob import glob
+import io
+import json
+import os
+import pickle
+import re
+import requests
+import shutil
+from subprocess import call, check_call, check_output, PIPE, STDOUT, CalledProcessError
+import sys
+
+import gh_api
+
+basedir = os.path.join(os.path.expanduser("~"), ".ipy_pr_tests")
+repodir = os.path.join(basedir, "ipython")
+ipy_repository = 'git://github.com/ipython/ipython.git'
+gh_project="ipython/ipython"
+
+supported_pythons = ['python2.6', 'python2.7', 'python3.1', 'python3.2']
+unavailable_pythons = []
+
+def available_python_versions():
+    """Get the executable names of available versions of Python on the system.
+    """
+    del unavailable_pythons[:]
+    for py in supported_pythons:
+        try:
+            check_call([py, '-c', 'import nose'], stdout=PIPE)
+            yield py
+        except (OSError, CalledProcessError):
+            unavailable_pythons.append(py)
+
+venvs = []
+
+def setup():
+    """Prepare the repository and virtualenvs."""
+    global venvs
+    
+    try:
+        os.mkdir(basedir)
+    except OSError as e:
+        if e.errno != errno.EEXIST:
+            raise
+    os.chdir(basedir)
+    
+    # Delete virtualenvs and recreate
+    for venv in glob('venv-*'):
+        shutil.rmtree(venv)
+    for py in available_python_versions():
+        check_call(['virtualenv', '-p', py, '--system-site-packages', 'venv-%s' % py])
+        venvs.append((py, 'venv-%s' % py))
+    
+    # Check out and update the repository
+    if not os.path.exists('ipython'):
+        check_call(['git', 'clone', ipy_repository])
+    os.chdir(repodir)
+    check_call(['git', 'checkout', 'master'])
+    check_call(['git', 'pull', ipy_repository, 'master'])
+    os.chdir(basedir)
+
+missing_libs_re = re.compile(r"Tools and libraries NOT available at test time:\n"
+                             r"\s*(.*?)\n")
+def get_missing_libraries(log):
+    m = missing_libs_re.search(log)
+    if m:
+        return m.group(1)
+
+def get_branch(repo, branch, owner, mergeable):
+    os.chdir(repodir)
+    if mergeable:
+        merged_branch = "%s-%s" % (owner, branch)
+        # Delete the branch first
+        call(['git', 'branch', '-D', merged_branch])
+        check_call(['git', 'checkout', '-b', merged_branch])
+        check_call(['git', 'pull', repo, branch])
+    else:
+        # Fetch the branch without merging it.
+        check_call(['git', 'fetch', repo, branch])
+        check_call(['git', 'checkout', 'FETCH_HEAD'])
+    os.chdir(basedir)
+
+def run_tests(venv):
+    py = os.path.join(basedir, venv, 'bin', 'python')
+    print(py)
+    os.chdir(repodir)
+    check_call([py, 'setup.py', 'install'])
+    os.chdir(basedir)
+    
+    iptest = os.path.join(basedir, venv, 'bin', 'iptest')
+    if not os.path.exists(iptest):
+        iptest = os.path.join(basedir, venv, 'bin', 'iptest3')
+        
+    print("\nRunning tests, this typically takes a few minutes...")
+    try:
+        return True, check_output([iptest], stderr=STDOUT).decode('utf-8')
+    except CalledProcessError as e:
+        return False, e.output.decode('utf-8')
+
+def markdown_format(pr, results_urls):
+    def format_result(py, passed, gist_url, missing_libraries):
+        s = "* %s: " % py
+        if passed:
+            s += "OK"
+        else:
+            s += "Failed, log at %s" % gist_url
+        if missing_libraries:
+            s += " (libraries not available: " + missing_libraries + ")"
+        return s
+    
+    if pr['mergeable']:
+        com = pr['head']['sha'][:7] + " merged into master"
+    else:
+        com = pr['head']['sha'][:7] + " (can't merge cleanly)"
+    lines = ["**Test results for commit %s**" % com,
+             "Platform: " + sys.platform,
+             ""] + \
+            [format_result(*r) for r in results_urls] + \
+            ["",
+             "Not available for testing: " + ", ".join(unavailable_pythons)] 
+    return "\n".join(lines)
+
+def post_results_comment(pr, results, num):
+    body = markdown_format(pr, results)
+    gh_api.post_issue_comment(gh_project, num, body)
+
+def print_results(pr, results_urls):
+    print("\n")
+    if pr['mergeable']:
+        print("**Test results for commit %s merged into master**" % pr['head']['sha'][:7])
+    else:
+        print("**Test results for commit %s (can't merge cleanly)**" % pr['head']['sha'][:7])
+    print("Platform:", sys.platform)
+    for py, passed, gist_url, missing_libraries in results_urls:
+        if passed:
+            print(py, ":", "OK")
+        else:
+            print(py, ":", "Failed")
+            print("    Test log:", gist_url)
+        if missing_libraries:
+            print("    Libraries not available:", missing_libraries)
+    print("Not available for testing:", ", ".join(unavailable_pythons))
+
+def dump_results(num, results, pr):
+    with open(os.path.join(basedir, 'lastresults.pkl'), 'wb') as f:
+        pickle.dump((num, results, pr), f)
+
+def load_results():
+    with open(os.path.join(basedir, 'lastresults.pkl'), 'rb') as f:
+        return pickle.load(f)
+
+def save_logs(results, pr):
+    results_paths = []
+    for py, passed, log, missing_libraries in results:
+        if passed:
+            results_paths.append((py, passed, None, missing_libraries))
+        else:
+            
+            result_locn = os.path.abspath(os.path.join('venv-%s' % py,
+                                        pr['head']['sha'][:7]+".log"))
+            with io.open(result_locn, 'w', encoding='utf-8') as f:
+                f.write(log)
+        
+            results_paths.append((py, False, result_locn, missing_libraries))
+    
+    return results_paths
+
+def post_logs(results):
+    results_urls = []
+    for py, passed, log, missing_libraries in results:
+        if passed:
+            results_urls.append((py, passed, None, missing_libraries))
+        else:
+            result_locn = gh_api.post_gist(log, description='IPython test log',
+                                            filename="results.log", auth=True)
+            results_urls.append((py, False, result_locn, missing_libraries))
+    
+    return results_urls
+
+def test_pr(num, post_results=True):
+    # Get Github authorisation first, so that the user is prompted straight away
+    # if their login is needed.
+    if post_results:
+        gh_api.get_auth_token()
+    
+    setup()
+    pr = gh_api.get_pull_request(gh_project, num)
+    get_branch(repo=pr['head']['repo']['clone_url'], 
+                 branch=pr['head']['ref'],
+                 owner=pr['head']['repo']['owner']['login'],
+                 mergeable=pr['mergeable'],
+              )
+    
+    results = []
+    for py, venv in venvs:
+        passed, log = run_tests(venv)
+        missing_libraries = get_missing_libraries(log)
+        if passed:
+            results.append((py, True, None, missing_libraries))
+        else:
+            results.append((py, False, log, missing_libraries))
+    
+    dump_results(num, results, pr)
+    
+    results_paths = save_logs(results, pr)
+    print_results(pr, results_paths)
+    
+    if post_results:
+        results_urls = post_logs(results)
+        post_results_comment(pr, results_urls, num)
+        print("(Posted to Github)")
+    else:
+        post_script = os.path.join(os.path.dirname(sys.argv[0]), "post_pr_test.py")
+        print("To post the results to Github, run", post_script)
+    
+
+if __name__ == '__main__':
+    import argparse
+    parser = argparse.ArgumentParser(description="Test an IPython pull request")
+    parser.add_argument('-p', '--publish', action='store_true',
+                        help="Publish the results to Github")
+    parser.add_argument('number', type=int, help="The pull request number")
+    
+    args = parser.parse_args()
+    test_pr(args.number, post_results=args.publish)