events.add_join((EVENTSAGGREGATE_TABLE, EVENTSAGGREGATE_TABLE.c.idevent == Event.idevent))

        events.add_join((CorrEvent, CorrEvent.idcorrevent == EVENTSAGGREGATE_TABLE.c.idcorrevent))

        events.add_join((EVENTSAGGREGATE_TABLE, EVENTSAGGREGATE_TABLE.c.idevent == Event.idevent))

        events.add_join((CorrEvent, CorrEvent.idcorrevent == EVENTSAGGREGATE_TABLE.c.idcorrevent))

# -*- coding: utf-8 -*-

# vim:set expandtab tabstop=4 shiftwidth=4:

"""

Test du tableau d'événements de Vigiboard

"""

from nose.tools import assert_true, assert_equal

from datetime import datetime

import transaction

from vigilo.models.configure import DBSession

from vigilo.models import Event, CorrEvent, \

                            Permission, StateName, \

                            Host, HostGroup, LowLevelService

from vigiboard.tests import TestController

def populate_DB():

    """ Peuple la base de données. """

    # On ajoute les groupes et leurs dépendances

    hosteditors = HostGroup(name=u'editorsgroup')

    DBSession.add(hosteditors)

    hostmanagers = HostGroup(name=u'managersgroup', parent=hosteditors)

    DBSession.add(hostmanagers)

    DBSession.flush()

    manage_perm = Permission.by_permission_name(u'manage')

    edit_perm = Permission.by_permission_name(u'edit')

    hostmanagers.permissions.append(manage_perm)

    hosteditors.permissions.append(edit_perm)

    DBSession.flush()

    # Création des hôtes de test.

    host_template = {

        'checkhostcmd': u'halt',

        'snmpcommunity': u'public',

        'hosttpl': u'/dev/null',

        'mainip': u'192.168.1.1',

        'snmpport': 42,

        'weight': 42,

    }

    managerhost = Host(name=u'managerhost', **host_template)

    editorhost = Host(name=u'editorhost', **host_template)

    DBSession.add(managerhost)

    DBSession.add(editorhost)

    # Affectation des hôtes aux groupes.

    hosteditors.hosts.append(editorhost)

    hostmanagers.hosts.append(managerhost)

    DBSession.flush()

    # Création des services techniques de test.

    service_template = {

        'command': u'halt',

        'op_dep': u'+',

        'weight': 42,

    }

    service1 = LowLevelService(

        host=managerhost,

        servicename=u'managerservice',

        **service_template

    )

    service2 = LowLevelService(

        host=editorhost,

        servicename=u'managerservice',

        **service_template

    )

    service3 = LowLevelService(

        host=managerhost,

        servicename=u'editorservice',

        **service_template

    )

    service4 = LowLevelService(

        host=editorhost,

        servicename=u'editorservice',

        **service_template

    )

    DBSession.add(service1)

    DBSession.add(service2)

    DBSession.add(service3)

    DBSession.add(service4)

    DBSession.flush()

    # Ajout des événements eux-mêmes

    event_template = {

        'message': u'foo',

        'current_state': StateName.statename_to_value(u'WARNING'),

    }

    event1 = Event(supitem=service1, **event_template)

    event2 = Event(supitem=service2, **event_template)

    event3 = Event(supitem=service3, **event_template)

    event4 = Event(supitem=service4, **event_template)

    event5 = Event(supitem=editorhost, **event_template)

    event6 = Event(supitem=managerhost, **event_template)

    DBSession.add(event1)

    DBSession.add(event2)

    DBSession.add(event3)

    DBSession.add(event4)

    DBSession.add(event5)

    DBSession.add(event6)

    DBSession.flush()

    # Ajout des événements corrélés

    aggregate_template = {

        'timestamp_active': datetime.now(),

        'priority': 1,

        'status': u'None',

    }

    aggregate1 = CorrEvent(

        idcause=event1.idevent, **aggregate_template)

    aggregate2 = CorrEvent(

        idcause=event4.idevent, **aggregate_template)

    aggregate3 = CorrEvent(

        idcause=event5.idevent, **aggregate_template)

    aggregate4 = CorrEvent(

        idcause=event6.idevent, **aggregate_template)

    aggregate1.events.append(event1)

    aggregate1.events.append(event3)

    aggregate2.events.append(event4)

    aggregate2.events.append(event2)

    aggregate3.events.append(event5)

    aggregate4.events.append(event6)

    DBSession.add(aggregate1)

    DBSession.add(aggregate2)

    DBSession.add(aggregate3)

    DBSession.add(aggregate4)

    DBSession.flush()

    transaction.commit()

