420 lines
16 KiB
Python
420 lines
16 KiB
Python
# 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
|