vigiboard / vigiboard / tests / functional / test_history_table.py @ 747e30c3
History | View | Annotate | Download (7.24 KB)
1 | 072f2a16 | Francois POIROTTE | # -*- coding: utf-8 -*-
|
---|---|---|---|
2 | # vim:set expandtab tabstop=4 shiftwidth=4:
|
||
3 | """
|
||
4 | Vérifie que la page qui affiche l'historique des actions sur un événement
|
||
5 | brut fonctionne correctement.
|
||
6 | """
|
||
7 | |||
8 | from nose.tools import assert_true, assert_equal |
||
9 | from datetime import datetime |
||
10 | import transaction |
||
11 | |||
12 | e7e3d45e | Francois POIROTTE | from vigilo.models.session import DBSession |
13 | from vigilo.models.tables import Event, EventHistory, CorrEvent, \ |
||
14 | 072f2a16 | Francois POIROTTE | Permission, StateName, \ |
15 | Host, HostGroup, LowLevelService, ServiceGroup |
||
16 | from vigiboard.tests import TestController |
||
17 | |||
18 | def populate_DB(): |
||
19 | """ Peuple la base de données. """
|
||
20 | # On ajoute un groupe d'hôtes et un groupe de services.
|
||
21 | hostmanagers = HostGroup(name = u'managersgroup')
|
||
22 | DBSession.add(hostmanagers) |
||
23 | servicemanagers = ServiceGroup(name = u'managersgroup')
|
||
24 | DBSession.add(servicemanagers) |
||
25 | DBSession.flush() |
||
26 | |||
27 | # On ajoute la permission 'manage' à ces deux groupes.
|
||
28 | manage_perm = Permission.by_permission_name(u'manage')
|
||
29 | hostmanagers.permissions.append(manage_perm) |
||
30 | servicemanagers.permissions.append(manage_perm) |
||
31 | DBSession.flush() |
||
32 | |||
33 | # On crée un hôte de test, et on l'ajoute au groupe d'hôtes.
|
||
34 | managerhost = Host( |
||
35 | name = u'managerhost',
|
||
36 | checkhostcmd = u'halt',
|
||
37 | snmpcommunity = u'public',
|
||
38 | hosttpl = u'/dev/null',
|
||
39 | mainip = u'192.168.1.1',
|
||
40 | snmpport = 42,
|
||
41 | weight = 42,
|
||
42 | ) |
||
43 | DBSession.add(managerhost) |
||
44 | hostmanagers.hosts.append(managerhost) |
||
45 | DBSession.flush() |
||
46 | |||
47 | # On crée un services de bas niveau, et on l'ajoute au groupe de services.
|
||
48 | managerservice = LowLevelService( |
||
49 | host = managerhost, |
||
50 | servicename = u'managerservice',
|
||
51 | command = u'halt',
|
||
52 | op_dep = u'+',
|
||
53 | weight = 42,
|
||
54 | ) |
||
55 | DBSession.add(managerservice) |
||
56 | servicemanagers.services.append(managerservice) |
||
57 | DBSession.flush() |
||
58 | |||
59 | return (managerhost, managerservice)
|
||
60 | |||
61 | def add_correvent_caused_by(supitem): |
||
62 | """
|
||
63 | Ajoute dans la base de données un évènement corrélé causé
|
||
64 | par un incident survenu sur l'item passé en paramètre.
|
||
65 | Génère un historique pour les tests.
|
||
66 | """
|
||
67 | |||
68 | # Ajout d'un événement
|
||
69 | event = Event( |
||
70 | supitem = supitem, |
||
71 | message = u'foo',
|
||
72 | current_state = StateName.statename_to_value(u"WARNING"),
|
||
73 | 0c8b0e15 | Francois POIROTTE | timestamp = datetime.now(), |
74 | 072f2a16 | Francois POIROTTE | ) |
75 | DBSession.add(event) |
||
76 | DBSession.flush() |
||
77 | |||
78 | # Ajout des historiques
|
||
79 | DBSession.add(EventHistory( |
||
80 | type_action=u'Nagios update state',
|
||
81 | idevent=event.idevent, |
||
82 | timestamp=datetime.now())) |
||
83 | DBSession.add(EventHistory( |
||
84 | type_action=u'Acknowlegement change state',
|
||
85 | idevent=event.idevent, |
||
86 | timestamp=datetime.now())) |
||
87 | DBSession.flush() |
||
88 | |||
89 | # Ajout d'un événement corrélé
|
||
90 | aggregate = CorrEvent( |
||
91 | idcause = event.idevent, |
||
92 | timestamp_active = datetime.now(), |
||
93 | priority = 1,
|
||
94 | status = u"None")
|
||
95 | aggregate.events.append(event) |
||
96 | DBSession.add(aggregate) |
||
97 | DBSession.flush() |
||
98 | |||
99 | return event.idevent
|
||
100 | |||
101 | |||
102 | class TestHistoryTable(TestController): |
||
103 | """
|
||
104 | 0c8b0e15 | Francois POIROTTE | Teste la table qui affiche l'historique des actions
|
105 | sur un événement brut.
|
||
106 | 072f2a16 | Francois POIROTTE | """
|
107 | |||
108 | def test_cause_host_history(self): |
||
109 | """Historique de la cause d'un événement corrélé sur un hôte."""
|
||
110 | |||
111 | # On peuple la BDD avec un hôte, un service de bas niveau,
|
||
112 | # et un groupe d'hôtes et de services associés à ces items.
|
||
113 | (managerhost, managerservice) = populate_DB() |
||
114 | |||
115 | # On ajoute un évènement corrélé causé par l'hôte
|
||
116 | idevent = add_correvent_caused_by(managerhost) |
||
117 | transaction.commit() |
||
118 | |||
119 | # L'utilisateur n'est pas authentifié.
|
||
120 | # On s'attend à ce qu'une erreur 401 soit renvoyée,
|
||
121 | # demandant à l'utilisateur de s'authentifier.
|
||
122 | response = self.app.get(
|
||
123 | '/event/%d' % idevent,
|
||
124 | status = 401)
|
||
125 | |||
126 | # L'utilisateur N'A PAS les bonnes permissions.
|
||
127 | environ = {'REMOTE_USER': 'editor'} |
||
128 | |||
129 | # On s'attend à ce qu'une erreur 302 soit renvoyée, et à
|
||
130 | # ce qu'un message d'erreur précise à l'utilisateur qu'il
|
||
131 | # n'a pas accès aux informations concernant cet évènement.
|
||
132 | response = self.app.get(
|
||
133 | '/event/%d' % idevent,
|
||
134 | status = 302,
|
||
135 | extra_environ = environ) |
||
136 | |||
137 | # On suit la redirection.
|
||
138 | response = response.follow(status = 200, extra_environ = environ)
|
||
139 | assert_true(len(response.lxml.xpath(
|
||
140 | '//div[@id="flash"]/div[@class="error"]')))
|
||
141 | |||
142 | # L'utilisateur a les bonnes permissions.
|
||
143 | environ = {'REMOTE_USER': 'manager'} |
||
144 | |||
145 | # On s'attend à ce que le statut de la requête soit 200.
|
||
146 | response = self.app.get(
|
||
147 | '/event/%d' % idevent,
|
||
148 | status = 200,
|
||
149 | extra_environ = environ) |
||
150 | |||
151 | # Il doit y avoir 2 lignes de résultats.
|
||
152 | 9211ef65 | Francois POIROTTE | # NB: la requête XPath est approchante, car XPath 1.0 ne permet pas
|
153 | # de rechercher directement une valeur dans une liste. Elle devrait
|
||
154 | # néanmoins suffire pour les besoins des tests.
|
||
155 | rows = response.lxml.xpath('//table[contains(@class, "vigitable")]/tbody/tr')
|
||
156 | 072f2a16 | Francois POIROTTE | assert_equal(len(rows), 2) |
157 | |||
158 | def test_cause_service_history(self): |
||
159 | """Historique de la cause d'un événement corrélé sur un LLS."""
|
||
160 | |||
161 | # On peuple la BDD avec un hôte, un service de bas niveau,
|
||
162 | # et un groupe d'hôtes et de services associés à ces items.
|
||
163 | (managerhost, managerservice) = populate_DB() |
||
164 | |||
165 | # On ajoute un évènement corrélé causé par le service
|
||
166 | idevent = add_correvent_caused_by(managerservice) |
||
167 | |||
168 | transaction.commit() |
||
169 | |||
170 | # L'utilisateur n'est pas authentifié.
|
||
171 | # On s'attend à ce qu'une erreur 401 soit renvoyée,
|
||
172 | # demandant à l'utilisateur de s'authentifier.
|
||
173 | response = self.app.get(
|
||
174 | '/event/%d' % idevent,
|
||
175 | status = 401)
|
||
176 | |||
177 | # L'utilisateur N'A PAS les bonnes permissions.
|
||
178 | environ = {'REMOTE_USER': 'editor'} |
||
179 | |||
180 | # On s'attend à ce qu'une erreur 302 soit renvoyée, et à
|
||
181 | # ce qu'un message d'erreur précise à l'utilisateur qu'il
|
||
182 | # n'a pas accès aux informations concernant cet évènement.
|
||
183 | response = self.app.get(
|
||
184 | '/event/%d' % idevent,
|
||
185 | status = 302,
|
||
186 | extra_environ = environ) |
||
187 | |||
188 | # On suit la redirection.
|
||
189 | response = response.follow(status = 200, extra_environ = environ)
|
||
190 | assert_true(len(response.lxml.xpath(
|
||
191 | '//div[@id="flash"]/div[@class="error"]')))
|
||
192 | |||
193 | # L'utilisateur a les bonnes permissions.
|
||
194 | environ = {'REMOTE_USER': 'manager'} |
||
195 | |||
196 | # On s'attend à ce que le statut de la requête soit 200.
|
||
197 | response = self.app.get(
|
||
198 | '/event/%d' % idevent,
|
||
199 | status = 200,
|
||
200 | extra_environ = environ) |
||
201 | |||
202 | # Il doit y avoir 2 lignes de résultats.
|
||
203 | 9211ef65 | Francois POIROTTE | # NB: la requête XPath est approchante, car XPath 1.0 ne permet pas
|
204 | # de rechercher directement une valeur dans une liste. Elle devrait
|
||
205 | # néanmoins suffire pour les besoins des tests.
|
||
206 | rows = response.lxml.xpath('//table[contains(@class,"vigitable")]/tbody/tr')
|
||
207 | 072f2a16 | Francois POIROTTE | assert_equal(len(rows), 2) |