class TestEventTable(TestController):

    """

    Test du tableau d'événements de Vigiboard

    """

    def test_event_table(self):

        """

        Test du tableau d'évènements de la page d'accueil de Vigiboard.

        """

        populate_DB()

        ### 1er cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'editor'.

        environ = {'REMOTE_USER': 'editor'}

        response = self.app.get('/', extra_environ=environ)

        # Il doit y avoir 2 lignes de résultats.

        rows = response.lxml.xpath('//table[@class="vigitable"]/tbody/tr')

        print "There are %d rows in the result set" % len(rows)

        assert_equal(len(rows), 2)

        # Il doit y avoir plusieurs colonnes dans la ligne de résultats.

        cols = response.lxml.xpath('//table[@class="vigitable"]/tbody/tr/td')

        print "There are %d columns in the result set" % len(cols)

        assert_true(len(cols) > 1)

        ### 2nd cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'manager'.

        environ = {'REMOTE_USER': 'manager'}

        response = self.app.get('/', extra_environ=environ)

        # Il doit y avoir 4 lignes de résultats.

        rows = response.lxml.xpath('//table[@class="vigitable"]/tbody/tr')

        print "There are %d rows in the result set" % len(rows)

        assert_equal(len(rows), 4)

        # Il doit y avoir plusieurs colonnes dans la ligne de résultats.

        cols = response.lxml.xpath('//table[@class="vigitable"]/tbody/tr/td')

        print "There are %d columns in the result set" % len(cols)

        assert_true(len(cols) > 1)

# -*- coding: utf-8 -*-

"""

Teste le formulaire donnant les liens vers les outils extérieurs

et les données de l'historique.

"""

from nose.tools import assert_true, assert_equal

from datetime import datetime

import transaction

from vigiboard.tests import TestController

from vigilo.models.configure import DBSession

from vigilo.models import ServiceGroup, HostGroup, \

                            Host, Permission, StateName, \

                            LowLevelService, Event, CorrEvent

def insert_deps(return_service):

    """Insère les dépendances nécessaires aux tests."""

    timestamp = datetime.now()

    hostgroup = HostGroup(

        name=u'foo',

    )

    DBSession.add(hostgroup)

    host = Host(

        name=u'bar',

        checkhostcmd=u'',

        description=u'',

        hosttpl=u'',

        mainip=u'127.0.0.1',

        snmpport=42,

        snmpcommunity=u'public',

        snmpversion=u'3',

        weight=42,

    )

    DBSession.add(host)

    DBSession.flush()

    hostgroup.hosts.append(host)

    DBSession.flush()

    servicegroup = ServiceGroup(

        name=u'foo',

    )

    DBSession.add(servicegroup)

    service = LowLevelService(

        host=host,

        command=u'',

        weight=42,

        servicename=u'baz',

        op_dep=u'&',

    )

    DBSession.add(service)

    DBSession.flush()

    servicegroup.services.append(service)

    DBSession.flush()

    event = Event(

        timestamp=timestamp,

        current_state=StateName.statename_to_value(u'WARNING'),

        message=u'Hello world',

    )

    if return_service:

        event.supitem = service

    else:

        event.supitem = host

    DBSession.add(event)

    DBSession.flush()

    correvent = CorrEvent(

        impact=42,

        priority=42,

        trouble_ticket=None,

        status=u'None',

        occurrence=42,

        timestamp_active=timestamp,

        cause=event,

    )

    correvent.events.append(event)

    DBSession.add(correvent)

    DBSession.flush()

    return (hostgroup, correvent.idcorrevent)

