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
import django.test as dtest
import rest_framework.test as rtest
import unittest.mock
from django.contrib.auth.models import User
import rest_framework.status as response_status
from lofardata.models import WorkSpecification, DataProduct
import lofardata.tasks as tasks
class TestWorkSpecificationRequest(rtest.APITestCase):
def setUp(self):
self.user = User.objects.create_superuser('admin')
self.client.force_authenticate(self.user)
@unittest.mock.patch('lofardata.tasks.define_work_specification.delay')
def test_insert_work_request(self, mocked_task):
# Mocking business
mocked_task.return_value = unittest.mock.MagicMock()
mocked_task.return_value.id = 'asyncresultid'
# ----------------------------------
# Insert test data
DataProduct.insert_dataproduct(12345, 'lta',
'myniceinstrument', 'ms', 'blubby', 'pipline',
'srm://lofarlta/mynice_file.tar.gz', 123456, {'dysco_compression': False})
response = self.client.post('/ldvspec/api/v1/workspecification/',
data={'filters': {'obs_id': 12345}}, format='json')
self.assertEqual(201, response.status_code)
inserted_work_specification = response.json()
mocked_task.assert_called_with(inserted_work_specification['id'])
self.assertEqual('asyncresultid', inserted_work_specification['async_task_result'])
tasks.define_work_specification(inserted_work_specification['id'])
work_spec_object = WorkSpecification.objects.get(pk=inserted_work_specification['id'])
self.assertTrue(work_spec_object.is_ready)
self.assertEqual({'surls': ['srm://lofarlta/mynice_file.tar.gz']}, work_spec_object.inputs)
@unittest.mock.patch('lofardata.tasks.define_work_specification.delay')
def test_insert_work_request_nested_fields(self, mocked_task):
# Mocking business
mocked_task.return_value = unittest.mock.MagicMock()
mocked_task.return_value.id = 'asyncresultid'
# ----------------------------------
# Insert test data
DataProduct.insert_dataproduct(12345, 'lta',
'myniceinstrument', 'ms', 'blubby', 'pipline',
'srm://lofarlta/mynice_file.tar.gz', 123456, {'dysco_compression': False})
DataProduct.insert_dataproduct(12345, 'lta',
'myniceinstrument', 'ms', 'blubby', 'pipline',
'srm://lofarlta/mynice_file_with_rhythm.tar.gz', 123456,
{'dysco_compression': True})
response = self.client.post('/ldvspec/api/v1/workspecification/',
data={'filters': {'obs_id': 12345, 'additional_meta__dysco_compression': True}},
format='json')
self.assertEqual(201, response.status_code)
inserted_work_specification = response.json()
mocked_task.assert_called_with(inserted_work_specification['id'])
self.assertEqual('asyncresultid', inserted_work_specification['async_task_result'])
tasks.define_work_specification(inserted_work_specification['id'])
work_spec_object = WorkSpecification.objects.get(pk=inserted_work_specification['id'])
self.assertTrue(work_spec_object.is_ready)
self.assertEqual({'surls': ['srm://lofarlta/mynice_file_with_rhythm.tar.gz']}, work_spec_object.inputs)