Skip to content

Commit

Permalink
[FIX] queue_job: fix test cases from TestDelayable being skipped
Browse files Browse the repository at this point in the history
  • Loading branch information
thienvh332 committed Oct 21, 2024
1 parent 68a72a6 commit bc90d0b
Showing 1 changed file with 44 additions and 28 deletions.
72 changes: 44 additions & 28 deletions queue_job/tests/test_delayable.py
Original file line number Diff line number Diff line change
@@ -1,29 +1,40 @@
# copyright 2019 Camptocamp
# license agpl-3.0 or later (http://www.gnu.org/licenses/agpl.html)

import unittest
import gc
import logging
from unittest import mock

from odoo.tests import common

from odoo.addons.queue_job.delay import Delayable, DelayableGraph


class TestDelayable(unittest.TestCase):
class TestDelayable(common.BaseCase):
def setUp(self):
super().setUp()
self.recordset = mock.MagicMock(name="recordset")

def test_delayable_set(self):
dl = Delayable(self.recordset)
dl.set(priority=15)
self.assertEqual(dl.priority, 15)
dl.set({"priority": 20, "description": "test"})
self.assertEqual(dl.priority, 20)
self.assertEqual(dl.description, "test")
# Use gc for garbage collection and use assertLogs to suppress WARNING
with self.assertLogs("odoo.addons.queue_job.delay", level=logging.WARNING):
dl = Delayable(self.recordset)
dl.set(priority=15)
self.assertEqual(dl.priority, 15)
dl.set({"priority": 20, "description": "test"})
self.assertEqual(dl.priority, 20)
self.assertEqual(dl.description, "test")
del dl
gc.collect()

def test_delayable_set_unknown(self):
dl = Delayable(self.recordset)
with self.assertRaises(ValueError):
dl.set(foo=15)
# Use gc for garbage collection and use assertLogs to suppress WARNING
with self.assertLogs("odoo.addons.queue_job.delay", level=logging.WARNING):
dl = Delayable(self.recordset)
with self.assertRaises(ValueError):
dl.set(foo=15)
del dl
gc.collect()

def test_graph_add_vertex_edge(self):
graph = DelayableGraph()
Expand Down Expand Up @@ -55,23 +66,28 @@ def test_graph_edges(self):
)

def test_graph_connect(self):
node_tail = Delayable(self.recordset)
node_tail2 = Delayable(self.recordset)
node_middle = Delayable(self.recordset)
node_top = Delayable(self.recordset)
node_middle.on_done(node_tail)
node_middle.on_done(node_tail2)
node_top.on_done(node_middle)
collected = node_top._graph._connect_graphs()
self.assertEqual(
collected._graph,
{
node_tail: set(),
node_tail2: set(),
node_middle: {node_tail, node_tail2},
node_top: {node_middle},
},
)
# Use gc for garbage collection and use assertLogs to suppress WARNING
with self.assertLogs("odoo.addons.queue_job.delay", level=logging.WARNING):
node_tail = Delayable(self.recordset)
node_tail2 = Delayable(self.recordset)
node_middle = Delayable(self.recordset)
node_top = Delayable(self.recordset)
node_middle.on_done(node_tail)
node_middle.on_done(node_tail2)
node_top.on_done(node_middle)
collected = node_top._graph._connect_graphs()
self.assertEqual(
collected._graph,
{
node_tail: set(),
node_tail2: set(),
node_middle: {node_tail, node_tail2},
node_top: {node_middle},
},
)

del node_tail, node_tail2, node_middle, node_top, collected
gc.collect()

def test_graph_paths(self):
graph = DelayableGraph(
Expand Down

0 comments on commit bc90d0b

Please sign in to comment.