class TestHistoryForm(TestController):

    """Teste le dialogue pour l'accès aux historiques."""

    def test_history_form_LLS_alert_when_allowed(self):

        """Teste le formulaire d'historique avec un LLS (alerte visible)."""

        hostgroup, idcorrevent = insert_deps(True)

        manage = Permission.by_permission_name(u'manage')

        manage.hostgroups.append(hostgroup)

        DBSession.flush()

        transaction.commit()

        response = self.app.post('/get_plugin_value', {

                'idcorrevent': idcorrevent,

                'plugin_name': 'history',

            }, extra_environ={'REMOTE_USER': 'manager'})

        json = response.json

        # Le contenu de "eventdetails" varie facilement.

        # On le teste séparément.

        json.pop('eventdetails', None)

        assert_true('eventdetails' in response.json)

        assert_equal(json, {

            "idcorrevent": idcorrevent,

            "service": "baz",

            "peak_state": "WARNING",

            "current_state": "WARNING",

            "host": "bar",

            "initial_state": "WARNING"

        })

    def test_history_form_host_alert_when_allowed(self):

        """Teste le formulaire d'historique avec un hôte (alerte visible)."""

        hostgroup, idcorrevent = insert_deps(False)

        manage = Permission.by_permission_name(u'manage')

        manage.hostgroups.append(hostgroup)

        DBSession.flush()

        transaction.commit()

        response = self.app.post('/get_plugin_value', {

                'idcorrevent': idcorrevent,

                'plugin_name': 'history',

            }, extra_environ={'REMOTE_USER': 'manager'})

        json = response.json

        # Le contenu de "eventdetails" varie facilement.

        # On le teste séparément.

        json.pop('eventdetails', None)

        assert_true('eventdetails' in response.json)

        assert_equal(json, {

            "idcorrevent": idcorrevent,

            "service": None,

            "peak_state": "WARNING",

            "current_state": "WARNING",

            "host": "bar",

            "initial_state": "WARNING"

        })

    def test_history_form_LLS_when_forbidden(self):

        """Teste le formulaire d'historique avec un LLS (alerte invisible)."""

        idcorrevent = insert_deps(True)[1]

        DBSession.flush()

        transaction.commit()

        self.app.post('/get_plugin_value', {

                'idcorrevent': idcorrevent,

                'plugin_name': 'history',

            }, extra_environ={'REMOTE_USER': 'manager'},

            status=302)

    def test_history_form_host_when_forbidden(self):

        """Teste le formulaire d'historique avec un hôte (alerte invisible)."""

        idcorrevent = insert_deps(False)[1]

        DBSession.flush()

        transaction.commit()

        self.app.post('/get_plugin_value', {

                'idcorrevent': idcorrevent,

                'plugin_name': 'history',

            }, extra_environ={'REMOTE_USER': 'manager'},

            status=302)

# -*- coding: utf-8 -*-

# vim:set expandtab tabstop=4 shiftwidth=4:

"""

Test du tableau d'événements de Vigiboard

"""

from nose.tools import assert_true, assert_equal

from datetime import datetime

import transaction

from vigilo.models.configure import DBSession

from vigilo.models import Event, EventHistory, CorrEvent, \

                            Permission, StateName, \

                            Host, HostGroup, LowLevelService, ServiceGroup

from vigiboard.tests import TestController

