Full coverage for job.

parent e4666a5e
......@@ -3,6 +3,7 @@
import datetime
import celery
from django.contrib.auth.models import User
from django.db import models
from jsonfield import JSONField
......@@ -26,19 +27,20 @@ class Job(models.Model):
created_by = models.ForeignKey(User, blank=True, null=True)
@classmethod
def create(cls, task, name, request, *args, **kwargs):
def create(cls, task: celery.Task, name: str, request, *args, **kwargs) -> 'Job':
"""Create job object and publish task on celery queue."""
# create database object
job = cls(task=str(task))
if request:
job.created_by = request.user
job.name = name[:255]
job.status = 'created'
job.save()
# publish original task which stores the result in this Job object
result_id = (task | cls.store_result.s(job_id=job.id)).apply_async(*args, **kwargs)
job.name = name[:255]
job = Job.objects.get(id=job.id)
job.result_id = result_id.id
job.status = 'created'
job.save()
return job
......@@ -54,6 +56,7 @@ class Job(models.Model):
job.result = result
job.status = 'completed'
job.finished_on = datetime.datetime.now()
print(job.status)
job.save()
def __str__(self):
......
import pytest
from django.contrib.auth.models import User
from ..celery import app
from .models import Job
@pytest.fixture
def celery(settings):
"""Common setup for running (simulated) celery tasks inside tests."""
# don't let celery contact broker, execute tasks directly
settings.CELERY_ALWAYS_EAGER = True
@app.task
def dummy(result=True):
"""Dummy celery task for testing."""
if result:
return 'result'
def test_job(db, mock, celery):
"""Test creating task wrapped in a job."""
request = mock.Mock()
user = User(username='testuser')
user.save()
request.user = user
job = Job.create(dummy.s(), 'a-name', request)
assert job.result_id
assert job.status == 'completed'
assert job.result == 'result'
assert job.task == 'failmap_admin.app.tests.dummy()'
assert job.created_by == user
assert str(job) == job.result_id
def test_job_no_result(db, celery):
"""Task returning no result should not break stuff."""
job = Job.create(dummy.s(False), 'a-name', None)
assert job.result_id
assert job.status == 'completed'
assert job.result == '-- task generated no result object --'
assert job.task == 'failmap_admin.app.tests.dummy(False)'
......@@ -3,6 +3,7 @@ from datetime import datetime
import pytz
from django.contrib import admin
from django.urls import reverse
from jet.admin import CompactInline
from failmap_admin.map.determineratings import (OrganizationRating, UrlRating, rate_organization,
......
......@@ -13,6 +13,7 @@ django-coverage
django-extensions
django-debug-toolbar
django-debug-toolbar-request-history
pytest-mock
# # dashboard for celery introspection
# # run failmap-admin celery flower
......
......@@ -11,3 +11,6 @@ linters = isort,pycodestyle,pyflakes
[isort]
line_length=100
known_first_party=failmap_admin
[tool:pytest]
python_files = tests.py test_*.py *_tests.py
......@@ -23,6 +23,8 @@ commands =
coverage run --include 'failmap_admin/*' -m pytest {posargs}
# generate coverage
coverage report
# and pretty html
coverage html
# run code quality check
[testenv:check]
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment