"""Dry-run executor guard tests — Work Unit 8. Verifies the `dry_run_signals_only` contract specified in the Unit 5 design (final-design.md §10.1 (d), §14 integration row on dry_run enforcement layer). Unit 5 (src/micro_trading.py) has NOT landed yet at the time this file was written. Tests are written against the **contract**, not the implementation. All tests in TestDryRunGuardAgainstContract are marked with pytest.mark.skipif(reason="requires Unit 5 — src/micro_trading.py") and will be skipped until that unit merges. TestDryRunContractStub tests the same invariants against a thin stub that mirrors the Unit 5 interface exactly. These run immediately without any dependency and serve as living documentation of the expected behaviour. Integrators: once Unit 5 lands, unskip the first suite and confirm both pass green. Contract under test: - MicroConfig(dry_run_signals_only=True) + valid entry signal: 1. executor (place_entry) is NOT called. 2. `entry_signal_would_fire` INFO log is emitted with full signal payload. 3. scan rollup is still emitted. 4. audit row is still written when audit_log_rejections=True. - MicroConfig(dry_run_signals_only=False) + same valid entry signal: 1. executor (place_entry) IS called. """ from __future__ import annotations import dataclasses import json import uuid from decimal import Decimal from typing import Any from unittest.mock import AsyncMock, MagicMock import pytest # --------------------------------------------------------------------------- # Minimal MicroConfig stub — mirrors Unit 1 contract. # If Unit 1 has already landed, import the real class instead. # --------------------------------------------------------------------------- try: from micro_scanner.config import MicroConfig as _RealMicroConfig # type: ignore[import] def _make_config(**overrides) -> Any: cfg = _RealMicroConfig() # Ensure new v2 fields exist; fall back to setattr if not yet on dataclass for k, v in overrides.items(): setattr(cfg, k, v) return cfg _USING_REAL_CONFIG = True except ImportError: _USING_REAL_CONFIG = False @dataclasses.dataclass class _StubMicroConfig: # type: ignore[no-redef] """Minimal stub matching the Unit 1 MicroConfig contract.""" dry_run_signals_only: bool = False audit_log_rejections: bool = False gate_rollup_every_n_scans: int = 1 composite_observer_enabled: bool = True composite_observer_threshold: float = 0.55 scan_deadline_seconds: float = 50.0 max_rejection_lines_per_scan: int = 20 score_entry_threshold: float = 0.55 score_full_position: float = 0.55 signal_max_age_seconds: int = 90 watchlist_max: int = 30 def _make_config(**overrides) -> _StubMicroConfig: cfg = _StubMicroConfig() for k, v in overrides.items(): setattr(cfg, k, v) return cfg # --------------------------------------------------------------------------- # Thin stub executor — models the Unit 5 micro_trading.py interface # so the contract tests can run without Unit 5 being present. # --------------------------------------------------------------------------- class _StubMicroTrader: """Minimal implementation of the dry-run executor guard contract. This class INTENTIONALLY mirrors only the behaviour mandated by the design spec so that contract tests are not coupled to Unit 5's internal implementation details. Integrator note: once Unit 5 lands, the TestDryRunGuardAgainstContract suite imports and tests the real class directly. These stub tests remain as a fast, dependency-free regression gate. """ def __init__(self, config: Any, place_entry_fn: Any, audit_writer: Any) -> None: self._config = config self._place_entry = place_entry_fn self._audit_writer = audit_writer self._scan_count = 0 self.log_records: list[dict] = [] def _log(self, event: str, level: str = "info", **fields: Any) -> None: self.log_records.append({"event": event, "level": level, **fields}) async def scan_and_trade(self, signal_payload: dict) -> dict: """Reduced scan loop for contract testing. Returns a rollup dict so tests can assert its contents. """ self._scan_count += 1 # --- Dry-run guard at the executor boundary --- if self._config.dry_run_signals_only: self._log( "entry_signal_would_fire", level="info", **signal_payload, ) # executor NOT called — fall through to rollup/audit else: # Real path — executor IS called await self._place_entry(**signal_payload) # --- Rollup (always emitted) --- rollup = { "scan_id": str(uuid.uuid4()), "signals_emitted": 0 if self._config.dry_run_signals_only else 1, "dry_run": self._config.dry_run_signals_only, } if self._scan_count % self._config.gate_rollup_every_n_scans == 0: self._log("micro_scan_rollup", level="info", **rollup) # --- Audit persistence (always attempted when enabled) --- if self._config.audit_log_rejections: try: self._audit_writer( event_type="scan_rollup", symbol="__SCAN__", detail=json.dumps(rollup), ) except Exception: pass # never crash the bot on audit failure return rollup # --------------------------------------------------------------------------- # Helper: build a representative signal payload # --------------------------------------------------------------------------- def _make_signal(symbol: str = "BLURUSDT") -> dict: return { "symbol": symbol, "score": 0.92, "confidence": 0.75, "signal_id": str(uuid.uuid4()), } # --------------------------------------------------------------------------- # Contract tests against the stub (run immediately, no Unit 5 dependency) # --------------------------------------------------------------------------- class TestDryRunContractStub: """Contract tests executed against _StubMicroTrader. These pass even before Unit 5 lands. They document the expected behaviour in executable form and remain as a fast regression gate after Unit 5 ships. """ @pytest.fixture() def place_entry(self) -> AsyncMock: return AsyncMock(return_value={"id": "order-ok"}) @pytest.fixture() def audit_writer(self) -> MagicMock: return MagicMock() def _make_trader( self, dry_run: bool, audit: bool, place_entry: AsyncMock, audit_writer: MagicMock, ) -> _StubMicroTrader: cfg = _make_config(dry_run_signals_only=dry_run, audit_log_rejections=audit) return _StubMicroTrader(cfg, place_entry, audit_writer) @pytest.mark.asyncio async def test_dry_run_true_executor_not_called( self, place_entry: AsyncMock, audit_writer: MagicMock ) -> None: """With dry_run_signals_only=True, place_entry must NOT be called.""" trader = self._make_trader( dry_run=True, audit=False, place_entry=place_entry, audit_writer=audit_writer, ) await trader.scan_and_trade(_make_signal()) place_entry.assert_not_called() @pytest.mark.asyncio async def test_dry_run_true_entry_signal_would_fire_logged( self, place_entry: AsyncMock, audit_writer: MagicMock ) -> None: """With dry_run_signals_only=True, 'entry_signal_would_fire' is logged at INFO.""" trader = self._make_trader( dry_run=True, audit=False, place_entry=place_entry, audit_writer=audit_writer, ) signal = _make_signal() await trader.scan_and_trade(signal) fired_logs = [ r for r in trader.log_records if r["event"] == "entry_signal_would_fire" ] assert len(fired_logs) == 1 assert fired_logs[0]["level"] == "info" # Full signal payload must appear in the log assert fired_logs[0]["symbol"] == signal["symbol"] assert fired_logs[0]["score"] == signal["score"] assert fired_logs[0]["signal_id"] == signal["signal_id"] @pytest.mark.asyncio async def test_dry_run_true_rollup_still_emitted( self, place_entry: AsyncMock, audit_writer: MagicMock ) -> None: """With dry_run_signals_only=True, micro_scan_rollup is still logged.""" trader = self._make_trader( dry_run=True, audit=False, place_entry=place_entry, audit_writer=audit_writer, ) await trader.scan_and_trade(_make_signal()) rollup_logs = [ r for r in trader.log_records if r["event"] == "micro_scan_rollup" ] assert len(rollup_logs) == 1 @pytest.mark.asyncio async def test_dry_run_true_audit_written_when_enabled( self, place_entry: AsyncMock, audit_writer: MagicMock ) -> None: """With dry_run_signals_only=True + audit_log_rejections=True, audit row is written.""" trader = self._make_trader( dry_run=True, audit=True, place_entry=place_entry, audit_writer=audit_writer ) await trader.scan_and_trade(_make_signal()) audit_writer.assert_called_once() kwargs = audit_writer.call_args.kwargs assert kwargs["event_type"] == "scan_rollup" assert kwargs["symbol"] == "__SCAN__" assert "scan_id" in json.loads(kwargs["detail"]) @pytest.mark.asyncio async def test_dry_run_false_executor_is_called( self, place_entry: AsyncMock, audit_writer: MagicMock ) -> None: """Counter-test: with dry_run_signals_only=False, place_entry IS called.""" trader = self._make_trader( dry_run=False, audit=False, place_entry=place_entry, audit_writer=audit_writer, ) signal = _make_signal() await trader.scan_and_trade(signal) place_entry.assert_called_once() @pytest.mark.asyncio async def test_dry_run_false_no_would_fire_log( self, place_entry: AsyncMock, audit_writer: MagicMock ) -> None: """Counter-test: with dry_run_signals_only=False, 'entry_signal_would_fire' is NOT logged.""" trader = self._make_trader( dry_run=False, audit=False, place_entry=place_entry, audit_writer=audit_writer, ) await trader.scan_and_trade(_make_signal()) fired_logs = [ r for r in trader.log_records if r["event"] == "entry_signal_would_fire" ] assert len(fired_logs) == 0 @pytest.mark.asyncio async def test_audit_not_written_when_disabled( self, place_entry: AsyncMock, audit_writer: MagicMock ) -> None: """Audit writer must NOT be called when audit_log_rejections=False.""" trader = self._make_trader( dry_run=True, audit=False, place_entry=place_entry, audit_writer=audit_writer, ) await trader.scan_and_trade(_make_signal()) audit_writer.assert_not_called() @pytest.mark.asyncio async def test_audit_write_failure_does_not_propagate( self, place_entry: AsyncMock, audit_writer: MagicMock ) -> None: """If the audit writer raises, scan_and_trade must not propagate the exception.""" audit_writer.side_effect = Exception("DB connection lost") trader = self._make_trader( dry_run=True, audit=True, place_entry=place_entry, audit_writer=audit_writer ) # Must not raise await trader.scan_and_trade(_make_signal()) # --------------------------------------------------------------------------- # Contract tests against the real Unit 5 implementation (skipped until landed) # --------------------------------------------------------------------------- _UNIT5_AVAILABLE = False try: import importlib.util _spec = importlib.util.find_spec("src.micro_trading") # type: ignore[attr-defined] if _spec is not None: _UNIT5_AVAILABLE = True except Exception: pass @pytest.mark.skipif( not _UNIT5_AVAILABLE, reason="requires Unit 5 — src/micro_trading.py", ) class TestDryRunGuardAgainstUnit5: """Integration-level contract tests against the real Unit 5 class. These tests exercise the actual dry-run guard Unit 5 implemented at ``MicroTradingManager._evaluate_entry_for_symbol`` (micro_trading.py line ~378), which is where the executor-boundary guard lives. The guard reads ``self._micro_cfg.dry_run_signals_only`` and, when True, emits an ``entry_signal_would_fire`` INFO log and returns BEFORE calling ``self._executor.place_entry(...)``. The MicroTradingManager constructor wires live exchange/DB/executor; we bypass it via ``__new__`` + hand-assignment (same pattern as tests/test_micro_trading_v2.py) so these tests run with zero external I/O. """ def _make_mgr( self, *, dry_run: bool, executor_mock: AsyncMock, ) -> Any: import asyncio as _asyncio # noqa: PLC0415 from dataclasses import dataclass, field # noqa: PLC0415 from src.micro_scanner.config import MicroConfig # noqa: PLC0415 from src.micro_trading import MicroTradingManager # noqa: PLC0415 @dataclass class _FakeSigilConfig: micro: MicroConfig = field(default_factory=MicroConfig) mode: str = "draft" micro = MicroConfig() micro.dry_run_signals_only = dry_run micro.audit_log_rejections = False mgr = MicroTradingManager.__new__(MicroTradingManager) mgr._config = _FakeSigilConfig(micro=micro, mode="draft") mgr._micro_cfg = micro mgr._db_conn = MagicMock() mgr._executor = MagicMock() mgr._executor.place_entry = executor_mock mgr._exchange = MagicMock() mgr._market_feed = None mgr._alert_fn = MagicMock() mgr._universe_scanner = MagicMock() mgr._watchlist = MagicMock() mgr._exit_managers = {} mgr._scan_lock = _asyncio.Lock() mgr._scan_counter = 0 mgr._audit_write_failed = 0 mgr._current_scan_state = { "scan_id": "test-scan", "evaluated": 0, "signals_emitted": 0, "passed_all_gates": 0, "rejected_score": 0, "rejected_stale": 0, "rejected_gates": { "trend": 0, "volume": 0, "spread": 0, "rsi": 0, "depth": 0, "resistance": 0, }, "composite_above_threshold": 0, "composite_below_threshold": 0, "symbol_faults": 0, "rejection_lines_emitted": 0, } mgr._symbol_protection_store = None # Mock the scoring/signal pipeline to feed a clean entry signal # through to the dry-run guard. class _FakeScore: score = 0.9 computed_at = None mgr._scorer = MagicMock() mgr._scorer.score = MagicMock(return_value=_FakeScore()) class _FakeSignal: signal_id = "sig-123" confidence = 0.75 mgr._signal_engine = MagicMock() mgr._signal_engine.evaluate_entry = MagicMock(return_value=_FakeSignal()) mgr._capital = MagicMock() mgr._capital.calculate_position_size = MagicMock(return_value=Decimal("50")) # Stub out coroutines that would otherwise hit the exchange. mgr._fetch_candles_1m = AsyncMock( return_value=[{"close": 1.0, "volume": 1000.0} for _ in range(30)] ) mgr._record_gate_telemetry = MagicMock() mgr._register_exit_manager = MagicMock() mgr._build_depth_from_ticker = MagicMock( return_value={ "bid": 1.0, "ask": 1.01, "bid_depth": 5000.0, "ask_depth": 5000.0, "from_ticker": True, } ) return mgr @pytest.mark.asyncio async def test_real_dry_run_executor_not_called(self) -> None: """Unit 5 guard: dry_run_signals_only=True → _executor.place_entry NOT called.""" place_entry_mock = AsyncMock(return_value=None) mgr = self._make_mgr(dry_run=True, executor_mock=place_entry_mock) tickers = {"FOOUSDT": {"quoteVolume": 500_000.0, "baseVolume": 100_000.0, "high": 1.03, "low": 1.0, "last": 1.0}} await mgr._evaluate_entry_for_symbol("FOOUSDT", tickers, max_volume=1_000_000.0) place_entry_mock.assert_not_called() @pytest.mark.asyncio async def test_real_dry_run_false_executor_called(self) -> None: """Unit 5 guard: dry_run_signals_only=False → _executor.place_entry IS called.""" place_entry_mock = AsyncMock(return_value=None) mgr = self._make_mgr(dry_run=False, executor_mock=place_entry_mock) tickers = {"FOOUSDT": {"quoteVolume": 500_000.0, "baseVolume": 100_000.0, "high": 1.03, "low": 1.0, "last": 1.0}} await mgr._evaluate_entry_for_symbol("FOOUSDT", tickers, max_volume=1_000_000.0) place_entry_mock.assert_called_once()