good/buy/tests/test_report_isolation.py

# -*- coding: utf-8 -*-
# Copyright 2026 南京欧度易网络科技有限公司 (http://www.intoerp.com)
# 作者: ccdos@intoerp.com 单雷平
"""Report results must follow the current transaction and own their row data."""
from odoo import api
from odoo.tests.common import TransactionCase


class TestReportIsolation(TransactionCase):
    def setUp(self):
        super().setUp()
        unit = self.env['uom'].create({'name': '报表隔离测试件'})
        category = self.env['core.category'].create({'name': '报表隔离测试类', 'type': 'goods'})
        optional_goods = {}
        if 'supplier_id' in self.env['goods']._fields:
            supplier_category = self.env['core.category'].create({'name': '原生回归供应商类', 'type': 'supplier'})
            supplier = self.env['partner'].create({'name': '原生回归供应商', 's_category_id': supplier_category.id})
            optional_goods['supplier_id'] = supplier.id
        goods = self.env['goods'].create({'name': '报表隔离测试物料', 'code': 'REPORT-ISOLATION',
            'uom_id': unit.id, 'category_id': category.id, **optional_goods})
        stock = self.env['warehouse'].create({'name': '报表隔离测试仓', 'type': 'stock'})
        supplier = self.env.ref('warehouse.warehouse_supplier')
        move = self.env['wh.move'].create({'warehouse_id': supplier.id, 'warehouse_dest_id': stock.id, 'origin': 'buy.receipt.buy'})
        # A report-only ledger fixture: this suite tests SQL visibility, not receipt approval.
        self.line = self.env['wh.move.line'].create({'move_id': move.id, 'type': 'in',
            'warehouse_id': supplier.id, 'warehouse_dest_id': stock.id,
            'goods_id': goods.id, 'uom_id': unit.id, 'goods_qty': 5,
            'cost_unit': 1, 'cost': 5, 'state': 'done', 'date': '2026-01-02'})
        self.report = self.env['buy.summary.goods'].with_context(
            goods_id=(goods.id, goods.name), warehouse_dest_id=(stock.id, stock.name),
            date_start='2026-01-01', date_end='2026-01-31')
        self.env.flush_all()

    def test_read_after_ledger_change(self):
        self.assertEqual(self.report.search_read([])[0]['qty'], 5)
        self.line.goods_qty = 7
        self.env.flush_all()
        self.assertEqual(self.report.search_read([])[0]['qty'], 7)

    def test_rows_are_not_shared_with_callers(self):
        rows = self.report.get_data_from_cache()
        rows[0]['qty'] = 999
        self.assertEqual(self.report.get_data_from_cache()[0]['qty'], 5)

    def test_new_transaction_does_not_receive_uncommitted_cached_rows(self):
        self.assertEqual(len(self.report.search_read([])), 1)
        with self.registry.cursor() as cr:
            other = api.Environment(cr, self.env.uid, dict(self.report.env.context))
            # This independent transaction cannot see the test's uncommitted ledger.
            self.assertFalse(other['buy.summary.goods'].search_read([]))