Skip to content

Commit 903ee87

Browse files
jjmaksoudjjmaksoud
committed
[ADD] testing: add unittest methods to test record updates
Two contextmanager methods added to UnitTestCase to assert wether records in the database are updated or not. The assert requires a table name and an optional list of ids. If ids are provided, those specific records will be checked if updated or not. If no ids are provided then the validation will be based on the insert or update of any record of that relation. Part-of: #224 Signed-off-by: Christophe Simonis (chs) <[email protected]>
1 parent 251bd2a commit 903ee87

File tree

2 files changed

+174
-0
lines changed

2 files changed

+174
-0
lines changed

‎src/base/tests/test_util.py

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2169,3 +2169,90 @@ def test_rename_xmlid(self):
21692169
util.invalidate(test_view_2)
21702170
self.assertIn('t-call="base.rename_view"', test_view_2.arch_db)
21712171
self.assertIn('t-name="base.rename_view"', test_view_1.arch_db)
2172+
2173+
2174+
class TestAssertUpdated(UnitTestCase):
2175+
def test_assert_updated(self):
2176+
p1 = self.env["res.partner"].create({"name": "Levi"})
2177+
p2 = self.env["res.partner"].create({"name": "Mikasa Ackerman"})
2178+
util.flush(p1)
2179+
util.flush(p2)
2180+
2181+
# when ids is None, assert any record is created or updated
2182+
with self.assertUpdated("res_partner"):
2183+
self.env["res.partner"].create({"name": "Sasha Braus"})
2184+
with self.assertUpdated("res_partner"):
2185+
p2.city = "Shiganshina"
2186+
util.flush(p2)
2187+
with self.assertRaises(AssertionError), self.assertUpdated("res_partner"):
2188+
pass
2189+
2190+
# when ids is [], assert a record is updated
2191+
with self.assertUpdated("res_partner", ids=[]):
2192+
p1.city = "Underground"
2193+
util.flush(p1)
2194+
with self.assertRaises(AssertionError), self.assertUpdated("res_partner", ids=[]):
2195+
self.env["res.partner"].create({"name": "Annie Leonhart"})
2196+
2197+
# when ids has multiple records, all records should be updated
2198+
with self.assertUpdated("res_partner", ids=[p1.id, p2.id]):
2199+
p1.company_name = "Survey Corps"
2200+
p2.company_name = "Survey Corps"
2201+
util.flush(p1)
2202+
util.flush(p2)
2203+
with self.assertRaises(AssertionError), self.assertUpdated("res_partner", ids=[p1.id, p2.id]):
2204+
p1.name = "Levi Ackerman"
2205+
util.flush(p1)
2206+
2207+
# when ids has a record, that record should be the one updated
2208+
with self.assertRaises(AssertionError), self.assertUpdated("res_partner", ids=[p1.id]):
2209+
p2.city = "Paradise Island"
2210+
util.flush(p2)
2211+
2212+
def test_assert_not_updated(self):
2213+
p1 = self.env["res.partner"].create({"name": "Eren Yeager"})
2214+
p2 = self.env["res.partner"].create({"name": "Armin Arlert"})
2215+
util.flush(p1)
2216+
util.flush(p2)
2217+
2218+
# when ids is None, assert no record is created or updated
2219+
with self.assertNotUpdated("res_partner"):
2220+
pass
2221+
with self.assertRaises(AssertionError), self.assertNotUpdated("res_partner"):
2222+
p1.city = "Shiganshina"
2223+
util.flush(p1)
2224+
with self.assertRaises(AssertionError), self.assertNotUpdated("res_partner"):
2225+
self.env["res.partner"].create({"name": "Bertolt Hoover"})
2226+
2227+
# when ids is [], assert no record is updated
2228+
with self.assertNotUpdated("res_partner", ids=[]):
2229+
self.env["res.partner"].create({"name": "Marco Bodt"})
2230+
with self.assertRaises(AssertionError), self.assertNotUpdated("res_partner", ids=[]):
2231+
p2.city = "Shiganshina"
2232+
util.flush(p2)
2233+
2234+
# when ids has a record, only that record should not be updated
2235+
with self.assertNotUpdated("res_partner", ids=[p2.id]):
2236+
p1.company_name = "Survey Corps"
2237+
util.flush(p1)
2238+
2239+
# when ids has multiple records, none of them should be updated
2240+
with self.assertRaises(AssertionError), self.assertNotUpdated("res_partner", ids=[p1.id, p2.id]):
2241+
p2.company_name = "Survey Corps"
2242+
util.flush(p2)
2243+
2244+
def test_assert_updated_combo(self):
2245+
p1 = self.env["res.partner"].create({"name": "Reiner Braun"})
2246+
p2 = self.env["res.partner"].create({"name": "Ymir Fritz"})
2247+
util.flush(p1)
2248+
util.flush(p2)
2249+
2250+
with self.assertUpdated("res_partner", ids=[p1.id]), self.assertNotUpdated("res_partner", ids=[p2.id]):
2251+
p1.company_name = "Marley Warriors"
2252+
util.flush(p1)
2253+
2254+
with self.assertRaises(AssertionError), self.assertUpdated("res_partner"), self.assertNotUpdated(
2255+
"res_partner", ids=[p2.id]
2256+
):
2257+
p2.city = "Niflheim"
2258+
util.flush(p2)

