Refactor test_pr.py to a more coherent design.

tools/gh_api.py

@@ -1,204 +1,215 b''
 """Functions for Github authorisation."""
 from __future__ import print_function
 
 try:
     input = raw_input
 except NameError:
     pass
 
 import os
 
 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'
 
+class Obj(dict):
+    """Dictionary with attribute access to names."""
+    def __getattr__(self, name):
+        try:
+            return self[name]
+        except KeyError:
+            raise AttributeError(name)
+    
+    def __setattr__(self, name, val):
+        self[name] = val
+
 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, github_api=3):
     """get pull request info  by number
 
     github_api : version of github api to use
     """
     if github_api==2 :
         url = "http://github.com/api/v2/json/pulls/{project}/{num}".format(project=project, num=num)
     elif github_api == 3:
         url = "https://api.github.com/repos/{project}/pulls/{num}".format(project=project, num=num)
     response = requests.get(url)
     response.raise_for_status()
     if github_api == 2 :
         return json.loads(response.text)['pull']
-    return json.loads(response.text)
+    return json.loads(response.text, object_hook=Obj)
 
 def get_pulls_list(project, github_api=3):
     """get pull request list
 
     github_api : version of github api to use
     """
     if github_api == 3 :
         url = "https://api.github.com/repos/{project}/pulls".format(project=project)
     else :
         url = "http://github.com/api/v2/json/pulls/{project}".format(project=project)
     response = requests.get(url)
     response.raise_for_status()
     if github_api == 2 :
         return json.loads(response.text)['pulls']
     return json.loads(response.text)
 
 # encode_multipart_formdata is from urllib3.filepost
 # The only change is to iter_fields, to enforce S3's required key ordering
 
 def iter_fields(fields):
     fields = fields.copy()
     for key in ('key', 'acl', 'Filename', 'success_action_status', 'AWSAccessKeyId',
         'Policy', 'Signature', 'Content-Type', 'file'):
         yield (key, fields.pop(key))
     for (k,v) in fields.items():
         yield k,v
 
 def encode_multipart_formdata(fields, boundary=None):
     """
     Encode a dictionary of ``fields`` using the multipart/form-data mime format.
 
     :param fields:
         Dictionary of fields or list of (key, value) field tuples.  The key is
         treated as the field name, and the value as the body of the form-data
         bytes. If the value is a tuple of two elements, then the first element
         is treated as the filename of the form-data section.
 
         Field names and filenames must be unicode.
 
     :param boundary:
         If not specified, then a random boundary will be generated using
         :func:`mimetools.choose_boundary`.
     """
     # copy requests imports in here:
     from io import BytesIO
     from requests.packages.urllib3.filepost import (
         choose_boundary, six, writer, b, get_content_type
     )
     body = BytesIO()
     if boundary is None:
         boundary = choose_boundary()
 
     for fieldname, value in iter_fields(fields):
         body.write(b('--%s\r\n' % (boundary)))
 
         if isinstance(value, tuple):
             filename, data = value
             writer(body).write('Content-Disposition: form-data; name="%s"; '
                                'filename="%s"\r\n' % (fieldname, filename))
             body.write(b('Content-Type: %s\r\n\r\n' %
                        (get_content_type(filename))))
         else:
             data = value
             writer(body).write('Content-Disposition: form-data; name="%s"\r\n'
                                % (fieldname))
             body.write(b'Content-Type: text/plain\r\n\r\n')
 
         if isinstance(data, int):
             data = str(data)  # Backwards compatibility
         if isinstance(data, six.text_type):
             writer(body).write(data)
         else:
             body.write(data)
 
         body.write(b'\r\n')
 
     body.write(b('--%s--\r\n' % (boundary)))
 
     content_type = b('multipart/form-data; boundary=%s' % boundary)
 
     return body.getvalue(), content_type
 
 
 def post_download(project, filename, name=None, description=""):
     """Upload a file to the GitHub downloads area"""
     if name is None:
         name = os.path.basename(filename)
     with open(filename, 'rb') as f:
         filedata = f.read()
     
     url = "https://api.github.com/repos/{project}/downloads".format(project=project)
     
     payload = json.dumps(dict(name=name, size=len(filedata),
                     description=description))
     response = requests.post(url, data=payload, headers=make_auth_header())
     response.raise_for_status()
     reply = json.loads(response.content)
     s3_url = reply['s3_url']
     
     fields = dict(
         key=reply['path'],
         acl=reply['acl'],
         success_action_status=201,
         Filename=reply['name'],
         AWSAccessKeyId=reply['accesskeyid'],
         Policy=reply['policy'],
         Signature=reply['signature'],
         file=(reply['name'], filedata),
     )
     fields['Content-Type'] = reply['mime_type']
     data, content_type = encode_multipart_formdata(fields)
     s3r = requests.post(s3_url, data=data, headers={'Content-Type': content_type})
     return s3r

