forked from mozilla/release-services
/
api.py
90 lines (67 loc) · 2.61 KB
/
api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
# -*- coding: utf-8 -*-
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import asyncio
from rq import Queue
from cli_common.phabricator import PhabricatorAPI
from cli_common.phabricator import revision_exists_on_central
from codecoverage_backend import coverage
from codecoverage_backend import coverage_by_changeset_impl
from codecoverage_backend import coverage_for_file_impl
from codecoverage_backend import coverage_summary_by_changeset_impl
from codecoverage_backend import secrets
from codecoverage_backend.worker import conn
q = Queue(connection=conn)
def coverage_for_file(changeset, path):
changeset = changeset[:12]
try:
return asyncio.get_event_loop().run_until_complete(coverage_for_file_impl.generate(changeset, path))
except Exception as e:
return {
'error': str(e)
}, 500
def coverage_by_changeset_job(changeset):
return asyncio.get_event_loop().run_until_complete(coverage_by_changeset_impl.generate(changeset))
def coverage_by_changeset(changeset):
changeset = changeset[:12]
job = q.fetch_job(changeset)
if job is None:
RESULT_TTL = 2 * 24 * 60 * 60
job = q.enqueue(
coverage_by_changeset_job,
changeset,
job_id=changeset,
result_ttl=RESULT_TTL
)
if job.result is not None:
return job.result, 200
if job.exc_info is not None:
return {
'error': str(job.exc_info)
}, 500
return '', 202
def coverage_summary_by_changeset(changeset):
result, code = coverage_by_changeset(changeset)
if code != 200:
return result, code
else:
return coverage_summary_by_changeset_impl.generate(result), 200
def coverage_supported_extensions():
return coverage.COVERAGE_EXTENSIONS
def coverage_latest():
return asyncio.get_event_loop().run_until_complete(coverage.get_latest_build_info())
def phabricator_base_revision_from_phid(revision_phid):
try:
phabricator = PhabricatorAPI(secrets.PHABRICATOR_TOKEN)
diffs = phabricator.search_diffs(revision_phid=revision_phid)
if len(diffs) > 0:
revision = diffs[-1]['baseRevision']
if revision and revision_exists_on_central(revision):
return {'revision': revision}, 200
return {'error': 'Base revision not found.'}, 404
except Exception as e:
return {
'error': str(e),
'error_code': getattr(e, 'error_code', 'unknown')
}, 500