Commit 69be81aa authored by Samuel GAIST's avatar Samuel GAIST Committed by Flavio TARSETTI
Browse files

[attestations][tests] Refactor tests and add email warning checks

parent 64bd2184
# vim: set fileencoding=utf-8 :
# encoding: utf-8
###############################################################################
# #
# Copyright (c) 2020 Idiap Research Institute, http://www.idiap.ch/ #
# Contact: beat.support@idiap.ch #
# #
# This file is part of the beat.web module of the BEAT platform. #
# #
# Commercial License Usage #
# Licensees holding valid commercial BEAT licenses may use this file in #
# accordance with the terms contained in a written agreement between you #
# and Idiap. For further information contact tto@idiap.ch #
# #
# Alternatively, this file may be used under the terms of the GNU Affero #
# Public License version 3 as published by the Free Software and appearing #
# in the file LICENSE.AGPL included in the packaging of this file. #
# The BEAT platform is distributed in the hope that it will be useful, but #
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY #
# or FITNESS FOR A PARTICULAR PURPOSE. #
# #
# You should have received a copy of the GNU Affero Public License along #
# with the BEAT platform. If not, see http://www.gnu.org/licenses/. #
# #
###############################################################################
import os
import shutil
from datetime import datetime
from django.conf import settings
from django.contrib.auth.models import User
from ...algorithms.models import Algorithm
from ...backend.models import Environment
from ...backend.models import Queue
from ...common.testutils import BaseTestCase
from ...common.testutils import tearDownModule # noqa test runner will call it
from ...databases.models import Database
from ...dataformats.models import DataFormat
from ...experiments.models import Experiment
from ...toolchains.models import Toolchain
TEST_PWD = "1234"
class AttestationsBaseTestCase(BaseTestCase):
ALGORITHM_TOOLCHAIN_DECLARATION = """{
"language": "python",
"schema_version": 1,
"splittable": false,
"groups": [
{
"inputs": {
"in": { "type": "%(user)s/float/1" }
},
"outputs": {
"out": { "type": "%(user)s/float/1" }
}
}
],
"parameters": {
}
}""" % {
"user": settings.SYSTEM_ACCOUNT
}
ALGORITHM_CODE = """class Algorithm:
def process(self, inputs, outputs):
return True
"""
ANALYZER_TOOLCHAIN_DECLARATION = """{
"language": "python",
"schema_version": 1,
"groups": [
{
"inputs": {
"in": { "type": "%(user)s/float/1" }
}
}
],
"results": {
"sum": {
"type": "%(user)s/float/1",
"display": false
}
},
"parameters": {
"score": { "type": "float32" }
}
}""" % {
"user": settings.SYSTEM_ACCOUNT
}
ANALYZER_CODE = """class Algorithm:
def process(self, inputs, output):
return True
"""
TOOLCHAIN_DECLARATION = {
"blocks": [
{
"name": "block1",
"inputs": ["in"],
"outputs": ["out"],
"synchronized_channel": "dataset1",
}
],
"datasets": [{"name": "dataset1", "outputs": ["out"]}],
"connections": [
{"from": "dataset1.out", "to": "block1.in", "channel": "dataset1"},
{"from": "block1.out", "to": "analyzer1.in", "channel": "dataset1"},
],
"analyzers": [
{"inputs": ["in"], "synchronized_channel": "dataset1", "name": "analyzer1"}
],
"representation": {"connections": {}, "blocks": {}, "channel_colors": {}},
}
EXPERIMENT_DECLARATION = {
"blocks": {
"block1": {
"algorithm": "jackdoe/algorithm1/1",
"parameters": {},
"inputs": {"in": "in"},
"outputs": {"out": "out"},
}
},
"analyzers": {
"analyzer1": {
"algorithm": "jackdoe/analyzer1/1",
"parameters": {},
"inputs": {"in": "in"},
}
},
"datasets": {
"dataset1": {
"database": "database1/1",
"protocol": "protocol1",
"set": "set1",
}
},
"globals": {
"environment": {"name": "env1", "version": "1.0"},
"queue": "queue1",
},
}
DATABASE = {
"root_folder": "/path/to/root/folder",
"protocols": [
{
"name": "protocol1",
"template": "test",
"sets": [
{
"name": "set1",
"template": "set",
"view": "dummy",
"outputs": {"out": settings.SYSTEM_ACCOUNT + "/float/1"},
}
],
}
],
}
def login_johndoe(self):
self.client.login(username="johndoe", password=TEST_PWD)
def login_jackdoe(self):
self.client.login(username="jackdoe", password=TEST_PWD)
def setUp(self):
if os.path.exists(settings.TOOLCHAINS_ROOT):
shutil.rmtree(settings.TOOLCHAINS_ROOT)
if os.path.exists(settings.EXPERIMENTS_ROOT):
shutil.rmtree(settings.EXPERIMENTS_ROOT)
if os.path.exists(settings.ALGORITHMS_ROOT):
shutil.rmtree(settings.ALGORITHMS_ROOT)
if os.path.exists(settings.CACHE_ROOT):
shutil.rmtree(settings.CACHE_ROOT)
# Users
system_user = User.objects.create_user(
settings.SYSTEM_ACCOUNT, settings.SYSTEM_ACCOUNT + "@test.org", TEST_PWD
)
user = User.objects.create_user("jackdoe", "jackdoe@test.org", TEST_PWD)
User.objects.create_user("johndoe", "johndoe@test.org", TEST_PWD)
# Create integer type for algorithm prototype loading
(integer, errors) = DataFormat.objects.create_dataformat(
author=system_user,
name="integer",
short_description="Default integer type need for algorithm prototype",
)
self.assertIsNotNone(integer, msg=errors)
integer.share()
# Create an environment and queue
environment = Environment(name="env1", version="1.0")
environment.save()
environment.share()
queue = Queue(
name="queue1",
memory_limit=1024,
time_limit=60,
cores_per_slot=1,
max_slots_per_user=10,
)
queue.save()
queue.environments.add(environment)
# Create a dataformat
(dataformat, errors) = DataFormat.objects.create_dataformat(
system_user, "float", ""
)
self.assertIsNotNone(dataformat, errors)
dataformat.share()
# Create a database
database, errors = Database.objects.create_database(
"database1", declaration=self.DATABASE
)
self.assertIsNotNone(database, errors)
# Create an algorithm
(algorithm1, errors) = Algorithm.objects.create_algorithm(
author=user,
name="algorithm1",
short_description="",
declaration=AttestationsBaseTestCase.ALGORITHM_TOOLCHAIN_DECLARATION,
code=AttestationsBaseTestCase.ALGORITHM_CODE,
)
self.assertIsNotNone(algorithm1, errors)
# Create an analyzer
(algorithm2, errors) = Algorithm.objects.create_algorithm(
author=user,
name="analyzer1",
short_description="",
declaration=AttestationsBaseTestCase.ANALYZER_TOOLCHAIN_DECLARATION,
code=AttestationsBaseTestCase.ANALYZER_CODE,
)
self.assertIsNotNone(algorithm2, errors)
# Create a toolchain
(toolchain, errors) = Toolchain.objects.create_toolchain(
author=user,
name="personal",
declaration=AttestationsBaseTestCase.TOOLCHAIN_DECLARATION,
)
self.assertIsNotNone(toolchain, errors)
# Create an experiment
(experiment, toolchain_instance, errors) = Experiment.objects.create_experiment(
author=user,
toolchain=toolchain,
name="experiment1",
declaration=AttestationsBaseTestCase.EXPERIMENT_DECLARATION,
)
self.assertIsNotNone(toolchain_instance, errors)
self.assertIsNotNone(experiment, errors)
experiment.start_date = datetime.now()
experiment.end_date = datetime.now()
experiment.save()
def tearDown(self):
if os.path.exists(settings.TOOLCHAINS_ROOT):
shutil.rmtree(settings.TOOLCHAINS_ROOT)
if os.path.exists(settings.EXPERIMENTS_ROOT):
shutil.rmtree(settings.EXPERIMENTS_ROOT)
if os.path.exists(settings.ALGORITHMS_ROOT):
shutil.rmtree(settings.ALGORITHMS_ROOT)
if os.path.exists(settings.CACHE_ROOT):
shutil.rmtree(settings.CACHE_ROOT)
......@@ -25,281 +25,20 @@
# #
###############################################################################
import io
import os
import shutil
from datetime import datetime
from datetime import timedelta
import simplejson as json
from django.conf import settings
from django.contrib.auth.models import User
from django.core.management import call_command
from django.urls import reverse
from ..algorithms.models import Algorithm
from ..backend.models import Environment
from ..backend.models import Queue
from ..common.models import Shareable
from ..common.testutils import BaseTestCase
from ..common.testutils import tearDownModule # noqa test runner will call it
from ..databases.models import Database
from ..dataformats.models import DataFormat
from ..experiments.models import Experiment
from ..toolchains.models import Toolchain
from .models import Attestation
from ...algorithms.models import Algorithm
from ...common.models import Shareable
from ...common.testutils import tearDownModule # noqa test runner will call it
from ..models import Attestation
from .core import AttestationsBaseTestCase
TEST_PWD = "1234"
class AttestationsAPIBase(BaseTestCase):
ALGORITHM_TOOLCHAIN_DECLARATION = """{
"language": "python",
"schema_version": 1,
"splittable": false,
"groups": [
{
"inputs": {
"in": { "type": "%(user)s/float/1" }
},
"outputs": {
"out": { "type": "%(user)s/float/1" }
}
}
],
"parameters": {
}
}""" % {
"user": settings.SYSTEM_ACCOUNT
}
ALGORITHM_CODE = """class Algorithm:
def process(self, inputs, outputs):
return True
"""
ANALYZER_TOOLCHAIN_DECLARATION = """{
"language": "python",
"schema_version": 1,
"groups": [
{
"inputs": {
"in": { "type": "%(user)s/float/1" }
}
}
],
"results": {
"sum": {
"type": "%(user)s/float/1",
"display": false
}
},
"parameters": {
"score": { "type": "float32" }
}
}""" % {
"user": settings.SYSTEM_ACCOUNT
}
ANALYZER_CODE = """class Algorithm:
def process(self, inputs, output):
return True
"""
TOOLCHAIN_DECLARATION = {
"blocks": [
{
"name": "block1",
"inputs": ["in"],
"outputs": ["out"],
"synchronized_channel": "dataset1",
}
],
"datasets": [{"name": "dataset1", "outputs": ["out"]}],
"connections": [
{"from": "dataset1.out", "to": "block1.in", "channel": "dataset1"},
{"from": "block1.out", "to": "analyzer1.in", "channel": "dataset1"},
],
"analyzers": [
{"inputs": ["in"], "synchronized_channel": "dataset1", "name": "analyzer1"}
],
"representation": {"connections": {}, "blocks": {}, "channel_colors": {}},
}
EXPERIMENT_DECLARATION = {
"blocks": {
"block1": {
"algorithm": "jackdoe/algorithm1/1",
"parameters": {},
"inputs": {"in": "in"},
"outputs": {"out": "out"},
}
},
"analyzers": {
"analyzer1": {
"algorithm": "jackdoe/analyzer1/1",
"parameters": {},
"inputs": {"in": "in"},
}
},
"datasets": {
"dataset1": {
"database": "database1/1",
"protocol": "protocol1",
"set": "set1",
}
},
"globals": {
"environment": {"name": "env1", "version": "1.0"},
"queue": "queue1",
},
}
DATABASE = {
"root_folder": "/path/to/root/folder",
"protocols": [
{
"name": "protocol1",
"template": "test",
"sets": [
{
"name": "set1",
"template": "set",
"view": "dummy",
"outputs": {"out": settings.SYSTEM_ACCOUNT + "/float/1"},
}
],
}
],
}
def login_johndoe(self):
self.client.login(username="johndoe", password=TEST_PWD)
def login_jackdoe(self):
self.client.login(username="jackdoe", password=TEST_PWD)
def setUp(self):
if os.path.exists(settings.TOOLCHAINS_ROOT):
shutil.rmtree(settings.TOOLCHAINS_ROOT)
if os.path.exists(settings.EXPERIMENTS_ROOT):
shutil.rmtree(settings.EXPERIMENTS_ROOT)
if os.path.exists(settings.ALGORITHMS_ROOT):
shutil.rmtree(settings.ALGORITHMS_ROOT)
if os.path.exists(settings.CACHE_ROOT):
shutil.rmtree(settings.CACHE_ROOT)
# Users
system_user = User.objects.create_user(
settings.SYSTEM_ACCOUNT, settings.SYSTEM_ACCOUNT + "@test.org", TEST_PWD
)
user = User.objects.create_user("jackdoe", "jackdoe@test.org", TEST_PWD)
User.objects.create_user("johndoe", "johndoe@test.org", TEST_PWD)
# Create integer type for algorithm prototype loading
(integer, errors) = DataFormat.objects.create_dataformat(
author=system_user,
name="integer",
short_description="Default integer type need for algorithm prototype",
)
self.assertIsNotNone(integer, msg=errors)
integer.share()
# Create an environment and queue
environment = Environment(name="env1", version="1.0")
environment.save()
environment.share()
queue = Queue(
name="queue1",
memory_limit=1024,
time_limit=60,
cores_per_slot=1,
max_slots_per_user=10,
)
queue.save()
queue.environments.add(environment)
# Create a dataformat
(dataformat, errors) = DataFormat.objects.create_dataformat(
system_user, "float", ""
)
self.assertIsNotNone(dataformat, errors)
dataformat.share()
# Create a database
database, errors = Database.objects.create_database(
"database1", declaration=self.DATABASE
)
self.assertIsNotNone(database, errors)
# Create an algorithm
(algorithm1, errors) = Algorithm.objects.create_algorithm(
author=user,
name="algorithm1",
short_description="",
declaration=AttestationsAPIBase.ALGORITHM_TOOLCHAIN_DECLARATION,
code=AttestationsAPIBase.ALGORITHM_CODE,
)
self.assertIsNotNone(algorithm1, errors)
# Create an analyzer
(algorithm2, errors) = Algorithm.objects.create_algorithm(
author=user,
name="analyzer1",
short_description="",
declaration=AttestationsAPIBase.ANALYZER_TOOLCHAIN_DECLARATION,
code=AttestationsAPIBase.ANALYZER_CODE,
)
self.assertIsNotNone(algorithm2, errors)
# Create a toolchain
(toolchain, errors) = Toolchain.objects.create_toolchain(
author=user,
name="personal",
declaration=AttestationsAPIBase.TOOLCHAIN_DECLARATION,
)
self.assertIsNotNone(toolchain, errors)
# Create an experiment
(experiment, toolchain_instance, errors) = Experiment.objects.create_experiment(
author=user,
toolchain=toolchain,
name="experiment1",
declaration=AttestationsAPIBase.EXPERIMENT_DECLARATION,
)
self.assertIsNotNone(toolchain_instance, errors)
self.assertIsNotNone(experiment, errors)
experiment.start_date = datetime.now()
experiment.end_date = datetime.now()
experiment.save()
def tearDown(self):
if os.path.exists(settings.TOOLCHAINS_ROOT):
shutil.rmtree(settings.TOOLCHAINS_ROOT)
if os.path.exists(settings.EXPERIMENTS_ROOT):
shutil.rmtree(settings.EXPERIMENTS_ROOT)
if os.path.exists(settings.ALGORITHMS_ROOT):
shutil.rmtree(settings.ALGORITHMS_ROOT)
if os.path.exists(settings.CACHE_ROOT):
shutil.rmtree(settings.CACHE_ROOT)
class AttestationCreationAPI(AttestationsAPIBase):
class AttestationCreationAPI(AttestationsBaseTestCase):
def setUp(self):
super(AttestationCreationAPI, self).setUp()
......@@ -370,7 +109,7 @@ class AttestationCreationAPI(AttestationsAPIBase):
self.assertTrue(analyzer in referenced_algorithms)
class AttestationUnlockingAPI(AttestationsAPIBase):
class AttestationUnlockingAPI(AttestationsBaseTestCase):
def setUp(self):
super(AttestationUnlockingAPI, self).setUp()
......@@ -493,7 +232,7 @@ class AttestationUnlockingAPI(AttestationsAPIBase):
self.assertEqual(algorithm.sharing, Shareable.USABLE)
class AttestationDeletionAPI(AttestationsAPIBase):
class AttestationDeletionAPI(AttestationsBaseTestCase):
def setUp(self):
super(AttestationDeletionAPI, self).setUp()
......@@ -551,39 +290,3 @@ class AttestationDeletionAPI(AttestationsAPIBase):
response = self.client.delete(self.url)
self.checkResponse(response, 204)
class CleanAttstationManagementCommandTestCase(AttestationsAPIBase):
def run_command(self):
"Runs the clean_attestation command"
new_io = io.StringIO()
call_command("clean_attestations", interactive=False, stdout=new_io)
return new_io.getvalue().strip()
def test_basic_usage(self):
"Check the operation of the clean_attestation management command"
command_output = self.run_command()
self.assertEqual(command_output, "0 attestation(s) successfully cleaned")