• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
6import os
7import os.path
8import sys
9import zipfile
10
11import requests
12
13
14def 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
25def make_url(url, **kwargs):
26    """Build an Appveyor API url."""
27    return "https://ci.appveyor.com/api" + url.format(**kwargs)
28
29
30def 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
37def 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
67def 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
74def 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
84def 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
94if __name__ == "__main__":
95    download_latest_artifacts(sys.argv[1])
96