147 lines
4.0 KiB
Python
147 lines
4.0 KiB
Python
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from k1link.device_plugins.xgrids_k1 import cli
|
|
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_serve_resolves_frontend_from_repository_root(monkeypatch: Any) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_run(application: str, **kwargs: object) -> None:
|
|
captured.update({"application": application, **kwargs})
|
|
|
|
monkeypatch.setattr(cli.uvicorn, "run", fake_run)
|
|
|
|
result = runner.invoke(app, ["serve"])
|
|
|
|
assert result.exit_code == 0
|
|
assert captured == {
|
|
"application": "k1link.web.app:app",
|
|
"host": "127.0.0.1",
|
|
"port": 8000,
|
|
"log_level": "info",
|
|
"access_log": True,
|
|
}
|
|
|
|
|
|
def test_authority_provision_requires_explicit_reviewed_value_confirmation() -> None:
|
|
result = runner.invoke(app, ["authority", "provision"])
|
|
|
|
assert result.exit_code == 2
|
|
assert "provisioning not confirmed" in result.stdout
|
|
|
|
|
|
def test_authority_provision_never_accepts_the_secret_as_a_cli_value(monkeypatch: Any) -> None:
|
|
calls = 0
|
|
|
|
class FakeSnapshot:
|
|
service = "fixed-service"
|
|
account = "fixed-account"
|
|
|
|
class FakeProvisioner:
|
|
def provision_interactively(self) -> FakeSnapshot:
|
|
nonlocal calls
|
|
calls += 1
|
|
return FakeSnapshot()
|
|
|
|
monkeypatch.setattr(
|
|
"k1link.device_plugins.xgrids_k1.cli.MacOSKeychainApplicationAuthorityProvisioner",
|
|
FakeProvisioner,
|
|
)
|
|
result = runner.invoke(
|
|
app,
|
|
["authority", "provision", "--confirm-reviewed-authority"],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert calls == 1
|
|
assert "Keychain item validated" in result.stdout
|
|
assert "No K1 command was sent" in " ".join(result.stdout.split())
|
|
|
|
|
|
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
|