91 lines
2.4 KiB
Python
91 lines
2.4 KiB
Python
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from k1link.device_plugins.xgrids_k1.cli import app
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
def test_help() -> None:
|
|
result = runner.invoke(app, ["--help"])
|
|
assert result.exit_code == 0
|
|
assert "evidence-led" in result.stdout
|
|
|
|
|
|
def test_doctor_json() -> None:
|
|
result = runner.invoke(app, ["doctor", "--json"])
|
|
assert result.exit_code == 0
|
|
payload = json.loads(result.stdout)
|
|
assert payload["k1link_version"] == "0.1.0"
|
|
assert isinstance(payload["tools"], list)
|
|
assert isinstance(payload["network"], dict)
|
|
assert any(item["name"] == "tcpdump" for item in payload["tools"])
|
|
|
|
|
|
def test_mqtt_capture_requires_owned_device_confirmation(tmp_path: Path) -> None:
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"net",
|
|
"mqtt-capture",
|
|
"--host",
|
|
"192.168.1.20",
|
|
"--out",
|
|
str(tmp_path / "capture"),
|
|
],
|
|
)
|
|
assert result.exit_code == 2
|
|
assert "ownership not confirmed" in result.stdout
|
|
assert not (tmp_path / "capture").exists()
|
|
|
|
|
|
def test_mqtt_capture_cli_uses_bounded_read_only_capture(
|
|
monkeypatch: Any,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_capture(host: str, out: Path, **kwargs: object) -> dict[str, object]:
|
|
on_ready = kwargs.pop("on_ready")
|
|
assert callable(on_ready)
|
|
on_ready()
|
|
captured.update({"host": host, "out": out, **kwargs})
|
|
return {
|
|
"stop_reason": "duration_elapsed",
|
|
"message_count": 2,
|
|
"payload_bytes": 128,
|
|
}
|
|
|
|
monkeypatch.setattr("k1link.device_plugins.xgrids_k1.cli.capture_mqtt", fake_capture)
|
|
out = tmp_path / "capture"
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"net",
|
|
"mqtt-capture",
|
|
"--host",
|
|
"10.0.0.42",
|
|
"--out",
|
|
str(out),
|
|
"--duration",
|
|
"5",
|
|
"--max-message-bytes",
|
|
"1024",
|
|
"--confirm-owned-device",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert captured == {
|
|
"host": "10.0.0.42",
|
|
"out": out,
|
|
"port": 1883,
|
|
"duration_seconds": 5.0,
|
|
"max_message_bytes": 1024,
|
|
}
|
|
assert "messages: 2" in result.stdout
|
|
assert "subscriptions active" in result.stdout
|