tests.py
4.47 KB
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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
from django.test import TestCase
import os
import shutil
import time
import alert_manager.tasks
from django.conf import settings
from pyrosapp.models import *
TEST_FILE = "unittest_voevent.xml"
TEST_FILE_PATH = os.path.join(alert_manager.tasks.VOEVENTS_PATH, TEST_FILE)
VOEVENTS_TO_SEND_PATH = "alert_manager/events_to_send"
class TestStrategyChange(TestCase):
fixtures = ['alert_mgr_test.json']
def setUp(self):
'''
Creates the base for the tests
'''
self.alert = Alert.objects.get()
self.strat1 = Alert.objects.get().strategyobs
self.strat2 = StrategyObs.objects.exclude(id=self.strat1.id)[0]
self.alert.strategyobs = self.strat1
self.alert.save()
def test_change_inexistant_strat(self):
self.client.login(username="test@test.test", password="test")
path = "/alert_manager/change_obs_strategy_validate/" + str(self.alert.id)
response = self.client.post(path, {"strategy_choice": 42})
self.assertTrue("error" in response.context.keys(), "There should be an error of non existant strategy")
def test_change_inexistant_alert(self):
self.client.login(username="test@test.test", password="test")
path = "/alert_manager/change_obs_strategy_validate/" + str(self.alert.id + 1)
response = self.client.post(path, {"strategy_choice": self.strat2.id})
self.assertTrue("error" in response.context.keys(), "There should be an error of non existant alert")
def test_change_same_strat(self):
self.client.login(username="test@test.test", password="test")
path = "/alert_manager/change_obs_strategy_validate/" + str(self.alert.id)
response = self.client.post(path, {"strategy_choice": self.strat1.id})
self.assertTrue("error" in response.context.keys(), "There should be an error of changing to same strategy")
def test_change_all_working(self):
self.client.login(username="test@test.test", password="test")
path = "/alert_manager/change_obs_strategy_validate/" + str(self.alert.id)
response = self.client.post(path, {"strategy_choice": self.strat2.id})
self.assertFalse("error" in response.context.keys(), "There shouldn't be an error")
self.assertEqual(Alert.objects.count(), 2, "There should be 2 alerts after the strategy change")
new_alert = Alert.objects.exclude(id=self.alert.id)[0]
self.assertEqual(new_alert.strategyobs.id, self.strat2.id, "The new alert should have the 'strat2' strategy")
class AlertListenerTestsCelery(TestCase):
'''
IMPORTANT : As soon as you do a DB request in a test, the test DB will no longer be synchronized with the DB used by celery
I have no idea why (Paul), but any call to the DB must be done in setup or after all celery actions are finished
'''
def setup(self):
default_strat = StrategyObs.objects.get(is_default=True)
default_strat.is_default = False
default_strat.save()
test_strat = StrategyObs.objects.get(name="strat_unittest")
test_strat.is_default = True
test_strat.save()
def test_alert_reception(self):
'''
Must be launched with scripts/celery_test.sh
Copy a VOEvent file in the events_received directory to start the workflow
Tests if the alert and children are well created
'''
if os.path.isfile(TEST_FILE_PATH):
os.remove(TEST_FILE_PATH)
time.sleep(3)
shutil.copyfile(os.path.join(VOEVENTS_TO_SEND_PATH, TEST_FILE),
TEST_FILE_PATH)
time.sleep(3)
self.assertEqual(Alert.objects.count(), 1)
alert = Alert.objects.all()[0]
self.assertEqual(alert.author, "ivo://nasa.gsfc.tan/gcn")
self.assertEqual(alert.burst_ra, 74.7412)
self.assertEqual(alert.burst_dec, 25.3137)
self.assertEqual(alert.trig_id, 532871)
self.assertEqual(alert.editor, "61")
self.assertEqual(alert.pkt_ser_num, 1)
self.assertEqual(Request.objects.count(), 1)
self.assertEqual(Sequence.objects.count(), 2)
self.assertEqual(Album.objects.count(), 3)
self.assertEqual(Plan.objects.count(), 6)
def test_basic(self):
'''
Used to test scripts/celery_test.sh
Only work if the initial_fixture is on the prod DB (an only it)
'''
Country.objects.create(name="TEEEEST")
self.assertEqual(Country.objects.count(), 2, "should be 2 countries")