#!/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 json
from datetime import datetime, timedelta
from django.contrib.auth.models import User
from django.conf import settings
from django.core.urlresolvers import reverse
from .models import Database
from ..dataformats.models import DataFormat
from ..common.testutils import BaseTestCase
from ..common.testutils import tearDownModule # noqa test runner will call it
TEST_PWD = "1234" # nosec
[docs]class DatabaseAPIBase(BaseTestCase):
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 setUp(self):
# Users
self.system_user = User.objects.create_user(
settings.SYSTEM_ACCOUNT, settings.SYSTEM_ACCOUNT + "@test.org", TEST_PWD
)
self.system_user.is_staff = True
self.system_user.save()
User.objects.create_user("jackdoe", "jackdoe@test.org", TEST_PWD)
User.objects.create_user("johndoe", "johndoe@test.org", TEST_PWD)
self.db_name = "test_db"
[docs] def tearDown(self):
pass
[docs]class DatabaseCreationAPI(DatabaseAPIBase):
[docs] def setUp(self):
super(DatabaseCreationAPI, self).setUp()
self.url = reverse("api_databases:all")
[docs] def test_no_access_for_anonymous_user(self):
response = self.client.post(self.url)
self.checkResponse(response, 403)
[docs] def test_no_access_for_non_admin_user(self):
self.client.login(username="jackdoe", password=TEST_PWD)
response = self.client.post(self.url)
self.checkResponse(response, 403)
[docs] def test_create_database_failure(self):
self.client.login(username=settings.SYSTEM_ACCOUNT, password=TEST_PWD)
response = self.client.post(
self.url,
json.dumps(
{"name": self.db_name, "version": 1, "declaration": self.DATABASE}
),
content_type="application/json",
)
self.checkResponse(response, 400)
[docs] def test_create_database(self):
# Create a dataformat
(dataformat, errors) = DataFormat.objects.create_dataformat(
self.system_user, "float", ""
)
self.assertIsNotNone(dataformat, errors)
dataformat.share()
self.client.login(username=settings.SYSTEM_ACCOUNT, password=TEST_PWD)
response = self.client.post(
self.url,
json.dumps(
{"name": self.db_name, "version": 1, "declaration": self.DATABASE}
),
content_type="application/json",
)
data = self.checkResponse(response, 201, content_type="application/json")
self.assertTrue(data["name"] == self.db_name)
databases = Database.objects.all()
self.assertEqual(databases.count(), 1)
databases.delete()
dataformat.delete()
[docs]class DatabaseRetrievalAPI(DatabaseAPIBase):
[docs] def setUp(self):
super(DatabaseRetrievalAPI, self).setUp()
(dataformat, errors) = DataFormat.objects.create_dataformat(
self.system_user, "float", ""
)
self.assertIsNotNone(dataformat, errors)
dataformat.share()
(self.database, errors) = Database.objects.create_database(
self.db_name, declaration=self.DATABASE
)
self.assertIsNotNone(self.database, errors)
[docs] def tearDown(self):
self.database.delete()
def __accessibility_test(self):
self.database.accessibility_start_date = datetime.now()
self.database.accessibility_end_date = (
self.database.accessibility_start_date + timedelta(days=1)
)
self.database.save()
object_url = reverse(
"api_databases:object", kwargs={"database_name": self.db_name, "version": 1}
)
all_url = reverse("api_databases:all")
response = self.client.get(object_url, format="json")
self.checkResponse(response, 200, content_type="application/json")
response = self.client.get(all_url, format="json")
self.checkResponse(response, 200, content_type="application/json")
self.assertEqual(len(response.json()), 1)
self.database.accessibility_start_date = datetime.now() - timedelta(days=2)
self.database.accessibility_end_date = (
self.database.accessibility_start_date + timedelta(days=1)
)
self.database.save()
response = self.client.get(object_url, format="json")
self.checkResponse(response, 404)
response = self.client.get(all_url, format="json")
self.checkResponse(response, 200, content_type="application/json")
self.assertEqual(len(response.json()), 0)
[docs] def test_retrieve_database(self):
self.database.share()
self.client.login(username=settings.SYSTEM_ACCOUNT, password=TEST_PWD)
url = reverse(
"api_databases:object", kwargs={"database_name": self.db_name, "version": 1}
)
response = self.client.get(url, format="json")
data = self.checkResponse(response, 200, content_type="application/json")
declaration = json.loads(data["declaration"])
self.assertTrue(declaration["root_folder"].startswith("/path_to_db_folder"))
[docs] def test_dated_database_for_user(self):
self.database.share(users=[settings.SYSTEM_ACCOUNT])
self.database.save()
self.client.login(username=settings.SYSTEM_ACCOUNT, password=TEST_PWD)
self.__accessibility_test()
[docs] def test_dated_database_for_public(self):
self.database.share()
self.database.save()
self.__accessibility_test()