...
  View open merge request
Commits (19)
#!/usr/bin/env python3
# AIMS : test the job API.
# Note : Should I separate get request from post ?
import logging
from shutil import rmtree
from os import makedirs
from tempfile import mkdtemp
import json
from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.http import JsonResponse
from django.test import (RequestFactory,
TestCase,)
from main.models import AllgoUser, Job, User
from main.tests import populate_db
log = logging.getLogger('allgo')
# ======================================
class ApiJobsTestCase(TestCase):
"""Base class for the Api job tests.
It defines an appli, some user allowed to run the appli or not.
It populates the DB with those info.
It ensures that the DB is cleaned between to TestCase.
Last it defines common methods used in daughter classes.
"""
APP_NAME = 'Fake'
USERNAME = 'Bob'
NEMO = 'Nemo'
# daughter class should set those attributes
# self._request initialized in create_request()
# name of the route view to test, used with 'reverse' to generate the url.
_ROUTE_NAME = None
@classmethod
def setUpTestData(cls):
log.info(":> AJ.setUpTestData.")
settings.DATASTORE = mkdtemp(prefix="api_test_datastore")
populate_db.with_vitals()
usernames = [cls.USERNAME, cls.NEMO]
populate_db.with_users(usernames, fake_verified_email=True)
populate_db.with_webapps({cls.USERNAME: [cls.APP_NAME]})
@classmethod
def tearDownClass(cls):
log.info(":> AJ.tearDownClass.")
rmtree(settings.DATASTORE)
makedirs(settings.DATASTORE)
User.objects.all().delete()
Job.objects.all().delete()
def setUp(self):
log.info(":> AJ.setUp.")
# Every test needs access to the request factory.
self._factory = RequestFactory()
@property
def route_name(self):
return self._ROUTE_NAME
def create_request(self, user=USERNAME, **kwargs):
"""Create the _request attribute, an HttpRequest object."""
raise NotImplementedError
def call_view(self):
"""Call the view."""
raise NotImplementedError
def _add_authorization_header(self, username):
"""authenticate the user by its token.
We though define the user as AnonymousUser
because AllgoValidAccountMixin require a user to be set in the request."""
self._request.user = AnonymousUser()
user_token = AllgoUser.objects.get(user__username=username).token
self._request.META['HTTP_AUTHORIZATION'] = "Token token={}".format(user_token)
def _test_access_without_user(self):
"""ensure the view return 401 when user is not authenticated."""
self.create_request(user=None)
o_response = self.call_view()
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 401, "without user = not authorized.")
def _test_simple_request(self):
"""Test with all the good params."""
o_response = self.call_view() # default = all good.
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 200, "simple case.")
def _get_json_output_as_dictionnary(self):
"""Simply get the output of the created view as dictionnary.
the json response is turned into a dictionnary."""
# response is a string
response = (self.call_view()
.content.decode('utf-8'))
return json.loads(response) # return a dictionnary
# ApiJobsTestCase
#!/usr/bin/env python3
# AIMS : test the job API.
import logging
from unittest import expectedFailure
from django.core.exceptions import PermissionDenied
from django.core.urlresolvers import reverse
from django.http import JsonResponse
from django.test import tag
from api.v1.views import APIJobView, APIDownloadView
from main.helpers import upload_data
from main.models import Job, User
from main.tests import populate_db
from .ApiJobsTestCase import ApiJobsTestCase
log = logging.getLogger('allgo')
# ======================================
class GetApiJobsTestCase(ApiJobsTestCase):
"""Base class for DownloadView and ViewJob.
It sets up the data, defines a job id, the create_request method
and the test for access with a wrong user.
Note: This class cannot be instanciated.
"""
__JOB_ID = None # set in setUpTestData
@classmethod
def setUpTestData(cls):
super().setUpTestData()
# job_ids is a dictionnary
# key = username, value = list of job ids
job_ids = populate_db.with_various_jobs_from_one_app()
cls.__JOB_ID = job_ids[cls.USERNAME][0]
job = Job.objects.get(id=cls.__JOB_ID)
upload_data([], job) # make the job data dir
@property
def _job_id(self):
return self .__JOB_ID
def _build_url(self, id_job=None):
raise NotImplementedError
def create_request(self,
user=ApiJobsTestCase.USERNAME,
id_job=None):
"""Create the _request attribute, an HttpRequest object through the RequestFactory object.
If a user is defined, it is associated to the request."""
url = self._build_url(id_job)
self._request = self._factory.get(url)
if user:
self._add_authorization_header(user)
def _test_access_with_user_with_no_right_to_do(self):
"""Test the view when accessing with a user that has no right to do it."""
self.create_request(self.NEMO)
with self.assertRaises(PermissionDenied, msg="User cannot access the job."):
self.call_view()
# GetApiJobsTestCase
class ViewJob(GetApiJobsTestCase):
"""Test for the Api 'APIJobView' view.
It checks a job to get the url of files.
"""
_ROUTE_NAME = 'api:job'
def _build_url(self, id_job=None):
if None is id_job:
id_job = self._job_id
# with kwargs, the url is ok, but the view does not have the pk param.
url = reverse(self.route_name, kwargs={'pk': id_job})
log.info("VJ.url = %s", str(url))
return url
def call_view(self, id_job=None):
"""Call the view.
Note: as mention in the comment, we have to set the args in the view call
even if they are already in the url.
"""
if None is id_job:
id_job = self._job_id
# why using pk here *and* in create_request ? :
# https://stackoverflow.com/questions/48580465/django-requestfactory-loses-url-kwargs
if not hasattr(self, '_request'):
self.create_request()
return APIJobView.as_view()(self._request, pk=id_job)
@expectedFailure # "Mixin pb cf BUG#304"
@tag('auth')
def test_access_without_user(self):
"""no user token."""
super()._test_access_without_user()
@tag('permission')
def test_access_with_user_with_no_right_to_do(self):
"""user with token has no right to view the job."""
super()._test_access_with_user_with_no_right_to_do()
# not validated user ? <- ImO in mixin test.
@expectedFailure # "Mixin pb cf BUG#305"
@tag('params')
def test_access_with_bad_param(self):
"""job_id is unknown."""
bad_id = 0
self.create_request(id_job=bad_id)
o_response = self.call_view(id_job=bad_id)
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 404, "job does not exist = not found.")
@tag('ok')
def test_simple_request(self):
"""Test with all the good params."""
super()._test_simple_request()
@tag('format')
def test_out_format(self):
"""Test the output format."""
json_dict = self._get_json_output_as_dictionnary()
self.assertTrue("status" in json_dict, "status in json output.")
# we expecte 2 keys, status and job id
json_dict.pop("status")
self.assertEqual(len(json_dict.keys()), 1, "only status and job_id as keys.")
j_id, files = json_dict.popitem()
self.assertEqual(int(j_id), self._job_id)
self.assertIsInstance(files, dict, "value associated with job id is a dict.")
@tag('state')
@expectedFailure # reason="#347- deleted job badly managed"
def test_deleted_job(self):
job = populate_db.create_single_job(result=Job.SUCCESS, state=Job.DELETED)
self.create_request(id_job=job.id)
o_response = self.call_view(id_job=job.id)
self.assertIsInstance(o_response, JsonResponse)
# ViewJob
class DownloadView(GetApiJobsTestCase):
"""Test for the Api 'APIDownloadView' view.
It checks the redirection of .
"""
_ROUTE_NAME = 'api:download'
__FILENAME = 'truc'
def _build_url(self, id_job=None):
if None is id_job:
id_job = self._job_id
args = [id_job, self.__FILENAME]
url = reverse(self.route_name, args=args)
log.info("DV.url = %s", str(url))
return url
def call_view(self, id_job=None, filename=None):
"""Call the view.
Note: as mention in the comment, we have to set the args in the view call
even if they are already in the url.
"""
if None is id_job:
id_job = self._job_id
if None is filename:
filename = self.__FILENAME
# why using pk here *and* in create_request ? :
# https://stackoverflow.com/questions/48580465/django-requestfactory-loses-url-kwargs
if not hasattr(self, '_request'):
self.create_request()
return APIDownloadView.as_view()(self._request, filename, pk=id_job)
@expectedFailure # "Mixin pb cf BUG#304"
@tag('auth')
def test_access_without_user(self):
"""no user logged."""
super()._test_access_without_user()
@tag('permission')
def test_access_with_user_with_no_right_to_do(self):
"""user logged has no right to view the job."""
super()._test_access_with_user_with_no_right_to_do()
@tag('params')
def test_access_with_bad_param(self):
"""job_id is unknown."""
bad_id = 0
self.create_request(id_job=bad_id)
with self.assertRaises(PermissionDenied, msg="User cannot access the job."):
self.call_view(id_job=bad_id)
# Not sure. I do not do this test
# because I do not check the redirection in the good condition.
# Here I just
# ~ @tag( 'params2' )
# ~ def test_access_bad_params(self):
# ~ fake_name = "Nada.rien"
# ~ o_response = self.create_view(filename=fake_name)
@expectedFailure # "View error. cf BUG#306"
@tag('ok')
def test_simple_request(self):
"""Test with all the good params."""
o_response = self.create_view() # default = all good.
# is it the correct way to get this expected url ?
# using reverse() has no sense ImO, as create_view already use it.
exp_url = "/datastore/%s/%s" % (self.__JOB_ID, self.__FILENAME)
self.assertRedirects(o_response, exp_url, fetch_redirect_response=False)
# DownloadView
#!/usr/bin/env python3
# AIMS : test the job API.
# USAGE : python3 manage.py test allgo.api.tests.test_ApiJob_submit
import logging
import tempfile
from unittest import expectedFailure
from django.core.files.uploadedfile import File
from django.core.urlresolvers import reverse
from django.http import JsonResponse
from django.test import tag
from api.v1.views import jobs
from main.models import User
from .ApiJobsTestCase import ApiJobsTestCase
log = logging.getLogger('allgo')
# ======================================
class SubmitJob(ApiJobsTestCase):
"""Test for the Api 'jobs' view.
It creates a job for a webapp given as POST arg (id or name).
"""
_ROUTE_NAME = 'api:jobs'
__PARAMS = {
'job[webapp_id]': ApiJobsTestCase.APP_NAME,
'job[param]' : '',
}
o_file = tempfile.NamedTemporaryFile()
__FILES = [o_file.name]
def create_request(self, user=ApiJobsTestCase.USERNAME,
params=None, files=None):
"""Create the _request attribute, an HttpRequest object through the RequestFactory object.
If a user is defined, it is associated to the request,
all files of files are uploaded."""
if params is None:
params = self.__PARAMS
if files is None:
files = self.__FILES
url = reverse(self.route_name)
self._request = self._factory.post(
url,
HTTP_USER_AGENT='Test',
data=params, # params is a dictionnary
# cannot set it, cf https://docs.djangoproject.com/en/2.1/topics/testing/tools/#django.test.Client.post
# probably due to the serialization made, but I can't get it work
# ~ content_type='application/json',
# ~ content_type = "multipart/form-data;boundary=----~~~~****----",
)
for filename in files:
# ~ with open(filename, "rb") as f: # does not work as it closes the file.
f = open(filename, "rb")
self._request.FILES.update({filename: File(f)})
# ... opened files are not closed, isn't it bad ?
if user:
self._add_authorization_header(user)
def call_view(self):
"""Call the view."""
if not hasattr(self, '_request'):
self.create_request()
return jobs(self._request)
@expectedFailure # Mixin pb cf BUG#304"
@tag('auth')
def test_access_without_user(self):
"""no user logged."""
super()._test_access_without_user()
@expectedFailure # "Private data can be submitted, BUG!303."
@tag('permission')
def test_access_with_user_with_no_right_to_do(self):
"""user logged has no right to launch a job."""
# log.info("-bad_params-" + str(self.D_PARAMS))
self.create_request(self.NEMO)
o_response = self.create_view()
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 404, "user without right = ?.")
@expectedFailure # "ApiJob do not manage this case. cf BUG#302"
@tag('no_app')
def test_access_no_app(self):
"""No job[webapp_id] is provided in post params."""
params = self.__PARAMS.copy() # params is a dictionnary
params.pop('job[webapp_id]')
self.create_request(params=params)
o_response = self.create_view()
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 404, "no app provided = Not found.")
# different test for no params and no FILES ?
@tag('params')
def test_access_with_bad_param(self):
"""No job[param] is provided in post params."""
log.info("-bad_params- %s", str(self.__PARAMS))
params = self.__PARAMS.copy() # params is a dictionnary
params.pop('job[param]')
self.create_request(params=params)
o_response = self.call_view()
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 200, "no param provided = managed.")
@tag('ok')
def test_simple_request(self):
"""Test with all the good params."""
super()._test_simple_request()
@tag('format')
def test_out_format(self):
"""Test the output format."""
json_dict = self._get_json_output_as_dictionnary()
expected_keys = set(["avg_time", "id", "url"])
self.assertEqual(json_dict.keys(), expected_keys, "expected keys")
self.assertIsInstance(json_dict['id'], int, "id is integer")
no_domain_url = reverse('api:job', kwargs={'pk':json_dict['id']})
expected_url = 'http://testserver' + no_domain_url
self.assertTrue(expected_url == json_dict['url'], "expected url")
try:
float(json_dict['avg_time'])
except ValueError:
self.fail("Avg_time is not float.")
# SubmitJob
......@@ -17,6 +17,12 @@ log = logging.getLogger('allgo')
DATASTORE = config.env.ALLGO_DATASTORE
BUF_SIZE = 65536
# Note about authentication
# We do not allow user authentication by cookie for the api
# because it can be accessed from anywhere (from other domains).
# User can only be authenticated by a token.
# cf main.helpers.get_request_user
# JobAuthMixin ensure the call to get_request_user
def get_link(jobid, dir, filename, request):
return '/'.join((get_base_url(request), "datastore", str(jobid), filename))
......
import logging
from django.conf import settings
from django.contrib.auth.mixins import UserPassesTestMixin, LoginRequiredMixin
from django.core.exceptions import PermissionDenied
......@@ -8,7 +10,10 @@ from .models import Job
from .helpers import get_request_user
LOG = logging.getLogger('allgo')
# ======================================
# FIXME: should we validate API calls with this mixin too ? The answer is not
# obvious because it would not be too good to break the API when the ToS are
# updated (at least there should be a grace period)
......@@ -94,6 +99,7 @@ class JobAuthMixin(AllgoValidAccountMixin, UserPassesTestMixin):
- allow access if user is the job owner or if user is a superuser
"""
user = get_request_user(self.request)
if user is None:
return False
self.raise_exception = True # to return a 403
......
#!/usr/bin/env python3
## AIMS : provide function to populate the DB for tests.
from datetime import datetime
from django.db import connection
from main.models import Job, \
User,AllgoUser, \
Webapp,WebappVersion, \
DockerOs,JobQueue, \
EmailAddress
def with_users( l_usernames, fake_verified_email=False ):
# Note : bulk_create should be the most efficient way to create several users.
# but AllgoUsers are created by catching post_save signals which is not emit in bulk_create.
for name in l_usernames:
user = User.objects.create_user(
username = name,
email = name+'@localhost',
password = '')
EmailAddress.objects.create(user=user, email=name, verified=True)
# with_users
def with_vitals():
"""Populate the DB with anything that is needed."""
DockerOs.objects.create(name="debian:stable")
JobQueue.objects.create( name = "default", is_default = True, )
def with_webapps( users_apps ):
"""users_apps= { user: l_app_names }
Will use the first docker_os."""
docker_os = DockerOs.objects.first()
job_queue = JobQueue.objects.first()
users_ids = {}
for username in users_apps:
user = User.objects.get(username=username)
ids = []
for app_name in users_apps[username]:
app = Webapp.objects.create(
name = app_name,
docker_os = docker_os,
job_queue = job_queue,
user = user,
)
WebappVersion.objects.create(
webapp = app,
number = 1,
state = WebappVersion.READY,
published = True,
)
ids.append(app.id)
users_ids[user] = ids
return users_ids # users_ids is a dictionnary
# with_webapps
# =================================
# More complex / specific datasets
def with_various_jobs_from_one_app():
"""create various jobs with given results.
All they state is NEW.
throw exception if you have less than 4 users in your DB."""
o_app = Webapp.objects.first()
o_job_queue = JobQueue.objects.first()
l_jobs = [
[ # all jobs for user1
[ "05/01/2019", "17/01/2019", Job.SUCCESS, ],
[ "07/01/2019", "10/01/2019", Job.SUCCESS, ],
[ "07/01/2019", "25/02/2019", Job.SUCCESS, ],
[ "10/01/2019", "15/01/2019", Job.ERROR, ],
[ "01/04/2019", "04/04/2019", Job.ERROR, ],
[ "01/04/2019", "10/04/2019", Job.ABORTED, ],
[ "02/04/2019", "14/04/2019", Job.SUCCESS, ],
[ "03/04/2019", "20/04/2019", Job.SUCCESS, ],
[ "10/04/2019", "27/04/2019", Job.SUCCESS, ],
],
[ # all jobs for user2
[ "03/02/2019", "10/02/2019", Job.SUCCESS, ],
[ "04/02/2019", "07/02/2019", Job.ABORTED, ],
[ "07/02/2019", "03/04/2019", Job.SUCCESS, ],
[ "20/02/2019", "12/03/2019", Job.TIMEOUT, ],
],
[ # all jobs for user3
[ "04/02/2019", "07/02/2019", Job.ABORTED, ],
[ "04/02/2019", "10/02/2019", Job.ERROR, ],
[ "02/03/2019", "17/03/2019", Job.SUCCESS, ],
[ "03/03/2019", "05/03/2019", Job.SUCCESS, ],
[ "05/03/2019", "06/03/2019", Job.SUCCESS, ],
[ "02/04/2019", "10/04/2019", Job.SUCCESS, ],
[ "04/04/2019", "12/04/2019", Job.SUCCESS, ],
[ "10/04/2019", "14/04/2019", Job.SUCCESS, ],
],
[ # all jobs for user4
[ "04/04/2019", "12/04/2019", Job.ERROR, ],
[ "10/04/2019", "20/04/2019", Job.ABORTED, ],
],
]
d_jobs = {}
lo_users = User.objects.all()
n = min(len(lo_users), len(l_jobs))
for i in range(n):
o_user = lo_users[i]
l_id = []
for start,end,result in l_jobs[i]:
o_job = Job.objects.create(
queue = o_job_queue,
webapp = o_app,
user = o_user,
result = result,
)
# Job.objects.create automatically set created_at & updated_at at the current time
# no edition is possible.
# ~ o_job.created_at = datetime.strptime(start, "%d/%m/%Y")
# ~ o_job.updated_at = datetime.strptime(end, "%d/%m/%Y")
with connection.cursor() as cursor:
cursor.execute(
"UPDATE dj_jobs SET created_at=%s, updated_at=%s \
WHERE id = %s",
[
datetime.strptime(start, "%d/%m/%Y"),
datetime.strptime(end, "%d/%m/%Y"),
o_job.id,
])
l_id.append(o_job.id)
d_jobs[o_user.username] = l_id
return d_jobs
# with_various_jobs_from_one_app
def create_single_job(result, state):
"""Create a job, don't specify app/queue/user/start/end."""
app = Webapp.objects.first()
job_queue = JobQueue.objects.first()
user = User.objects.first()
job = Job.objects.create(
queue = job_queue,
webapp = app,
user = user,
result = result,
state = state,
)
return job