forked from nautobot/nautobot-app-data-validation-engine
-
Notifications
You must be signed in to change notification settings - Fork 0
/
tasks.py
451 lines (354 loc) · 14 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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
"""Tasks for use with Invoke.
(c) 2020-2021 Network To Code
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from distutils.util import strtobool
from invoke import Collection, task as invoke_task
import os
def is_truthy(arg):
"""Convert "truthy" strings into Booleans.
Examples:
>>> is_truthy('yes')
True
Args:
arg (str): Truthy string (True values are y, yes, t, true, on and 1; false values are n, no,
f, false, off and 0. Raises ValueError if val is anything else.
"""
if isinstance(arg, bool):
return arg
return bool(strtobool(arg))
# Use pyinvoke configuration for default values, see http://docs.pyinvoke.org/en/stable/concepts/configuration.html
# Variables may be overwritten in invoke.yml or by the environment variables INVOKE_NAUTOBOT_DATA_VALIDATION_ENGINE_xxx
namespace = Collection("nautobot_data_validation_engine")
namespace.configure(
{
"nautobot_data_validation_engine": {
"nautobot_ver": "latest",
"project_name": "nautobot_data_validation_engine",
"python_ver": "3.8",
"local": False,
"compose_dir": os.path.join(os.path.dirname(__file__), "development"),
"compose_files": [
"docker-compose.base.yml",
"docker-compose.redis.yml",
"docker-compose.postgres.yml",
"docker-compose.dev.yml",
],
"compose_http_timeout": "86400",
}
}
)
def task(function=None, *args, **kwargs):
"""Task decorator to override the default Invoke task decorator and add each task to the invoke namespace."""
def task_wrapper(function=None):
"""Wrapper around invoke.task to add the task to the namespace as well."""
if args or kwargs:
task_func = invoke_task(*args, **kwargs)(function)
else:
task_func = invoke_task(function)
namespace.add_task(task_func)
return task_func
if function:
# The decorator was called with no arguments
return task_wrapper(function)
# The decorator was called with arguments
return task_wrapper
def docker_compose(context, command, **kwargs):
"""Helper function for running a specific docker-compose command with all appropriate parameters and environment.
Args:
context (obj): Used to run specific commands
command (str): Command string to append to the "docker-compose ..." command, such as "build", "up", etc.
**kwargs: Passed through to the context.run() call.
"""
build_env = {
# Note: 'docker-compose logs' will stop following after 60 seconds by default,
# so we are overriding that by setting this environment variable.
"COMPOSE_HTTP_TIMEOUT": context.nautobot_data_validation_engine.compose_http_timeout,
"NAUTOBOT_VER": context.nautobot_data_validation_engine.nautobot_ver,
"PYTHON_VER": context.nautobot_data_validation_engine.python_ver,
}
compose_command_tokens = [
"docker-compose",
f"--project-name {context.nautobot_data_validation_engine.project_name}",
f'--project-directory "{context.nautobot_data_validation_engine.compose_dir}"',
]
for compose_file in context.nautobot_data_validation_engine.compose_files:
compose_file_path = os.path.join(context.nautobot_data_validation_engine.compose_dir, compose_file)
compose_command_tokens.append(f' -f "{compose_file_path}"')
compose_command_tokens.append(command)
# If `service` was passed as a kwarg, add it to the end.
service = kwargs.pop("service", None)
if service is not None:
compose_command_tokens.append(service)
print(f'Running docker-compose command "{command}"')
compose_command = " ".join(compose_command_tokens)
return context.run(compose_command, env=build_env, **kwargs)
def run_command(context, command, **kwargs):
"""Wrapper to run a command locally or inside the nautobot container."""
if is_truthy(context.nautobot_data_validation_engine.local):
context.run(command, **kwargs)
else:
# Check if nautobot is running, no need to start another nautobot container to run a command
docker_compose_status = "ps --services --filter status=running"
results = docker_compose(context, docker_compose_status, hide="out")
if "nautobot" in results.stdout:
compose_command = f"exec nautobot {command}"
else:
compose_command = f"run --entrypoint '{command}' nautobot"
docker_compose(context, compose_command, pty=True)
# ------------------------------------------------------------------------------
# BUILD
# ------------------------------------------------------------------------------
@task(
help={
"force_rm": "Always remove intermediate containers",
"cache": "Whether to use Docker's cache when building the image (defaults to enabled)",
}
)
def build(context, force_rm=False, cache=True):
"""Build Nautobot docker image."""
command = "build"
if not cache:
command += " --no-cache"
if force_rm:
command += " --force-rm"
print(f"Building Nautobot with Python {context.nautobot_data_validation_engine.python_ver}...")
docker_compose(context, command)
@task
def generate_packages(context):
"""Generate all Python packages inside docker and copy the file locally under dist/."""
command = "poetry build"
run_command(context, command)
# ------------------------------------------------------------------------------
# START / STOP / DEBUG
# ------------------------------------------------------------------------------
@task
def debug(context):
"""Start Nautobot and its dependencies in debug mode."""
print("Starting Nautobot in debug mode...")
docker_compose(context, "up")
@task(help={"service": "If specified, only affect this service."})
def start(context, service=None):
"""Start Nautobot and its dependencies in detached mode."""
print("Starting Nautobot in detached mode...")
docker_compose(context, "up --detach", service=service)
@task
def restart(context):
"""Gracefully restart all containers."""
print("Restarting Nautobot...")
docker_compose(context, "restart")
@task
def stop(context):
"""Stop Nautobot and its dependencies."""
print("Stopping Nautobot...")
docker_compose(context, "down")
@task
def destroy(context):
"""Destroy all containers and volumes."""
print("Destroying Nautobot...")
docker_compose(context, "down --volumes")
@task
def vscode(context):
"""Launch Visual Studio Code with the appropriate Environment variables to run in a container."""
command = "code nautobot.code-workspace"
context.run(command)
@task(
help={
"service": "Docker-compose service name to view (default: nautobot)",
"follow": "Follow logs",
"tail": "Tail N number of lines or 'all'",
}
)
def logs(context, service="nautobot", follow=False, tail=None):
"""View the logs of a docker-compose service."""
command = "logs "
if follow:
command += "--follow "
if tail:
command += f"--tail={tail} "
command += service
docker_compose(context, command)
# ------------------------------------------------------------------------------
# ACTIONS
# ------------------------------------------------------------------------------
@task
def nbshell(context):
"""Launch an interactive nbshell session."""
command = "nautobot-server nbshell"
run_command(context, command)
@task
def shell_plus(context):
"""Launch an interactive shell_plus session."""
command = "nautobot-server shell_plus"
run_command(context, command)
@task
def cli(context):
"""Launch a bash shell inside the running Nautobot container."""
run_command(context, "bash")
@task(
help={
"user": "name of the superuser to create (default: admin)",
}
)
def createsuperuser(context, user="admin"):
"""Create a new Nautobot superuser account (default: "admin"), will prompt for password."""
command = f"nautobot-server createsuperuser --username {user}"
run_command(context, command)
@task(
help={
"name": "name of the migration to be created; if unspecified, will autogenerate a name",
}
)
def makemigrations(context, name=""):
"""Perform makemigrations operation in Django."""
command = "nautobot-server makemigrations nautobot_data_validation_engine"
if name:
command += f" --name {name}"
run_command(context, command)
@task
def migrate(context):
"""Perform migrate operation in Django."""
command = "nautobot-server migrate"
run_command(context, command)
@task(help={})
def post_upgrade(context):
"""
Performs Nautobot common post-upgrade operations using a single entrypoint.
This will run the following management commands with default settings, in order:
- migrate
- trace_paths
- collectstatic
- remove_stale_contenttypes
- clearsessions
- invalidate all
"""
command = "nautobot-server post_upgrade"
run_command(context, command)
# ------------------------------------------------------------------------------
# DOCS
# ------------------------------------------------------------------------------
@task
def docs(context):
"""Build and serve docs locally for development."""
command = "mkdocs serve -v"
if is_truthy(context.nautobot_data_validation_engine.local):
print(">>> Serving Documentation at http://localhost:8001")
run_command(context, command)
else:
start(context, service="docs")
# ------------------------------------------------------------------------------
# TESTS
# ------------------------------------------------------------------------------
@task(
help={
"autoformat": "Apply formatting recommendations automatically, rather than failing if formatting is incorrect.",
}
)
def black(context, autoformat=False):
"""Check Python code style with Black."""
if autoformat:
black_command = "black"
else:
black_command = "black --check --diff"
command = f"{black_command} ."
run_command(context, command)
@task
def flake8(context):
"""Check for PEP8 compliance and other style issues."""
command = "flake8 . --config .flake8"
run_command(context, command)
@task
def hadolint(context):
"""Check Dockerfile for hadolint compliance and other style issues."""
command = "hadolint development/Dockerfile"
run_command(context, command)
@task
def pylint(context):
"""Run pylint code analysis."""
command = (
'pylint --init-hook "import nautobot; nautobot.setup()" --rcfile pyproject.toml nautobot_data_validation_engine'
)
run_command(context, command)
@task
def pydocstyle(context):
"""Run pydocstyle to validate docstring formatting adheres to NTC defined standards."""
# We exclude the /migrations/ directory since it is autogenerated code
command = "pydocstyle ."
run_command(context, command)
@task
def bandit(context):
"""Run bandit to validate basic static code security analysis."""
command = "bandit --recursive . --configfile .bandit.yml"
run_command(context, command)
@task
def yamllint(context):
"""Run yamllint to validate formating adheres to NTC defined YAML standards.
Args:
context (obj): Used to run specific commands
"""
command = "yamllint . --format standard"
run_command(context, command)
@task
def check_migrations(context):
"""Check for missing migrations."""
command = "nautobot-server --config=nautobot/core/tests/nautobot_config.py makemigrations --dry-run --check"
run_command(context, command)
@task(
help={
"keepdb": "save and re-use test database between test runs for faster re-testing.",
"label": "specify a directory or module to test instead of running all Nautobot tests",
"failfast": "fail as soon as a single test fails don't run the entire test suite",
"buffer": "Discard output from passing tests",
}
)
def unittest(context, keepdb=False, label="nautobot_data_validation_engine", failfast=False, buffer=True):
"""Run Nautobot unit tests."""
command = f"coverage run --module nautobot.core.cli test {label}"
if keepdb:
command += " --keepdb"
if failfast:
command += " --failfast"
if buffer:
command += " --buffer"
run_command(context, command)
@task
def unittest_coverage(context):
"""Report on code test coverage as measured by 'invoke unittest'."""
command = "coverage report --skip-covered --include 'nautobot_data_validation_engine/*' --omit *migrations*"
run_command(context, command)
@task(
help={
"failfast": "fail as soon as a single test fails don't run the entire test suite",
}
)
def tests(context, failfast=False):
"""Run all tests for this plugin."""
# If we are not running locally, start the docker containers so we don't have to for each test
if not is_truthy(context.nautobot_data_validation_engine.local):
print("Starting Docker Containers...")
start(context)
# Sorted loosely from fastest to slowest
print("Running black...")
black(context)
print("Running flake8...")
flake8(context)
print("Running bandit...")
bandit(context)
print("Running pydocstyle...")
pydocstyle(context)
print("Running yamllint...")
yamllint(context)
print("Running pylint...")
pylint(context)
print("Running unit tests...")
unittest(context, failfast=failfast)
print("All tests have passed!")
unittest_coverage(context)