Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[17.0][FIX] queue_job: fix test cases from TestDelayable being skipped #703

Merged
merged 1 commit into from
Oct 21, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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
Loading