test_related_action.py 6.25 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 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184
# -*- coding: utf-8 -*-

import mock
import unittest

import openerp.tests.common as common
from ..connector import Binder
from ..queue.job import (Job,
                         OpenERPJobStorage,
                         related_action)
from ..session import ConnectorSession
from ..related_action import unwrap_binding


def task_no_related(session, model_name):
    pass


def task_related_none(session, model_name):
    pass


def task_related_return(session, model_name):
    pass


def task_related_return_kwargs(session, model_name):
    pass


def open_url(session, job, url=None):
    subject = job.args[0]
    return {
        'type': 'ir.actions.act_url',
        'target': 'new',
        'url': url.format(subject=subject),
    }


@related_action(action=open_url, url='https://en.wikipedia.org/wiki/{subject}')
def task_wikipedia(session, subject):
    pass


@related_action(action=unwrap_binding)
def try_unwrap_binding(session, model_name, binding_id):
    pass


class test_related_action(unittest.TestCase):
    """ Test Related Actions """

    def setUp(self):
        super(test_related_action, self).setUp()
        self.session = mock.MagicMock()

    def test_no_related_action(self):
        """ Job without related action """
        job = Job(func=task_no_related)
        self.assertIsNone(job.related_action(self.session))

    def test_return_none(self):
        """ Job with related action returning None """
        # default action returns None
        job = Job(func=related_action()(task_related_none))
        self.assertIsNone(job.related_action(self.session))

    def test_return(self):
        """ Job with related action check if action returns correctly """
        def action(session, job):
            return session, job
        job = Job(func=related_action(action=action)(task_related_return))
        act_session, act_job = job.related_action(self.session)
        self.assertEqual(act_session, self.session)
        self.assertEqual(act_job, job)

    def test_kwargs(self):
        """ Job with related action check if action propagates kwargs """
        def action(session, job, a=1, b=2):
            return a, b
        task = task_related_return_kwargs
        job_func = related_action(action=action, b=4)(task)
        job = Job(func=job_func)
        self.assertEqual(job.related_action(self.session), (1, 4))

    def test_unwrap_binding(self):
        """ Call the unwrap binding related action """
        class TestBinder(Binder):
            _model_name = 'binding.res.users'

            def unwrap_binding(self, binding_id, browse=False):
                return 42

            def unwrap_model(self):
                return 'res.users'

        job = Job(func=try_unwrap_binding, args=('res.users', 555))
        session = mock.MagicMock(name='session')
        backend_record = mock.Mock(name='backend_record')
        backend = mock.Mock(name='backend')
        browse_record = mock.Mock(name='browse_record')
        backend.get_class.return_value = TestBinder
        backend_record.get_backend.return_value = backend
        browse_record.exists.return_value = True
        browse_record.backend_id = backend_record
        recordset = mock.Mock(name='recordset')
        session.env.__getitem__.return_value = recordset
        recordset.browse.return_value = browse_record
        action = unwrap_binding(session, job)
        expected = {
            'name': mock.ANY,
            'type': 'ir.actions.act_window',
            'view_type': 'form',
            'view_mode': 'form',
            'res_id': 42,
            'res_model': 'res.users',
        }
        self.assertEquals(action, expected)

    def test_unwrap_binding_direct_binding(self):
        """ Call the unwrap binding related action """
        class TestBinder(Binder):
            _model_name = 'res.users'

            def unwrap_binding(self, binding_id, browse=False):
                raise ValueError('Not an inherits')

            def unwrap_model(self):
                raise ValueError('Not an inherits')

        job = Job(func=try_unwrap_binding, args=('res.users', 555))
        session = mock.MagicMock(name='session')
        backend_record = mock.Mock(name='backend_record')
        backend = mock.Mock(name='backend')
        browse_record = mock.Mock(name='browse_record')
        backend.get_class.return_value = TestBinder
        backend_record.get_backend.return_value = backend
        browse_record.exists.return_value = True
        browse_record.backend_id = backend_record
        recordset = mock.Mock(name='recordset')
        session.env.__getitem__.return_value = recordset
        recordset.browse.return_value = browse_record
        action = unwrap_binding(session, job)
        expected = {
            'name': mock.ANY,
            'type': 'ir.actions.act_window',
            'view_type': 'form',
            'view_mode': 'form',
            'res_id': 555,
            'res_model': 'res.users',
        }
        self.assertEquals(action, expected)


class test_related_action_storage(common.TransactionCase):
    """ Test related actions on stored jobs """

    def setUp(self):
        super(test_related_action_storage, self).setUp()
        self.session = ConnectorSession(self.cr, self.uid)
        self.queue_job = self.env['queue.job']

    def test_store_related_action(self):
        """ Call the related action on the model """
        job = Job(func=task_wikipedia, args=('Discworld',))
        storage = OpenERPJobStorage(self.session)
        storage.store(job)
        stored_job = self.queue_job.search([('uuid', '=', job.uuid)])
        self.assertEqual(len(stored_job), 1)
        expected = {'type': 'ir.actions.act_url',
                    'target': 'new',
                    'url': 'https://en.wikipedia.org/wiki/Discworld',
                    }
        self.assertEquals(stored_job.open_related_action(), expected)

    def test_unwrap_binding_not_exists(self):
        """ Call the related action on the model on non-existing record """
        job = Job(func=try_unwrap_binding, args=('res.users', 555))
        storage = OpenERPJobStorage(self.session)
        storage.store(job)
        stored_job = self.queue_job.search([('uuid', '=', job.uuid)])
        stored_job.unlink()
        self.assertFalse(stored_job.exists())
        self.assertEquals(unwrap_binding(self.session, job), None)