[IMP] {util/pg,base/tests}: pick parallel_execute impl at runtime

If we can force a serial implementation for `parallel_execute` we can
avoid unexpected `cr.commit` calls during tests. In effect, committing
breaks the savepoint mechanic in tests and thus causes hard to
understand test failures.

Another benefit of choosing the implementation at runtime is that we can
now test both implementations for correctness.

closes #33

Signed-off-by: Christophe Simonis (chs) <[email protected]>
Co-authored-by: "Christophe Simonis" <[email protected]>

Author: aj-fuentes

src/base/tests/test_util.py

@@ -1,5 +1,6 @@
 import operator
 import re
+import threading
 import unittest
 import uuid
 from ast import literal_eval
@@ -446,6 +447,11 @@ def test_parallel_rowcount(self):
             rowcount = util.explode_execute(cr, query, table="res_lang", bucket_size=10)
         self.assertEqual(rowcount, expected)
 
+    def test_parallel_rowcount_threaded(self):
+        threading.current_thread().testing = False
+        self.test_parallel_rowcount()
+        threading.current_thread().testing = True
+
     def test_create_column_with_fk(self):
         cr = self.env.cr
         self.assertFalse(util.column_exists(cr, "res_partner", "_test_lang_id"))

src/util/pg.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import re
+import threading
 import time
 import uuid
 import warnings
@@ -61,18 +62,17 @@ def savepoint(cr):
         yield
 
 
-if ThreadPoolExecutor is None:
+def _parallel_execute_serial(cr, queries, logger=_logger):
+    cnt = 0
+    for query in log_progress(queries, logger, qualifier="queries", size=len(queries)):
+        cr.execute(query)
+        cnt += cr.rowcount
+    return cnt
 
-    def parallel_execute(cr, queries, logger=_logger):
-        cnt = 0
-        for query in log_progress(queries, logger, qualifier="queries", size=len(queries)):
-            cr.execute(query)
-            cnt += cr.rowcount
-        return cnt
 
-else:
+if ThreadPoolExecutor is not None:
 
-    def parallel_execute(cr, queries, logger=_logger):
+    def _parallel_execute_threaded(cr, queries, logger=_logger):
         """
         Execute queries in parallel
         Use a maximum of 8 workers (but not more than the number of CPUs)
@@ -118,6 +118,18 @@ def execute(query):
                 )
             )
 
+else:
+    _parallel_execute_threaded = _parallel_execute_serial
+
+
+def parallel_execute(cr, queries, logger=_logger):
+    parallel_execute_impl = (
+        _parallel_execute_serial
+        if getattr(threading.current_thread(), "testing", False)
+        else _parallel_execute_threaded
+    )
+    return parallel_execute_impl(cr, queries, logger=_logger)
+
 
 def format_query(cr, query, *args, **kwargs):
     """