282 lines
10 KiB
Python
282 lines
10 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.pool import Pool
|
|
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,
|
|
}.__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_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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
del ModuleTestCase
|