#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
###############################################################################
# #
# Copyright (c) 2016 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 simplejson as json
import shutil
from datetime import datetime, timedelta
from django.contrib.auth.models import User
from django.conf import settings
from django.core.urlresolvers import reverse
from django.utils import six
from django.core.management import call_command
from ..experiments.models import Experiment
from ..algorithms.models import Algorithm
from ..common.models import Shareable
from ..backend.models import Environment, Queue
from ..dataformats.models import DataFormat
from ..toolchains.models import Toolchain
from ..databases.models import Database
from ..common.testutils import BaseTestCase
from ..common.testutils import tearDownModule # noqa test runner will call it
from .models import Attestation
TEST_PWD = "1234"
[docs]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"},
}
],
}
],
}
[docs] def login_johndoe(self):
self.client.login(username="johndoe", password=TEST_PWD)
[docs] def login_jackdoe(self):
self.client.login(username="jackdoe", password=TEST_PWD)
[docs] 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()
[docs] 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)
[docs]class AttestationCreationAPI(AttestationsAPIBase):
[docs] def setUp(self):
super(AttestationCreationAPI, self).setUp()
self.url = reverse("api_attestations:create")
[docs] def test_no_access_for_anonymous_user(self):
response = self.client.post(self.url)
self.checkResponse(response, 403)
[docs] def test_bad_request_with_invalid_content_type(self):
self.login_jackdoe()
response = self.client.post(
self.url,
{"experiment": "jackdoe/personal/1/experiment1"},
content_type="image/png",
)
self.checkResponse(response, 415, content_type="application/json")
[docs] def test_bad_request_with_unknown_experiment(self):
self.login_jackdoe()
response = self.client.post(
self.url,
json.dumps({"experiment": "jackdoe/personal/1/unknown"}),
content_type="application/json",
)
self.checkResponse(response, 404)
[docs] def test_create_attestation(self):
self.login_jackdoe()
response = self.client.post(
self.url,
json.dumps({"experiment": "jackdoe/personal/1/experiment1"}),
content_type="application/json",
)
data = self.checkResponse(response, 201, content_type="application/json")
self.assertTrue(isinstance(data["number"], int))
self.assertTrue(data["number"] > 0)
attestations = Attestation.objects.all()
self.assertEqual(attestations.count(), 1)
attestation = attestations[0]
self.assertEqual(attestation.number, data["number"])
self.assertEqual(attestation.experiment.name, "experiment1")
self.assertTrue(attestation.locked)
self.assertTrue(attestation.creation_date is not None)
self.assertTrue(attestation.publication_date is None)
algorithm = Algorithm.objects.get(name="algorithm1")
analyzer = Algorithm.objects.get(name="analyzer1")
referenced_algorithms = attestation.experiment.referenced_algorithms
self.assertEqual(referenced_algorithms.count(), 2)
self.assertTrue(algorithm in referenced_algorithms.all())
self.assertTrue(analyzer in referenced_algorithms.all())
referenced_dataformats = attestation.dataformats.all()
self.assertEqual(referenced_dataformats.count(), 1)
referenced_algorithms = attestation.algorithms.all()
self.assertEqual(referenced_algorithms.count(), 2)
self.assertTrue(algorithm in referenced_algorithms)
self.assertTrue(analyzer in referenced_algorithms)
[docs]class AttestationUnlockingAPI(AttestationsAPIBase):
[docs] def setUp(self):
super(AttestationUnlockingAPI, self).setUp()
self.login_jackdoe()
response = self.client.post(
reverse("api_attestations:create"),
json.dumps({"experiment": "jackdoe/personal/1/experiment1"}),
content_type="application/json",
)
data = self.checkResponse(response, 201, content_type="application/json")
self.number = data["number"]
self.url = reverse("api_attestations:unlock", args=[self.number])
self.client.logout()
[docs] def test_no_access_for_anonymous_user(self):
response = self.client.post(
self.url, json.dumps({}), content_type="application/json"
)
self.checkResponse(response, 403)
[docs] def test_bad_request_with_invalid_content_type(self):
self.login_jackdoe()
response = self.client.post(
self.url,
{"experiment": "jackdoe/personal/1/experiment1"},
content_type="image/png",
)
self.checkResponse(response, 415, content_type="application/json")
[docs] def test_bad_request_with_unknown_number(self):
self.login_jackdoe()
response = self.client.post(
reverse("api_attestations:unlock", args=[self.number - 1]),
json.dumps({}),
content_type="application/json",
)
self.checkResponse(response, 404)
[docs] def test_fail_to_unlock_attestation_of_other_user(self):
self.login_johndoe()
response = self.client.post(
self.url, json.dumps({}), content_type="application/json"
)
self.checkResponse(response, 403)
[docs] def test_unlock_attestation(self):
self.login_jackdoe()
response = self.client.post(
self.url,
json.dumps({"toolchain_shared_name": "shared"}),
content_type="application/json",
)
self.checkResponse(response, 204)
attestations = Attestation.objects.all()
self.assertEqual(attestations.count(), 1)
attestation = attestations[0]
self.assertEqual(attestation.number, self.number)
self.assertFalse(attestation.locked)
self.assertTrue(attestation.creation_date is not None)
self.assertTrue(attestation.publication_date is not None)
self.assertEqual(attestation.toolchain.sharing, Shareable.PUBLIC)
self.assertEqual(attestation.toolchain.shared_with.count(), 0)
self.assertEqual(attestation.experiment.sharing, Shareable.PUBLIC)
self.assertEqual(attestation.experiment.shared_with.count(), 0)
algorithm = Algorithm.objects.get(name="algorithm1")
self.assertEqual(algorithm.sharing, Shareable.PUBLIC)
algorithm = Algorithm.objects.get(name="analyzer1")
self.assertEqual(algorithm.sharing, Shareable.PUBLIC)
[docs] def test_unlock_attestation_with_visible_algorithm(self):
self.login_jackdoe()
response = self.client.post(
self.url,
json.dumps(
{
"visible_algorithms": ["jackdoe/algorithm1/1"],
"toolchain_shared_name": "shared",
}
),
content_type="application/json",
)
self.checkResponse(response, 204)
attestations = Attestation.objects.all()
self.assertEqual(attestations.count(), 1)
attestation = attestations[0]
self.assertEqual(attestation.number, self.number)
self.assertFalse(attestation.locked)
self.assertTrue(attestation.creation_date is not None)
self.assertTrue(attestation.publication_date is not None)
self.assertEqual(attestation.toolchain.sharing, Shareable.PUBLIC)
self.assertEqual(attestation.experiment.sharing, Shareable.PUBLIC)
self.assertEqual(attestation.experiment.shared_with.count(), 0)
algorithm = Algorithm.objects.get(name="algorithm1")
self.assertEqual(algorithm.sharing, Shareable.USABLE)
[docs]class AttestationDeletionAPI(AttestationsAPIBase):
[docs] def setUp(self):
super(AttestationDeletionAPI, self).setUp()
self.login_jackdoe()
response = self.client.post(
reverse("api_attestations:create"),
json.dumps({"experiment": "jackdoe/personal/1/experiment1"}),
content_type="application/json",
)
data = self.checkResponse(response, 201, content_type="application/json")
self.number = data["number"]
self.url = reverse("api_attestations:delete", args=[self.number])
self.client.logout()
[docs] def test_no_access_for_anonymous_user(self):
response = self.client.delete(
self.url, json.dumps({}), content_type="application/json"
)
self.checkResponse(response, 403)
[docs] def test_with_other_user(self):
self.login_johndoe()
response = self.client.delete(self.url)
self.checkResponse(response, 403, content_type="application/json")
[docs] def test_with_invalid_number(self):
self.login_jackdoe()
url = reverse("api_attestations:delete", args=[999999])
response = self.client.delete(url)
self.checkResponse(response, 404, content_type="application/json")
[docs] def test_unlocked_attestation(self):
attestation = Attestation.objects.get(number=self.number)
attestation.locked = False
attestation.save()
self.login_jackdoe()
response = self.client.delete(self.url)
self.checkResponse(response, 403, content_type="application/json")
attestation.locked = True
attestation.save()
[docs] def test_normal_case(self):
self.login_jackdoe()
response = self.client.delete(self.url)
self.checkResponse(response, 204)
[docs]class CleanAttstationManagementCommandTestCase(AttestationsAPIBase):
[docs] def run_command(self):
"Runs the clean_attestation command"
new_io = six.StringIO()
call_command("clean_attestations", interactive=False, stdout=new_io)
return new_io.getvalue().strip()
[docs] 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")
[docs] def test_valid_attestation(self):
experiment = Experiment.objects.all()[0]
Attestation.objects.create_attestation(experiment)
command_output = self.run_command()
self.assertEqual(command_output, "0 attestation(s) successfully cleaned")
[docs] def test_outdated_attestation(self):
experiment = Experiment.objects.all()[0]
attestation = Attestation.objects.create_attestation(experiment)
attestation.expiration_date = attestation.expiration_date - timedelta(days=200)
attestation.save()
command_output = self.run_command()
self.assertEqual(command_output, "1 attestation(s) successfully cleaned")
[docs] def test_unlocked_attestation(self):
experiment = Experiment.objects.all()[0]
attestation = Attestation.objects.create_attestation(experiment)
attestation.expiration_date = attestation.expiration_date - timedelta(days=100)
attestation.locked = False
attestation.save()
command_output = self.run_command()
self.assertEqual(command_output, "0 attestation(s) successfully cleaned")