mirror of
https://github.com/l5yth/potato-mesh.git
synced 2026-10-31 03:51:58 +01:00
726 lines
29 KiB
Python
726 lines
29 KiB
Python
# Copyright © 2025-26 l5yth & contributors
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
"""Unit tests for the ``DEBUG=1`` ignored-packet capture (SPEC DC1-DC3).
|
|
|
|
With ``DEBUG=1`` the ingestor appends one line per dropped packet to
|
|
``ignored-meshtastic.txt`` (:mod:`data.mesh_ingestor.handlers.ignored`). A
|
|
line holds the drop reason and the packet's metadata only: port, channel,
|
|
sender and recipient ids, packet id, receive time and payload size. It
|
|
never holds the text, the payload, the meshtastic library's ``raw``
|
|
``MeshPacket``, other decoded content or a position's coordinates; the
|
|
review found every one of them there (M12), for hidden and filtered
|
|
channels, via_mqtt traffic and direct messages, MeshCore's included. The
|
|
file rotates at 10 MB and keeps one previous file.
|
|
|
|
The Meshtastic cases hand real ``MeshPacket`` protobufs to the pinned
|
|
library through the ``radio`` fixture of ``tests/test_channel_scope_unit.py``
|
|
(SPEC CF8); the MeshCore cases run the real companion reader and event
|
|
handlers (``tests/meshcore_frames.py``). Only the HTTP queue is replaced.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(REPO_ROOT) not in sys.path: # pragma: no cover - conftest adds it
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
import meshcore_frames as frames # noqa: E402 - pytest puts tests/ on sys.path
|
|
|
|
import data.mesh_ingestor.channels as channels # noqa: E402
|
|
import data.mesh_ingestor.config as config # noqa: E402
|
|
import data.mesh_ingestor.handlers.ignored as ignored_mod # noqa: E402
|
|
import data.mesh_ingestor.protocols.meshcore as meshcore_mod # noqa: E402
|
|
from data.mesh_ingestor.protocols.meshcore import ( # noqa: E402
|
|
_MeshcoreInterface,
|
|
_make_event_handlers,
|
|
)
|
|
from test_channel_scope_unit import ( # noqa: E402,F401 - shared fixtures
|
|
PRIMARY,
|
|
SECONDARY,
|
|
SENDER,
|
|
SENDER_ID,
|
|
_mesh_packet,
|
|
_payload,
|
|
radio,
|
|
scope,
|
|
)
|
|
|
|
CAP = 10_000_000
|
|
"""The documented size, in bytes, at which the capture file rotates (DC2)."""
|
|
|
|
TEXT = _payload("TEXT_MESSAGE_APP").decode()
|
|
"""The text every Meshtastic probe message carries (CF8's ``_payload``)."""
|
|
|
|
DM_TEXT = "the door code is 4711"
|
|
"""Text of the MeshCore direct message."""
|
|
|
|
HOST_NUM = 0x0D15EA5E
|
|
"""Node number of the ingestor's own radio, the direct messages' recipient."""
|
|
|
|
HOST_ID = "!0d15ea5e"
|
|
"""Canonical id of :data:`HOST_NUM`."""
|
|
|
|
MESHCORE_KEY = "aabbccddee11" + "00" * 26
|
|
"""Public key of the MeshCore contact that sends the direct message."""
|
|
|
|
GATE_CASES = {
|
|
"hidden-channel": ({"HIDDEN_CHANNELS": ("Secret",)}, SECONDARY, False),
|
|
"disallowed-channel": ({"ALLOWED_CHANNELS": ("LongFast",)}, SECONDARY, False),
|
|
"non-primary-channel": ({"PRIMARY_CHANNEL_ONLY": True}, SECONDARY, False),
|
|
"via_mqtt": ({"DROP_VIA_MQTT": True}, PRIMARY, True),
|
|
}
|
|
"""Each CF2 gate reason, with the configuration and the reception it drops:
|
|
``(config overrides, channel index, via_mqtt)``."""
|
|
|
|
LISTED_PACKET = {
|
|
"from": SENDER,
|
|
"fromId": SENDER_ID,
|
|
"to": 0xFFFFFFFF,
|
|
"toId": "^all",
|
|
"id": 77,
|
|
"channel": SECONDARY,
|
|
"rxTime": 1_791_000_000,
|
|
"decoded": {"portnum": "TEXT_MESSAGE_APP", "payload": b"hello", "text": "hello"},
|
|
}
|
|
"""A library-shaped packet carrying a source for every listed field."""
|
|
|
|
LISTED_ENTRY = {
|
|
"reason": "hidden-channel",
|
|
"id": 77,
|
|
"rx_time": 1_791_000_000,
|
|
"from_id": SENDER_ID,
|
|
"to_id": "^all",
|
|
"channel": SECONDARY,
|
|
"portnum": "TEXT_MESSAGE_APP",
|
|
"payload_size": 5,
|
|
}
|
|
"""The line :data:`LISTED_PACKET` leaves, without its ``timestamp``."""
|
|
|
|
_RECORD = ignored_mod._record_ignored_packet
|
|
"""The production recorder, which the CF8 ``scope`` fixture replaces."""
|
|
|
|
|
|
@pytest.fixture
|
|
def capture(scope, monkeypatch, tmp_path):
|
|
"""Run the real recorder under ``DEBUG=1``, writing into ``tmp_path``.
|
|
|
|
``scope`` (neutral filters, POSTs recorded) replaces the recorder with a
|
|
list of reasons; this puts the real one back. The MeshCore capture is
|
|
sent to ``tmp_path`` too, so no case writes beside the checkout.
|
|
|
|
Returns:
|
|
A namespace whose ``path`` is the capture file.
|
|
"""
|
|
|
|
path = tmp_path / "ignored-meshtastic.txt"
|
|
monkeypatch.setattr(ignored_mod, "_record_ignored_packet", _RECORD)
|
|
monkeypatch.setattr(ignored_mod, "_IGNORED_PACKET_LOG_PATH", path)
|
|
monkeypatch.setattr(
|
|
meshcore_mod, "_IGNORED_MESSAGE_LOG_PATH", tmp_path / "ignored-meshcore.txt"
|
|
)
|
|
monkeypatch.setattr(config, "DEBUG", True)
|
|
return SimpleNamespace(path=path)
|
|
|
|
|
|
def _hear(radio, packet) -> None:
|
|
"""Deliver ``packet`` from the known :data:`SENDER`, as ``radio.hear`` does."""
|
|
|
|
radio.iface._getOrCreateByNum(SENDER)
|
|
radio.iface._handlePacketFromRadio(packet)
|
|
|
|
|
|
def _meshcore_hidden_channel_message(scope) -> None:
|
|
"""Hear the real reader's ``#test`` channel message, ``#test`` hidden."""
|
|
|
|
# A MeshCore radio names its own channels, as the real reader does.
|
|
channels._reset_channel_cache()
|
|
scope.apply(HIDDEN_CHANNELS=(frames.CHANNEL_NAME,))
|
|
hmap = _make_event_handlers(_MeshcoreInterface(target=None), "/dev/ttyUSB0")
|
|
raw_frames = [frames.channel_info_frame(), frames.channel_msg_v3_frame(path_len=3)]
|
|
asyncio.run(frames.feed_reader(raw_frames, hmap))
|
|
|
|
|
|
def _meshcore_direct_message(*, rostered: bool) -> None:
|
|
"""Hear :data:`DM_TEXT` sent to the host, its sender in the roster or not."""
|
|
|
|
iface = _MeshcoreInterface(target=None)
|
|
iface.host_node_id = HOST_ID
|
|
if rostered:
|
|
iface._update_contact({"public_key": MESHCORE_KEY, "adv_name": "Alice"})
|
|
hmap = _make_event_handlers(iface, "/dev/ttyUSB0")
|
|
event = SimpleNamespace(
|
|
payload={
|
|
"sender_timestamp": frames.SENDER_TS,
|
|
"text": DM_TEXT,
|
|
"pubkey_prefix": MESHCORE_KEY[:12],
|
|
}
|
|
)
|
|
asyncio.run(hmap["CONTACT_MSG_RECV"](event))
|
|
|
|
|
|
def _b64(data: bytes) -> str:
|
|
"""Return ``data`` as the old capture wrote bytes: Base64 text."""
|
|
|
|
return base64.b64encode(data).decode("ascii")
|
|
|
|
|
|
def _leaks(capture, **forms) -> list[str]:
|
|
"""Name each content form of a dropped packet the capture file holds.
|
|
|
|
Parameters:
|
|
capture: The :func:`capture` fixture's namespace.
|
|
**forms: A name for each form and the string, or tuple of strings,
|
|
any one of which in the file betrays it: the text, the payload as
|
|
Base64, a coordinate.
|
|
|
|
Returns:
|
|
The names found, in argument order, then ``raw`` when a line carries
|
|
the library's ``raw`` ``MeshPacket``.
|
|
"""
|
|
|
|
content = capture.path.read_text(encoding="utf-8")
|
|
found = []
|
|
for name, needles in forms.items():
|
|
needles = (needles,) if isinstance(needles, str) else needles
|
|
if any(needle in content for needle in needles):
|
|
found.append(name)
|
|
if '"raw"' in content:
|
|
found.append("raw")
|
|
return found
|
|
|
|
|
|
def _only_entry(capture) -> dict:
|
|
"""Return the file's single line, parsed, its ISO ``timestamp`` removed."""
|
|
|
|
lines = capture.path.read_text(encoding="utf-8").splitlines()
|
|
assert len(lines) == 1, lines
|
|
entry = json.loads(lines[0])
|
|
assert datetime.fromisoformat(entry.pop("timestamp")).tzinfo is not None
|
|
return entry
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Every drop reason the review found keeps metadata only (SPEC DC1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDroppedPacketsKeepNoContent:
|
|
"""A dropped packet's line holds its metadata, never its content."""
|
|
|
|
@pytest.mark.parametrize("reason", list(GATE_CASES))
|
|
def test_text_dropped_at_the_filter_gate(self, radio, capture, reason):
|
|
"""A text the CF2 gate drops leaves no text, payload or ``raw``."""
|
|
overrides, channel, via_mqtt = GATE_CASES[reason]
|
|
radio.apply(**overrides)
|
|
packet = _mesh_packet("TEXT_MESSAGE_APP", channel=channel, via_mqtt=via_mqtt)
|
|
_hear(radio, packet)
|
|
assert radio.posts == []
|
|
assert _leaks(capture, text=TEXT, payload=_b64(packet.decoded.payload)) == []
|
|
expected = {
|
|
"reason": reason,
|
|
"id": packet.id,
|
|
"rx_time": packet.rx_time,
|
|
"from_id": SENDER_ID,
|
|
"to_id": "^all",
|
|
"portnum": "TEXT_MESSAGE_APP",
|
|
"payload_size": len(TEXT.encode()),
|
|
}
|
|
if channel:
|
|
# proto3 omits the channel field at index 0 (SPEC CF2).
|
|
expected["channel"] = channel
|
|
assert _only_entry(capture) == expected
|
|
|
|
def test_hidden_channel_position(self, radio, capture):
|
|
"""A hidden channel's position leaves no coordinates or payload."""
|
|
radio.apply("hidden")
|
|
packet = _mesh_packet("POSITION_APP", channel=SECONDARY)
|
|
_hear(radio, packet)
|
|
assert radio.posts == []
|
|
assert (
|
|
_leaks(
|
|
capture,
|
|
coordinates=("525200000", "52.52", "134050000", "13.405"),
|
|
payload=_b64(packet.decoded.payload),
|
|
)
|
|
== []
|
|
)
|
|
entry = _only_entry(capture)
|
|
assert (entry["reason"], entry["portnum"], entry["channel"]) == (
|
|
"hidden-channel",
|
|
"POSITION_APP",
|
|
SECONDARY,
|
|
)
|
|
assert entry["payload_size"] == len(packet.decoded.payload)
|
|
|
|
def test_meshtastic_direct_message(self, radio, capture):
|
|
"""A direct message the ingestor skips leaves no text, payload or ``raw``."""
|
|
packet = _mesh_packet("TEXT_MESSAGE_APP", channel=PRIMARY)
|
|
packet.to = HOST_NUM
|
|
_hear(radio, packet)
|
|
assert radio.posts == []
|
|
assert _leaks(capture, text=TEXT, payload=_b64(packet.decoded.payload)) == []
|
|
entry = _only_entry(capture)
|
|
assert entry["reason"] == "skipped-direct-message"
|
|
assert (entry["from_id"], entry["to_id"]) == (SENDER_ID, HOST_ID)
|
|
|
|
def test_meshcore_hidden_channel_message(self, scope, capture):
|
|
"""A MeshCore message on a hidden channel leaves no text (SC8)."""
|
|
_meshcore_hidden_channel_message(scope)
|
|
assert scope.posts == []
|
|
assert _leaks(capture, text=("hello path", frames.TEXT)) == []
|
|
entry = _only_entry(capture)
|
|
assert (entry["reason"], entry["channel"], entry["portnum"]) == (
|
|
"hidden-channel",
|
|
0,
|
|
"TEXT_MESSAGE_APP",
|
|
)
|
|
assert entry["payload_size"] == len(frames.TEXT.encode())
|
|
|
|
def test_meshcore_direct_message(self, scope, capture):
|
|
"""A MeshCore direct message the ingestor skips leaves no text."""
|
|
_meshcore_direct_message(rostered=True)
|
|
assert scope.posts == []
|
|
assert _leaks(capture, text=DM_TEXT) == []
|
|
entry = _only_entry(capture)
|
|
assert entry["reason"] == "skipped-direct-message"
|
|
assert (entry["from_id"], entry["to_id"]) == ("!aabbccdd", HOST_ID)
|
|
assert entry["payload_size"] == len(DM_TEXT.encode())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DEBUG log lines keep metadata only (SPEC DC7)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _log_line(out: str, message: str) -> str:
|
|
"""Return the one line of ``out`` that ``config._debug_log`` printed for
|
|
``message``, which it writes after the fields."""
|
|
|
|
[line] = [line for line in out.splitlines() if line.endswith(message)]
|
|
return line
|
|
|
|
|
|
class TestDebugLogLinesKeepNoContent:
|
|
"""A log line that describes a dropped packet prints no content."""
|
|
|
|
def test_packet_missing_from_id_prints_metadata(self, scope, capture, capsys):
|
|
"""A MeshCore direct message from outside the roster has no sender id;
|
|
its ``Packet missing from_id`` line names the packet, not its text."""
|
|
_meshcore_direct_message(rostered=False)
|
|
out = capsys.readouterr().out
|
|
assert DM_TEXT not in out
|
|
line = _log_line(out, "Packet missing from_id")
|
|
assert f"to_id='{HOST_ID}'" in line
|
|
assert f"payload_size={len(DM_TEXT.encode())}" in line
|
|
assert "from_id" not in line.removesuffix("Packet missing from_id")
|
|
assert scope.posts == []
|
|
assert _only_entry(capture)["reason"] == "skipped-direct-message"
|
|
|
|
def test_meshcore_channel_message_names_no_sender(self, scope, capture, capsys):
|
|
"""The ``MeshCore channel message`` line keeps the channel and the
|
|
sender id; the name parsed from a hidden channel's text stays out."""
|
|
_meshcore_hidden_channel_message(scope)
|
|
out = capsys.readouterr().out
|
|
assert "Alice" not in out
|
|
assert "hello path" not in out
|
|
line = _log_line(out, "MeshCore channel message")
|
|
assert "channel=0" in line
|
|
assert "from_id='!" in line
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The allowlist (SPEC DC1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCaptureAllowlist:
|
|
"""Only the listed fields are written; a new field is excluded by default."""
|
|
|
|
def test_unknown_field_is_left_out(self, capture):
|
|
"""A field the allowlist does not name never reaches the file, at the
|
|
top level or inside ``decoded``; the listed fields still do."""
|
|
packet = {
|
|
**LISTED_PACKET,
|
|
"futureField": "zz-unlisted-zz",
|
|
"decoded": {
|
|
**LISTED_PACKET["decoded"],
|
|
"futureSection": {"note": "zz-unlisted-zz"},
|
|
},
|
|
}
|
|
ignored_mod._record_ignored_packet(packet, reason="hidden-channel")
|
|
assert "zz-unlisted-zz" not in capture.path.read_text(encoding="utf-8")
|
|
assert _only_entry(capture) == LISTED_ENTRY
|
|
|
|
def test_allowlist_is_the_documented_metadata(self):
|
|
"""The allowlist names the metadata DC1 lists, in DC1's order."""
|
|
assert list(ignored_mod._CAPTURE_FIELDS) == [
|
|
"id",
|
|
"rx_time",
|
|
"from_id",
|
|
"to_id",
|
|
"channel",
|
|
"portnum",
|
|
"payload_size",
|
|
]
|
|
|
|
def test_non_scalar_values_are_left_out(self, capture):
|
|
"""A listed field whose value is not a string or number is left out."""
|
|
nested = {"text": "zz-nested-zz"}
|
|
packet = {
|
|
"id": nested,
|
|
"rxTime": ["zz-nested-zz"],
|
|
"fromId": nested,
|
|
"toId": b"zz-nested-zz",
|
|
"channel": nested,
|
|
"decoded": {"portnum": nested, "payload": nested},
|
|
}
|
|
ignored_mod._record_ignored_packet(packet, reason="unsupported-port")
|
|
assert "zz-nested-zz" not in capture.path.read_text(encoding="utf-8")
|
|
assert _only_entry(capture) == {"reason": "unsupported-port"}
|
|
|
|
def test_packet_that_is_no_mapping_leaves_its_reason(self, capture):
|
|
"""A packet that is not a mapping leaves the reason and nothing else."""
|
|
ignored_mod._record_ignored_packet("zz-opaque-zz", reason="unsupported-port")
|
|
assert "zz-opaque-zz" not in capture.path.read_text(encoding="utf-8")
|
|
assert _only_entry(capture) == {"reason": "unsupported-port"}
|
|
|
|
@pytest.mark.parametrize(
|
|
"packet, size",
|
|
[
|
|
({"decoded": {"payload": b"\x00\x01\x02", "text": "zz"}}, 3),
|
|
({"decoded": {"payload": _b64(b"abcd")}}, 4),
|
|
({"decoded": {"payload": {"__bytes_b64__": _b64(b"abcde")}}}, 5),
|
|
({"decoded": {"text": "grüße"}}, 7),
|
|
({"encrypted": _b64(b"\x07" * 9)}, 9),
|
|
({"encrypted": True, "decoded": {"portnum": "TEXT_MESSAGE_APP"}}, None),
|
|
({"decoded": {"payload": "not base64!"}}, None),
|
|
({}, None),
|
|
],
|
|
ids=[
|
|
"library-bytes",
|
|
"udp-base64",
|
|
"bytes-mapping",
|
|
"handler-text",
|
|
"ciphertext",
|
|
"encrypted-flag",
|
|
"unreadable",
|
|
"none",
|
|
],
|
|
)
|
|
def test_payload_size(self, capture, packet, size):
|
|
"""The size counts the payload bytes, then the ciphertext, then the
|
|
text in UTF-8; a packet with none of them has no size."""
|
|
ignored_mod._record_ignored_packet(packet, reason="no-message-payload")
|
|
assert _only_entry(capture).get("payload_size") == size
|
|
|
|
def test_numeric_sender_and_recipient_are_canonical(self, capture):
|
|
"""A numeric ``from``/``to`` is written as a ``!xxxxxxxx`` id."""
|
|
packet = {"from": SENDER, "fromId": "!0badbeef", "to": HOST_NUM}
|
|
ignored_mod._record_ignored_packet(packet, reason="skipped-direct-message")
|
|
entry = _only_entry(capture)
|
|
assert (entry["from_id"], entry["to_id"]) == (SENDER_ID, HOST_ID)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rotation (SPEC DC2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _fill(path: Path, size: int) -> None:
|
|
"""Make ``path`` ``size`` bytes long without writing them (a sparse file)."""
|
|
|
|
with path.open("wb") as handle:
|
|
handle.truncate(size)
|
|
|
|
|
|
_WRITER = """
|
|
import json, os, sys, time
|
|
from pathlib import Path
|
|
|
|
repo, path, cap, count, ready, go = sys.argv[1:]
|
|
sys.path.insert(0, repo)
|
|
import data.mesh_ingestor.config as config
|
|
import data.mesh_ingestor.handlers.ignored as ignored
|
|
|
|
config.DEBUG = True
|
|
ignored._IGNORED_PACKET_LOG_PATH = Path(path)
|
|
ignored._IGNORED_PACKET_LOG_MAX_BYTES = int(cap)
|
|
moved, discarded = [], []
|
|
replace = os.replace
|
|
|
|
|
|
def spy(src, dst):
|
|
if os.path.exists(dst):
|
|
data = Path(dst).read_bytes()
|
|
discarded.append(data.rstrip(b"\\n").rfind(b"\\n") + 1)
|
|
replace(src, dst)
|
|
moved.append(os.stat(dst).st_size)
|
|
|
|
|
|
os.replace = spy
|
|
errors = []
|
|
Path(ready).touch()
|
|
while not os.path.exists(go):
|
|
time.sleep(0.001)
|
|
for number in range(int(count)):
|
|
try:
|
|
ignored._record_ignored_packet({"id": number}, reason="hidden-channel")
|
|
except OSError as exc:
|
|
errors.append(type(exc).__name__)
|
|
print(json.dumps({"errors": errors, "moved": moved, "discarded": discarded}))
|
|
"""
|
|
"""A capture writer run as its own process: after the shared start it writes
|
|
``count`` lines to ``path`` at a cap of ``cap`` bytes and prints the errors it
|
|
caught, the size of each file it rotated into ``.1``, and for each ``.1`` it
|
|
replaced the offset of that file's last line."""
|
|
|
|
|
|
def _open_files() -> set[str]:
|
|
"""Return the path of every file this process holds open (Linux)."""
|
|
|
|
names = set()
|
|
for fd in os.listdir("/proc/self/fd"):
|
|
with contextlib.suppress(OSError):
|
|
names.add(os.readlink(f"/proc/self/fd/{fd}"))
|
|
return names
|
|
|
|
|
|
class TestCaptureRotation:
|
|
"""The file rotates at 10 MB and keeps exactly one previous file; it is
|
|
created, with its directory, on the first write."""
|
|
|
|
def test_write_that_reaches_the_cap_rotates(self, capture):
|
|
"""The write that takes the file to 10 MB moves it to ``.1``; the next
|
|
write starts a new file."""
|
|
previous = capture.path.with_name("ignored-meshtastic.txt.1")
|
|
_fill(capture.path, CAP - 1)
|
|
ignored_mod._record_ignored_packet({"id": 1}, reason="hidden-channel")
|
|
assert previous.exists()
|
|
assert previous.stat().st_size >= CAP
|
|
assert not capture.path.exists()
|
|
ignored_mod._record_ignored_packet({"id": 2}, reason="hidden-channel")
|
|
assert _only_entry(capture) == {"reason": "hidden-channel", "id": 2}
|
|
|
|
def test_write_below_the_cap_appends(self, capture):
|
|
"""A write that leaves the file under 10 MB appends to it."""
|
|
_fill(capture.path, CAP - 1000)
|
|
ignored_mod._record_ignored_packet({"id": 3}, reason="hidden-channel")
|
|
assert not capture.path.with_name("ignored-meshtastic.txt.1").exists()
|
|
size = capture.path.stat().st_size
|
|
assert CAP - 1000 < size < CAP
|
|
with capture.path.open("rb") as handle:
|
|
handle.seek(CAP - 1000)
|
|
assert json.loads(handle.read())["id"] == 3
|
|
|
|
def test_only_one_previous_file_is_kept(self, capture, tmp_path):
|
|
"""A second rotation replaces the previous file; nothing else is kept."""
|
|
previous = capture.path.with_name("ignored-meshtastic.txt.1")
|
|
previous.write_text("older capture\n", encoding="utf-8")
|
|
_fill(capture.path, CAP - 1)
|
|
ignored_mod._record_ignored_packet({"id": 4}, reason="hidden-channel")
|
|
# The 14-byte older file is gone: ``.1`` is now the rotated 10 MB one.
|
|
assert previous.stat().st_size >= CAP
|
|
assert sorted(p.name for p in tmp_path.iterdir()) == [
|
|
"ignored-meshtastic.txt.1"
|
|
]
|
|
|
|
def test_write_creates_a_missing_directory(self, capture, monkeypatch, tmp_path):
|
|
"""A capture path whose directory is missing is created on the write."""
|
|
path = tmp_path / "gone" / "ignored-meshtastic.txt"
|
|
monkeypatch.setattr(ignored_mod, "_IGNORED_PACKET_LOG_PATH", path)
|
|
ignored_mod._record_ignored_packet({"id": 5}, reason="hidden-channel")
|
|
assert json.loads(path.read_text(encoding="utf-8"))["id"] == 5
|
|
|
|
def test_two_processes_share_one_rotation(self, tmp_path):
|
|
"""Two ingestors writing one checkout's capture never fail on a
|
|
rotation, and every rotation moves a full file to ``.1``: never a
|
|
fresh one over it (SPEC DC2)."""
|
|
path = tmp_path / "ignored-meshtastic.txt"
|
|
cap, count, go = 20_000, 10_000, tmp_path / "go"
|
|
ready = [tmp_path / f"ready-{index}" for index in range(2)]
|
|
writers = [
|
|
subprocess.Popen(
|
|
[sys.executable, "-c", _WRITER, str(REPO_ROOT), str(path)]
|
|
+ [str(cap), str(count), str(flag), str(go)],
|
|
cwd=tmp_path,
|
|
stdout=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
for flag in ready
|
|
]
|
|
deadline = time.monotonic() + 60
|
|
while not all(flag.exists() for flag in ready):
|
|
assert time.monotonic() < deadline, "the writers never started"
|
|
time.sleep(0.01)
|
|
go.touch()
|
|
reports = [json.loads(writer.communicate(timeout=120)[0]) for writer in writers]
|
|
assert [writer.returncode for writer in writers] == [0, 0]
|
|
assert [report["errors"] for report in reports] == [[], []]
|
|
moved = [size for report in reports for size in report["moved"]]
|
|
assert moved, "no rotation happened"
|
|
assert min(moved) >= cap
|
|
assert path.with_name("ignored-meshtastic.txt.1").stat().st_size >= cap
|
|
# A file is rotated by the write that filled it, so every line but its
|
|
# last went in below the cap: no line was appended after a rotation.
|
|
discarded = [offset for report in reports for offset in report["discarded"]]
|
|
assert discarded, "no previous file was replaced"
|
|
assert max(discarded) < cap
|
|
|
|
@pytest.mark.parametrize("fresh", [True, False], ids=["new-file", "no-file"])
|
|
def test_writer_that_waited_appends_to_the_current_file(
|
|
self, capture, monkeypatch, fresh
|
|
):
|
|
"""A writer that waited for the lock while another process rotated the
|
|
full file appends to the file now at the path, never to the rotated
|
|
one, which the next rotation would discard."""
|
|
fcntl = pytest.importorskip("fcntl")
|
|
previous = capture.path.with_name("ignored-meshtastic.txt.1")
|
|
_fill(capture.path, CAP)
|
|
waits = []
|
|
|
|
def rotated_while_waiting(fd, operation):
|
|
"""On the first wait let another process rotate the full file."""
|
|
if not waits:
|
|
waits.append(fd)
|
|
os.replace(capture.path, previous)
|
|
if fresh:
|
|
capture.path.write_text("fresh\n", encoding="utf-8")
|
|
fcntl.flock(fd, operation)
|
|
|
|
monkeypatch.setattr(
|
|
ignored_mod,
|
|
"fcntl",
|
|
SimpleNamespace(LOCK_EX=fcntl.LOCK_EX, flock=rotated_while_waiting),
|
|
)
|
|
ignored_mod._record_ignored_packet({"id": 6}, reason="hidden-channel")
|
|
assert previous.stat().st_size == CAP
|
|
lines = capture.path.read_text(encoding="utf-8").splitlines()
|
|
assert lines[:-1] == (["fresh"] if fresh else [])
|
|
assert json.loads(lines[-1])["id"] == 6
|
|
|
|
def test_without_fcntl_no_handle_is_open_at_the_rename(self, capture, monkeypatch):
|
|
"""Without ``fcntl`` (Windows) the file is closed before it is renamed,
|
|
as Windows refuses to rename an open file."""
|
|
if not os.path.isdir("/proc/self/fd"):
|
|
pytest.skip("needs /proc to list the open files")
|
|
monkeypatch.setattr(ignored_mod, "fcntl", None)
|
|
replace, seen = os.replace, []
|
|
|
|
def checked_replace(src, dst):
|
|
"""Record whether either file is open in this process, then rename."""
|
|
files = _open_files()
|
|
seen.append(
|
|
(os.path.realpath(src) in files, os.path.realpath(dst) in files)
|
|
)
|
|
replace(src, dst)
|
|
|
|
monkeypatch.setattr(os, "replace", checked_replace)
|
|
_fill(capture.path, CAP - 1)
|
|
ignored_mod._record_ignored_packet({"id": 8}, reason="hidden-channel")
|
|
assert seen == [(False, False)]
|
|
assert capture.path.with_name("ignored-meshtastic.txt.1").stat().st_size >= CAP
|
|
|
|
def test_with_fcntl_the_rename_runs_under_the_lock(self, capture, monkeypatch):
|
|
"""With ``fcntl`` the file is renamed while this process holds its lock,
|
|
so no other process appends to it or rotates it meanwhile."""
|
|
fcntl = pytest.importorskip("fcntl")
|
|
replace, held = os.replace, []
|
|
|
|
def checked_replace(src, dst):
|
|
"""Try to take the lock from another handle, then rename."""
|
|
with open(src, "ab") as probe:
|
|
try:
|
|
fcntl.flock(probe.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
held.append(False)
|
|
except BlockingIOError:
|
|
held.append(True)
|
|
replace(src, dst)
|
|
|
|
monkeypatch.setattr(os, "replace", checked_replace)
|
|
_fill(capture.path, CAP - 1)
|
|
ignored_mod._record_ignored_packet({"id": 11}, reason="hidden-channel")
|
|
assert held == [True]
|
|
assert capture.path.with_name("ignored-meshtastic.txt.1").stat().st_size >= CAP
|
|
|
|
@pytest.mark.parametrize("locks", [True, False], ids=["fcntl", "no-fcntl"])
|
|
def test_refused_rename_is_retried_at_the_next_write(
|
|
self, capture, monkeypatch, locks
|
|
):
|
|
"""A rename the system refuses, as Windows does while another process
|
|
holds the file open, is no error; the next write rotates the file."""
|
|
if not locks:
|
|
monkeypatch.setattr(ignored_mod, "fcntl", None)
|
|
previous = capture.path.with_name("ignored-meshtastic.txt.1")
|
|
replace, refusals = os.replace, [PermissionError(13, "in use")]
|
|
|
|
def refusing_once(src, dst):
|
|
"""Refuse the first rename, then rename."""
|
|
if refusals:
|
|
raise refusals.pop()
|
|
replace(src, dst)
|
|
|
|
monkeypatch.setattr(os, "replace", refusing_once)
|
|
_fill(capture.path, CAP - 1)
|
|
ignored_mod._record_ignored_packet({"id": 9}, reason="hidden-channel")
|
|
assert not previous.exists()
|
|
assert capture.path.stat().st_size >= CAP
|
|
ignored_mod._record_ignored_packet({"id": 10}, reason="hidden-channel")
|
|
assert previous.stat().st_size >= CAP
|
|
assert not capture.path.exists()
|
|
|
|
def test_rotating_a_file_already_gone_is_no_error(self, capture):
|
|
"""A capture file removed before its rotation leaves nothing to move."""
|
|
ignored_mod._rotate(capture.path)
|
|
assert not capture.path.with_name("ignored-meshtastic.txt.1").exists()
|
|
|
|
def test_without_file_locks_one_process_still_rotates(self, capture, monkeypatch):
|
|
"""Where ``fcntl`` is missing (not POSIX), rotation works per process:
|
|
the write that fills the file rotates it, the next one appends."""
|
|
monkeypatch.setattr(ignored_mod, "fcntl", None)
|
|
_fill(capture.path, CAP - 1)
|
|
ignored_mod._record_ignored_packet({"id": 7}, reason="hidden-channel")
|
|
assert capture.path.with_name("ignored-meshtastic.txt.1").stat().st_size >= CAP
|
|
assert not capture.path.exists()
|
|
ignored_mod._record_ignored_packet({"id": 12}, reason="hidden-channel")
|
|
assert _only_entry(capture) == {"reason": "hidden-channel", "id": 12}
|
|
|
|
def test_meshcore_capture_rotates_the_same_way(self, capture, tmp_path):
|
|
"""``ignored-meshcore.txt`` rotates at 10 MB too and keeps one previous
|
|
file; its lines keep the content SC8 gives them."""
|
|
path = tmp_path / "ignored-meshcore.txt"
|
|
previous = path.with_name("ignored-meshcore.txt.1")
|
|
_fill(path, CAP - 1)
|
|
frame = {"payload_typename": "GRP_TXT", "msg_hash": 7}
|
|
meshcore_mod._record_meshcore_message(frame, source="auto:RX_LOG_DATA")
|
|
assert previous.exists()
|
|
assert previous.stat().st_size >= CAP
|
|
assert not path.exists()
|
|
meshcore_mod._record_meshcore_message(frame, source="auto:RX_LOG_DATA")
|
|
[line] = path.read_text(encoding="utf-8").splitlines()
|
|
assert json.loads(line)["message"] == frame
|