Commit cfc59d11 authored by sebastien letort's avatar sebastien letort

API metrics tests.

parent ba3337fe
#!/usr/bin/env python3
## AIMS : provide function to populate the DB for tests.
## USAGE :
## AUTHORS : sletort@irisa.fr
from datetime import datetime
from django.db import connection
from main.models import Job, User, Webapp,DockerOs,JobQueue,AllgoUser
def with_users( l_usernames ):
# Note : bulk_create should be the most efficient way to create several users.
# but I cannot managed to build AllgoUser with this.
# ~ User.objects.bulk_create([
# ~ User( username='Bob', email='bob@localhost', password='' ),
# ~ User( username='Liz', email='Liz@localhost', password='' ),
# ~ User( username='Zaza', email='Zaza@localhost', password='' ),
# ~ User( username='Lol', email='lol@localhost', password='' ),
# ~ ])
# Note: It seems AllgoUsers are created without asking !
for name in l_usernames:
o_user = User.objects.create_user(
username = name,
email = name+'@localhost',
password = '')
# ~ print( name + " id=" + str( o_user.id ) )
# ~ l_ = AllgoUser.objects.all()
# ~ [ print(str(x)) for x in l_ ]
# with_users
def with_vitals():
"""Populate the DB with anything that is needed, but not specified."""
DockerOs.objects.create(name="debian:stable")
JobQueue.objects.create( name = "default", is_default = True, )
def with_webapps( d_apps ):
"""d_apps= { user: l_app_names }
Will use the first docker_os."""
o_docker_os = DockerOs.objects.first()
o_job_queue = JobQueue.objects.first()
for user in d_apps:
o_user = User.objects.get(username=user)
for app_name in d_apps[user]:
Webapp.objects.create(
name = app_name,
docker_os = o_docker_os,
job_queue = o_job_queue,
user = o_user,
)
# with_webapps
# =================================
# More complex / specific datasets
def with_various_jobs_from_one_app():
"""
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, ],
],
]
lo_users = User.objects.all()
for i in range(len(l_jobs)):
o_user = lo_users[i]
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,
])
# with_various_jobs_from_one_app
#!/usr/bin/env python3
## AIMS : test the metric API.
## USAGE :
## AUTHORS : sletort@irisa.fr
import logging
from unittest import skip
import json
import random
from datetime import datetime
from django.core.exceptions import PermissionDenied
from django.test import RequestFactory, TestCase, tag
from django.contrib.auth.models import User
from django.http import JsonResponse
from main.models import AllgoUser
from api.v1.views import Metrics
from . import populate_db
log = logging.getLogger('test')
# ======================================
class ApiMetricsTestCase(TestCase):
APP_NAME = 'Fake'
API_URL = '/api/v1/metrics'
@classmethod
def setUpTestData(cls):
log.info(":> setUpTestData.\nCreate 4 users, one own the app, and various jobs are created.")
populate_db.with_vitals()
populate_db.with_users([ 'Bob', 'Liz', 'Zaza', 'Lol' ])
populate_db.with_webapps({ 'Bob': [cls.APP_NAME] })
populate_db.with_various_jobs_from_one_app()
def setUp(self):
log.info(":> setUp.")
# Every test needs access to the request factory.
self.request = RequestFactory().get(self.API_URL)
self.req_factory = RequestFactory()
@classmethod
def _build_url(cls, d_params=None):
if d_params is None:
return cls.API_URL
l_args = []
for k,v in d_params.items():
l_args.append( "{}={}".format(k,v) )
api_url = cls.API_URL + "=?" + "&".join(l_args)
return api_url
def create_view(self, what):
SLEEP_ID = '1'
return Metrics.as_view()(self.request, SLEEP_ID, what)
def assert_json_struct(self, d_json):
l_keys = list(d_json.keys())
self.assertEqual(l_keys, [self.APP_NAME], "WebApp is the only key.")
d_apps = d_json[self.APP_NAME]
self.assertIn('from', d_apps)
self.assertIn('to', d_apps)
self.assertIn('data', d_apps)
def assert_record(self, d_record, d_format):
for k,type_ in d_format.items():
self.assertIn(k, d_record )
self.assertTrue(isinstance(d_record[k], type_))
def check_api(self, what, d_record_fmt):
url = self._build_url()
self.request = self.req_factory.get(url)
self.request.user = User.objects.get(username='Bob')
o_response = self.create_view(what)
self.assertIsInstance(o_response, JsonResponse)
str_response = o_response.content.decode('utf-8')
d_json = json.loads( str_response )
self.assert_json_struct(d_json)
# check one random record of data
d_record = random.choice( d_json[self.APP_NAME]['data'] )
self.assert_record( d_record, d_record_fmt )
# TODO: manage time_period as date !
# TODO : tester l'erreur sur faute de format de date dans url
@tag( 'auth' )
def test_access_no_user(self ):
url = self._build_url()
self.request = self.req_factory.get(url)
o_response = self.create_view('per_user')
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 401, "without user = not authorized.")
@tag( 'permission' )
def test_access_bad_user(self):
url = self._build_url()
self.request = self.req_factory.get(url)
self.request.user = User.objects.get(username='Lol')
o_response = self.create_view('per_user')
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 403, "user without app = permission denied.")
@tag( 'params' )
def test_access_bad_webapp(self):
url = self._build_url()
self.request = self.req_factory.get(url)
self.request.user = User.objects.get(username='Lol')
NON_EXISTING_APP_ID = 0
o_response = Metrics.as_view()(self.request, NON_EXISTING_APP_ID, 'per_user')
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 404, "Unknown app = not found.")
# test with assertJSONEqual ?
@tag( 'params' )
def test_access_bad_metrics(self):
o_user = User.objects.get(username='Bob')
self.request.user = o_user
o_response = self.create_view('peureuse')
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 404, "Unknown metrics type = not found.")
@tag( 'format' )
def test_per_user(self):
d_format = { 'n': int, 'uname': str, 'user': int, 'time_period': str }
self.check_api( 'per_user', d_format )
# I cannot manage to make "01-01-1 00:00" be validated with this.
# It seems that only 4 digits years are allowed.
# ~ with not self.assertRaises(ValueError):
# ~ datetime.strptime( d_apps['from'], DATE_FMT )
# ~ with not self.assertRaises(ValueError):
# ~ datetime.strptime( d_apps['to'], DATE_FMT )
@tag( 'format' )
def test_per_state(self):
d_format = { 'n': int, 'result': str, 'time_period': str }
self.check_api( 'per_state', d_format )
@tag( 'format' )
def test_created(self):
d_format = { 'n': int, 'time_period': str }
self.check_api( 'created', d_format )
def check_results(self, what, ld_expected):
url = self._build_url({ 'step': 'month', 'from': '2019-01-01', 'to':'2019-05-01' })
self.request = self.req_factory.get(url)
self.request.user = User.objects.get(username='Bob')
o_response = self.create_view(what)
str_response = o_response.content.decode('utf-8')
d_json = json.loads(str_response)
l_data = d_json[self.APP_NAME]['data']
N = len(ld_expected)
self.assertTrue(N==len(l_data), "number of elements")
for i in range(N):
self.assertIn(ld_expected[i], l_data)
@tag( 'results' )
def test_per_state_results(self):
ld_expected = [
{'time_period': '2019-01-01', 'result': 'SUCCESS', 'n': 2},
{'time_period': '2019-01-01', 'result': 'ERROR', 'n': 1},
{'time_period': '2019-02-01', 'result': 'SUCCESS', 'n': 2},
{'time_period': '2019-02-01', 'result': 'ERROR', 'n': 1},
{'time_period': '2019-02-01', 'result': 'ABORTED', 'n': 2},
{'time_period': '2019-03-01', 'result': 'SUCCESS', 'n': 3},
{'time_period': '2019-03-01', 'result': 'TIMEOUT', 'n': 1},
{'time_period': '2019-04-01', 'result': 'SUCCESS', 'n': 7},
{'time_period': '2019-04-01', 'result': 'ERROR', 'n': 2},
{'time_period': '2019-04-01', 'result': 'ABORTED', 'n': 2},
]
self.check_results('per_state', ld_expected)
@tag( 'result' )
def test_per_user_results(self):
ld_expected = [
{'uname': 'Bob', 'user': 1, 'n': 3, 'time_period': '2019-01-01'},
{'uname': 'Bob', 'user': 1, 'n': 1, 'time_period': '2019-02-01'},
{'uname': 'Liz', 'user': 2, 'n': 2, 'time_period': '2019-02-01'},
{'uname': 'Zaza', 'user': 3, 'n': 2, 'time_period': '2019-02-01'},
{'uname': 'Liz', 'user': 2, 'n': 1, 'time_period': '2019-03-01'},
{'uname': 'Zaza', 'user': 3, 'n': 3, 'time_period': '2019-03-01'},
{'uname': 'Bob', 'user': 1, 'n': 5, 'time_period': '2019-04-01'},
{'uname': 'Liz', 'user': 2, 'n': 1, 'time_period': '2019-04-01'},
{'uname': 'Zaza', 'user': 3, 'n': 3, 'time_period': '2019-04-01'},
{'uname': 'Lol', 'user': 4, 'n': 2, 'time_period': '2019-04-01'},
]
self.check_results('per_user', ld_expected)
@tag( 'results' )
def test_created_results(self):
ld_expected = [
{'time_period': '2019-01-01', 'n': 4},
{'time_period': '2019-02-01', 'n': 6},
{'time_period': '2019-03-01', 'n': 3},
{'time_period': '2019-04-01', 'n': 10},
]
self.check_results('created', ld_expected)
if __name__ == '__main__':
import sys
print( "ok" )
sys.exit( 0 )
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