Shadowbroker/backend/tests/mesh/test_s12b_transport_truth.py

"""S12B Transport Truth Enforcement.

Tests:
- transport_tier_from_state returns correct tiers for all states
- tier ordering remains coherent for route checks
- DM MLS PRIVATE_STRONG expectations do not regress
- wormhole status endpoints expose the new tier honestly
- mesh DM send responses include machine-readable carrier truth
"""

import pytest
from typing import Any

from services.wormhole_supervisor import transport_tier_from_state


# ── transport_tier_from_state returns correct tiers ────────────────────


def test_tier_public_degraded_not_configured():
    assert transport_tier_from_state({"configured": False, "ready": False}) == "public_degraded"


def test_tier_public_degraded_not_ready():
    assert transport_tier_from_state({"configured": True, "ready": False}) == "public_degraded"


def test_tier_control_only_no_carriers():
    """Configured+ready but neither Arti nor RNS -> private_control_only."""
    state = {"configured": True, "ready": True, "arti_ready": False, "rns_ready": False}
    assert transport_tier_from_state(state) == "private_control_only"


def test_tier_transitional_arti_only():
    state = {"configured": True, "ready": True, "arti_ready": True, "rns_ready": False}
    assert transport_tier_from_state(state) == "private_transitional"


def test_tier_transitional_rns_only():
    state = {"configured": True, "ready": True, "arti_ready": False, "rns_ready": True}
    assert transport_tier_from_state(state) == "private_transitional"


def test_tier_strong_both():
    state = {"configured": True, "ready": True, "arti_ready": True, "rns_ready": True}
    assert transport_tier_from_state(state) == "private_strong"


def test_tier_none_state():
    assert transport_tier_from_state(None) == "public_degraded"


def test_tier_empty_state():
    assert transport_tier_from_state({}) == "public_degraded"


# ── Tier ordering coherence ────────────────────────────────────────────


def test_tier_ordering_coherence():
    """Tier ordering must be: public_degraded < private_control_only < private_transitional < private_strong."""
    from auth import _TRANSPORT_TIER_ORDER

    assert _TRANSPORT_TIER_ORDER["public_degraded"] < _TRANSPORT_TIER_ORDER["private_control_only"]
    assert _TRANSPORT_TIER_ORDER["private_control_only"] < _TRANSPORT_TIER_ORDER["private_transitional"]
    assert _TRANSPORT_TIER_ORDER["private_transitional"] < _TRANSPORT_TIER_ORDER["private_strong"]


def test_tier_sufficiency_checks():
    """Route-tier sufficiency must respect the new ordering."""
    from auth import _transport_tier_is_sufficient

    # private_control_only is NOT sufficient for private_transitional routes
    assert not _transport_tier_is_sufficient("private_control_only", "private_transitional")

    # private_control_only IS sufficient for private_control_only
    assert _transport_tier_is_sufficient("private_control_only", "private_control_only")

    # private_transitional is sufficient for private_transitional
    assert _transport_tier_is_sufficient("private_transitional", "private_transitional")

    # private_strong is sufficient for everything
    assert _transport_tier_is_sufficient("private_strong", "private_strong")
    assert _transport_tier_is_sufficient("private_strong", "private_transitional")
    assert _transport_tier_is_sufficient("private_strong", "private_control_only")

    # public_degraded is not sufficient for anything private
    assert not _transport_tier_is_sufficient("public_degraded", "private_control_only")
    assert not _transport_tier_is_sufficient("public_degraded", "private_transitional")


def test_control_only_blocks_transitional_routes():
    """private_control_only must NOT satisfy private_transitional route requirements."""
    from auth import _transport_tier_is_sufficient

    assert not _transport_tier_is_sufficient("private_control_only", "private_transitional")
    assert not _transport_tier_is_sufficient("private_control_only", "private_strong")


def test_control_only_satisfies_gate_lifecycle_routes():
    """Gate entry/persona lifecycle can proceed once Wormhole is ready, even without a private carrier."""
    from auth import _minimum_transport_tier, _transport_tier_is_sufficient

    assert _minimum_transport_tier("/api/wormhole/gate/enter", "POST") == "private_control_only"
    assert _minimum_transport_tier("/api/wormhole/gate/persona/create", "POST") == "private_control_only"
    assert _transport_tier_is_sufficient("private_control_only", "private_control_only")


