Revision 08495a93
Ajout import oublié + correction de l'indentation dans un des tests.
git-svn-id: https://vigilo-dev.si.c-s.fr/svn@5878 b22e2e97-25c9-44ff-b637-2e5ceca36478
| vigiboard/controllers/root.py | ||
|---|---|---|
| 41 | 41 |
from vigilo.models.session import DBSession |
| 42 | 42 |
from vigilo.models.tables import Event, EventHistory, CorrEvent, Host, \ |
| 43 | 43 |
SupItem, SupItemGroup, LowLevelService, \ |
| 44 |
StateName |
|
| 44 |
StateName, State
|
|
| 45 | 45 |
from vigilo.models.tables.grouphierarchy import GroupHierarchy |
| 46 | 46 |
from vigilo.models.functions import sql_escape_like |
| 47 | 47 |
from vigilo.models.tables.secondary_tables import EVENTSAGGREGATE_TABLE |
| vigiboard/tests/functional/test_raw_events_table.py | ||
|---|---|---|
| 15 | 15 |
from vigiboard.tests import TestController |
| 16 | 16 |
|
| 17 | 17 |
def populate_accounts(): |
| 18 |
perm = Permission.by_permission_name(u'vigiboard-access')
|
|
| 19 |
|
|
| 20 |
user = User(
|
|
| 21 |
user_name=u'access',
|
|
| 22 |
fullname=u'',
|
|
| 23 |
email=u'user.has@access',
|
|
| 24 |
)
|
|
| 25 |
usergroup = UserGroup(
|
|
| 26 |
group_name=u'users_with_access',
|
|
| 27 |
)
|
|
| 28 |
usergroup.permissions.append(perm)
|
|
| 29 |
user.usergroups.append(usergroup)
|
|
| 30 |
DBSession.add(user)
|
|
| 31 |
DBSession.add(usergroup)
|
|
| 32 |
DBSession.flush()
|
|
| 33 |
|
|
| 34 |
user = User(
|
|
| 35 |
user_name=u'limited_access',
|
|
| 36 |
fullname=u'',
|
|
| 37 |
email=u'user.has.no@access',
|
|
| 38 |
)
|
|
| 39 |
usergroup = UserGroup(
|
|
| 40 |
group_name=u'users_with_limited_access',
|
|
| 41 |
)
|
|
| 42 |
usergroup.permissions.append(perm)
|
|
| 43 |
user.usergroups.append(usergroup)
|
|
| 44 |
DBSession.add(user)
|
|
| 45 |
DBSession.add(usergroup)
|
|
| 46 |
DBSession.flush()
|
|
| 47 |
transaction.commit()
|
|
| 18 |
perm = Permission.by_permission_name(u'vigiboard-access') |
|
| 19 |
|
|
| 20 |
user = User( |
|
| 21 |
user_name=u'access', |
|
| 22 |
fullname=u'', |
|
| 23 |
email=u'user.has@access', |
|
| 24 |
) |
|
| 25 |
usergroup = UserGroup( |
|
| 26 |
group_name=u'users_with_access', |
|
| 27 |
) |
|
| 28 |
usergroup.permissions.append(perm) |
|
| 29 |
user.usergroups.append(usergroup) |
|
| 30 |
DBSession.add(user) |
|
| 31 |
DBSession.add(usergroup) |
|
| 32 |
DBSession.flush() |
|
| 33 |
|
|
| 34 |
user = User( |
|
| 35 |
user_name=u'limited_access', |
|
| 36 |
fullname=u'', |
|
| 37 |
email=u'user.has.no@access', |
|
| 38 |
) |
|
| 39 |
usergroup = UserGroup( |
|
| 40 |
group_name=u'users_with_limited_access', |
|
| 41 |
) |
|
| 42 |
usergroup.permissions.append(perm) |
|
| 43 |
user.usergroups.append(usergroup) |
|
| 44 |
DBSession.add(user) |
|
| 45 |
DBSession.add(usergroup) |
|
| 46 |
DBSession.flush() |
|
| 47 |
transaction.commit() |
|
| 48 | 48 |
|
| 49 | 49 |
def populate_DB(caused_by_service): |
| 50 | 50 |
""" Peuple la base de données. """ |
| ... | ... | |
| 63 | 63 |
|
| 64 | 64 |
# On crée un hôte de test, et on l'ajoute au groupe d'hôtes. |
| 65 | 65 |
managerhost = Host( |
| 66 |
name = u'managerhost',
|
|
| 66 |
name = u'managerhost', |
|
| 67 | 67 |
checkhostcmd = u'halt', |
| 68 | 68 |
snmpcommunity = u'public', |
| 69 | 69 |
hosttpl = u'/dev/null', |
| ... | ... | |
| 93 | 93 |
|
| 94 | 94 |
# Ajout d'un événement |
| 95 | 95 |
event = Event( |
| 96 |
supitem = supitem,
|
|
| 96 |
supitem = supitem, |
|
| 97 | 97 |
message = u'foo', |
| 98 | 98 |
current_state = StateName.statename_to_value(u"WARNING"), |
| 99 | 99 |
timestamp = datetime.now(), |
| ... | ... | |
| 104 | 104 |
# Ajout des historiques |
| 105 | 105 |
DBSession.add(EventHistory( |
| 106 | 106 |
type_action=u'Nagios update state', |
| 107 |
idevent=event.idevent,
|
|
| 107 |
idevent=event.idevent, |
|
| 108 | 108 |
timestamp=datetime.now())) |
| 109 | 109 |
DBSession.add(EventHistory( |
| 110 | 110 |
type_action=u'Acknowlegement change state', |
| 111 |
idevent=event.idevent,
|
|
| 111 |
idevent=event.idevent, |
|
| 112 | 112 |
timestamp=datetime.now())) |
| 113 | 113 |
DBSession.flush() |
| 114 | 114 |
|
| 115 | 115 |
# Ajout d'un événement corrélé |
| 116 | 116 |
aggregate = CorrEvent( |
| 117 |
idcause = event.idevent,
|
|
| 117 |
idcause = event.idevent, |
|
| 118 | 118 |
timestamp_active = datetime.now(), |
| 119 | 119 |
priority = 1, |
| 120 | 120 |
status = u"None") |
| ... | ... | |
| 130 | 130 |
nb_hosts = DBSession.query(Host).count() |
| 131 | 131 |
|
| 132 | 132 |
masked_host = Host( |
| 133 |
name = u'masked host #%d' % nb_hosts,
|
|
| 133 |
name = u'masked host #%d' % nb_hosts, |
|
| 134 | 134 |
checkhostcmd = u'halt', |
| 135 | 135 |
snmpcommunity = u'public', |
| 136 | 136 |
hosttpl = u'/dev/null', |
| ... | ... | |
| 220 | 220 |
# n'a pas accès aux informations concernant cet évènement. |
| 221 | 221 |
response = self.app.get( |
| 222 | 222 |
'/masked_events/%d' % idcorrevent, |
| 223 |
status = 302,
|
|
| 223 |
status = 302, |
|
| 224 | 224 |
extra_environ = environ) |
| 225 | 225 |
response = response.follow(status = 200, extra_environ = environ) |
| 226 | 226 |
assert_true(len(response.lxml.xpath( |
| ... | ... | |
| 234 | 234 |
add_masked_event(idcorrevent) |
| 235 | 235 |
response = self.app.get( |
| 236 | 236 |
'/masked_events/%d' % idcorrevent, |
| 237 |
status = 302,
|
|
| 237 |
status = 302, |
|
| 238 | 238 |
extra_environ = environ) |
| 239 | 239 |
response = response.follow(status = 200, extra_environ = environ) |
| 240 | 240 |
assert_true(len(response.lxml.xpath( |
| ... | ... | |
| 305 | 305 |
def test_table(self): |
| 306 | 306 |
"""Événements masqués d'un agrégat sur un hôte avec permissions.""" |
| 307 | 307 |
super(TestRawEventsTableWithPermsHost, self).test_table() |
| 308 |
|
|
Also available in: Unified diff