Celes Renata
143 lines
5.1 KiB
Python
143 lines
5.1 KiB
Python
"""Tests for the operator approval workflow for live trading mode.
|
|
|
|
Validates:
|
|
- requires_approval logic for paper/live/disabled modes
|
|
- ApprovalRequest model behavior (pending, expired)
|
|
- compute_expiry calculation
|
|
- Integration with broker service process_order_job flow
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from services.risk.approval import (
|
|
ApprovalRequest,
|
|
ApprovalStatus,
|
|
compute_expiry,
|
|
requires_approval,
|
|
)
|
|
from services.risk.engine import (
|
|
OperatorApproval,
|
|
PortfolioRiskConfig,
|
|
TradingMode,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# requires_approval tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRequiresApproval:
|
|
def test_paper_mode_auto_approved(self):
|
|
"""Paper orders are auto-approved by default."""
|
|
config = PortfolioRiskConfig(trading_mode=TradingMode.PAPER)
|
|
assert requires_approval(config) is False
|
|
|
|
def test_paper_mode_approval_required_when_auto_approve_off(self):
|
|
"""Paper orders need approval when auto_approve_paper is False."""
|
|
config = PortfolioRiskConfig(
|
|
trading_mode=TradingMode.PAPER,
|
|
operator_approval=OperatorApproval(auto_approve_paper=False),
|
|
)
|
|
assert requires_approval(config) is True
|
|
|
|
def test_live_mode_requires_approval_by_default(self):
|
|
"""Live orders require approval by default."""
|
|
config = PortfolioRiskConfig(trading_mode=TradingMode.LIVE)
|
|
assert requires_approval(config) is True
|
|
|
|
def test_live_mode_no_approval_when_disabled(self):
|
|
"""Live orders skip approval when require_approval_for_live is False."""
|
|
config = PortfolioRiskConfig(
|
|
trading_mode=TradingMode.LIVE,
|
|
operator_approval=OperatorApproval(require_approval_for_live=False),
|
|
)
|
|
assert requires_approval(config) is False
|
|
|
|
def test_disabled_mode_never_requires_approval(self):
|
|
"""Disabled mode never requires approval (blocked upstream)."""
|
|
config = PortfolioRiskConfig(trading_mode=TradingMode.DISABLED)
|
|
assert requires_approval(config) is False
|
|
|
|
def test_override_trading_mode_parameter(self):
|
|
"""The trading_mode parameter overrides config.trading_mode."""
|
|
config = PortfolioRiskConfig(trading_mode=TradingMode.PAPER)
|
|
# Override to live — should require approval
|
|
assert requires_approval(config, trading_mode=TradingMode.LIVE) is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# compute_expiry tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestComputeExpiry:
|
|
def test_default_30_minutes(self):
|
|
now = datetime(2026, 4, 11, 12, 0, 0, tzinfo=timezone.utc)
|
|
config = PortfolioRiskConfig()
|
|
expiry = compute_expiry(config, now=now)
|
|
assert expiry == now + timedelta(minutes=30)
|
|
|
|
def test_custom_timeout(self):
|
|
now = datetime(2026, 4, 11, 12, 0, 0, tzinfo=timezone.utc)
|
|
config = PortfolioRiskConfig(
|
|
operator_approval=OperatorApproval(approval_timeout_minutes=60),
|
|
)
|
|
expiry = compute_expiry(config, now=now)
|
|
assert expiry == now + timedelta(minutes=60)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ApprovalRequest model tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestApprovalRequest:
|
|
def test_defaults(self):
|
|
req = ApprovalRequest(ticker="AAPL")
|
|
assert req.ticker == "AAPL"
|
|
assert req.status == ApprovalStatus.PENDING
|
|
assert req.is_pending is True
|
|
assert req.approval_id # auto-generated UUID
|
|
|
|
def test_is_expired_when_past_expiry(self):
|
|
past = datetime.now(timezone.utc) - timedelta(minutes=5)
|
|
req = ApprovalRequest(ticker="AAPL", expires_at=past)
|
|
assert req.is_expired is True
|
|
|
|
def test_not_expired_when_future_expiry(self):
|
|
future = datetime.now(timezone.utc) + timedelta(minutes=30)
|
|
req = ApprovalRequest(ticker="AAPL", expires_at=future)
|
|
assert req.is_expired is False
|
|
|
|
def test_approved_is_not_expired(self):
|
|
past = datetime.now(timezone.utc) - timedelta(minutes=5)
|
|
req = ApprovalRequest(
|
|
ticker="AAPL",
|
|
status=ApprovalStatus.APPROVED,
|
|
expires_at=past,
|
|
)
|
|
assert req.is_expired is False
|
|
|
|
def test_to_dict_roundtrip(self):
|
|
req = ApprovalRequest(
|
|
ticker="MSFT",
|
|
side="sell",
|
|
quantity=50.0,
|
|
estimated_value=15000.0,
|
|
recommendation_id="rec-123",
|
|
)
|
|
d = req.to_dict()
|
|
assert d["ticker"] == "MSFT"
|
|
assert d["side"] == "sell"
|
|
assert d["quantity"] == 50.0
|
|
assert d["status"] == "pending"
|
|
assert d["recommendation_id"] == "rec-123"
|
|
|
|
def test_explicit_expired_status(self):
|
|
req = ApprovalRequest(
|
|
ticker="AAPL",
|
|
status=ApprovalStatus.EXPIRED,
|
|
)
|
|
assert req.is_expired is True
|
|
assert req.is_pending is False
|