Newer
Older
import types
from unittest import mock
import django.test as dtest
from django.contrib.auth.models import User
from lofardata.models import WorkSpecification, Group, ATDBProcessingSite
from lofardata.view_helpers.specification import create_work_specifications_for_group, \
create_work_specification_for_group
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
class WorkSpecificationCreationDatabaseInteraction(dtest.TestCase):
def setUp(self):
self.user, _ = User.objects.get_or_create()
self.site, _ = ATDBProcessingSite.objects.get_or_create(
name="marvel", url="http://marvel.com", access_token="dummy"
)
self.group, _ = Group.objects.get_or_create(name="x-men", processing_site=self.site, selected_workflow="claw",
selected_workflow_tag="attack")
self.delay_return_value = types.SimpleNamespace(id='5')
@mock.patch('lofardata.tasks.define_work_specification.delay')
def test_create_group_with_one_obs_id(self, mock_delay):
mock_delay.return_value = self.delay_return_value
create_work_specification_for_group(self.group, self.user, "123")
work_specification = list(WorkSpecification.objects.filter(group=self.group.id))
self.assertEqual(len(work_specification), 1)
actual_inherited_group_values = [(x.processing_site, x.selected_workflow, x.selected_workflow_tag) for x in
work_specification if x.group is self.group]
for actual_inherited_group_value in actual_inherited_group_values:
self.assertTupleEqual(actual_inherited_group_value,
(self.site, self.group.selected_workflow, self.group.selected_workflow_tag))
@mock.patch('lofardata.tasks.define_work_specification.delay')
def test_create_group_with_multiple_obs_ids(self, mock_delay):
mock_delay.return_value = self.delay_return_value
create_work_specifications_for_group(self.group, self.user, ["123", "456", " 781 "])
work_specification = list(WorkSpecification.objects.filter(group=self.group.id))
self.assertEqual(len(work_specification), 3)
actual_inherited_group_values = [(x.processing_site, x.selected_workflow, x.selected_workflow_tag) for x in
work_specification if x.group is self.group]
for actual_inherited_group_value in actual_inherited_group_values:
self.assertTupleEqual(actual_inherited_group_value,
(self.site, self.group.selected_workflow, self.group.selected_workflow_tag))
@mock.patch('lofardata.tasks.define_work_specification.delay')
def test_create_group_with_filters(self, mock_delay):
mock_delay.return_value = self.delay_return_value
obs_ids = ['123', '456']
create_work_specifications_for_group(self.group, self.user, obs_ids)
work_specification = list(WorkSpecification.objects.filter(group=self.group.id))
actual_obs_ids = [x.filters['obs_id'] for x in work_specification if x.filters['obs_id'] in obs_ids]
# assertCountEqual: Asserts that two iterables have the same elements, the same number of
# times, without regard to order.
self.assertCountEqual(actual_obs_ids, obs_ids)
@mock.patch("lofardata.view_helpers.specification.set_post_submit_values")
@mock.patch('lofardata.tasks.define_work_specification.delay')
def test_create_group_with_post_submit_values(self, mock_delay, post_submit_mock):
mock_delay.return_value = self.delay_return_value
create_work_specifications_for_group(self.group, self.user, ["123"])
self.assertTrue(post_submit_mock.called)