def populate_DB():

    """ Peuple la base de données. """

    # On ajoute un groupe d'hôtes et un groupe de services.

    hostmanagers = HostGroup(name = u'managersgroup')

    DBSession.add(hostmanagers)

    servicemanagers = ServiceGroup(name = u'managersgroup')

    DBSession.add(servicemanagers)

    DBSession.flush()

    # On ajoute la permission 'manage' à ces deux groupes.

    manage_perm = Permission.by_permission_name(u'manage')

    hostmanagers.permissions.append(manage_perm)

    servicemanagers.permissions.append(manage_perm)

    DBSession.flush()

    # On crée un hôte de test, et on l'ajoute au groupe d'hôtes.

    managerhost = Host(

        name = u'managerhost',      

        checkhostcmd = u'halt',

        snmpcommunity = u'public',

        hosttpl = u'/dev/null',

        mainip = u'192.168.1.1',

        snmpport = 42,

        weight = 42,

    )

    DBSession.add(managerhost)

    hostmanagers.hosts.append(managerhost)

    DBSession.flush()

    # On crée un services de bas niveau, et on l'ajoute au groupe de services.

    managerservice = LowLevelService(

        host = managerhost,

        servicename = u'managerservice',

        command = u'halt',

        op_dep = u'+',

        weight = 42,

    )

    DBSession.add(managerservice)

    servicemanagers.services.append(managerservice)

    DBSession.flush()

    return (managerhost, managerservice)

def add_correvent_caused_by(supitem, 

        correvent_status=u"None", event_status=u"WARNING"):

    """

    Ajoute dans la base de données un évènement corrélé causé 

    par un incident survenu sur l'item passé en paramètre.

    Génère un historique pour les tests.

    """

    # Ajout d'un événement

    event = Event(

        supitem = supitem, 

        message = u'foo',

        current_state = StateName.statename_to_value(event_status),

    )

    DBSession.add(event)

    DBSession.flush()

    # Ajout des historiques

    DBSession.add(EventHistory(

        type_action=u'Nagios update state',

        idevent=event.idevent, 

        timestamp=datetime.now()))

    DBSession.add(EventHistory(

        type_action=u'Acknowlegement change state',

        idevent=event.idevent, 

        timestamp=datetime.now()))

    DBSession.flush()

    # Ajout d'un événement corrélé

    aggregate = CorrEvent(

        idcause = event.idevent, 

        timestamp_active = datetime.now(),

        priority = 1,

        status = correvent_status)

    aggregate.events.append(event)

    DBSession.add(aggregate)

    DBSession.flush()

    return aggregate.idcorrevent

