-
Jorrit Schaap authoredJorrit Schaap authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
t_tmssapp_scheduling_django.py 38.90 KiB
#!/usr/bin/env python3
# Copyright (C) 2018 ASTRON (Netherlands Institute for Radio Astronomy)
# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
#
# This file is part of the LOFAR software suite.
# The LOFAR software suite is free software: you can redistribute it and/or
# modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# The LOFAR software suite 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. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
# $Id: $
import unittest
from datetime import datetime
# use this to create timezone-aware datetime objects: from django.utils import timezone
# todo: Tags? -> Decide how to deal with them first.
# todo: Immutability of Blueprints on db level?
# use setup/teardown magic for tmss test database
from lofar.sas.tmss.test.tmss_database_unittest_setup import *
# now it's safe to import django modules
from lofar.sas.tmss.tmss.tmssapp import models
from django.db.utils import IntegrityError
from lofar.sas.tmss.test.t_tmssapp_specification_django import TaskBlueprint_test_data, TaskRelationBlueprint_test_data
# TODO: rest API testing should be moved out of this test module.
# import rest_framework.test
# client = rest_framework.test.APIClient()
# from lofar.sas.tmss.test.test_utils import assertDataWithUrls, assertUrlList
def SubtaskTemplate_test_data():
return {"type": models.SubtaskType.objects.get(value='copy'),
"name": "observation",
"description": 'My one observation',
"version": 'v0.314159265359',
"schema": {"mykey": "my value"},
"realtime": True,
"queue": False,
"tags": ["TMSS", "TESTING"]}
class SubtaskTemplateTest(unittest.TestCase):
def test_SubtaskTemplate_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.SubtaskTemplate.objects.create(**SubtaskTemplate_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_SubtaskTemplate_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.SubtaskTemplate.objects.create(**SubtaskTemplate_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_SubtaskTemplate_list_view_shows_entry(self):
#
# # setup
# entry = models.SubtaskTemplate.objects.create(**self.get_test_data())
#
# # assert
# response = client.get('/subtask_template/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], self.get_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_SubtaskTemplate_view_returns_correct_entry(self):
#
# # setup
# id1 = models.SubtaskTemplate.objects.create(**self.get_test_data()).id
# id2 = models.SubtaskTemplate.objects.create(**self.get_test_data()).id
#
# # assert
# response1 = client.get('/subtask_template/%s/' % id1, format='json', follow=True)
# response2 = client.get('/subtask_template/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, self.get_test_data())
# assertDataWithUrls(self, response2.data, self.get_test_data())
def DataproductSpecificationsTemplate_test_data():
return {"name": "data",
"description": 'My one date',
"version": 'v0.314159265359',
"schema": {"mykey": "my value"},
"tags": ["TMSS", "TESTING"]}
class DataproductSpecificationsTemplateTest(unittest.TestCase):
def test_DataproductSpecificationsTemplate_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.DataproductSpecificationsTemplate.objects.create(**DataproductSpecificationsTemplate_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_DataproductSpecificationsTemplate_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.DataproductSpecificationsTemplate.objects.create(**DataproductSpecificationsTemplate_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_DataproductSpecificationsTemplate_list_view_shows_entry(self):
#
# # setup
# entry = models.DataproductSpecificationsTemplate.objects.create(**self.get_test_data())
#
# # assert
# response = client.get('/dataproduct_specifications_template/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# for item in self.get_test_data().items():
# self.assertIn(item, response.data['results'][0].items())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_DataproductSpecificationsTemplate_view_returns_correct_entry(self):
#
# # setup
# id1 = models.DataproductSpecificationsTemplate.objects.create(**self.get_test_data()).id
# id2 = models.DataproductSpecificationsTemplate.objects.create(**self.get_test_data()).id
#
# # assert
# response1 = client.get('/dataproduct_specifications_template/%s/' % id1, format='json', follow=True)
# response2 = client.get('/dataproduct_specifications_template/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# for item in self.get_test_data().items():
# self.assertIn(item, response1.data.items())
# for item in self.get_test_data().items():
# self.assertIn(item, response2.data.items())
def SubtaskInputSelectionTemplate_test_data():
return {"name": "data",
"description": 'My one date',
"version": 'v0.314159265359',
"schema": {"mykey": "my value"},
"tags": ["TMSS", "TESTING"]}
class SubtaskInputSelectionTemplateTest(unittest.TestCase):
# This currently adds nothing on top of the template base class, so nothing new to test here.
pass
def DataproductFeedbackTemplate_test_data():
return {"name": "data",
"description": 'My one date',
"version": 'v0.314159265359',
"schema": {"mykey": "my value"},
"tags": ["TMSS", "TESTING"]}
class DataproductFeedbackTemplateTest(unittest.TestCase):
# This currently adds nothing on top of the template base class, so nothing new to test here.
pass
def SubtaskOutput_test_data():
return {"subtask": models.Subtask.objects.create(**Subtask_test_data()),
"connector": models.SubtaskConnector.objects.create(**SubtaskConnector_test_data()),
"tags":[]}
class SubtaskOutputTest(unittest.TestCase):
def test_SubtaskOutput_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.SubtaskOutput.objects.create(**SubtaskOutput_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_SubtaskOutput_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.SubtaskOutput.objects.create(**SubtaskOutput_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_SubtaskOutput_list_view_shows_entry(self):
# # setup
# models.SubtaskOutput.objects.create(**SubtaskOutput_test_data())
#
# # assert
# response = client.get('/subtask_output/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], SubtaskOutput_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_SubtaskOutput_view_returns_correct_entry(self):
# # setup
# id1 = models.SubtaskOutput.objects.create(**SubtaskOutput_test_data()).id
# id2 = models.SubtaskOutput.objects.create(**SubtaskOutput_test_data()).id
#
# # assert
# response1 = client.get('/subtask_output/%s/' % id1, format='json', follow=True)
# response2 = client.get('/subtask_output/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, SubtaskOutput_test_data())
# assertDataWithUrls(self, response2.data, SubtaskOutput_test_data())
def test_SubtaskOutput_prevents_missing_subtask(self):
# setup
test_data = dict(SubtaskOutput_test_data())
test_data['subtask'] = None
# assert
with self.assertRaises(IntegrityError):
models.SubtaskOutput.objects.create(**test_data)
def SubtaskInput_test_data():
# test data
return {"subtask": models.Subtask.objects.create(**Subtask_test_data()),
"task_relation_blueprint": models.TaskRelationBlueprint.objects.create(**TaskRelationBlueprint_test_data()),
"connector": models.SubtaskConnector.objects.create(**SubtaskConnector_test_data()),
"producer": models.SubtaskOutput.objects.create(**SubtaskOutput_test_data()),
#"dataproducts": models.Dataproduct.objects.create(**dpt.get_test_data()),
"selection_doc": "{}",
"selection_template": models.SubtaskInputSelectionTemplate.objects.create(**SubtaskInputSelectionTemplate_test_data()),
"tags":[]}
class SubtaskInputTest(unittest.TestCase):
def test_SubtaskInput_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.SubtaskInput.objects.create(**SubtaskInput_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_SubtaskInput_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.SubtaskInput.objects.create(**SubtaskInput_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_SubtaskInput_list_view_shows_entry(self):
# # setup
# models.SubtaskInput.objects.create(**SubtaskInput_test_data())
#
# # assert
# response = client.get('/subtask_input/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], SubtaskInput_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_SubtaskInput_view_returns_correct_entry(self):
# # setup
# id1 = models.SubtaskInput.objects.create(**SubtaskInput_test_data()).id
# id2 = models.SubtaskInput.objects.create(**SubtaskInput_test_data()).id
#
# # assert
# response1 = client.get('/subtask_input/%s/' % id1, format='json', follow=True)
# response2 = client.get('/subtask_input/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, SubtaskInput_test_data())
# assertDataWithUrls(self, response2.data, SubtaskInput_test_data())
def test_SubtaskInput_prevents_missing_subtask(self):
# setup
test_data = dict(SubtaskInput_test_data())
test_data['subtask'] = None
# assert
with self.assertRaises(IntegrityError):
models.SubtaskInput.objects.create(**test_data)
# TODO: rest API testing should be moved out of this test module.
# def test_SubtaskInput_allows_setting_dataproducts(self):
# # Other then through the API view, we cannot assign ManyToMany on creation, but have to set it later
# si = models.SubtaskInput.objects.create(**SubtaskInput_test_data())
# si.dataproducts.set([models.Dataproduct.objects.create(**Dataproduct_test_data()),
# models.Dataproduct.objects.create(**Dataproduct_test_data())])
# si.save()
#
# # assert
# response = client.get('/subtask_input/%s' % si.id, format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data, SubtaskInput_test_data())
def Subtask_test_data():
return { "start_time": datetime.utcnow().isoformat(),
"stop_time": datetime.utcnow().isoformat(),
"state": models.SubtaskState.objects.get(value='scheduling'),
"specifications_doc": "{}",
"task_blueprint": models.TaskBlueprint.objects.create(**TaskBlueprint_test_data()),
"specifications_template": models.SubtaskTemplate.objects.create(**SubtaskTemplate_test_data()),
"tags": ["TMSS", "TESTING"],
"do_cancel": datetime.utcnow().isoformat(),
"priority": 1,
"schedule_method": models.ScheduleMethod.objects.get(value='manual'),
"cluster": models.Cluster.objects.create(location="downstairs", tags=[]),
"scheduler_input_doc": "{}"}
class SubtaskTest(unittest.TestCase):
def test_Subtask_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.Subtask.objects.create(**Subtask_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_Subtask_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.Subtask.objects.create(**Subtask_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_Subtask_list_view_shows_entry(self):
#
# # setup
# entry = models.Subtask.objects.create(**Subtask_test_data())
#
# # assert
# response = client.get('/subtask/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], Subtask_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_Subtask_view_returns_correct_entry(self):
#
# # setup
# id1 = models.Subtask.objects.create(**Subtask_test_data()).id
# id2 = models.Subtask.objects.create(**Subtask_test_data()).id
#
# # assert
# response1 = client.get('/subtask/%s/' % id1, format='json', follow=True)
# response2 = client.get('/subtask/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, Subtask_test_data())
# assertDataWithUrls(self, response2.data, Subtask_test_data())
def test_Subtask_prevents_missing_template(self):
# setup
test_data = dict(Subtask_test_data())
test_data['specifications_template'] = None
# assert
with self.assertRaises(IntegrityError):
models.Subtask.objects.create(**test_data)
# TODO: rest API testing should be moved out of this test module.
# def test_nested_Subtask_are_filtered_according_to_TaskBlueprint(self):
# tbt = TaskBlueprintTest()
# tbt.setUp(populate=False)
#
# # setup
# task_blueprint_1 = models.TaskBlueprint.objects.create(**tbt.get_test_data())
# task_blueprint_2 = models.TaskBlueprint.objects.create(**tbt.get_test_data())
# test_data_1 = dict(Subtask_test_data())
# test_data_1['task_blueprint'] = task_blueprint_1
# subtask_1 = models.Subtask.objects.create(**test_data_1)
# test_data_2 = dict(Subtask_test_data())
# test_data_2['task_blueprint'] = task_blueprint_2
# subtask_2 = models.Subtask.objects.create(**test_data_2)
#
# # assert the returned list contains related items
# response = client.get('/task_blueprint/%s/subtask/' % task_blueprint_2.id, format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# self.assertEqual(len(response.data['results']), 1)
# assertDataWithUrls(self, response.data['results'][0], test_data_2)
#
# # assert an existing related item is returned
# response = client.get(
# '/task_blueprint/%s/subtask/%s/' % (task_blueprint_2.id, subtask_2.id), format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data, test_data_2)
#
# # assert an existing unrelated item is not returned
# response = client.get( '/task_blueprint/%s/subtask/%s/' % (task_blueprint_2.id, subtask_1.id), format='json', follow=True)
# self.assertEqual(response.status_code, 404)
def Dataproduct_test_data():
return {"filename": "my.file",
"directory": "/home/boskabouter/",
"dataformat": models.Dataformat.objects.get(value='Beamformed'),
"deleted_since": datetime.utcnow().isoformat(),
"pinned_since": datetime.utcnow().isoformat(),
"specifications_doc": "{}",
"specifications_template": models.DataproductSpecificationsTemplate.objects.create(**DataproductSpecificationsTemplate_test_data()),
"tags": ["TMSS", "TESTING"],
"producer": models.SubtaskOutput.objects.create(**SubtaskOutput_test_data()),
"do_cancel": datetime.utcnow().isoformat(),
"expected_size": 1234,
"size": 123,
"feedback_doc": "{}",
"feedback_template": models.DataproductFeedbackTemplate.objects.create(**DataproductFeedbackTemplate_test_data())}
class DataproductTest(unittest.TestCase):
def test_Dataproduct_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.Dataproduct.objects.create(**Dataproduct_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_Dataproduct_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.Dataproduct.objects.create(**Dataproduct_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_Dataproduct_list_view_shows_entry(self):
#
# # setup
# entry = models.Dataproduct.objects.create(**Dataproduct_test_data())
#
# # assert
# response = client.get('/dataproduct/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], Dataproduct_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_Dataproduct_view_returns_correct_entry(self):
#
# # setup
# id1 = models.Dataproduct.objects.create(**Dataproduct_test_data()).id
# id2 = models.Dataproduct.objects.create(**Dataproduct_test_data()).id
#
# # assert
# response1 = client.get('/dataproduct/%s/' % id1, format='json', follow=True)
# response2 = client.get('/dataproduct/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, Dataproduct_test_data())
# assertDataWithUrls(self, response2.data, Dataproduct_test_data())
def test_Dataproduct_prevents_missing_specifications_template(self):
# setup
test_data = dict(Dataproduct_test_data())
test_data['specifications_template'] = None
# assert
with self.assertRaises(IntegrityError):
models.Dataproduct.objects.create(**test_data)
def SubtaskConnector_test_data():
return {"role": models.Role.objects.get(value='calibrator'),
"datatype": models.Datatype.objects.get(value='instrument model'),
# "dataformats": [models.Dataformat.objects.get(value='Beamformed')], # -> use set()
"output_of": models.SubtaskTemplate.objects.create(**SubtaskTemplate_test_data()),
"input_of": models.SubtaskTemplate.objects.create(**SubtaskTemplate_test_data()),
"tags": []}
class SubtaskConnectorTest(unittest.TestCase):
def test_SubtaskConnector_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.SubtaskConnector.objects.create(**SubtaskConnector_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_SubtaskConnector_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.SubtaskConnector.objects.create(**SubtaskConnector_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_SubtaskConnector_list_view_shows_entry(self):
# # setup
# models.SubtaskConnector.objects.create(**SubtaskConnector_test_data())
#
# # assert
# response = client.get('/subtask_connector/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], SubtaskConnector_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_SubtaskConnector_view_returns_correct_entry(self):
# # setup
# id1 = models.SubtaskConnector.objects.create(**SubtaskConnector_test_data()).id
# id2 = models.SubtaskConnector.objects.create(**SubtaskConnector_test_data()).id
#
# # assert
# response1 = client.get('/subtask_connector/%s/' % id1, format='json', follow=True)
# response2 = client.get('/subtask_connector/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, SubtaskConnector_test_data())
# assertDataWithUrls(self, response2.data, SubtaskConnector_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_SubtaskConnector_allows_setting_dataformats(self):
# # Other then through the API view, we cannot assign ManyToMany on creation, but have to set it later
#
# test_data_1 = dict(SubtaskConnector_test_data())
# test_data_1['inputs'] = None
# tior = models.SubtaskConnector.objects.create(**SubtaskConnector_test_data())
# tior.dataformats.set([models.Dataformat.objects.get(value='Beamformed'),
# models.Dataformat.objects.get(value='MeasurementSet')])
# tior.save()
#
# # assert
# response = client.get('/subtask_connector/%s' % tior.id, format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data, SubtaskConnector_test_data())
def AntennaSet_test_data():
return {"name": "observation",
"description": 'My one observation',
"station_type": models.StationType.objects.get(value='core'),
"rcus": [1,2,3,4,5],
"inputs": ['input1', 'input2'],
"tags": ['tmss', 'testing']}
class AntennaSetTest(unittest.TestCase):
def test_AntennaSet_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.AntennaSet.objects.create(**AntennaSet_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_AntennaSet_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.AntennaSet.objects.create(**AntennaSet_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_AntennaSet_list_view_shows_entry(self):
# # setup
# models.AntennaSet.objects.create(**AntennaSet_test_data())
#
# # assert
# response = client.get('/antenna_set/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], AntennaSet_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_AntennaSet_view_returns_correct_entry(self):
# # setup
# id1 = models.AntennaSet.objects.create(**AntennaSet_test_data()).id
# id2 = models.AntennaSet.objects.create(**AntennaSet_test_data()).id
#
# # assert
# response1 = client.get('/antenna_set/%s/' % id1, format='json', follow=True)
# response2 = client.get('/antenna_set/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, AntennaSet_test_data())
# assertDataWithUrls(self, response2.data, AntennaSet_test_data())
def DataproductTransform_test_data():
return {"input": models.Dataproduct.objects.create(**Dataproduct_test_data()),
"output": models.Dataproduct.objects.create(**Dataproduct_test_data()),
"identity": True,
"tags": ['tmss', 'testing']}
class DataproductTransformTest(unittest.TestCase):
def test_DataproductTransform_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.DataproductTransform.objects.create(**DataproductTransform_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_DataproductTransform_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.DataproductTransform.objects.create(**DataproductTransform_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_DataproductTransform_list_view_shows_entry(self):
# # setup
# models.DataproductTransform.objects.create(**DataproductTransform_test_data())
#
# # assert
# response = client.get('/dataproduct_transform/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], DataproductTransform_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_DataproductTransform_view_returns_correct_entry(self):
# # setup
# id1 = models.DataproductTransform.objects.create(**DataproductTransform_test_data()).id
# id2 = models.DataproductTransform.objects.create(**DataproductTransform_test_data()).id
#
# # assert
# response1 = client.get('/dataproduct_transform/%s/' % id1, format='json', follow=True)
# response2 = client.get('/dataproduct_transform/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, DataproductTransform_test_data())
# assertDataWithUrls(self, response2.data, DataproductTransform_test_data())
def Filesystem_test_data():
return {"capacity": 1111111111,
"cluster": models.Cluster.objects.create(**Cluster_test_data()),
"tags": ['tmss', 'testing']}
class FilesystemTest(unittest.TestCase):
def test_Filesystem_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.Filesystem.objects.create(**Filesystem_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_Filesystem_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.Filesystem.objects.create(**Filesystem_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_Filesystem_list_view_shows_entry(self):
# # setup
# models.Filesystem.objects.create(**Filesystem_test_data())
#
# # assert
# response = client.get('/filesystem/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], Filesystem_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_Filesystem_view_returns_correct_entry(self):
# # setup
# id1 = models.Filesystem.objects.create(**Filesystem_test_data()).id
# id2 = models.Filesystem.objects.create(**Filesystem_test_data()).id
#
# # assert
# response1 = client.get('/filesystem/%s/' % id1, format='json', follow=True)
# response2 = client.get('/filesystem/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, Filesystem_test_data())
# assertDataWithUrls(self, response2.data, Filesystem_test_data())
def Cluster_test_data():
return {"location": "upstairs",
"tags": ['tmss', 'testing']}
class ClusterTest(unittest.TestCase):
def test_Cluster_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.Cluster.objects.create(**Cluster_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_Cluster_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.Cluster.objects.create(**Cluster_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_Cluster_list_view_shows_entry(self):
# # setup
# models.Cluster.objects.create(**Cluster_test_data())
#
# # assert
# response = client.get('/cluster/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], Cluster_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_Cluster_view_returns_correct_entry(self):
# # setup
# id1 = models.Cluster.objects.create(**Cluster_test_data()).id
# id2 = models.Cluster.objects.create(**Cluster_test_data()).id
#
# # assert
# response1 = client.get('/cluster/%s/' % id1, format='json', follow=True)
# response2 = client.get('/cluster/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, Cluster_test_data())
# assertDataWithUrls(self, response2.data, Cluster_test_data())
def DataproductArchiveInfo_test_data():
return {"dataproduct": models.Dataproduct.objects.create(**Dataproduct_test_data()),
"storage_ticket": "myticket_1",
"public_since": datetime.utcnow().isoformat(),
"corrupted_since": datetime.utcnow().isoformat(),
"tags": ['tmss', 'testing']}
class DataproductArchiveInfoTest(unittest.TestCase):
def test_DataproductArchiveInfo_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.DataproductArchiveInfo.objects.create(**DataproductArchiveInfo_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_DataproductArchiveInfo_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.DataproductArchiveInfo.objects.create(**DataproductArchiveInfo_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_DataproductArchiveInfo_list_view_shows_entry(self):
# # setup
# models.DataproductArchiveInfo.objects.create(**DataproductArchiveInfo_test_data())
#
# # assert
# response = client.get('/dataproduct_archive_info/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], DataproductArchiveInfo_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_DataproductArchiveInfo_view_returns_correct_entry(self):
# # setup
# id1 = models.DataproductArchiveInfo.objects.create(**DataproductArchiveInfo_test_data()).id
# id2 = models.DataproductArchiveInfo.objects.create(**DataproductArchiveInfo_test_data()).id
#
# # assert
# response1 = client.get('/dataproduct_archive_info/%s/' % id1, format='json', follow=True)
# response2 = client.get('/dataproduct_archive_info/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, DataproductArchiveInfo_test_data())
# assertDataWithUrls(self, response2.data, DataproductArchiveInfo_test_data())
def DataproductHash_test_data():
return {"dataproduct": models.Dataproduct.objects.create(**Dataproduct_test_data()),
"algorithm": models.Algorithm.objects.get(value='md5'),
"hash": "myhash_1",
"tags": ['tmss', 'testing']}
class DataproductHashTest(unittest.TestCase):
def test_DataproductHash_gets_created_with_correct_creation_timestamp(self):
# setup
before = datetime.utcnow()
entry = models.DataproductHash.objects.create(**DataproductHash_test_data())
after = datetime.utcnow()
# assert
self.assertLess(before, entry.created_at)
self.assertGreater(after, entry.created_at)
def test_DataproductHash_update_timestamp_gets_changed_correctly(self):
# setup
entry = models.DataproductHash.objects.create(**DataproductHash_test_data())
before = datetime.utcnow()
entry.save()
after = datetime.utcnow()
# assert
self.assertLess(before, entry.updated_at)
self.assertGreater(after, entry.updated_at)
# TODO: rest API testing should be moved out of this test module.
# def test_GET_DataproductHash_list_view_shows_entry(self):
# # setup
# models.DataproductHash.objects.create(**DataproductHash_test_data())
#
# # assert
# response = client.get('/dataproduct_hash/', format='json', follow=True)
# self.assertEqual(response.status_code, 200)
# assertDataWithUrls(self, response.data['results'][0], DataproductHash_test_data())
# TODO: rest API testing should be moved out of this test module.
# def test_GET_DataproductHash_view_returns_correct_entry(self):
# # setup
# id1 = models.DataproductHash.objects.create(**DataproductHash_test_data()).id
# id2 = models.DataproductHash.objects.create(**DataproductHash_test_data()).id
#
# # assert
# response1 = client.get('/dataproduct_hash/%s/' % id1, format='json', follow=True)
# response2 = client.get('/dataproduct_hash/%s/' % id2, format='json', follow=True)
# self.assertEqual(response1.status_code, 200)
# self.assertEqual(response2.status_code, 200)
# assertDataWithUrls(self, response1.data, DataproductHash_test_data())
# assertDataWithUrls(self, response2.data, DataproductHash_test_data())