251 lines
7.2 KiB
Python
251 lines
7.2 KiB
Python
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 _pythonpath_entry_path(entry):
|
|
if not entry:
|
|
# Empty PYTHONPATH entries mean the subprocess cwd.
|
|
return os.path.realpath(REPO_ROOT)
|
|
if os.path.isabs(entry):
|
|
return os.path.realpath(entry)
|
|
return os.path.realpath(os.path.join(REPO_ROOT, entry))
|
|
|
|
|
|
def _is_cpp_backend_path(entry):
|
|
path = _pythonpath_entry_path(entry)
|
|
cpp_dir = os.path.realpath(CPP_BUILD_DIR)
|
|
return path == cpp_dir or path.startswith(cpp_dir + os.sep)
|
|
|
|
|
|
def _parent_pythonpath_entries(include_cpp):
|
|
entries = os.environ.get("PYTHONPATH", "").split(os.pathsep)
|
|
if include_cpp:
|
|
return [entry for entry in entries if entry]
|
|
return [entry for entry in entries if entry and not _is_cpp_backend_path(entry)]
|
|
|
|
|
|
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)
|
|
pythonpath.extend(_parent_pythonpath_entries(include_cpp))
|
|
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_include_cpp_false_filters_parent_pythonpath(monkeypatch):
|
|
parent_pythonpath = os.pathsep.join(
|
|
[
|
|
"migration/protocol_core",
|
|
SRC_DIR,
|
|
os.path.join(CPP_BUILD_DIR, "build", "lib.fake-platform"),
|
|
]
|
|
)
|
|
monkeypatch.setenv("PYTHONPATH", parent_pythonpath)
|
|
|
|
result = run_probe("auto", include_cpp=False)
|
|
|
|
assert result["backend"] == "python"
|
|
assert result["fragmenter_module"] == "ble_reticulum.BLEFragmentation"
|
|
|
|
|
|
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
|