# ── DM MLS PRIVATE_STRONG expectation does not regress ────────────────


def test_dm_mls_tier_order_coherent():
    """mesh_dm_mls._TRANSPORT_TIER_ORDER must match auth ordering."""
    from services.mesh.mesh_dm_mls import _TRANSPORT_TIER_ORDER

    assert _TRANSPORT_TIER_ORDER["public_degraded"] < _TRANSPORT_TIER_ORDER["private_control_only"]
    assert _TRANSPORT_TIER_ORDER["private_control_only"] < _TRANSPORT_TIER_ORDER["private_transitional"]
    assert _TRANSPORT_TIER_ORDER["private_transitional"] < _TRANSPORT_TIER_ORDER["private_strong"]


def test_dm_mls_private_control_only_is_weaker_than_private_strong():
    """DM MLS now opens at PRIVATE_CONTROL_ONLY, and PRIVATE_STRONG remains stronger."""
    from services.mesh.mesh_dm_mls import _TRANSPORT_TIER_ORDER

    assert _TRANSPORT_TIER_ORDER["private_control_only"] < _TRANSPORT_TIER_ORDER["private_strong"]
    assert _TRANSPORT_TIER_ORDER["private_transitional"] < _TRANSPORT_TIER_ORDER["private_strong"]


# ── Wormhole status endpoints expose the new tier ─────────────────────


def test_status_snapshot_includes_transport_tier():
    """_current_runtime_state snapshot must include transport_tier."""
    # We can't call the full runtime state, but we can verify
    # transport_tier_from_state produces the new tier.
    state = {"configured": True, "ready": True, "arti_ready": False, "rns_ready": False}
    tier = transport_tier_from_state(state)
    assert tier == "private_control_only"
    # Verify it would be included in a snapshot dict.
    snapshot = {**state, "transport_tier": tier}
    assert snapshot["transport_tier"] == "private_control_only"


def test_status_snapshot_private_strong():
    state = {"configured": True, "ready": True, "arti_ready": True, "rns_ready": True}
    snapshot = {**state, "transport_tier": transport_tier_from_state(state)}
    assert snapshot["transport_tier"] == "private_strong"


def test_status_snapshot_transitional():
    state = {"configured": True, "ready": True, "arti_ready": True, "rns_ready": False}
    snapshot = {**state, "transport_tier": transport_tier_from_state(state)}
    assert snapshot["transport_tier"] == "private_transitional"


# ── Mesh DM send carrier truth ────────────────────────────────────────


def test_dm_send_reticulum_direct_response_shape():
    """Reticulum direct DM responses must include carrier and NOT overstate hidden transport."""
    # Simulate the response structure from the reticulum direct path.
    # Direct RNS is private carriage, not hidden transport (Tor/I2P/mixnet).
    response = {
        "ok": True,
        "msg_id": "test-123",
        "transport": "reticulum",
        "carrier": "reticulum_direct",
        "hidden_transport_effective": False,
        "detail": "Delivered via Reticulum",
    }
    assert response["carrier"] == "reticulum_direct"
    assert response["hidden_transport_effective"] is False


def test_dm_send_relay_response_shape():
    """Relay DM responses must include carrier and hidden_transport_effective."""
    response = {
        "ok": True,
        "msg_id": "test-456",
        "transport": "relay",
        "carrier": "relay",
        "hidden_transport_effective": False,
    }
    assert response["carrier"] == "relay"
    assert response["hidden_transport_effective"] is False


def test_dm_send_anonymous_relay_response_shape():
    """Anonymous-mode relay DM responses must show hidden_transport_effective=True."""
    response = {
        "ok": True,
        "msg_id": "test-789",
        "transport": "relay",
        "carrier": "relay",
        "hidden_transport_effective": True,
        "detail": "Anonymous mode keeps private DMs off direct transport; delivered via hidden relay path",
    }
    assert response["carrier"] == "relay"
    assert response["hidden_transport_effective"] is True


# ── Do not overclaim hidden transport ─────────────────────────────────


def test_control_only_is_not_private_transport():
    """private_control_only must not be treated as having actual private carriage."""
    tier = transport_tier_from_state({
        "configured": True, "ready": True,
        "arti_ready": False, "rns_ready": False,
    })
    assert tier == "private_control_only"
    assert not tier.endswith("transitional")
    assert not tier.endswith("strong")