ble-reticulum/migration/tests/test_fragmentation_backend_shim.py

import json
import os
import subprocess
import sys

import pytest


REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
SRC_DIR = os.path.join(REPO_ROOT, "src")
CPP_BUILD_DIR = os.path.join(REPO_ROOT, "migration", "protocol_core")


PROBE = r"""
import json
from ble_reticulum.BLEFragmentationBackend import (
    BACKEND,
    BLEFragmenter,
    BLEReassembler,
    HDLCFramer,
)

fragmenter = BLEFragmenter(mtu=20)
reassembler = BLEReassembler(timeout=0.1)
packet = b"abc"
fragments = fragmenter.fragment_packet(packet)
result = reassembler.receive_fragment(fragments[0], "device1")
framed = HDLCFramer.frame_packet(packet)
deframed = HDLCFramer.deframe_packet(framed)

print(json.dumps({
    "backend": BACKEND,
    "fragmenter_module": BLEFragmenter.__module__,
    "reassembler_module": BLEReassembler.__module__,
    "hdlc_module": HDLCFramer.__module__,
    "fragment": fragments[0].hex(),
    "result": result.hex(),
    "deframed": deframed.hex(),
    "stats_keys": sorted(reassembler.get_statistics().keys()),
    "fragmenter_methods": sorted(
        name for name in dir(BLEFragmenter) if name in {
            "fragment_packet",
            "get_fragment_overhead",
        }
    ),
    "reassembler_methods": sorted(
        name for name in dir(BLEReassembler) if name in {
            "receive_fragment",
            "_reassemble",
            "cleanup_stale_buffers",
            "get_statistics",
            "reset_statistics",
        }
    ),
    "hdlc_methods": sorted(
        name for name in dir(HDLCFramer) if name in {
            "frame_packet",
            "deframe_packet",
        }
    ),
}))
"""


def run_probe(backend, include_cpp):
    env = os.environ.copy()
    env["BLE_RETICULUM_FRAGMENTATION_BACKEND"] = backend
    pythonpath = [SRC_DIR]
    if include_cpp:
        pythonpath.append(CPP_BUILD_DIR)
    if env.get("PYTHONPATH"):
        pythonpath.append(env["PYTHONPATH"])
    env["PYTHONPATH"] = os.pathsep.join(pythonpath)

    completed = subprocess.run(
        [sys.executable, "-c", PROBE],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        capture_output=True,
        check=True,
    )
    return json.loads(completed.stdout)


def test_cpp_backend_loads_when_available():
    result = run_probe("cpp", include_cpp=True)

    assert result["backend"] == "cpp"
    assert result["fragmenter_module"] == "ble_protocol_core_cpp"
    assert result["reassembler_module"] == "ble_protocol_core_cpp"
    assert result["hdlc_module"] == "ble_protocol_core_cpp"
    assert result["fragment"] == "0100000001616263"
    assert result["result"] == "616263"
    assert result["deframed"] == "616263"


def test_python_backend_still_works_when_cpp_backend_is_unavailable():
    result = run_probe("auto", include_cpp=False)

    assert result["backend"] == "python"
    assert result["fragmenter_module"] == "ble_reticulum.BLEFragmentation"
    assert result["fragment"] == "0100000001616263"
    assert result["result"] == "616263"
    assert result["deframed"] == "616263"


def test_auto_backend_prefers_cpp_when_available():
    result = run_probe("auto", include_cpp=True)

    assert result["backend"] == "cpp"
    assert result["fragmenter_module"] == "ble_protocol_core_cpp"


def test_python_backend_can_be_forced_even_when_cpp_backend_is_available():
    result = run_probe("python", include_cpp=True)

    assert result["backend"] == "python"
    assert result["fragmenter_module"] == "ble_reticulum.BLEFragmentation"


def test_cpp_backend_request_fails_clearly_when_unavailable():
    env = os.environ.copy()
    env["BLE_RETICULUM_FRAGMENTATION_BACKEND"] = "cpp"
    env["PYTHONPATH"] = SRC_DIR

    completed = subprocess.run(
        [sys.executable, "-c", "import ble_reticulum.BLEFragmentationBackend"],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        capture_output=True,
    )

    assert completed.returncode != 0
    assert "C++ BLE fragmentation backend is not available" in completed.stderr


def test_bleinterface_does_not_silently_fallback_when_cpp_requested_but_unavailable():
    env = os.environ.copy()
    env["BLE_RETICULUM_FRAGMENTATION_BACKEND"] = "cpp"
    env["PYTHONPATH"] = SRC_DIR

    completed = subprocess.run(
        [
            sys.executable,
            "-c",
            """
import sys, types
RNS = types.ModuleType('RNS')
sys.modules['RNS'] = RNS
transport = types.ModuleType('RNS.Transport')
sys.modules['RNS.Transport'] = transport
interfaces = types.ModuleType('RNS.Interfaces')
sys.modules['RNS.Interfaces'] = interfaces
interface_mod = types.ModuleType('RNS.Interfaces.Interface')
class Interface:
    MODE_FULL = 0
interface_mod.Interface = Interface
sys.modules['RNS.Interfaces.Interface'] = interface_mod
import ble_reticulum.BLEInterface
""",
        ],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        capture_output=True,
    )

    assert completed.returncode != 0
    assert "C++ BLE fragmentation backend is not available" in completed.stderr


def test_selected_backend_exposes_methods_expected_by_current_python_code():
    result = run_probe("cpp", include_cpp=True)

    assert result["fragmenter_methods"] == [
        "fragment_packet",
        "get_fragment_overhead",
    ]
    assert result["reassembler_methods"] == [
        "_reassemble",
        "cleanup_stale_buffers",
        "get_statistics",
        "receive_fragment",
        "reset_statistics",
    ]
    assert result["hdlc_methods"] == [
        "deframe_packet",
        "frame_packet",
    ]
    assert result["stats_keys"] == [
        "fragments_received",
        "packets_reassembled",
        "packets_timeout",
        "pending_packets",
    ]


def test_invalid_backend_setting_fails_clearly():
    env = os.environ.copy()
    env["BLE_RETICULUM_FRAGMENTATION_BACKEND"] = "rust"
    env["PYTHONPATH"] = SRC_DIR

    completed = subprocess.run(
        [sys.executable, "-c", "import ble_reticulum.BLEFragmentationBackend"],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        capture_output=True,
    )

    assert completed.returncode != 0
    assert "expected 'auto', 'cpp', or 'python'" in completed.stderr