[MIG] test_workflow: convert code to the new API

This commit is contained in:
Raphael Collet
2016-08-10 13:14:16 +02:00
parent b225d1119b
commit 8acfb3afcd
5 changed files with 112 additions and 159 deletions
+3 -1
View File
@@ -1,2 +1,4 @@
# -*- coding: utf-8 -*-
import models
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from . import models
@@ -1,4 +1,5 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
{
'name': 'test-workflow',
'version': '0.1',
+40 -29
View File
@@ -1,70 +1,81 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import logging
import openerp.osv.orm
from odoo import models, fields, api
from odoo.workflow import trg_trigger
_logger = logging.getLogger(__name__)
class m(openerp.osv.orm.Model):
class m(models.Model):
""" A model for which we will define a workflow (see data.xml). """
_name = 'test.workflow.model'
def print_(self, cr, uid, ids, s, context=None):
_logger.info('Running activity `%s` for record %s', s, ids)
@api.multi
def _print(self, s):
_logger.info('Running activity `%s` for record %s', s, self.ids)
return True
def print_a(self, cr, uid, ids, context=None):
return self.print_(cr, uid, ids, 'a', context)
@api.multi
def print_a(self):
return self._print('a')
def print_b(self, cr, uid, ids, context=None):
return self.print_(cr, uid, ids, 'b', context)
@api.multi
def print_b(self):
return self._print('b')
def print_c(self, cr, uid, ids, context=None):
return self.print_(cr, uid, ids, 'c', context)
@api.multi
def print_c(self):
return self._print('c')
def condition(self, cr, uid, ids, context=None):
m = self.pool['test.workflow.trigger']
for r in m.browse(cr, uid, [1], context=context):
if not r.value:
return False
return True
@api.multi
def condition(self):
record = self.env['test.workflow.trigger'].browse(1)
return bool(record.value)
def trigger(self, cr, uid, context=None):
return openerp.workflow.trg_trigger(uid, 'test.workflow.trigger', 1, cr)
@api.model
def trigger(self):
return trg_trigger(self._uid, 'test.workflow.trigger', 1, self._cr)
class n(openerp.osv.orm.Model):
class n(models.Model):
""" A model used for the trigger feature. """
_name = 'test.workflow.trigger'
_columns = { 'value': openerp.osv.fields.boolean('Value') }
_defaults = { 'value': False }
value = fields.Boolean(default=False)
class a(openerp.osv.orm.Model):
class a(models.Model):
_name = 'test.workflow.model.a'
_columns = { 'value': openerp.osv.fields.integer('Value') }
_defaults = { 'value': 0 }
value = fields.Integer(default=0)
class b(openerp.osv.orm.Model):
class b(models.Model):
_name = 'test.workflow.model.b'
_inherit = 'test.workflow.model.a'
class c(openerp.osv.orm.Model):
class c(models.Model):
_name = 'test.workflow.model.c'
_inherit = 'test.workflow.model.a'
class d(openerp.osv.orm.Model):
class d(models.Model):
_name = 'test.workflow.model.d'
_inherit = 'test.workflow.model.a'
class e(openerp.osv.orm.Model):
class e(models.Model):
_name = 'test.workflow.model.e'
_inherit = 'test.workflow.model.a'
for name in 'bcdefghijkl':
#
# Do not use type() to create the class here, but use the class construct.
# This is because the __module__ of the new class would be the one of the
# metaclass that provides method __new__!
#
class NewModel(openerp.osv.orm.Model):
class NewModel(models.Model):
_name = 'test.workflow.model.%s' % name
_inherit = 'test.workflow.model.a'
@@ -1,4 +1,5 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from . import test_workflow
@@ -1,183 +1,121 @@
# -*- coding: utf-8 -*-
import openerp
from openerp import SUPERUSER_ID
from openerp.tests import common
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo.tests import common
class test_workflows(common.TransactionCase):
def check_activities(self, model_name, i, names):
""" Check that the record i has workitems in the given activity names.
def check_activities(self, record, names):
""" Check that the record has workitems in the given activity names.
"""
instance = self.registry('workflow.instance')
workitem = self.registry('workflow.workitem')
Instance = self.env['workflow.instance']
Workitem = self.env['workflow.workitem']
# Given the workflow instance associated to the record ...
instance_id = instance.search(
self.cr, SUPERUSER_ID,
[('res_type', '=', model_name), ('res_id', '=', i)])
self.assertTrue( instance_id, 'A workflow instance is expected.')
instance = Instance.search([('res_type', '=', record._name), ('res_id', '=', record.id)])
self.assertTrue(instance, 'A workflow instance is expected.')
# ... get all its workitems ...
workitem_ids = workitem.search(
self.cr, SUPERUSER_ID,
[('inst_id', '=', instance_id[0])])
self.assertTrue(
workitem_ids,
'The workflow instance should have workitems.')
workitems = Workitem.search([('inst_id', '=', instance.id)])
self.assertTrue(workitems, 'The workflow instance should have workitems.')
# ... and check the activity the are in against the provided names.
workitem_records = workitem.browse(
self.cr, SUPERUSER_ID, workitem_ids)
self.assertEqual(
sorted([item.act_id.name for item in workitem_records]),
sorted([item.act_id.name for item in workitems]),
sorted(names))
def check_value(self, model_name, i, value):
""" Check that the record i has the given value.
def check_value(self, record, value):
""" Check that the record has the given value.
"""
model = self.registry(model_name)
record = model.read(self.cr, SUPERUSER_ID, [i], ['value'])[0]
self.assertEqual(record['value'], value)
self.assertEqual(record.value, value)
def test_workflow(self):
model = self.registry('test.workflow.model')
trigger = self.registry('test.workflow.trigger')
model = self.env['test.workflow.model']
trigger = self.env['test.workflow.trigger']
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['a'])
record = model.create({})
self.check_activities(record, ['a'])
# a -> b is just a signal.
model.signal_workflow(self.cr, SUPERUSER_ID, [i], 'a-b')
self.check_activities(model._name, i, ['b'])
record.signal_workflow('a-b')
self.check_activities(record, ['b'])
# b -> c is a trigger (which is False),
# so we remain in the b activity.
model.trigger(self.cr, SUPERUSER_ID)
self.check_activities(model._name, i, ['b'])
record.trigger()
self.check_activities(record, ['b'])
# b -> c is a trigger (which is set to True).
# so we go in c when the trigger is called.
trigger.write(self.cr, SUPERUSER_ID, [1], {'value': True})
model.trigger(self.cr, SUPERUSER_ID)
self.check_activities(model._name, i, ['c'])
trigger.browse(1).write({'value': True})
record.trigger()
self.check_activities(record, ['c'])
self.assertEqual(
True,
True)
model.unlink(self.cr, SUPERUSER_ID, [i])
self.assertEqual(True, True)
def test_workflow_a(self):
model = self.registry('test.workflow.model.a')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['a'])
self.check_value(model._name, i, 0)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.a'].create({})
self.check_activities(record, ['a'])
self.check_value(record, 0)
def test_workflow_b(self):
model = self.registry('test.workflow.model.b')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['a'])
self.check_value(model._name, i, 1)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.b'].create({})
self.check_activities(record, ['a'])
self.check_value(record, 1)
def test_workflow_c(self):
model = self.registry('test.workflow.model.c')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['a'])
self.check_value(model._name, i, 0)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.c'].create({})
self.check_activities(record, ['a'])
self.check_value(record, 0)
def test_workflow_d(self):
model = self.registry('test.workflow.model.d')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['a'])
self.check_value(model._name, i, 1)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.d'].create({})
self.check_activities(record, ['a'])
self.check_value(record, 1)
def test_workflow_e(self):
model = self.registry('test.workflow.model.e')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['b'])
self.check_value(model._name, i, 2)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.e'].create({})
self.check_activities(record, ['b'])
self.check_value(record, 2)
def test_workflow_f(self):
model = self.registry('test.workflow.model.f')
record = self.env['test.workflow.model.f'].create({})
self.check_activities(record, ['a'])
self.check_value(record, 1)
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['a'])
self.check_value(model._name, i, 1)
model.signal_workflow(self.cr, SUPERUSER_ID, [i], 'a-b')
self.check_activities(model._name, i, ['b'])
self.check_value(model._name, i, 2)
model.unlink(self.cr, SUPERUSER_ID, [i])
record.signal_workflow('a-b')
self.check_activities(record, ['b'])
self.check_value(record, 2)
def test_workflow_g(self):
model = self.registry('test.workflow.model.g')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['a'])
self.check_value(model._name, i, 1)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.g'].create({})
self.check_activities(record, ['a'])
self.check_value(record, 1)
def test_workflow_h(self):
model = self.registry('test.workflow.model.h')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['b', 'c'])
self.check_value(model._name, i, 2)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.h'].create({})
self.check_activities(record, ['b', 'c'])
self.check_value(record, 2)
def test_workflow_i(self):
model = self.registry('test.workflow.model.i')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['b'])
self.check_value(model._name, i, 2)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.i'].create({})
self.check_activities(record, ['b'])
self.check_value(record, 2)
def test_workflow_j(self):
model = self.registry('test.workflow.model.j')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['a'])
self.check_value(model._name, i, 1)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.j'].create({})
self.check_activities(record, ['a'])
self.check_value(record, 1)
def test_workflow_k(self):
model = self.registry('test.workflow.model.k')
i = model.create(self.cr, SUPERUSER_ID, {})
record = self.env['test.workflow.model.k'].create({})
# Non-determinisitic: can be b or c
# self.check_activities(model._name, i, ['b'])
# self.check_activities(model._name, i, ['c'])
self.check_value(model._name, i, 2)
model.unlink(self.cr, SUPERUSER_ID, [i])
# self.check_activities(record, ['b'])
# self.check_activities(record, ['c'])
self.check_value(record, 2)
def test_workflow_l(self):
model = self.registry('test.workflow.model.l')
i = model.create(self.cr, SUPERUSER_ID, {})
self.check_activities(model._name, i, ['c', 'c', 'd'])
self.check_value(model._name, i, 3)
model.unlink(self.cr, SUPERUSER_ID, [i])
record = self.env['test.workflow.model.l'].create({})
self.check_activities(record, ['c', 'c', 'd'])
self.check_value(record, 3)