# This file is part of Tryton. The COPYRIGHT file at the top level of # this repository contains the full copyright notices and license terms. from decimal import Decimal from datetime import date from types import SimpleNamespace from unittest.mock import Mock, patch from trytond.model.exceptions import ValidationError from trytond.pool import Pool from trytond.modules.price.price_value import PriceValue from trytond.tests.test_tryton import ModuleTestCase, with_transaction from trytond.modules.purchase_trade.pricing import ImportPrices class PurchaseTradeTestCase(ModuleTestCase): 'Test purchase_trade module' module = 'purchase_trade' @with_transaction() def test_get_totals_without_rows(self): 'get_totals returns zeros when the query has no row' Valuation = Pool().get('valuation.valuation') cursor = Mock() cursor.fetchone.return_value = None connection = Mock() connection.cursor.return_value = cursor with patch( 'trytond.modules.purchase_trade.valuation.Transaction' ) as Transaction, patch.object( Valuation, '__table__', return_value='valuation_valuation'): Transaction.return_value.connection = connection self.assertEqual( Valuation.get_totals(), (Decimal(0), Decimal(0))) @with_transaction() def test_get_totals_without_previous_total(self): 'get_totals converts null variation to zero' Valuation = Pool().get('valuation.valuation') cursor = Mock() cursor.fetchone.return_value = (Decimal('42.50'), None) connection = Mock() connection.cursor.return_value = cursor with patch( 'trytond.modules.purchase_trade.valuation.Transaction' ) as Transaction, patch.object( Valuation, '__table__', return_value='valuation_valuation'): Transaction.return_value.connection = connection self.assertEqual( Valuation.get_totals(), (Decimal('42.50'), Decimal(0))) @with_transaction() def test_get_mtm_applies_component_ratio_as_percentage(self): 'get_mtm treats component ratio as a percentage' Strategy = Pool().get('mtm.strategy') strategy = Strategy() strategy.scenario = Mock( valuation_date='2026-03-29', use_last_price=True, ) strategy.currency = Mock() strategy.components = [Mock( price_source_type='curve', price_index=Mock(get_price=Mock(return_value=Decimal('100'))), price_matrix=None, ratio=Decimal('25'), )] line = Mock(unit=Mock()) self.assertEqual( strategy.get_mtm(line, Decimal('10')), Decimal('250.00')) def test_import_prices_skips_missing_index_without_create_option(self): 'import prices skips rows when the price index is missing' price_model = _FakePriceModel() price_value_model = _FakePriceValueModel() with patch('trytond.modules.purchase_trade.pricing.Pool') as pool: pool.return_value.get.side_effect = { 'price.price': price_model, 'price.price_value': price_value_model, }.__getitem__ stats = ImportPrices._import_rows([{ 'price_index': 'LME Copper', 'price_date': date(2026, 3, 27), 'high_price': '100', 'low_price': '90', 'open_price': '95', 'price_value': '98', }]) self.assertEqual(len(stats['imported']), 0) self.assertEqual(len(stats['skipped']), 1) self.assertEqual(stats['skipped'][0]['detail'], 'price_index missing') def test_import_prices_creates_index_and_imports_new_price(self): 'import prices creates missing index when requested' price_model = _FakePriceModel() price_value_model = _FakePriceValueModel() with patch('trytond.modules.purchase_trade.pricing.Pool') as pool: pool.return_value.get.side_effect = { 'price.price': price_model, 'price.price_value': price_value_model, 'price.fixtype': _FakeRecordModel([ SimpleNamespace(id=10, name='Market price')]), 'currency.currency': _FakeRecordModel([ SimpleNamespace(id=20, code='USD')]), 'price.calendar': _FakeRecordModel([ SimpleNamespace(id=30, name='Argus EU')]), 'product.uom': _FakeRecordModel([ SimpleNamespace(id=40, name='Mt')]), 'product.month': _FakeRecordModel(), }.__getitem__ stats = ImportPrices._import_rows([{ 'price_index': 'LME Copper', 'price_date': date(2026, 3, 27), 'high_price': '100', 'low_price': '90', 'open_price': '95', 'price_value': '98', }], create_missing_price_index=True) self.assertEqual(len(stats['created_indexes']), 1) self.assertEqual(len(stats['imported']), 1) self.assertEqual(price_model.records[0].price_index, 'LME Copper') self.assertEqual(price_value_model.records[0].price_value, 98.0) def test_import_prices_defaults_created_price_index(self): 'import prices defaults newly created price index fields' price_model = _FakePriceModel() price_value_model = _FakePriceValueModel() period_model = _FakeRecordModel() with patch('trytond.modules.purchase_trade.pricing.Pool') as pool: pool.return_value.get.side_effect = { 'price.price': price_model, 'price.price_value': price_value_model, 'price.fixtype': _FakeRecordModel([ SimpleNamespace(id=10, name='Market price')]), 'currency.currency': _FakeRecordModel([ SimpleNamespace(id=20, code='USD')]), 'price.calendar': _FakeRecordModel([ SimpleNamespace(id=30, name='Argus EU')]), 'product.uom': _FakeRecordModel([ SimpleNamespace(id=40, name='Mt')]), 'product.month': period_model, }.__getitem__ ImportPrices._import_rows([{ 'price_index': 'Argus Copper 2026-07', 'price_date': date(2026, 3, 27), 'price_value': '98', }], create_missing_price_index=True) price = price_model.records[0] period = period_model.records[0] self.assertEqual(price.price_type, 10) self.assertEqual(price.price_currency, 20) self.assertEqual(price.price_calendar, 30) self.assertEqual(price.price_curve_type, 'future') self.assertEqual(price.price_unit, 40) self.assertEqual(price.price_period, period.id) self.assertEqual(period.month_name, 'JUL26') self.assertEqual(period.beg_date, date(2026, 7, 1)) self.assertEqual(period.end_date, date(2026, 7, 31)) self.assertTrue(period.is_cotation) def test_import_prices_uses_existing_period_from_price_index(self): 'import prices reuses existing period parsed from the price index' price_model = _FakePriceModel() price_value_model = _FakePriceValueModel() existing_period = SimpleNamespace( id=55, month_name='APR25', description='APR25') period_model = _FakeRecordModel([existing_period]) with patch('trytond.modules.purchase_trade.pricing.Pool') as pool: pool.return_value.get.side_effect = { 'price.price': price_model, 'price.price_value': price_value_model, 'price.fixtype': _FakeRecordModel(), 'currency.currency': _FakeRecordModel(), 'price.calendar': _FakeRecordModel(), 'product.uom': _FakeRecordModel(), 'product.month': period_model, }.__getitem__ ImportPrices._import_rows([{ 'price_index': 'Argus Copper 2025-04', 'price_date': date(2026, 3, 27), 'price_value': '98', }], create_missing_price_index=True) self.assertEqual(price_model.records[0].price_period, 55) self.assertEqual(period_model.records, [existing_period]) def test_import_prices_skips_existing_date_without_overwrite(self): 'import prices skips existing price dates unless overwrite is enabled' price = SimpleNamespace(id=1, price_index='LME Copper') price_model = _FakePriceModel([price]) price_value_model = _FakePriceValueModel([ SimpleNamespace( id=1, price=1, price_date=date(2026, 3, 27), price_value=97.0)]) with patch('trytond.modules.purchase_trade.pricing.Pool') as pool: pool.return_value.get.side_effect = { 'price.price': price_model, 'price.price_value': price_value_model, }.__getitem__ stats = ImportPrices._import_rows([{ 'price_index': 'LME Copper', 'price_date': date(2026, 3, 27), 'price_value': '98', }]) self.assertEqual(len(stats['updated']), 0) self.assertEqual(price_value_model.records[0].price_value, 97.0) self.assertEqual( stats['skipped'][0]['detail'], 'price_date already exists') def test_import_prices_overwrites_existing_date_when_requested(self): 'import prices updates existing price dates when requested' price = SimpleNamespace(id=1, price_index='LME Copper') price_model = _FakePriceModel([price]) price_value_model = _FakePriceValueModel([ SimpleNamespace( id=1, price=1, price_date=date(2026, 3, 27), price_value=97.0)]) with patch('trytond.modules.purchase_trade.pricing.Pool') as pool: pool.return_value.get.side_effect = { 'price.price': price_model, 'price.price_value': price_value_model, }.__getitem__ stats = ImportPrices._import_rows([{ 'price_index': 'LME Copper', 'price_date': date(2026, 3, 27), 'price_value': '98', }], overwrite_existing_price=True) self.assertEqual(len(stats['updated']), 1) self.assertEqual(price_value_model.records[0].price_value, 98.0) def test_import_prices_collects_row_errors(self): 'import prices reports invalid row values without hiding the result' price = SimpleNamespace(id=1, price_index='LME Copper') price_model = _FakePriceModel([price]) price_value_model = _FakePriceValueModel() with patch('trytond.modules.purchase_trade.pricing.Pool') as pool: pool.return_value.get.side_effect = { 'price.price': price_model, 'price.price_value': price_value_model, }.__getitem__ stats = ImportPrices._import_rows([{ 'price_index': 'LME Copper', 'price_date': date(2026, 3, 27), 'price_value': 'not-a-number', }]) self.assertEqual(len(stats['errors']), 1) self.assertIn('not-a-number', stats['errors'][0]['detail']) def test_import_prices_result_lists_all_outcome_sections(self): 'import prices formats detailed result sections' message = ImportPrices._format_result({ 'created_indexes': [{ 'row': 2, 'price_index': 'LME Copper', 'price_date': None, 'detail': 'price index created', }], 'imported': [{ 'row': 2, 'price_index': 'LME Copper', 'price_date': date(2026, 3, 27), 'detail': 'price_value=98.0', }], 'updated': [], 'skipped': [{ 'row': 3, 'price_index': 'LME Copper', 'price_date': date(2026, 3, 27), 'detail': 'price_date already exists', }], 'errors': [{ 'row': 4, 'price_index': 'LME Zinc', 'price_date': 'bad-date', 'detail': 'Invalid price_date: bad-date', }], }) self.assertIn('Created price indexes:', message) self.assertIn('Successfully imported prices:', message) self.assertIn('Skipped records:', message) self.assertIn('Errors:', message) self.assertIn('Row 2 - LME Copper / 2026-03-27', message) def test_price_value_unique_rejects_duplicate_create(self): 'price value uniqueness rejects duplicate model creation' price_value = SimpleNamespace( id=None, price=SimpleNamespace(id=1), price_date=date(2026, 3, 27), ) with patch.object( PriceValue, 'search', return_value=[SimpleNamespace(id=2)]): with self.assertRaises(ValidationError): PriceValue.check_unique_price_date([price_value]) def test_price_value_unique_rejects_duplicate_write(self): 'price value uniqueness excludes current record on write' price_value = SimpleNamespace( id=2, price=SimpleNamespace(id=1), price_date=date(2026, 3, 27), ) with patch.object( PriceValue, 'search', return_value=[SimpleNamespace(id=3)]) as search: with self.assertRaises(ValidationError): PriceValue.check_unique_price_date([price_value]) domain = search.call_args[0][0] self.assertIn(('id', '!=', 2), domain[1]) class _FakePriceModel: def __init__(self, records=None): self.records = records or [] def search(self, domain, limit=None): price_index = domain[0][2] records = [ record for record in self.records if record.price_index == price_index] return records[:limit] if limit else records def create(self, values): records = [] for value in values: record = SimpleNamespace(id=len(self.records) + 1, **value) self.records.append(record) records.append(record) return records class _FakePriceValueModel: def __init__(self, records=None): self.records = records or [] def search(self, domain, limit=None): price = domain[0][2] price_date = domain[1][2] records = [ record for record in self.records if record.price == price and record.price_date == price_date] return records[:limit] if limit else records def create(self, values): records = [] for value in values: record = SimpleNamespace(id=len(self.records) + 1, **value) self.records.append(record) records.append(record) return records def write(self, records, values): for record in records: for name, value in values.items(): setattr(record, name, value) class _FakeRecordModel: def __init__(self, records=None): self.records = records or [] def search(self, domain, limit=None): records = self.records for name, operator, value in domain: self.assert_operator(operator) records = [ record for record in records if getattr(record, name, None) == value] return records[:limit] if limit else records def create(self, values): records = [] for value in values: record = SimpleNamespace(id=len(self.records) + 1, **value) self.records.append(record) records.append(record) return records @staticmethod def assert_operator(operator): if operator != '=': raise AssertionError('Unsupported operator: %s' % operator) del ModuleTestCase