siloqy/prod/clean_arch/violet/test_violet_domain.py

"""V2a: V-TYPES domain layer — refined types reject by construction."""

from __future__ import annotations

import importlib
import math
import re
import sys
from pathlib import Path

sys.path.insert(0, "/mnt/dolphinng5_predict")

import pytest
from pydantic import TypeAdapter, ValidationError

from prod.clean_arch.violet.domain import (
    BarsHeld,
    DivergenceRow,
    ExecDriverSettings,
    Px,
    Symbol,
)

DDL_PATH = Path(
    "/mnt/dolphinng5_predict/prod/clickhouse/violet/20_violet_feed_divergence.sql"
)

VALID_ROW = dict(
    ts=1_781_300_000_000, session_id="sess", asset="FETUSDT",
    scan_price=0.2176, venue_mid=0.1878, divergence_bps=-1369.49,
    scan_seq=1, venue_seq=7, mono_ns=123,
)


def test_bars_held_kills_the_incident():
    """bars_held=-106 (the 2026-06-12 zombie-trades poison) must be
    UNREPRESENTABLE — and so must the UInt16 overflow side."""
    ta = TypeAdapter(BarsHeld)
    assert ta.validate_python(0) == 0
    assert ta.validate_python(65535) == 65535
    with pytest.raises(ValidationError):
        ta.validate_python(-106)
    with pytest.raises(ValidationError):
        ta.validate_python(65536)


def test_px_rejects_nonfinite_and_nonpositive():
    ta = TypeAdapter(Px)
    assert ta.validate_python(0.2176) == 0.2176
    for bad in (0.0, -1.0, math.nan, math.inf, -math.inf):
        with pytest.raises(ValidationError):
            ta.validate_python(bad)


def test_symbol_shape():
    ta = TypeAdapter(Symbol)
    assert ta.validate_python("BTCUSDT") == "BTCUSDT"
    assert ta.validate_python("1000PEPEUSDT") == "1000PEPEUSDT"
    assert ta.validate_python("BTC-USDT") == "BTC-USDT"
    for bad in ("", "btcusdt", "BTC USDT", "X" * 33):
        with pytest.raises(ValidationError):
            ta.validate_python(bad)


def test_divergence_row_valid_and_frozen():
    row = DivergenceRow(**VALID_ROW)
    assert row.model_dump() == VALID_ROW            # exact round-trip
    with pytest.raises(ValidationError):
        row.ts = 1  # frozen


@pytest.mark.parametrize("field,bad", [
    ("ts", 0), ("ts", -5),
    ("session_id", ""),
    ("asset", "fetusdt"),
    ("scan_price", 0.0), ("scan_price", float("nan")),
    ("venue_mid", float("inf")),
    ("divergence_bps", float("nan")),
    ("scan_seq", -1), ("venue_seq", -1), ("mono_ns", -1),
])
def test_divergence_row_rejects(field, bad):
    with pytest.raises(ValidationError):
        DivergenceRow(**{**VALID_ROW, field: bad})


def test_divergence_row_forbids_extras():
    with pytest.raises(ValidationError):
        DivergenceRow(**VALID_ROW, bars_held=1)


def test_divergence_row_fields_match_ddl_columns():
    cols = set(re.findall(r"`(\w+)`", DDL_PATH.read_text()))
    assert set(DivergenceRow.model_fields) == cols, (
        set(DivergenceRow.model_fields) ^ cols
    )


def test_exec_settings_defaults_and_ttl_logic():
    s = ExecDriverSettings()
    assert s.ttl_override_ms == 100.0
    assert s.requote_hot_window_ns == 5_000_000_000
    # router maker plan (8s entry TTL) bows to the 100ms override
    assert s.ttl_ms_for(8.0) == 100.0
    # plan shorter than override wins
    assert ExecDriverSettings(ttl_override_ms=10_000.0).ttl_ms_for(5.0) == 5_000.0
    # taker ttl_s=0 → override alone
    assert s.ttl_ms_for(0.0) == 100.0
    # override disabled → plan verbatim
    p = ExecDriverSettings(ttl_override_ms=None)
    assert p.ttl_ms_for(8.0) == 8_000.0


def test_exec_settings_from_env():
    s = ExecDriverSettings.from_env({"DOLPHIN_VIOLET_EXEC_TTL_MS": "250",
                                     "DOLPHIN_VIOLET_EXEC_REQUOTE_HOT_S": "2"})
    assert s.ttl_override_ms == 250.0
    assert s.requote_hot_window_ns == 2_000_000_000
    assert ExecDriverSettings.from_env(
        {"DOLPHIN_VIOLET_EXEC_TTL_MS": "plan"}).ttl_override_ms is None
    # malformed env value raises at boot — loud reject at the source
    with pytest.raises(ValueError):
        ExecDriverSettings.from_env({"DOLPHIN_VIOLET_EXEC_TTL_MS": "fast"})
    with pytest.raises(ValidationError):
        ExecDriverSettings.from_env({"DOLPHIN_VIOLET_EXEC_TTL_MS": "-100"})


def test_typed_enforces_and_kill_switch(monkeypatch):
    import prod.clean_arch.violet.domain as domain

    @domain.typed
    def f(x: int) -> int:
        return x

    assert f(3) == 3
    with pytest.raises(Exception):     # BeartypeCallHintParamViolation
        f("not-an-int")

    monkeypatch.setenv("DOLPHIN_VIOLET_BEARTYPE", "0")
    try:
        importlib.reload(domain)

        @domain.typed
        def g(x: int) -> int:
            return x

        assert g("passes-when-killed") == "passes-when-killed"
    finally:
        monkeypatch.delenv("DOLPHIN_VIOLET_BEARTYPE", raising=False)
        importlib.reload(domain)


def test_divergence_monitor_rejects_at_source():
    """End-to-end: a poisoned session_id means zero rows reach the sink and
    the reject counter advances — the spool never sees the row."""
    from types import SimpleNamespace
    from prod.clean_arch.violet.clock import PlaneClock
    from prod.clean_arch.violet.divergence import FeedDivergenceMonitor

    rows = []
    m = FeedDivergenceMonitor(
        sink=lambda t, r: rows.append(r),
        scan_clock=PlaneClock("scan", 12_000_000_000),
        venue_clock=PlaneClock("venue", 2_000_000_000),
        session_id="",                         # invalid by construction
    )
    m.on_venue_tick("BTC-USDT", 100.0, 100.0)
    m.on_scan(SimpleNamespace(scan_payload={"assets": ["BTCUSDT"],
                                            "asset_prices": [100.0]}))
    assert rows == []
    assert m.rows_rejected == 1
    assert m.rows_emitted == 0


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-v"]))