tools/post_pr_test.py

@@ -1,13 +1,13 b''
 #!/usr/bin/env python
 """Post the results of a pull request test to Github.
 """
-from test_pr import load_results, post_logs, post_results_comment, print_results
+from test_pr import TestRun
 
-num, results, pr, unavailable_pythons = load_results()
-results_urls = post_logs(results)
-print_results(pr, results_urls, unavailable_pythons)
-post_results_comment(pr, results_urls, num, unavailable_pythons)
+testrun = TestRun.load_results()
+testrun.post_logs()
+testrun.print_results()
+testrun.post_results_comment()
 
 print()
 print("Posted test results to pull request")
-print("  " + pr['html_url'])
+print("  " + testrun.pr['html_url'])

tools/test_pr.py

@@ -1,259 +1,266 b''
 #!/usr/bin/env python
 """
 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
 import time
 from subprocess import call, check_call, check_output, PIPE, STDOUT, CalledProcessError
 import sys
 
 import gh_api
+from gh_api import Obj
 
 basedir = os.path.join(os.path.expanduser("~"), ".ipy_pr_tests")
 repodir = os.path.join(basedir, "ipython")
 ipy_repository = 'git://github.com/ipython/ipython.git'
 ipy_http_repository = 'http://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():
+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)
+
+class TestRun(object):
+    def __init__(self, pr_num):
+        self.unavailable_pythons = []
+        self.venvs = []
+        self.pr_num = pr_num
+        
+        self.pr = gh_api.get_pull_request(gh_project, pr_num)
+        
+        self.setup()
+        
+        self.results = []
+    
+    def available_python_versions(self):
     """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 = []
+                self.unavailable_pythons.append(py)
 
