Files
calminer/tests/test_pricing_settings_repository.py
zwitschi 795a9f99f4 feat: Enhance currency handling and validation across scenarios
- Updated form template to prefill currency input with default value and added help text for clarity.
- Modified integration tests to assert more descriptive error messages for invalid currency codes.
- Introduced new tests for currency normalization and validation in various scenarios, including imports and exports.
- Added comprehensive tests for pricing calculations, ensuring defaults are respected and overrides function correctly.
- Implemented unit tests for pricing settings repository, ensuring CRUD operations and default settings are handled properly.
- Enhanced scenario pricing evaluation tests to validate currency handling and metadata defaults.
- Added simulation tests to ensure Monte Carlo runs are accurate and handle various distribution scenarios.
2025-11-11 18:29:59 +01:00

210 lines
6.9 KiB
Python

from __future__ import annotations
from collections.abc import Iterator
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from config.database import Base
from models import PricingImpuritySettings, PricingMetalSettings, PricingSettings
from services.pricing import PricingMetadata
from services.repositories import (
PricingSettingsRepository,
ensure_default_pricing_settings,
)
from services.unit_of_work import UnitOfWork
@pytest.fixture()
def engine() -> Iterator:
engine = create_engine("sqlite:///:memory:", future=True)
Base.metadata.create_all(bind=engine)
try:
yield engine
finally:
Base.metadata.drop_all(bind=engine)
@pytest.fixture()
def session(engine) -> Iterator[Session]:
TestingSession = sessionmaker(
bind=engine, expire_on_commit=False, future=True)
db = TestingSession()
try:
yield db
finally:
db.close()
def test_pricing_settings_repository_crud(session: Session) -> None:
repo = PricingSettingsRepository(session)
settings = PricingSettings(
name="Contract A",
slug="Contract-A",
default_currency="usd",
default_payable_pct=95.0,
moisture_threshold_pct=7.5,
moisture_penalty_per_pct=1500.0,
)
repo.create(settings)
metal_override = PricingMetalSettings(
metal_code="Copper",
payable_pct=96.0,
moisture_threshold_pct=None,
moisture_penalty_per_pct=None,
)
repo.attach_metal_override(settings, metal_override)
impurity_override = PricingImpuritySettings(
impurity_code="as",
threshold_ppm=100.0,
penalty_per_ppm=3.5,
)
repo.attach_impurity_override(settings, impurity_override)
retrieved = repo.get_by_slug("CONTRACT-A", include_children=True)
assert retrieved.slug == "contract-a"
assert retrieved.default_currency == "USD"
assert len(retrieved.metal_overrides) == 1
assert retrieved.metal_overrides[0].metal_code == "copper"
assert len(retrieved.impurity_overrides) == 1
assert retrieved.impurity_overrides[0].impurity_code == "AS"
listed = repo.list(include_children=True)
assert len(listed) == 1
assert listed[0].id == settings.id
def test_ensure_default_pricing_settings_creates_and_updates(session: Session) -> None:
repo = PricingSettingsRepository(session)
metadata_initial = PricingMetadata(
default_payable_pct=100.0,
default_currency="USD",
moisture_threshold_pct=8.0,
moisture_penalty_per_pct=0.0,
impurity_thresholds={"As": 50.0},
impurity_penalty_per_ppm={"As": 2.0},
)
result_create = ensure_default_pricing_settings(
repo,
metadata=metadata_initial,
name="Seeded Pricing",
description="Seeded from defaults",
)
assert result_create.created is True
assert result_create.settings.slug == "default"
assert result_create.settings.default_currency == "USD"
assert len(result_create.settings.impurity_overrides) == 1
assert result_create.settings.impurity_overrides[0].penalty_per_ppm == 2.0
metadata_update = PricingMetadata(
default_payable_pct=97.0,
default_currency="EUR",
moisture_threshold_pct=6.5,
moisture_penalty_per_pct=250.0,
impurity_thresholds={"As": 45.0, "Pb": 12.0},
impurity_penalty_per_ppm={"As": 3.0, "Pb": 1.25},
)
result_update = ensure_default_pricing_settings(
repo,
metadata=metadata_update,
name="Seeded Pricing",
description="Seeded from defaults",
)
assert result_update.created is False
assert result_update.updated_fields > 0
assert result_update.impurity_upserts >= 1
updated = repo.get_by_slug("default", include_children=True)
assert updated.default_currency == "EUR"
as_override = {
item.impurity_code: item for item in updated.impurity_overrides}["AS"]
assert float(as_override.threshold_ppm) == 45.0
assert float(as_override.penalty_per_ppm) == 3.0
pb_override = {
item.impurity_code: item for item in updated.impurity_overrides}["PB"]
assert float(pb_override.threshold_ppm) == 12.0
def test_unit_of_work_exposes_pricing_settings(engine) -> None:
TestingSession = sessionmaker(
bind=engine, expire_on_commit=False, future=True)
metadata = PricingMetadata(
default_payable_pct=99.0,
default_currency="USD",
moisture_threshold_pct=7.0,
moisture_penalty_per_pct=125.0,
impurity_thresholds={"Zn": 80.0},
impurity_penalty_per_ppm={"Zn": 0.5},
)
with UnitOfWork(session_factory=TestingSession) as uow:
assert uow.pricing_settings is not None
result = uow.ensure_default_pricing_settings(
metadata=metadata,
slug="contract-core",
name="Contract Core",
)
assert result.settings.slug == "contract-core"
assert result.created is True
with UnitOfWork(session_factory=TestingSession) as uow:
assert uow.pricing_settings is not None
stored = uow.pricing_settings.get_by_slug(
"contract-core", include_children=True)
assert stored.default_payable_pct == 99.0
assert stored.impurity_overrides[0].impurity_code == "ZN"
def test_unit_of_work_get_pricing_metadata_returns_defaults(engine) -> None:
TestingSession = sessionmaker(
bind=engine, expire_on_commit=False, future=True)
seeded_metadata = PricingMetadata(
default_payable_pct=96.5,
default_currency="aud",
moisture_threshold_pct=6.25,
moisture_penalty_per_pct=210.0,
impurity_thresholds={"As": 45.0, "Pb": 15.0},
impurity_penalty_per_ppm={"As": 1.75, "Pb": 0.9},
)
with UnitOfWork(session_factory=TestingSession) as uow:
result = uow.ensure_default_pricing_settings(
metadata=seeded_metadata,
slug="default",
name="Default Contract",
description="Primary contract defaults",
)
assert result.created is True
with UnitOfWork(session_factory=TestingSession) as uow:
retrieved = uow.get_pricing_metadata()
assert retrieved is not None
assert retrieved.default_currency == "AUD"
assert retrieved.default_payable_pct == 96.5
assert retrieved.moisture_threshold_pct == 6.25
assert retrieved.moisture_penalty_per_pct == 210.0
assert retrieved.impurity_thresholds["AS"] == 45.0
assert retrieved.impurity_thresholds["PB"] == 15.0
assert retrieved.impurity_penalty_per_ppm["AS"] == 1.75
assert retrieved.impurity_penalty_per_ppm["PB"] == 0.9
def test_unit_of_work_get_pricing_metadata_returns_none_when_missing(engine) -> None:
TestingSession = sessionmaker(
bind=engine, expire_on_commit=False, future=True)
with UnitOfWork(session_factory=TestingSession) as uow:
missing = uow.get_pricing_metadata(slug="non-existent")
assert missing is None