class TestEventTable(TestController):

    """

    Test des historiques de Vigiboard.

    """

    def setUp(self):

        super(TestEventTable, self).setUp()

    def test_host_event_history(self):

        """Affichage de l'historique d'un évènement corrélé pour un hôte."""

        # On peuple la BDD avec un hôte, un service de bas niveau,

        # et un groupe d'hôtes et de services associés à ces items.

        (managerhost, managerservice) = populate_DB()

        # On ajoute un évènement corrélé causé par l'hôte

        aggregate_id = add_correvent_caused_by(managerhost)

        transaction.commit()

        ### 1er cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'editor'.

        environ = {'REMOTE_USER': 'editor'}

        # On s'attend à ce qu'une erreur 302 soit renvoyée, et à

        # ce qu'un message d'erreur précise à l'utilisateur qu'il

        # n'a pas accès aux informations concernant cet évènement.

        response = self.app.get(

            '/event?idcorrevent=' + str(aggregate_id),

            status = 302, 

            extra_environ = environ)

        response = self.app.get(

            '/', 

            status = 200, 

            extra_environ = environ)

        assert_true(response.lxml.xpath(

            '//div[@id="flash"]/div[@class="error"]'))

        ### 2nd cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'manager'.

        environ = {'REMOTE_USER': 'manager'}

        # On s'attend à ce que le statut de la requête soit 200.

        response = self.app.get(

            '/event?idcorrevent=' + str(aggregate_id),

            status = 200, 

            extra_environ = environ)

        # Il doit y avoir 2 lignes de résultats.

        rows = response.lxml.xpath('//table[@class="history_table"]/tbody/tr')

        assert_equal(len(rows), 2)

    def test_service_event_history(self):

        """Affichage de l'historique d'un évènement corrélé pour un SBN."""

        # On peuple la BDD avec un hôte, un service de bas niveau,

        # et un groupe d'hôtes et de services associés à ces items.

        (managerhost, managerservice) = populate_DB()

        # On ajoute un évènement corrélé causé par le service

        aggregate_id = add_correvent_caused_by(managerservice)

        transaction.commit()

        ### 1er cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'editor'.

        environ = {'REMOTE_USER': 'editor'}

        # On s'attend à ce qu'une erreur 302 soit renvoyée, et à

        # ce qu'un message d'erreur précise à l'utilisateur qu'il

        # n'a pas accès aux informations concernant cet évènement.

        response = self.app.get(

            '/event?idcorrevent=' + str(aggregate_id),

            status = 302, 

            extra_environ = environ)

        response = self.app.get(

            '/', 

            status = 200, 

            extra_environ = environ)

        assert_true(response.lxml.xpath(

            '//div[@id="flash"]/div[@class="error"]'))

        ### 2nd cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'manager'.

        environ = {'REMOTE_USER': 'manager'}

        # On s'attend à ce que le statut de la requête soit 200.

        response = self.app.get(

            '/event?idcorrevent=' + str(aggregate_id),

            status = 200, 

            extra_environ = environ)

        # Il doit y avoir 2 lignes de résultats.

        rows = response.lxml.xpath('//table[@class="history_table"]/tbody/tr')

        assert_equal(len(rows), 2)

    def test_host_history(self):

        """Affichage de l'historique d'un hôte."""

        # On peuple la BDD avec un hôte, un service de bas niveau,

        # et un groupe d'hôtes et de services associés à ces items.

        (managerhost, managerservice) = populate_DB()

        # On ajoute deux évènements corrélés causés par l'hôte :

        # le premier encore ouvert, le second clos par un utilisateur.

        add_correvent_caused_by(managerhost)

        add_correvent_caused_by(managerhost, u"AAClosed", u"OK")

        transaction.commit()

        DBSession.add(managerhost)

        ### 1er cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'editor'.

        environ = {'REMOTE_USER': 'editor'}

        # On s'attend à ce qu'une erreur 302 soit renvoyée, et à

        # ce qu'un message d'erreur précise à l'utilisateur qu'il

        # n'a pas accès aux informations concernant cet évènement.

        response = self.app.get(

            '/host_service/' + managerhost.name,

            status = 302, 

            extra_environ = environ)

        response = self.app.get(

            '/', 

            status = 200, 

            extra_environ = environ)

        assert_true(response.lxml.xpath(

            '//div[@id="flash"]/div[@class="error"]'))

        ### 2nd cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'manager'.

        environ = {'REMOTE_USER': 'manager'}

        # On s'attend à ce que le statut de la requête soit 200.

        response = self.app.get(

            '/host_service/' + managerhost.name,

            status = 200, 

            extra_environ = environ)

        # Il doit y avoir 2 lignes d'évènements 

        # + 2 lignes contenant les tableaux d'historiques.

        rows = response.lxml.xpath('//table[@class="vigitable"]/tbody/tr')

        assert_equal(len(rows), 2 + 2)

        # Et 4 lignes d'historiques dans les tableaux d'historiques.

        rows = response.lxml.xpath('//table[@class="history_table"]/tbody/tr')

        assert_equal(len(rows), 4)

    def test_service_history(self):

        """Affichage de l'historique d'un service de bas niveau."""

        # On peuple la BDD avec un hôte, un service de bas niveau,

        # et un groupe d'hôtes et de services associés à ces items.

        (managerhost, managerservice) = populate_DB()

        # On ajoute deux évènements corrélés causés par le service :

        # le premier encore ouvert, le second clos par un utilisateur.

        add_correvent_caused_by(managerservice)

        add_correvent_caused_by(managerservice, u"AAClosed", u"OK")

        transaction.commit()

        DBSession.add(managerhost)

        DBSession.add(managerservice)

        ### 1er cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'editor'.

        environ = {'REMOTE_USER': 'editor'}

        # On s'attend à ce qu'une erreur 302 soit renvoyée, et à

        # ce qu'un message d'erreur précise à l'utilisateur qu'il

        # n'a pas accès aux informations concernant cet évènement.

        response = self.app.get(

            '/host_service/' + managerhost.name 

                                + '/' + managerservice.servicename,

            status = 302, 

            extra_environ = environ)

        response = self.app.get(

            '/', 

            status = 200, 

            extra_environ = environ)

        assert_true(response.lxml.xpath(

            '//div[@id="flash"]/div[@class="error"]'))

        ### 2nd cas : L'utilisateur utilisé pour

        # se connecter à Vigiboard est 'manager'.

        environ = {'REMOTE_USER': 'manager'}

        # On s'attend à ce que le statut de la requête soit 200.

        response = self.app.get(

            '/host_service/' + managerhost.name 

                                + '/' + managerservice.servicename,

            status = 200, 

            extra_environ = environ)

        # Il doit y avoir 2 lignes d'évènements 

        # + 2 lignes contenant les tableaux d'historiques.

        rows = response.lxml.xpath('//table[@class="vigitable"]/tbody/tr')

        assert_equal(len(rows), 2 + 2)

        # Et 4 lignes d'historiques dans les tableaux d'historiques.

        rows = response.lxml.xpath('//table[@class="history_table"]/tbody/tr')

        assert_equal(len(rows), 4)

