ITSA - Import Prices
This commit is contained in:
AzureAD\SylvainDUVERNAY
@@ -2,10 +2,13 @@
|
||||
# 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):
|
||||
@@ -70,5 +73,152 @@ class PurchaseTradeTestCase(ModuleTestCase):
|
||||
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(stats['imported'], 0)
|
||||
self.assertEqual(len(stats['skipped']), 1)
|
||||
self.assertEqual(stats['skipped'][0]['reason'], '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(stats['created_indexes'], 1)
|
||||
self.assertEqual(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(stats['updated'], 0)
|
||||
self.assertEqual(price_value_model.records[0].price_value, 97.0)
|
||||
self.assertEqual(
|
||||
stats['skipped'][0]['reason'], '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(stats['updated'], 1)
|
||||
self.assertEqual(price_value_model.records[0].price_value, 98.0)
|
||||
|
||||
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user