test_event.py 5.7 KB
Newer Older
François C. committed
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 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156
# -*- coding: utf-8 -*-

import mock

import openerp.tests.common as common
from openerp.addons.connector.event import Event
from openerp.addons.connector.session import ConnectorSession


class test_event(common.TransactionCase):
    """ Test Event """

    def setUp(self):
        super(test_event, self).setUp()
        self.consumer1 = lambda session, model_name: None
        self.consumer2 = lambda session, model_name: None
        self.event = Event()
        self.session = ConnectorSession(self.cr,
                                        self.uid)

    def test_subscribe(self):
        self.event.subscribe(self.consumer1)
        self.assertIn(self.consumer1, self.event._consumers[None])

    def test_subscribe_decorator(self):
        @self.event
        def consumer():
            pass
        self.assertIn(consumer, self.event._consumers[None])

    def test_subscribe_model(self):
        self.event.subscribe(self.consumer1, model_names=['res.users'])
        self.assertIn(self.consumer1, self.event._consumers['res.users'])

    def test_subscribe_decorator_model(self):
        @self.event(model_names=['res.users'])
        def consumer():
            pass
        self.assertIn(consumer, self.event._consumers['res.users'])

    def test_unsubscribe(self):
        self.event.subscribe(self.consumer1)
        self.event.unsubscribe(self.consumer1)
        self.assertNotIn(self.consumer1, self.event._consumers[None])

    def test_unsubscribe_model(self):
        self.event.subscribe(self.consumer1, model_names=['res.users'])
        self.event.unsubscribe(self.consumer1, model_names=['res.users'])
        self.assertNotIn(self.consumer1, self.event._consumers['res.users'])

    def test_unsubscribe_not_existing(self):
        """ Discard without error """
        self.event.unsubscribe(self.consumer1)

    def test_unsubscribe_not_existing_model(self):
        """ Discard without error """
        self.event.unsubscribe(self.consumer1, model_names=['res.users'])

    def test_replacing(self):
        self.event.subscribe(self.consumer1)
        self.event.subscribe(self.consumer2, replacing=self.consumer1)
        self.assertNotIn(self.consumer1, self.event._consumers[None])
        self.assertIn(self.consumer2, self.event._consumers[None])

    def test_replacing_decorator(self):
        @self.event
        def consumer1(session, model_name):
            pass

        @self.event(replacing=consumer1)
        def consumer2(session, model_name):
            pass
        self.assertNotIn(consumer1, self.event._consumers[None])
        self.assertIn(consumer2, self.event._consumers[None])

    def test_replacing_model(self):
        self.event.subscribe(self.consumer1, model_names=['res.users'])
        self.event.subscribe(self.consumer2, replacing=self.consumer1,
                             model_names=['res.users'])
        self.assertNotIn(self.consumer1, self.event._consumers['res.users'])
        self.assertIn(self.consumer2, self.event._consumers['res.users'])

    def test_fire(self):
        """ Fire a consumer """
        class Recipient(object):
            def __init__(self):
                self.message = None

            def set_message(self, message):
                self.message = message

        @self.event
        def set_message(session, model_name, recipient, message):
            recipient.set_message(message)
        recipient = Recipient()
        # an event is fired on a model name
        session = mock.Mock()
        self.event.fire(session, 'res.users', recipient, 'success')
        self.assertEquals(recipient.message, 'success')

    def test_fire_several_consumers(self):
        """ Fire several consumers """
        class Recipient(object):
            def __init__(self):
                self.message = None

            def set_message(self, message):
                self.message = message

        recipient = Recipient()
        recipient2 = Recipient()

        @self.event
        def set_message(session, model_name, message):
            recipient.set_message(message)

        @self.event
        def set_message2(session, model_name, message):
            recipient2.set_message(message)

        # an event is fired on a model name
        session = mock.Mock()
        self.event.fire(session, 'res.users', 'success')
        self.assertEquals(recipient.message, 'success')
        self.assertEquals(recipient2.message, 'success')

    def test_has_consumer_for(self):
        @self.event(model_names=['product.product'])
        def consumer1(session, model_name):
            pass
        self.assertTrue(self.event.has_consumer_for(self.session,
                                                    'product.product'))
        self.assertFalse(self.event.has_consumer_for(self.session,
                                                     'res.partner'))

    def test_has_consumer_for_global(self):
        @self.event
        def consumer1(session, model_name):
            pass
        self.assertTrue(self.event.has_consumer_for(self.session,
                                                    'product.product'))
        self.assertTrue(self.event.has_consumer_for(self.session,
                                                    'res.partner'))

    def test_consumer_uninstalled_module(self):
        """A consumer in a uninstalled module should not be fired"""
        @self.event
        def consumer1(session, model_name):
            pass
        # devious way to test it: the __module__ of a mock is 'mock'
        # and we use __module__ to know the module of the event
        # so, here func is considered to be in a uninstalled module
        func = mock.Mock()
        func.side_effect = Exception('Should not be called')
        self.event(func)
        self.event.fire(self.session, 'res.users')