-def setup():
+    def setup(self):
     """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():
+        for py in self.available_python_versions():
         check_call(['virtualenv', '-p', py, '--system-site-packages', 'venv-%s' % py])
-        venvs.append((py, 'venv-%s' % py))
+            self.venvs.append((py, 'venv-%s' % py))
     
     # Check out and update the repository
     if not os.path.exists('ipython'):
         try :
             check_call(['git', 'clone', ipy_repository])
         except CalledProcessError :
             check_call(['git', 'clone', ipy_http_repository])
     os.chdir(repodir)
     check_call(['git', 'checkout', 'master'])
     try :
         check_call(['git', 'pull', ipy_repository, 'master'])
     except CalledProcessError :
         check_call(['git', 'pull', ipy_http_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(self):
+        repo = self.pr['head']['repo']['clone_url']
+        branch = self.pr['head']['ref']
+        owner = self.pr['head']['repo']['owner']['login']
+        mergeable = self.pr['mergeable']
 
-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', '--no-ff', '--no-commit', repo, branch])
         check_call(['git', 'commit', '-m', "merge %s/%s" % (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)
-    # cleanup build-dir
-    if os.path.exists('build'):
-        shutil.rmtree('build')
-    check_call([py, 'setup.py', 'install'])
-    os.chdir(basedir)
-    
-    # Environment variables:
-    orig_path = os.environ["PATH"]
-    os.environ["PATH"] = os.path.join(basedir, venv, 'bin') + ':' + os.environ["PATH"]
-    os.environ.pop("PYTHONPATH", None)
-    
-    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')
-    finally:
-        # Restore $PATH
-        os.environ["PATH"] = orig_path
-
-def markdown_format(pr, results_urls, unavailable_pythons):
-    def format_result(py, passed, gist_url, missing_libraries):
-        s = "* %s: " % py
-        if passed:
+    def markdown_format(self):
+        def format_result(result):
+            s = "* %s: " % result.py
+            if result.passed:
             s += "OK"
         else:
-            s += "Failed, log at %s" % gist_url
-        if missing_libraries:
-            s += " (libraries not available: " + missing_libraries + ")"
+                s += "Failed, log at %s" % result.log_url
+            if result.missing_libraries:
+                s += " (libraries not available: " + result.missing_libraries + ")"
         return s
     
-    if pr['mergeable']:
-        com = pr['head']['sha'][:7] + " merged into master"
+        if self.pr['mergeable']:
+            com = self.pr['head']['sha'][:7] + " merged into master"
     else:
-        com = pr['head']['sha'][:7] + " (can't merge cleanly)"
+            com = self.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] + \
+                [format_result(r) for r in self.results] + \
             ["",
-             "Not available for testing: " + ", ".join(unavailable_pythons)] 
+                 "Not available for testing: " + ", ".join(self.unavailable_pythons)]
     return "\n".join(lines)
 
-def post_results_comment(pr, results, num, unavailable_pythons=unavailable_pythons):
-    body = markdown_format(pr, results, unavailable_pythons)
-    gh_api.post_issue_comment(gh_project, num, body)
+    def post_results_comment(self):
+        body = self.markdown_format()
+        gh_api.post_issue_comment(gh_project, self.pr_num, body)
+    
+    def print_results(self):
+        pr = self.pr
 
-def print_results(pr, results_urls, unavailable_pythons=unavailable_pythons):
     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")
+        for result in self.results:
+            if result.passed:
+                print(result.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))
+                print(result.py, ":", "Failed")
+                print("    Test log:", result.get('log_url') or result.log_file)
+            if result.missing_libraries:
+                print("    Libraries not available:", result.missing_libraries)
+        print("Not available for testing:", ", ".join(self.unavailable_pythons))
 
-def dump_results(num, results, pr):
+    def dump_results(self):
     with open(os.path.join(basedir, 'lastresults.pkl'), 'wb') as f:
-        pickle.dump((num, results, pr, unavailable_pythons), f)
+            pickle.dump(self, f)
 
+    @staticmethod
 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"))
+    def save_logs(self):
+        for result in self.results:
+            if not result.passed:
+                result_locn = os.path.abspath(os.path.join('venv-%s' % result.py,
+                                            self.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))
+                    f.write(result.log)
     
-    return results_paths
+                result.log_file = result_locn
 
-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',
+    def post_logs(self):
+        for result in self.results:
+            if not result.passed:
+                result.log_url = gh_api.post_gist(result.log,
+                                                description='IPython test log',
                                             filename="results.log", auth=True)
-            results_urls.append((py, False, result_locn, missing_libraries))
     
-    return results_urls
+    def run(self):
+        for py, venv in self.venvs:
+            tic = time.time()
+            passed, log = run_tests(venv)
+            elapsed = int(time.time() - tic)
+            print("Ran tests with %s in %is" % (py, elapsed))
+            missing_libraries = get_missing_libraries(log)
+            
+            self.results.append(Obj(py=py,
+                                   passed=passed,
+                                   log=log,
+                                   missing_libraries=missing_libraries
+                                  )
+                               )
+
+
+def run_tests(venv):
+    py = os.path.join(basedir, venv, 'bin', 'python')
+    print(py)
+    os.chdir(repodir)
+    # cleanup build-dir
+    if os.path.exists('build'):
+        shutil.rmtree('build')
+    check_call([py, 'setup.py', 'install'])
+    os.chdir(basedir)
+    
+    # Environment variables:
+    orig_path = os.environ["PATH"]
+    os.environ["PATH"] = os.path.join(basedir, venv, 'bin') + ':' + os.environ["PATH"]
+    os.environ.pop("PYTHONPATH", None)
+    
+    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')
+    finally:
+        # Restore $PATH
+        os.environ["PATH"] = orig_path
+
 
 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'],
-              )
+    testrun = TestRun(num)
     
-    results = []
-    for py, venv in venvs:
-        tic = time.time()
-        passed, log = run_tests(venv)
-        elapsed = int(time.time() - tic)
-        print("Ran tests with %s in %is" % (py, elapsed))
-        missing_libraries = get_missing_libraries(log)
-        if passed:
-            results.append((py, True, None, missing_libraries))
-        else:
-            results.append((py, False, log, missing_libraries))
+    testrun.get_branch()
+    
+    testrun.run()
     
-    dump_results(num, results, pr)
+    testrun.dump_results()
     
-    results_paths = save_logs(results, pr)
-    print_results(pr, results_paths)
+    testrun.save_logs()
+    testrun.print_results()
     
     if post_results:
-        results_urls = post_logs(results)
-        post_results_comment(pr, results_urls, num)
+        results_urls = testrun.post_logs
+        testrun.post_results_comment()
         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)