Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import django.test as dtest
import rest_framework.test as rtest
from django.contrib.auth.models import User
import rest_framework.status as response_status
from lofardata.models import DataProduct, DataLocation
test_object_value = dict(obs_id='12345', oid_source='SAS', dataproduct_source='lofar',
dataproduct_type='observation',
project='LT10_10',
location='srm://surfsara.nl:4884/',
activity='observation',
surl='srm://surfsara.nl:4884/...',
filesize=40,
additional_meta={'dysco_compression': True})
class TestDatabaseInteraction(dtest.TestCase):
def test_insert(self):
location = DataLocation(name='surfsara', uri='srm://surfsara.nl')
location.save()
test_values = dict(test_object_value)
test_values['location'] = location
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_values = dict(test_object_value)
test_values.pop('location')
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('surfsara.nl:4884', dp.location.name)
self.assertEqual( 'srm://surfsara.nl:4884', dp.location.uri)
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()
response = client.post('/ldvspec/api/v1/data/insert/', data={}, format='json')
self.assertEqual(response_status.HTTP_403_FORBIDDEN, response.status_code)
def test_insert_flat_error(self):
response = self.client.post('/ldvspec/api/v1/data/insert/', 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)
test_payload.pop('location')
response = self.client.post('/ldvspec/api/v1/data/insert/', data=test_payload, format='json')
self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
def test_insert_flat_multi(self):
test_payload = dict(test_object_value)
test_payload.pop('location')
response = self.client.post('/ldvspec/api/v1/data/insert/', data=[test_payload, test_payload], format='json')
self.assertEqual(response_status.HTTP_201_CREATED, response.status_code)
self.assertTrue(DataProduct.objects.count() == 2, 'Not all dataproduct have been inserted')
self.assertTrue(DataLocation.objects.count() == 1, 'Not all dataproduct have been inserted')
def test_insert_flat_multi_insert_single(self):
test_payload = dict(test_object_value)
test_payload.pop('location')
response = self.client.post('/ldvspec/api/v1/data/insert/', 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')
self.assertTrue(DataLocation.objects.count() == 1, 'Not all dataproduct have been inserted')