from odoo import Command from odoo.tests import tagged, common @tagged('at_install', '-post_install') # LEGACY at_install class TestPrivateReadGroupingSets(common.TransactionCase): def test_simple_read_grouping_sets(self): Model = self.env['test_read_group.aggregate'] Partner = self.env['res.partner'] partner_1 = Partner.create({'name': 'z_one'}) partner_2 = Partner.create({'name': 'a_two'}) Model.create({'key': 1, 'partner_id': partner_1.id, 'value': 1}) Model.create({'key': 1, 'partner_id': partner_1.id, 'value': 2}) Model.create({'key': 1, 'partner_id': partner_2.id, 'value': 3}) Model.create({'key': 2, 'partner_id': partner_2.id, 'value': 4}) Model.create({'key': 2, 'partner_id': partner_2.id}) Model.create({'key': 2, 'value': 5}) Model.create({'partner_id': partner_2.id, 'value': 5}) Model.create({'value': 6}) Model.create({}) grouping_sets = [['key', 'partner_id'], ['key'], ['partner_id'], []] expected_result = [ Model._read_group([], grouping_set, aggregates=['value:sum']) for grouping_set in grouping_sets ] with self.assertQueries([""" SELECT GROUPING( "test_read_group_aggregate"."key", "test_read_group_aggregate"."partner_id" ), "test_read_group_aggregate"."key", "test_read_group_aggregate"."partner_id", SUM("test_read_group_aggregate"."value") FROM "test_read_group_aggregate" GROUP BY GROUPING SETS ( ("test_read_group_aggregate"."key", "test_read_group_aggregate"."partner_id"), ("test_read_group_aggregate"."key"), ("test_read_group_aggregate"."partner_id"), () ) ORDER BY "test_read_group_aggregate"."key" ASC, "test_read_group_aggregate"."partner_id" ASC """]): self.assertEqual( Model._read_grouping_sets([], grouping_sets, aggregates=['value:sum']), expected_result, ) grouping_sets = [['key', 'partner_id'], ['key'], ['partner_id'], []] orders = ["partner_id, key", "key", 'partner_id', ""] expected_result = [ Model._read_group([], grouping_set, aggregates=['value:sum'], order=order) for grouping_set, order in zip(grouping_sets, orders) ] # Forcing order with many2one, traverse use the order of the comodel (res.partner) with self.assertQueries([""" SELECT GROUPING( "test_read_group_aggregate"."key", "test_read_group_aggregate"."partner_id" ), "test_read_group_aggregate"."key", "test_read_group_aggregate"."partner_id", SUM("test_read_group_aggregate"."value") FROM "test_read_group_aggregate" LEFT JOIN "res_partner" AS "test_read_group_aggregate__partner_id" ON ( "test_read_group_aggregate"."partner_id" = "test_read_group_aggregate__partner_id"."id" ) GROUP BY GROUPING SETS ( ( "test_read_group_aggregate"."key", "test_read_group_aggregate"."partner_id", "test_read_group_aggregate__partner_id"."complete_name", "test_read_group_aggregate__partner_id"."id" ), ("test_read_group_aggregate"."key"), ( "test_read_group_aggregate"."partner_id", "test_read_group_aggregate__partner_id"."complete_name", "test_read_group_aggregate__partner_id"."id" ), () ) ORDER BY "test_read_group_aggregate__partner_id"."complete_name" ASC, "test_read_group_aggregate__partner_id"."id" DESC, "test_read_group_aggregate"."key" ASC """]): self.assertEqual( Model._read_grouping_sets([], grouping_sets, aggregates=['value:sum'], order="partner_id, key"), expected_result, ) def test_many2many_read_grouping_sets(self): User = self.env['test_read_group.user'] mario, luigi = User.create([{'name': 'Mario'}, {'name': 'Luigi'}]) tasks = self.env['test_read_group.task'].create([ { # both users 'name': "Super Mario Bros.", 'user_ids': [Command.set((mario + luigi).ids)], 'integer': 1, }, { # mario only 'name': "Paper Mario", 'user_ids': [Command.set(mario.ids)], 'customer_ids': [Command.set(luigi.ids)], 'integer': 2, }, { # luigi only 'name': "Luigi's Mansion", 'user_ids': [Command.set(luigi.ids)], 'customer_ids': [Command.set(mario.ids)], 'integer': 3, }, { # no user 'name': 'Donkey Kong', 'customer_ids': [Command.set((mario + luigi).ids)], }, ]) domain = [('id', 'in', tasks.ids)] grouping_sets = [['user_ids', 'key'], ['key'], ['user_ids'], []] aggregates = ['name:array_agg', '__count', 'integer:sum'] expected_result = [ tasks._read_group(domain, groupby, aggregates) for groupby in grouping_sets ] with self.assertQueries([ """ SELECT GROUPING("test_read_group_task__user_ids__rel"."user_id", "test_read_group_task"."key"), "test_read_group_task__user_ids__rel"."user_id", "test_read_group_task"."key", ARRAY_AGG("test_read_group_task"."name" ORDER BY "test_read_group_task"."id"), COUNT(*), SUM("test_read_group_task"."integer") FROM "test_read_group_task" LEFT JOIN "test_read_group_task_user_rel" AS "test_read_group_task__user_ids__rel" ON ( "test_read_group_task"."id" = "test_read_group_task__user_ids__rel"."task_id" ) WHERE "test_read_group_task"."id" IN %s GROUP BY GROUPING SETS ( ("test_read_group_task__user_ids__rel"."user_id", "test_read_group_task"."key"), ("test_read_group_task__user_ids__rel"."user_id")) ORDER BY "test_read_group_task__user_ids__rel"."user_id" ASC, "test_read_group_task"."key" ASC """, """ SELECT GROUPING("test_read_group_task"."key"), "test_read_group_task"."key", ARRAY_AGG("test_read_group_task"."name" ORDER BY "test_read_group_task"."id"), COUNT(*), SUM("test_read_group_task"."integer") FROM "test_read_group_task" WHERE "test_read_group_task"."id" IN %s GROUP BY GROUPING SETS (("test_read_group_task"."key"), ()) ORDER BY "test_read_group_task"."key" ASC """, ]): self.assertEqual( tasks._read_grouping_sets(domain, grouping_sets, aggregates), expected_result, ) complete_order = "user_ids DESC, key" grouping_sets = [['user_ids', 'key'], ['key'], ['user_ids'], []] orders = ["user_ids DESC, key", "key", "user_ids DESC", ''] aggregates = ['name:array_agg', '__count', 'integer:sum'] expected_result = [ tasks._read_group(domain, groupby, aggregates, order=order) for groupby, order in zip(grouping_sets, orders) ] with self.assertQueries([ """ SELECT GROUPING("test_read_group_task__user_ids__rel"."user_id", "test_read_group_task"."key"), "test_read_group_task__user_ids__rel"."user_id", "test_read_group_task"."key", ARRAY_AGG("test_read_group_task"."name" ORDER BY "test_read_group_task"."id"), COUNT(*), SUM("test_read_group_task"."integer") FROM "test_read_group_task" LEFT JOIN "test_read_group_task_user_rel" AS "test_read_group_task__user_ids__rel" ON ( "test_read_group_task"."id" = "test_read_group_task__user_ids__rel"."task_id" ) WHERE "test_read_group_task"."id" IN %s GROUP BY GROUPING SETS ( ("test_read_group_task__user_ids__rel"."user_id", "test_read_group_task"."key"), ("test_read_group_task__user_ids__rel"."user_id")) ORDER BY "test_read_group_task__user_ids__rel"."user_id" DESC, "test_read_group_task"."key" ASC """, """ SELECT GROUPING("test_read_group_task"."key"), "test_read_group_task"."key", ARRAY_AGG("test_read_group_task"."name" ORDER BY "test_read_group_task"."id"), COUNT(*), SUM("test_read_group_task"."integer") FROM "test_read_group_task" WHERE "test_read_group_task"."id" IN %s GROUP BY GROUPING SETS (("test_read_group_task"."key"), ()) ORDER BY "test_read_group_task"."key" ASC """, ]): self.assertEqual( tasks._read_grouping_sets(domain, grouping_sets, aggregates, order=complete_order), expected_result, ) cases = [ { # Test 2 many2manys 'grouping_sets': [ ['user_ids', 'customer_ids'], ['key'], ['user_ids'], ['customer_ids'], ['key', 'customer_ids'], [] ], 'aggregates': ['__count', 'integer:sum'], # 1 for ('user_ids', 'customer_ids') + 1 for ('user_ids',) + 1 for ('customer_ids',) + 1 for remaining (key, []) 'nb_queries': 4, }, { # Test that __count doesn't make a extra query 'grouping_sets': [['user_ids', 'key'], ['key'], ['user_ids'], []], 'aggregates': ['__count'], 'nb_queries': 1, }, { # Test that __count as order 'grouping_sets': [['user_ids', 'customer_ids'], ['key'], ['user_ids'], []], 'read_group_orders': [ "__count, user_ids, customer_ids", "__count, key", "__count, user_ids", "__count", ], 'complete_order': "__count, user_ids, customer_ids, key", 'aggregates': ['__count', 'integer:min', 'integer:max', 'integer:count_distinct'], 'nb_queries': 1, }, { # Everything 'grouping_sets': [ ['user_ids', 'customer_ids'], ['key'], ['integer'], [], ['customer_ids'], ['key', 'customer_ids'], ['integer', 'customer_ids'], ], 'read_group_orders': [ "__count DESC, customer_ids DESC, user_ids", "key, __count DESC", "__count DESC, integer", "__count DESC", "__count DESC, customer_ids DESC", "key, __count DESC, customer_ids DESC", "__count DESC, customer_ids DESC, integer", ], 'complete_order': "key, __count DESC, customer_ids DESC, integer, user_ids", 'aggregates': ['integer:sum', '__count'], # 1 for ('user_ids', 'customer_ids') + 1 for ('customer_ids',) + 1 for remainings grouping set 'nb_queries': 3, }, ] for i, case in enumerate(cases): nb_queries = case['nb_queries'] aggregates = case['aggregates'] grouping_sets = case['grouping_sets'] read_group_orders = case.get('read_group_orders', [""] * len(grouping_sets)) complete_order = case.get('complete_order') expected_result = [ tasks._read_group(domain, groupby, aggregates, order=order) for groupby, order in zip(grouping_sets, read_group_orders) ] with self.subTest(f"Case {i} - {grouping_sets!r} - {aggregates!r}"), self.assertQueryCount(nb_queries): result = tasks._read_grouping_sets(domain, grouping_sets, aggregates, order=complete_order) self.assertEqual(result, expected_result)