37 lines
1.4 KiB
Python
37 lines
1.4 KiB
Python
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_production_main_uses_rust_receiver_not_python_packet_receiver():
|
|
main = (ROOT / "app" / "main.py").read_text()
|
|
assert "from .tzsp_rust import RustTZSPReceiver" in main
|
|
assert "from .tzsp import TZSPReceiver" not in main
|
|
assert "TapDevice(" not in main
|
|
assert "flow_tracker.observe(" not in main
|
|
|
|
|
|
def test_docker_builds_and_tests_rust_receiver_before_runtime_copy():
|
|
dockerfile = (ROOT / "Dockerfile").read_text()
|
|
assert "AS rust-builder" in dockerfile
|
|
assert "cargo test --release" in dockerfile
|
|
assert "cargo build --release" in dockerfile
|
|
assert "COPY --from=rust-builder" in dockerfile
|
|
assert "/usr/local/bin/mikrosuricata-tzsp" in dockerfile
|
|
|
|
|
|
def test_rust_receiver_is_dependency_free_and_uses_batched_linux_receive():
|
|
cargo = (ROOT / "rust" / "tzsp-receiver" / "Cargo.toml").read_text()
|
|
source = (ROOT / "rust" / "tzsp-receiver" / "src" / "main.rs").read_text()
|
|
assert "[dependencies]" not in cargo
|
|
assert "recvmmsg(" in source
|
|
assert "MSG_WAITFORONE" in source
|
|
assert "sync_channel::<BatchBuffer>" in source
|
|
assert 'name("tzsp-udp-rx"' in source
|
|
assert 'name("tzsp-tap-worker"' in source
|
|
assert "queue_dropped_datagrams" in source
|
|
assert "SO_RCVBUFFORCE" in source
|
|
assert "TUNSETIFF" in source
|
|
assert "tap.write(frame)" in source
|
|
assert "write_all(frame)" not in source
|