Home | History | Annotate | Download | only in ci 
      1 # Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
      2 # For details: https://bitbucket.org/ned/coveragepy/src/default/NOTICE.txt
      3 
      4 """Use the Appveyor API to download Windows artifacts."""
      5 
      6 import os
      7 import os.path
      8 import sys
      9 import zipfile
     10 
     11 import requests
     12 
     13 
     14 def make_auth_headers():
     15     """Make the authentication headers needed to use the Appveyor API."""
     16     with open("ci/appveyor.token") as f:
     17         token = f.read().strip()
     18 
     19     headers = {
     20         'Authorization': 'Bearer {0}'.format(token),
     21     }
     22     return headers
     23 
     24 
     25 def make_url(url, **kwargs):
     26     """Build an Appveyor API url."""
     27     return "https://ci.appveyor.com/api" + url.format(**kwargs)
     28 
     29 
     30 def get_project_build(account_project):
     31     """Get the details of the latest Appveyor build."""
     32     url = make_url("/projects/{account_project}", account_project=account_project)
     33     response = requests.get(url, headers=make_auth_headers())
     34     return response.json()
     35 
     36 
     37 def download_latest_artifacts(account_project):
     38     """Download all the artifacts from the latest build."""
     39     build = get_project_build(account_project)
     40     jobs = build['build']['jobs']
     41     print "Build {0[build][version]}, {1} jobs: {0[build][message]}".format(build, len(jobs))
     42     for job in jobs:
     43         name = job['name'].partition(':')[2].split(',')[0].strip()
     44         print "  {0}: {1[status]}, {1[artifactsCount]} artifacts".format(name, job)
     45 
     46         url = make_url("/buildjobs/{jobid}/artifacts", jobid=job['jobId'])
     47         response = requests.get(url, headers=make_auth_headers())
     48         artifacts = response.json()
     49 
     50         for artifact in artifacts:
     51             is_zip = artifact['type'] == "Zip"
     52             filename = artifact['fileName']
     53             print "    {0}, {1} bytes".format(filename, artifact['size'])
     54 
     55             url = make_url(
     56                 "/buildjobs/{jobid}/artifacts/{filename}",
     57                 jobid=job['jobId'],
     58                 filename=filename
     59             )
     60             download_url(url, filename, make_auth_headers())
     61 
     62             if is_zip:
     63                 unpack_zipfile(filename)
     64                 os.remove(filename)
     65 
     66 
     67 def ensure_dirs(filename):
     68     """Make sure the directories exist for `filename`."""
     69     dirname, _ = os.path.split(filename)
     70     if dirname and not os.path.exists(dirname):
     71         os.makedirs(dirname)
     72 
     73 
     74 def download_url(url, filename, headers):
     75     """Download a file from `url` to `filename`."""
     76     ensure_dirs(filename)
     77     response = requests.get(url, headers=headers, stream=True)
     78     if response.status_code == 200:
     79         with open(filename, 'wb') as f:
     80             for chunk in response.iter_content(16*1024):
     81                 f.write(chunk)
     82 
     83 
     84 def unpack_zipfile(filename):
     85     """Unpack a zipfile, using the names in the zip."""
     86     with open(filename, 'rb') as fzip:
     87         z = zipfile.ZipFile(fzip)
     88         for name in z.namelist():
     89             print "      extracting {0}".format(name)
     90             ensure_dirs(name)
     91             z.extract(name)
     92 
     93 
     94 if __name__ == "__main__":
     95     download_latest_artifacts(sys.argv[1])
     96