‎src/testing.py

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import logging
44
import os
55
import re
6+
from contextlib import contextmanager
67

78
import odoo
89
from odoo import api, release
@@ -122,6 +123,92 @@ def setUpClass(cls):
122123
bv = os.getenv("ODOO_BASE_VERSION", release.series)
123124
util.ENVIRON["__base_version"] = parse_version(bv)
124125

126+
@contextmanager
127+
def assertNotUpdated(self, table, ids=None, msg=None):
128+
cr = self.env.cr
129+
cr.execute(util.format_query(cr, "DROP TRIGGER IF EXISTS no_update ON {}", table))
130+
cr.execute(
131+
"""
132+
DROP TABLE IF EXISTS _upg_test_no_upd_id;
133+
CREATE UNLOGGED TABLE _upg_test_no_upd_id(id int PRIMARY KEY, record json);
134+
CREATE OR REPLACE
135+
FUNCTION fail_assert_not_updated() RETURNS TRIGGER AS $$
136+
BEGIN
137+
INSERT INTO _upg_test_no_upd_id VALUES (NEW.id, row_to_json(NEW, true))
138+
ON CONFLICT DO NOTHING;
139+
RETURN NEW;
140+
END
141+
$$ LANGUAGE PLPGSQL
142+
""",
143+
)
144+
cr.execute(
145+
util.format_query(
146+
cr,
147+
"""
148+
CREATE TRIGGER no_update
149+
BEFORE {when}
150+
ON {table}
151+
FOR EACH ROW {cond} EXECUTE
152+
FUNCTION fail_assert_not_updated()
153+
""",
154+
when=util.SQLStr("UPDATE" if ids is not None else "UPDATE or INSERT"),
155+
table=table,
156+
cond=util.SQLStr("WHEN (new.id = ANY(%s))" if ids else ""),
157+
),
158+
[list(ids) if ids is not None else None],
159+
)
160+
self.addCleanup(cr.execute, "DROP TABLE IF EXISTS _upg_test_no_upd_id")
161+
self.addCleanup(cr.execute, util.format_query(cr, "DROP TRIGGER IF EXISTS no_update ON {}", table))
162+
yield
163+
cr.execute("SELECT record FROM _upg_test_no_upd_id")
164+
updated_records = [r[0] for r in cr.fetchall()]
165+
if updated_records:
166+
raise AssertionError(msg or "Some {} records were updated {}".format(table, updated_records))
167+
168+
@contextmanager
169+
def assertUpdated(self, table, ids=None, msg=None):
170+
cr = self.env.cr
171+
cr.execute(util.format_query(cr, "DROP TRIGGER IF EXISTS assert_update ON {}", table))
172+
cr.execute(
173+
"""
174+
DROP TABLE IF EXISTS _upg_test_upd_id;
175+
CREATE UNLOGGED TABLE _upg_test_upd_id(id int PRIMARY KEY);
176+
CREATE OR REPLACE
177+
FUNCTION save_updated() RETURNS TRIGGER AS $$
178+
BEGIN
179+
INSERT INTO _upg_test_upd_id VALUES (NEW.id)
180+
ON CONFLICT DO NOTHING;
181+
RETURN NEW;
182+
END
183+
$$ LANGUAGE PLPGSQL
184+
""",
185+
)
186+
cr.execute(
187+
util.format_query(
188+
cr,
189+
"""
190+
CREATE TRIGGER assert_update
191+
BEFORE {when}
192+
ON {table}
193+
FOR EACH ROW {cond} EXECUTE
194+
FUNCTION save_updated()
195+
""",
196+
when=util.SQLStr("UPDATE" if ids is not None else "UPDATE or INSERT"),
197+
table=table,
198+
cond=util.SQLStr("WHEN (NEW.id = ANY(%s))" if ids else ""),
199+
),
200+
[list(ids) if ids is not None else None],
201+
)
202+
self.addCleanup(cr.execute, "DROP TABLE IF EXISTS _upg_test_upd_id")
203+
self.addCleanup(cr.execute, util.format_query(cr, "DROP TRIGGER IF EXISTS assert_update ON {}", table))
204+
yield
205+
cr.execute("SELECT id FROM _upg_test_upd_id")
206+
updated_ids = [r[0] for r in cr.fetchall()]
207+
if not ids:
208+
self.assertTrue(updated_ids, msg or "No record was updated.")
209+
else:
210+
self.assertEqual(set(updated_ids), set(ids), msg or "Records were not updated.")
211+
125212

126213
class UpgradeCommon(BaseCase):
127214
__initialized = False

0 commit comments

Comments
 (0)