[12.0][IMP] kpi_dashboard: Compute the KPI on a separate cursor and rollback it

kpi_dashboard/__manifest__.py

@@ -21,4 +21,5 @@
         "views/kpi_dashboard.xml",
     ],
     "demo": ["demo/demo_dashboard.xml"],
+    "maintainers": ["etobella"],
 }

kpi_dashboard/demo/demo_dashboard.xml

@@ -87,7 +87,6 @@ result = {"graphs": [
     <function model="kpi.kpi" name="compute"
               eval="[[ref('widget_graph')]]"/>
 
-
     <record id="dashboard_widget_text" model="kpi.dashboard.item">
         <field name="name">Dashboard title</field>
         <field name="dashboard_id" ref="demo_dashboard"/>

kpi_dashboard/models/kpi_kpi.py

@@ -1,9 +1,11 @@
 # Copyright 2020 Creu Blanca
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
 
-from odoo import api, fields, models
+from odoo import api, fields, models, _
+from odoo.exceptions import ValidationError
 import ast
 from odoo.tools.safe_eval import safe_eval
+import re
 
 
 class KpiKpi(models.Model):
@@ -89,12 +91,28 @@ class KpiKpi(models.Model):
     def _get_code_input_dict(self):
         return {
             "self": self,
-            "model": self,
+            "model": self.browse(),
         }
 
+    def _forbidden_code(self):
+        return ["commit", "rollback", "getattr", "execute"]
+
     def _compute_value_code(self):
+        forbidden = self._forbidden_code()
+        search_terms = "(" + ("|".join(forbidden)) + ")"
+        if re.search(search_terms, (self.code or "").lower()):
+            message = ", ".join(forbidden[:-1]) or ""
+            if len(message) > 0:
+                message += _(" or ")
+            message += forbidden[-1]
+            raise ValidationError(_(
+                "The code cannot contain the following terms: %s."
+            ) % message)
         results = self._get_code_input_dict()
+        savepoint = "kpi_formula_%s" % self.id
+        self.env.cr.execute("savepoint %s" % savepoint)
         safe_eval(self.code or "", results, mode="exec", nocopy=True)
+        self.env.cr.execute("rollback to %s" % savepoint)
         return results.get("result", {})
 
 

kpi_dashboard/tests/test_formula.py

@@ -2,30 +2,58 @@
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
 
 from odoo.tests.common import TransactionCase
+from odoo.exceptions import ValidationError
 
 
 class TestFormula(TransactionCase):
-    def test_computation(self):
-        kpi = self.env["kpi.kpi"].create(
+    def setUp(self):
+        super().setUp()
+        self.kpi = self.env["kpi.kpi"].create(
             {
                 "name": "DEMO KPI",
                 "widget": "number",
                 "computation_method": "code",
             }
         )
-        self.assertFalse(kpi.value)
-        kpi.compute()
-        self.assertEqual(kpi.value, {})
-        kpi.code = """
+
+    def test_forbidden_words_01(self):
+        self.kpi.code = """
+result = {"value": 0}
+self.env.cr.commit()
+"""
+        with self.assertRaises(ValidationError):
+            self.kpi.compute()
+
+    def test_forbidden_words_02(self):
+        self.kpi.code = """
+result = {"value": 0}
+self.env.cr.rollback()
+"""
+        with self.assertRaises(ValidationError):
+            self.kpi.compute()
+
+    def test_forbidden_words_03(self):
+        self.kpi.code = """
+result = {"value": 0}
+self.env.cr.execute("CoMMiT")
+"""
+        with self.assertRaises(ValidationError):
+            self.kpi.compute()
+
+    def test_computation(self):
+        self.assertFalse(self.kpi.value)
+        self.kpi.compute()
+        self.assertEqual(self.kpi.value, {})
+        self.kpi.code = """
 result = {}
 result['value'] = len(model.search([('id', '=', %s)]))
 result['previous'] = len(model.search([('id', '!=', %s)]))
         """ % (
-            kpi.id,
-            kpi.id,
+            self.kpi.id,
+            self.kpi.id,
         )
-        kpi.compute()
-        value = kpi.value
+        self.kpi.compute()
+        value = self.kpi.value
         self.assertTrue(value.get("value"))
         self.assertEqual(value.get("value"), 1)
-        self.assertEqual(value.get("previous"), kpi.search_count([]) - 1)
+        self.assertEqual(value.get("previous"), self.kpi.search_count([]) - 1)