# -*- coding: utf-8 -*-

# vim:set expandtab tabstop=4 shiftwidth=4:

"""

Test de la classe Vigiboard Request pour des requêtes concernant les hôtes

"""

from nose.tools import assert_true

from datetime import datetime

import tg

import transaction

from vigilo.models.configure import DBSession

from vigilo.models import Event, EventHistory, CorrEvent, \

                            Permission, User, StateName, \

                            Host, HostGroup

from vigiboard.tests import TestController

from vigiboard.controllers.vigiboardrequest import VigiboardRequest

class TestHostVigiboardRequest(TestController):

    """ Préparation de la base de données en vue des tests. """

    def setUp(self):

        super(TestHostVigiboardRequest, self).setUp()

        # On peuple la base de données.

        # On ajoute les groupes et leurs dépendances

        self.hosteditors = HostGroup(name=u'editorsgroup')

        DBSession.add(self.hosteditors)

        self.hostmanagers = HostGroup(name=u'managersgroup', 

                                        parent=self.hosteditors)

        DBSession.add(self.hostmanagers)

        DBSession.flush()

        manage_perm = Permission.by_permission_name(u'manage')

        edit_perm = Permission.by_permission_name(u'edit')

        self.hostmanagers.permissions.append(manage_perm)

        self.hosteditors.permissions.append(edit_perm)

        DBSession.flush()

        # Création des hôtes de test.

        host_template = {

            'checkhostcmd': u'halt',

            'snmpcommunity': u'public',

            'hosttpl': u'/dev/null',

            'mainip': u'192.168.1.1',

            'snmpport': 42,

            'weight': 42,

        }

        managerhost = Host(name=u'managerhost', **host_template)

        editorhost = Host(name=u'editorhost', **host_template)

        DBSession.add(managerhost)

        DBSession.add(editorhost)

        # Affectation des hôtes aux groupes.

        self.hosteditors.hosts.append(editorhost)

        self.hostmanagers.hosts.append(managerhost)

        DBSession.flush()

        # Ajout des événements eux-mêmes

        event_template = {

            'message': u'foo',

            'current_state': StateName.statename_to_value(u'WARNING'),

        }

        event1 = Event(supitem=managerhost, **event_template)

        event2 = Event(supitem=editorhost, **event_template)

        DBSession.add(event1)

        DBSession.add(event2)

        DBSession.flush()

        # Ajout des historiques

        DBSession.add(EventHistory(type_action=u'Nagios update state',

            idevent=event1.idevent, timestamp=datetime.now()))

        DBSession.add(EventHistory(type_action=u'Acknowledgement change state',

            idevent=event1.idevent, timestamp=datetime.now()))

        DBSession.add(EventHistory(type_action=u'Nagios update state',

            idevent=event2.idevent, timestamp=datetime.now()))

        DBSession.add(EventHistory(type_action=u'Acknowledgement change state',

            idevent=event2.idevent, timestamp=datetime.now()))

        DBSession.flush()

        # Ajout des événements corrélés

        aggregate_template = {

            'timestamp_active': datetime.now(),

            'priority': 1,

            'status': u'None',

        }

        self.aggregate1 = CorrEvent(

            idcause=event1.idevent, **aggregate_template)

        self.aggregate2 = CorrEvent(

            idcause=event2.idevent, **aggregate_template)

        self.aggregate1.events.append(event1)

        self.aggregate2.events.append(event2)

        DBSession.add(self.aggregate1)

        DBSession.add(self.aggregate2)

        DBSession.flush()

        transaction.commit()

    def tearDown(self):

        """ Nettoyage de la base de données après les tests. """

        super(TestHostVigiboardRequest, self).tearDown()

    def test_request_creation(self):

        """Génération d'une requête avec plugin et permissions."""

        # On indique qui on est et on envoie une requête sur 

        # l'index pour obtenir toutes les variables de sessions.

        environ = {'REMOTE_USER': 'editor'}

        response = self.app.get('/', extra_environ=environ)

        tg.request = response.request

        # Derrière, VigiboardRequest doit charger le plugin de tests tout seul

        vigi_req = VigiboardRequest(User.by_user_name(u'editor'))

        vigi_req.add_table(

            CorrEvent,

            vigi_req.items.c.hostname,

            vigi_req.items.c.servicename,

        )

        vigi_req.add_join((Event, CorrEvent.idcause == Event.idevent))

        vigi_req.add_join((vigi_req.items, 

            Event.idsupitem == vigi_req.items.c.idsupitem))

        # On vérifie que le nombre d'événements corrélés 

        # trouvés par la requête est bien égal à 1.

        num_rows = vigi_req.num_rows()

        assert_true(num_rows == 1, 

            msg = "One history should be available for " +

            "the user 'editor' but there are %d" % num_rows)

        vigi_req.format_events(0, 10)

        vigi_req.format_history()

        assert_true(len(vigi_req.events) == 1 + 1,

            msg = "One history should be available for user " +

            "'editor' but there are %d" % (len(vigi_req.events) - 1))

        # On recommence les tests précédents avec l'utilisateur

        # manager (qui dispose de plus de droits).

        environ = {'REMOTE_USER': 'manager'}

        response = self.app.get('/', extra_environ=environ)

        tg.request = response.request

        vigi_req = VigiboardRequest(User.by_user_name(u'manager'))

        vigi_req.add_plugin(MonPlugin)

        vigi_req.add_table(

            CorrEvent,

            vigi_req.items.c.hostname,

            vigi_req.items.c.servicename,

        )

        vigi_req.add_join((Event, CorrEvent.idcause == Event.idevent))

        vigi_req.add_join((vigi_req.items, 

            Event.idsupitem == vigi_req.items.c.idsupitem))

        # On vérifie que le nombre d'événements corrélés 

        # trouvés par la requête est bien égal à 2.

        num_rows = vigi_req.num_rows()

        assert_true(num_rows == 2,  

            msg = "2 histories should be available for " +

            "the user 'manager' but there are %d" % num_rows)

        vigi_req.format_events(0, 10)

        vigi_req.format_history()

        assert_true(len(vigi_req.events) == 2 + 1,

            msg = "2 histories should be available for user " +