hoangvv
148 lines
6.8 KiB
Python
148 lines
6.8 KiB
Python
from odoo.tests import common, tagged
|
|
from odoo import exceptions, api, registry
|
|
from odoo.tools import mute_logger
|
|
from psycopg2 import ProgrammingError
|
|
import uuid
|
|
import logging
|
|
from ..models.web_progress import last_report_time
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
@tagged('at_install', '-post_install')
|
|
class WebProgressTest(common.TransactionCase):
|
|
|
|
def check_all_progress_data_empty(self):
|
|
"""
|
|
Check that all global progress data is empty after tests
|
|
"""
|
|
self.assertFalse(last_report_time, msg="Global variable last_report_time shall be empty by now")
|
|
|
|
def setUp(self):
|
|
super(WebProgressTest, self).setUp()
|
|
self.maxDiff = None
|
|
self.partner_obj = self.env['res.partner']
|
|
self.web_progress_obj = self.env['web.progress']
|
|
self.partner_ids = self.partner_obj
|
|
self.partner_vals = {}
|
|
for idx in range(20):
|
|
self.partner_vals[idx] = dict(name='Test{}'.format(idx),
|
|
email='email{}@test.me'.format(idx))
|
|
self.partner_ids |= self.partner_obj.create(dict(self.partner_vals[idx]))
|
|
self.addCleanup(self.check_all_progress_data_empty)
|
|
|
|
def _check_web_progress_iter_recordset(self, total, recur_level=0):
|
|
"""
|
|
Check that web_progress_iter works correctly for a recordset
|
|
:param total: total number of collection elements
|
|
"""
|
|
progress_iter = self.partner_ids[:total].with_progress(msg="Total {} Level {}".format(total,
|
|
recur_level))
|
|
self.assertEqual(len(progress_iter), total, msg="Length shall be accessible")
|
|
if total > 0:
|
|
self.assertEqual(progress_iter[0], self.partner_ids[0], msg="Indexing shall be accessible")
|
|
self.assertEqual(progress_iter._name, self.partner_ids._name, msg="Attributes shall be accessible")
|
|
if total == len(self.partner_ids):
|
|
self.assertEqual(progress_iter.ids, self.partner_ids.ids, msg="Attributes shall be accessible")
|
|
count = 0
|
|
for idx, partner_id in zip(range(total),progress_iter):
|
|
self.assertEqual(partner_id.name, self.partner_vals[idx]['name'].format(idx), msg="Wrong name")
|
|
self.assertEqual(partner_id.email, self.partner_vals[idx]['email'].format(idx), msg="Wrong email")
|
|
count += 1
|
|
if recur_level > 0:
|
|
self._check_web_progress_iter_recordset(total, recur_level - 1)
|
|
self.assertEqual(count, total, msg="Not all elements are yielded from a collection")
|
|
|
|
def _check_web_progress_iter_recordset_many(self, recur_level=0):
|
|
"""
|
|
Iterate recordsets of different lengths
|
|
:param recur_level: recursion level of iterations
|
|
"""
|
|
# iterate all partners
|
|
self._check_web_progress_iter_recordset(len(self.partner_ids), recur_level)
|
|
# iterate half of all partners
|
|
self._check_web_progress_iter_recordset(round(len(self.partner_ids)/2), recur_level)
|
|
# iterate again all partners (no recursion)
|
|
self._check_web_progress_iter_recordset(len(self.partner_ids))
|
|
# iterate one partner
|
|
self._check_web_progress_iter_recordset(1, recur_level)
|
|
# iterate empty recordset
|
|
self._check_web_progress_iter_recordset(0, recur_level)
|
|
|
|
def _check_web_progress_cancelled(self):
|
|
"""
|
|
Checks that the current operation has been cancelled
|
|
"""
|
|
code = self.partner_ids._context.get('progress_code', None)
|
|
self.assertIsNotNone(code, msg="Progress code shall be in the context")
|
|
cancelled = self.web_progress_obj._check_cancelled(dict(code=code))
|
|
self.assertTrue(cancelled, msg="Currect operation should have been cancelled")
|
|
|
|
def test_web_progress_iter_without_web_progress_code(self):
|
|
"""
|
|
Check that web_progress_iter works correctly without a progress_code in context
|
|
"""
|
|
self._check_web_progress_iter_recordset_many(0)
|
|
self._check_web_progress_iter_recordset_many(1)
|
|
|
|
def test_web_progress_iter_with_web_progress_code(self):
|
|
"""
|
|
Check that web_progress_iter works correctly with a progress_code in context
|
|
"""
|
|
progress_code = str(uuid.uuid4())
|
|
self.partner_ids = self.partner_ids.with_context(progress_code=progress_code)
|
|
self._check_web_progress_iter_recordset_many(0)
|
|
self._check_web_progress_iter_recordset_many(1)
|
|
|
|
def test_web_progress_iter_with_web_progress_code_cancel(self):
|
|
"""
|
|
Check that cancel request is respected by web_progress_iter
|
|
"""
|
|
progress_code = str(uuid.uuid4())
|
|
self.partner_ids = self.partner_ids.with_context(progress_code=progress_code)
|
|
self._check_web_progress_iter_recordset_many(0)
|
|
self.partner_ids.web_progress_cancel()
|
|
self._check_web_progress_cancelled()
|
|
# any further iteration shall raise UserError
|
|
with self.assertRaises(exceptions.UserError, msg="Exception UserErro shall have been raised"):
|
|
self._check_web_progress_iter_recordset_many(0)
|
|
self._check_web_progress_cancelled()
|
|
|
|
def test_web_progress_percent(self):
|
|
"""
|
|
Check web_progress_percent
|
|
"""
|
|
progress_code = str(uuid.uuid4())
|
|
self.partner_ids = self.partner_ids.with_context(progress_code=progress_code)
|
|
self.partner_ids.web_progress_percent(0, "Start")
|
|
self.partner_ids.web_progress_percent(50, "Middle")
|
|
self.partner_ids.web_progress_percent(100, "End")
|
|
|
|
|
|
class WebProgressTestAllProgress(common.TransactionCase):
|
|
at_install = True
|
|
post_install = False
|
|
|
|
@mute_logger('odoo.sql_db')
|
|
def test_get_all_progress(self):
|
|
"""
|
|
Check call to get_all_progress without and with parameters.
|
|
Verify if the parameter is properly escaped in the internal SQL query.
|
|
"""
|
|
progress_code = str(uuid.uuid4())
|
|
partner_obj = self.env['res.partner'].with_context(progress_code=progress_code)
|
|
partner_obj.web_progress_percent(0, "Start")
|
|
with registry(self.env.cr.dbname).cursor() as new_cr:
|
|
# Create a new environment with a new cursor
|
|
new_env = api.Environment(new_cr, self.env.uid, self.env.context)
|
|
progress_obj = self.env['web.progress'].with_env(new_env)
|
|
res = progress_obj.get_all_progress()
|
|
self.assertEqual(res, [{'code': progress_code}])
|
|
res = progress_obj.get_all_progress(0)
|
|
self.assertEqual(res, [])
|
|
with self.assertRaises(ProgrammingError) as e:
|
|
progress_obj.get_all_progress("0 SECOND' GROUP BY code; "
|
|
"SELECT code, array_agg(state) FROM web_progress "
|
|
"WHERE create_date > timezone('utc', now()) - INTERVAL '10")
|
|
new_cr.rollback()
|