-
Notifications
You must be signed in to change notification settings - Fork 1
/
tasks.py
139 lines (114 loc) · 3.54 KB
/
tasks.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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
from invoke import task
@task
def test(c, cov=False, cov_report=False, junit=False, enforce_percent=0):
""" Run unit tests. """
command = "pytest --disable-warnings"
if cov_report:
command += " --cov-report=html"
if junit:
command += " --junitxml=test-results/results.xml"
if enforce_percent > 0:
command += f" --cov-fail-under={enforce_percent}"
if cov or cov_report or junit or enforce_percent:
command += " --cov app"
else:
command += " app tests"
c.run(command)
@task
def style(c, fix=False):
""" Run black checks against the codebase. """
command = "black"
if not fix:
command += " --check"
c.run(f"{command} app tests")
@task
def lint(c, fail_under=0):
""" Run pylint checks against the codebase """
command = "pylint --rcfile=.pylintrc"
if fail_under > 0:
command += f" --fail-under={fail_under}"
c.run(f"{command} app")
@task
def precommit(c, fix=False):
test(c, junit=True, enforce_percent=99)
style(c, fix=fix)
lint(c, fail_under=9)
@task
def create_application_key(c, name, permission):
""" Create an application api_key.
name = name of the application
permission = permission to support.
"""
from app import create_app
from application_roles.services import create_application
from app.model_utils import SystemPermissionEnum
(app, db, _, _) = create_app()
with app.app_context():
application = create_application(name, SystemPermissionEnum[permission])
db.session.commit()
print(f"API_TOKEN={application.api_key}")
@task
def run_worker(
c,
input_directory,
docker_image,
repository_url,
repository_branch,
repository_script,
):
""" Run a pipeline repository locally. This runs a repository with sample
data provided in `input_directory` using the same logic used by the workflow
server's celery task.
"""
from app.tasks import execute_pipeline
from unittest.mock import patch
import logging
import os
import sys
logging.basicConfig(level=logging.DEBUG, stream=sys.stdout)
logger = logging.getLogger("run_task")
@patch("app.tasks.get_task_logger")
@patch("app.tasks.RunExecutor._make_request")
def fake_execute(make_request_mock, task_logger_mock):
def note_call(*args, **kwargs):
print(args)
return None
task_logger_mock.return_value = logger
make_request_mock().side_effect = note_call
execute_pipeline(
"pipeline-uuid",
"pipeline-run-uuid",
[
{"name": f, "url": f"file://{input_directory}/{f}"}
for f in os.listdir(input_directory)
],
docker_image,
repository_url,
repository_branch,
repository_script,
)
fake_execute()
@task
def create_pipeline(
c,
name,
docker_image_url,
repository_ssh_url,
repository_branch="master",
repository_script="openfido.sh",
):
""" Create a pipeline. """
from app import create_app
from app.pipelines.services import create_pipeline
(app, db, _, _) = create_app()
with app.app_context():
pipeline = create_pipeline(
{
"name": name,
"docker_image_url": docker_image_url,
"repository_ssh_url": repository_ssh_url,
"repository_branch": repository_branch,
"repository_script": repository_script,
}
)
print(pipeline.uuid)