Newer
Older
import rest_framework.test as rtest
from django.contrib.auth.models import User
import rest_framework.status as response_status
from django.db import IntegrityError
import django.test as dtest
test_object_value = dict(obs_id='12345', oid_source='SAS', dataproduct_source='lofar',
dataproduct_type='observation',
project='LT10_10',
surl='srm://lta-head.lofar.psnc.pl:4884/subfolder/some_nice.tar',
antenna_set="HBA Dual Inner",
instrument_filter= "110-190 MHz",
dysco_compression=True)
class TestDatabaseInteraction(dtest.TestCase):
def test_insert(self):
test_values = dict(test_object_value)
dp = DataProduct(**test_values)
dp.save()
self.assertTrue(dp.pk is not None, 'Failed saving object')
dp = DataProduct.objects.get(obs_id='12345')
for field in test_values:
self.assertEqual(getattr(dp, field), test_values.get(field))
def test_insert_with_helper_function(self):
test_object_value['surl'] = test_object_value['surl'] + 'rrr'
test_values = dict(test_object_value)
dp = DataProduct.insert_dataproduct(**test_values)
for field in test_values:
self.assertEqual(test_object_value.get(field), getattr(dp, field), msg=f'Field {field} does not coincide')
self.assertEqual('lta-head.lofar.psnc.pl', dp.location)
def test_insert_non_unique_surl(self):
test_values = dict(test_object_value)
DataProduct.insert_dataproduct(**test_values)
try:
DataProduct.insert_dataproduct(**test_values)
except IntegrityError:
self.fail(
"Inserting a dataproduct with a surl that's already present raised an Integrity error unexpectedly. It should pass quietly!")
class TestRESTAPI(rtest.APITestCase):
def setUp(self):
self.user = User.objects.create_superuser('admin')
self.client.force_authenticate(self.user)
def test_insert_not_allowed(self):
client = rtest.APIClient()
Roy de Goei
committed
response = client.post('/ldvspec/api/v1/insert_dataproduct/', data={}, format='json')
self.assertEqual(response_status.HTTP_403_FORBIDDEN, response.status_code)
def test_insert_flat_error(self):
Roy de Goei
committed
response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data={}, format='json')
self.assertEqual(response_status.HTTP_400_BAD_REQUEST, response.status_code)
def test_insert_flat_single(self):
test_payload = dict(test_object_value)
Roy de Goei
committed
response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=test_payload, format='json')
self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
def test_insert_flat_multi(self):
test_payload_one = dict(test_object_value)
test_payload_two = dict(test_object_value)
test_payload_two['obs_id'] = '9876'
test_payload_two['surl'] = 'srm://surfsara.nl:4884/x-men/wolverine.tar'
response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=[test_payload_one, test_payload_two],
self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
self.assertTrue(DataProduct.objects.count() == 2, 'Not all dataproduct have been inserted')
def test_insert_flat_multi_insert_single(self):
test_payload = dict(test_object_value)
Roy de Goei
committed
response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=test_payload, format='json')
self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
self.assertTrue(DataProduct.objects.count() == 1, 'Not all dataproduct have been inserted')
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def test_insert_flat_all_valid_locations(self):
test_payload_one = dict(test_object_value)
test_payload_two = dict(test_object_value)
test_payload_three = dict(test_object_value)
test_payload_four = dict(test_object_value)
test_payload_one['location'] = 'lta-head.lofar.psnc.pl'
test_payload_two['location'] = 'srm.grid.sara.nl'
test_payload_two['surl'] = 'srm://surfsara.nl:4884/subfolder/some_nice_two.tar'
test_payload_three['location'] = 'lofar-srm.fz-juelich.de'
test_payload_three['surl'] = 'srm://surfsara.nl:4884/subfolder/some_nice_three.tar'
test_payload_four['location'] = 'srm.target.rug.nl'
test_payload_four['surl'] = 'srm://surfsara.nl:4884/subfolder/some_nice_four.tar'
payload = [test_payload_one, test_payload_two, test_payload_three, test_payload_four]
response = self.client.post('/ldvspec/api/v1/insert_dataproduct/',
data=payload,
format='json')
self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
self.assertTrue(DataProduct.objects.count() == 4, 'Not all dataproduct have been inserted')
locations = [x['location'] for x in DataProduct.objects.values('location')]
for pl in payload:
self.assertTrue(pl['location'] in locations, 'Location not present in the db')
def test_insert_invalid_location(self):
test_payload = dict(test_object_value)
test_payload['location'] = 'srm.grid.invalid.nl'
response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=test_payload, format='json')
self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
self.assertTrue(DataProduct.objects.count() == 1, 'The invalid ')
locations = [x['location'] for x in DataProduct.objects.values('location')]
self.assertEqual(1, len(locations))
self.assertEqual(locations[0], 'Unknown')
def test_insert_missing_location(self):
test_payload = dict(test_object_value)
test_payload.pop('location')
response = self.client.post('/ldvspec/api/v1/insert_dataproduct/', data=test_payload, format='json')
self.assertEqual(response_status.HTTP_400_BAD_REQUEST, response.status_code)