diff --git a/solstone/apps/link/call.py b/solstone/apps/link/call.py index 7ffbaf843..3ef387c3a 100644 --- a/solstone/apps/link/call.py +++ b/solstone/apps/link/call.py @@ -389,6 +389,28 @@ def list_devices() -> None: printed_section = True +@app.command("authorized-clients") +@convey_cli +def authorized_clients() -> None: + """List every authorized client cert: fingerprint, label, last-seen (flat view).""" + devices = get_client().request("GET", "/app/link/api/devices")["devices"] + if not devices: + typer.echo("No authorized clients.") + return + for device in devices: + label = device.get("display_label") or device["device_label"] + typer.echo( + f"{device['fingerprint']} {label}" + f" last seen {_relative_time(device['last_seen_at'])}" + ) + + +@app.command("observer-pause") +def observer_pause() -> None: + """Pause linked observers (not yet available).""" + typer.echo("observer-pause is not yet available.") + + @app.command() @convey_cli def unpair( diff --git a/solstone/think/link/README.md b/solstone/think/link/README.md index 6082c9893..95308ec29 100644 --- a/solstone/think/link/README.md +++ b/solstone/think/link/README.md @@ -10,7 +10,7 @@ The `spl` repo's `home/` continues as the open-source reference implementation o | File | Purpose | |------|---------| -| `cli.py` | Entry point for caller-side `sol link join`, `sol link list`, and `sol link serve`. | +| `cli.py` | Entry point for caller-side `sol link join` and `sol link serve`. | | `serve_cli.py` | Loopback proxy over the PL tunnel for paired caller access. | | `observer_paths.py` | Shared observer SPL bundle path helpers. | | `ca.py` | Local CA lifecycle + CSR signing + home-attestation minting. | diff --git a/solstone/think/link/__init__.py b/solstone/think/link/__init__.py index ae8cc62df..ec2406492 100644 --- a/solstone/think/link/__init__.py +++ b/solstone/think/link/__init__.py @@ -3,9 +3,9 @@ """Caller-side link commands and shared pairing/runtime helpers. -`sol link` dispatches caller-side commands such as `join`, `list`, and `serve`. The -supervised home-side spl rendezvous daemon lives in `solstone.think.spl` and is -launched as `journal spl`. +`sol link` dispatches caller-side commands such as `join` and `serve`. The +supervised home-side spl rendezvous daemon lives in `solstone.think.spl` and +runs as `journal spl`. """ __version__ = "0.1.0" diff --git a/solstone/think/link/__main__.py b/solstone/think/link/__main__.py index 67666b435..c868068d6 100644 --- a/solstone/think/link/__main__.py +++ b/solstone/think/link/__main__.py @@ -1,7 +1,7 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc -"""python -m solstone.think.link — entry point for `sol link`.""" +"""python -m solstone.think.link — entry point for caller-side `sol link`.""" from solstone.think.link.cli import main diff --git a/solstone/think/link/cli.py b/solstone/think/link/cli.py index bf2b1cc52..e46ba26b3 100644 --- a/solstone/think/link/cli.py +++ b/solstone/think/link/cli.py @@ -8,14 +8,14 @@ from __future__ import annotations import argparse import sys -from solstone.think.link import join_cli, list_cli, serve_cli +from solstone.think.link import join_cli, serve_cli def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="solstone link access commands") subparsers = parser.add_subparsers( dest="command", - metavar="{join,list,serve}", + metavar="{join,serve}", title="commands", ) join_parser = subparsers.add_parser( @@ -23,11 +23,6 @@ def _build_parser() -> argparse.ArgumentParser: help="join a solstone with a short code or pair link", ) join_cli.add_arguments(join_parser) - list_parser = subparsers.add_parser( - "list", - help="list caller-side link bundles", - ) - list_cli.add_arguments(list_parser) serve_parser = subparsers.add_parser( "serve", help="serve a loopback proxy over a link tunnel", @@ -38,15 +33,20 @@ def _build_parser() -> argparse.ArgumentParser: def main(argv: list[str] | None = None) -> int: """CLI entry point for `sol link`.""" + args = list(sys.argv[1:] if argv is None else argv) + if sys.argv[0] == "journal link": + from solstone.apps.link.call import app as link_management_app + + link_management_app(args=args, prog_name="journal link") + return 0 # unreachable: the Typer app raises SystemExit + parser = _build_parser() - args = parser.parse_args(list(sys.argv[1:] if argv is None else argv)) - if args.command is None: + namespace = parser.parse_args(args) + if namespace.command is None: parser.print_help() return 0 - if args.command == "join": - return join_cli.main(args) - if args.command == "list": - return list_cli.main(args) - if args.command == "serve": - return serve_cli.main(args) + if namespace.command == "join": + return join_cli.main(namespace) + if namespace.command == "serve": + return serve_cli.main(namespace) return 0 diff --git a/solstone/think/link/list_cli.py b/solstone/think/link/list_cli.py deleted file mode 100644 index 1dcf44ab8..000000000 --- a/solstone/think/link/list_cli.py +++ /dev/null @@ -1,235 +0,0 @@ -# SPDX-License-Identifier: AGPL-3.0-only -# Copyright (c) 2026 sol pbc - -"""Caller-side bundle enumeration for `sol link list`; reads peer bundles under the journal and, with `--observers`, observer bundles under the solstone-observer config tree, without network access or writes.""" - -from __future__ import annotations - -import argparse -import datetime as dt -import json -import sys -from dataclasses import dataclass -from pathlib import Path -from typing import Any - -from solstone.think.link.observer_paths import observer_spl_root -from solstone.think.utils import get_journal - -DASH = "—" -BULLET = "•" - - -@dataclass(frozen=True) -class _Bundle: - kind: str - label: str - instance_id: str - home_label: str | None - paired_at: str | None - fingerprint: str | None - bundle_dir: str - - -def add_arguments(parser: argparse.ArgumentParser) -> None: - parser.add_argument( - "--observers", - action="store_true", - default=False, - help="Include observer bundles from the solstone-observer config tree", - ) - parser.add_argument( - "--json", - action="store_true", - default=False, - help="Output a JSON array", - ) - - -def main(args: argparse.Namespace) -> int: - peers = _sorted_bundles(_walk_bundle_root(Path(get_journal()) / "peers", "peer")) - observers = ( - _sorted_bundles(_walk_bundle_root(observer_spl_root(), "observer")) - if args.observers - else [] - ) - - if args.json: - records = _json_records(_sorted_bundles([*peers, *observers])) - print(json.dumps(records)) - return 0 - - _print_human(peers, observers, include_observers=args.observers) - return 0 - - -def _walk_bundle_root(root: Path, kind: str) -> list[_Bundle]: - if not root.exists(): - return [] - if not root.is_dir(): - _warn( - f"warning: {kind} bundle root {_absolute(root)} exists but is not a " - "directory; treating as empty" - ) - return [] - - bundles: list[_Bundle] = [] - for bundle_dir in sorted(root.iterdir()): - if not bundle_dir.is_dir(): - continue - peer_json = bundle_dir / "peer.json" - if not peer_json.exists(): - continue - try: - raw = peer_json.read_text(encoding="utf-8") - except (OSError, UnicodeDecodeError) as exc: - _warn( - f"warning: skipping {kind} bundle {_absolute(bundle_dir)}: " - f"cannot read peer.json ({type(exc).__name__})" - ) - continue - try: - peer = json.loads(raw) - except json.JSONDecodeError: - _warn( - f"warning: skipping {kind} bundle {_absolute(bundle_dir)}: " - "peer.json is not valid JSON" - ) - continue - - if not isinstance(peer, dict): - _warn_missing_required(kind, bundle_dir, "label") - continue - label = _required_str(peer, "label") - if label is None: - _warn_missing_required(kind, bundle_dir, "label") - continue - instance_id = _required_str(peer, "instance_id") - if instance_id is None: - _warn_missing_required(kind, bundle_dir, "instance_id") - continue - - bundles.append( - _Bundle( - kind=kind, - label=label, - instance_id=instance_id, - home_label=_optional_str(peer, "home_label"), - paired_at=_optional_str(peer, "paired_at"), - fingerprint=_optional_str(peer, "fingerprint"), - bundle_dir=_absolute(bundle_dir), - ) - ) - return bundles - - -def _required_str(peer: dict[str, Any], field: str) -> str | None: - value = peer.get(field) - if not isinstance(value, str) or not value: - return None - return value - - -def _optional_str(peer: dict[str, Any], field: str) -> str | None: - value = peer.get(field) - if not isinstance(value, str) or not value: - return None - return value - - -def _warn_missing_required(kind: str, bundle_dir: Path, field: str) -> None: - _warn( - f"warning: skipping {kind} bundle {_absolute(bundle_dir)}: " - f"peer.json missing required field '{field}'" - ) - - -def _warn(msg: str) -> None: - sys.stderr.write(f"{msg}\n") - - -def _absolute(path: Path) -> str: - return str(path.resolve()) - - -def _relative_time(iso: str | None) -> str: - from solstone.convey.utils import relative_time - - if not iso: - return "never" - try: - then = dt.datetime.strptime(iso, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=dt.UTC) - except ValueError: - return iso - now = dt.datetime.now(dt.UTC) - delta_seconds = max(0, (now - then).total_seconds()) - return f"{relative_time(delta_seconds)} ago" - - -def _sort_key(bundle: _Bundle) -> tuple[int, float | str, str]: - basename = Path(bundle.bundle_dir).name - if bundle.paired_at: - try: - then = dt.datetime.strptime(bundle.paired_at, "%Y-%m-%dT%H:%M:%SZ").replace( - tzinfo=dt.UTC - ) - except ValueError: - pass - else: - return (0, -then.timestamp(), basename) - return (1, "", basename) - - -def _sorted_bundles(bundles: list[_Bundle]) -> list[_Bundle]: - return sorted(bundles, key=_sort_key) - - -def _display(value: str | None) -> str: - return value or DASH - - -def _print_human( - peers: list[_Bundle], - observers: list[_Bundle], - *, - include_observers: bool, -) -> None: - if not include_observers: - if not peers: - print("No peers paired yet.") - return - _print_section("Peers", peers, "No peers paired yet.") - return - - _print_section("Peers", peers, "No peers paired yet.") - print() - _print_section("Observers", observers, "No observers paired yet.") - - -def _print_section(heading: str, bundles: list[_Bundle], empty_message: str) -> None: - print(f"{heading}:") - if not bundles: - print(f" {empty_message}") - return - for bundle in bundles: - print(f" {bundle.label} ({bundle.instance_id})") - print( - f" paired {_relative_time(bundle.paired_at)} {BULLET} " - f"home: {_display(bundle.home_label)} {BULLET} " - f"fingerprint: {_display(bundle.fingerprint)}" - ) - - -def _json_records(bundles: list[_Bundle]) -> list[dict[str, str | None]]: - return [ - { - "kind": bundle.kind, - "label": bundle.label, - "instance_id": bundle.instance_id, - "home_label": bundle.home_label, - "paired_at": bundle.paired_at, - "fingerprint": bundle.fingerprint, - "bundle_dir": bundle.bundle_dir, - } - for bundle in bundles - ] diff --git a/solstone/think/sol_cli.py b/solstone/think/sol_cli.py index dd8122862..fe9142496 100644 --- a/solstone/think/sol_cli.py +++ b/solstone/think/sol_cli.py @@ -120,7 +120,7 @@ COMMANDS: dict[str, Command] = { "facet-candidates": Command("solstone.think.facet_candidates_cli", "service"), "cortex": Command("solstone.think.cortex", "service"), "talent": Command("solstone.think.talent_cli", "service"), - "link": Command("solstone.think.link", "access"), + "link": Command("solstone.think.link", "universal"), "spl": Command("solstone.think.spl", "service"), "call": Command("solstone.think.call", "access"), "navigate": Command("solstone.think.tools.navigate", "service"), diff --git a/tests/link/test_link_cli_namespace.py b/tests/link/test_link_cli_namespace.py index 1fa888a6d..9ab131ea5 100644 --- a/tests/link/test_link_cli_namespace.py +++ b/tests/link/test_link_cli_namespace.py @@ -3,6 +3,8 @@ from __future__ import annotations +import sys + import pytest from solstone.think.link import cli @@ -37,46 +39,57 @@ def test_link_join_dispatches_to_join_cli(monkeypatch: pytest.MonkeyPatch) -> No assert calls == [("http://receiver", "ABCD-EFGH", "observer", "laptop")] -def test_link_list_dispatches_to_list_cli(monkeypatch: pytest.MonkeyPatch) -> None: - calls = [] - - def fake_list(args) -> int: - calls.append((args.command, args.observers, args.json)) - return 0 - - monkeypatch.setattr("solstone.think.link.list_cli.main", fake_list) - - assert cli.main(["list"]) == 0 +def test_link_no_subcommand_help_lists_commands( + capsys: pytest.CaptureFixture[str], +) -> None: + assert cli.main([]) == 0 - assert calls == [("list", False, False)] + out = capsys.readouterr().out + assert "{join,serve}" in out + assert "join" in out + assert "serve" in out + assert "list" not in out -def test_link_list_dispatches_flags_to_list_cli( +def test_journal_link_routes_to_management_app_help( monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], ) -> None: - calls = [] + from typer.main import get_command - def fake_list(args) -> int: - calls.append((args.command, args.observers, args.json)) - return 0 + from solstone.apps.link import call as link_call - monkeypatch.setattr("solstone.think.link.list_cli.main", fake_list) + monkeypatch.setattr(sys, "argv", ["journal link", "--help"]) - assert cli.main(["list", "--observers", "--json"]) == 0 + with pytest.raises(SystemExit) as exc: + cli.main(["--help"]) - assert calls == [("list", True, True)] + assert exc.value.code == 0 + out = capsys.readouterr().out + management_commands = set(get_command(link_call.app).commands) + assert "Usage: journal link" in out + assert "unpair" in out + assert "authorized-clients" in out + assert "join" not in management_commands + assert "serve" not in management_commands + assert "join a solstone with a short code or pair link" not in out + assert "serve a loopback proxy over a link tunnel" not in out -def test_link_no_subcommand_help_lists_commands( +def test_sol_link_list_is_unknown_client_subcommand( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: - assert cli.main([]) == 0 + monkeypatch.setattr(sys, "argv", ["sol link", "list"]) - out = capsys.readouterr().out - assert "{join,list,serve}" in out - assert "join" in out - assert "list" in out - assert "serve" in out + with pytest.raises(SystemExit) as exc: + cli.main(["list"]) + + assert exc.value.code == 2 + captured = capsys.readouterr() + output = captured.out + captured.err + assert "invalid choice: 'list'" in captured.err + assert "Traceback" not in output def test_link_serve_dispatches_to_serve_cli(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/link/test_link_import_hygiene.py b/tests/link/test_link_import_hygiene.py new file mode 100644 index 000000000..60b0c8c74 --- /dev/null +++ b/tests/link/test_link_import_hygiene.py @@ -0,0 +1,46 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import ast +from pathlib import Path + +import solstone.think.link + +ALLOWED_CONVEY_IMPORTS = {"solstone.convey.secure_listener.framing"} + + +def _convey_imports(path: Path) -> set[str]: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + imports: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name == "solstone.convey" or alias.name.startswith( + "solstone.convey." + ): + imports.add(alias.name) + elif isinstance(node, ast.ImportFrom): + if node.level != 0 or node.module is None: + continue + if node.module == "solstone.convey" or node.module.startswith( + "solstone.convey." + ): + imports.add(node.module) + return imports + + +def test_think_link_convey_imports_stay_wire_protocol_only() -> None: + """Keep presentation imports off `sol link`, while allowlisting framing. + + The removed `solstone.convey.utils` dependency was a presentation leak on the + caller-side client path. `solstone.convey.secure_listener.framing` remains + legitimate because the surviving serve/dialer path shares the wire protocol. + """ + package_dir = Path(solstone.think.link.__file__).resolve().parent + imports: set[str] = set() + for path in package_dir.rglob("*.py"): + imports.update(_convey_imports(path)) + + assert imports == ALLOWED_CONVEY_IMPORTS diff --git a/tests/link/test_link_list.py b/tests/link/test_link_list.py deleted file mode 100644 index 0d645e08f..000000000 --- a/tests/link/test_link_list.py +++ /dev/null @@ -1,427 +0,0 @@ -# SPDX-License-Identifier: AGPL-3.0-only -# Copyright (c) 2026 sol pbc - -from __future__ import annotations - -import argparse -import json -from pathlib import Path -from typing import Any - -import pytest - -from solstone.think import utils as think_utils -from solstone.think.link import list_cli - -OLD = "2026-01-01T00:00:00Z" -NEW = "2026-05-01T00:00:00Z" -SAME = "2026-03-01T00:00:00Z" -_MISSING = object() - - -def _args(**overrides: bool) -> argparse.Namespace: - values = {"observers": False, "json": False} - values.update(overrides) - return argparse.Namespace(**values) - - -def _set_journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: - journal = tmp_path / "journal" - monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) - think_utils._journal_path_cache = None - return journal - - -def _write_bundle(bundle_dir: Path, **fields: Any) -> dict[str, Any]: - peer: dict[str, Any] = { - "label": bundle_dir.name, - "instance_id": bundle_dir.name, - "home_label": "solstone", - "paired_at": OLD, - "fingerprint": "sha256:abcdef123456", - } - for key, value in fields.items(): - if value is _MISSING: - peer.pop(key, None) - else: - peer[key] = value - bundle_dir.mkdir(parents=True, exist_ok=True) - (bundle_dir / "peer.json").write_text( - json.dumps(peer, sort_keys=True), - encoding="utf-8", - ) - return peer - - -def test_no_peers_directory_prints_empty( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - _set_journal(tmp_path, monkeypatch) - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr() - assert out.out == "No peers paired yet.\n" - assert out.err == "" - - -def test_empty_peers_directory_prints_empty( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - (journal / "peers").mkdir(parents=True) - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr() - assert out.out == "No peers paired yet.\n" - assert out.err == "" - - -def test_one_peer_human_output( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - _write_bundle( - journal / "peers" / "inst-1", - label="alpha", - instance_id="inst-1", - home_label="home-a", - fingerprint="sha256:0123456789abcdef", - ) - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr() - assert out.err == "" - assert "Peers:\n" in out.out - assert " alpha (inst-1)\n" in out.out - assert "home: home-a" in out.out - assert "fingerprint: sha256:0123456789abcdef" in out.out - - -def test_multiple_peers_output_is_stable( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - _write_bundle(journal / "peers" / "inst-b", label="beta", instance_id="inst-b") - _write_bundle(journal / "peers" / "inst-a", label="alpha", instance_id="inst-a") - - assert list_cli.main(_args()) == 0 - out1 = capsys.readouterr() - assert list_cli.main(_args()) == 0 - out2 = capsys.readouterr() - - assert out1.out == out2.out - assert out1.err == out2.err - - -def test_sort_newest_first_then_basename( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - _write_bundle( - journal / "peers" / "z-old", - label="old", - instance_id="z-old", - paired_at=OLD, - ) - _write_bundle( - journal / "peers" / "b-same", - label="same-b", - instance_id="b-same", - paired_at=SAME, - ) - _write_bundle( - journal / "peers" / "a-same", - label="same-a", - instance_id="a-same", - paired_at=SAME, - ) - _write_bundle( - journal / "peers" / "m-new", - label="new", - instance_id="m-new", - paired_at=NEW, - ) - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr().out - assert out.index("new (m-new)") < out.index("same-a (a-same)") - assert out.index("same-a (a-same)") < out.index("same-b (b-same)") - assert out.index("same-b (b-same)") < out.index("old (z-old)") - - -def test_missing_paired_at_renders_never_and_sorts_last( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - _write_bundle( - journal / "peers" / "inst-valid", - label="valid", - instance_id="inst-valid", - paired_at=OLD, - ) - _write_bundle( - journal / "peers" / "inst-missing", - label="missing", - instance_id="inst-missing", - paired_at=_MISSING, - ) - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr().out - assert "missing (inst-missing)\n paired never" in out - assert out.index("valid (inst-valid)") < out.index("missing (inst-missing)") - - -def test_empty_home_label_renders_dash( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - bundle = journal / "peers" / "inst-1" - _write_bundle( - bundle, - label="alpha", - instance_id="inst-1", - home_label="", - ) - - assert list_cli.main(_args()) == 0 - human = capsys.readouterr() - assert "home: —" in human.out - - assert list_cli.main(_args(json=True)) == 0 - json_out = capsys.readouterr() - records = json.loads(json_out.out) - assert records == [ - { - "kind": "peer", - "label": "alpha", - "instance_id": "inst-1", - "home_label": None, - "paired_at": OLD, - "fingerprint": "sha256:abcdef123456", - "bundle_dir": str(bundle.resolve()), - } - ] - - -def test_malformed_peer_json_warns_and_skips( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - bad = journal / "peers" / "bad" - bad.mkdir(parents=True) - (bad / "peer.json").write_text("{", encoding="utf-8") - _write_bundle(journal / "peers" / "good", label="good", instance_id="good") - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr() - assert out.err == ( - f"warning: skipping peer bundle {bad.resolve()}: peer.json is not valid JSON\n" - ) - assert "good (good)" in out.out - assert "bad" not in out.out - - -def test_all_malformed_peers_returns_empty_and_warns( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - bad_json = journal / "peers" / "bad-json" - bad_json.mkdir(parents=True) - (bad_json / "peer.json").write_text("{", encoding="utf-8") - missing_label = journal / "peers" / "missing-label" - missing_instance = journal / "peers" / "missing-instance" - _write_bundle(missing_label, label=_MISSING) - _write_bundle(missing_instance, instance_id=_MISSING) - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr() - assert out.out == "No peers paired yet.\n" - assert ( - f"warning: skipping peer bundle {bad_json.resolve()}: " - "peer.json is not valid JSON\n" - ) in out.err - assert ( - f"warning: skipping peer bundle {missing_label.resolve()}: " - "peer.json missing required field 'label'\n" - ) in out.err - assert ( - f"warning: skipping peer bundle {missing_instance.resolve()}: " - "peer.json missing required field 'instance_id'\n" - ) in out.err - - -def test_missing_label_warns_and_skips( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - bad = journal / "peers" / "bad" - _write_bundle(bad, label=_MISSING) - _write_bundle(journal / "peers" / "good", label="good", instance_id="good") - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr() - assert out.err == ( - f"warning: skipping peer bundle {bad.resolve()}: " - "peer.json missing required field 'label'\n" - ) - assert "good (good)" in out.out - assert "bad" not in out.out - - -def test_missing_instance_id_warns_and_skips( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - bad = journal / "peers" / "bad" - _write_bundle(bad, instance_id=_MISSING) - _write_bundle(journal / "peers" / "good", label="good", instance_id="good") - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr() - assert out.err == ( - f"warning: skipping peer bundle {bad.resolve()}: " - "peer.json missing required field 'instance_id'\n" - ) - assert "good (good)" in out.out - assert "bad" not in out.out - - -def test_missing_peer_json_silent_skip( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - (journal / "peers" / "inst-empty").mkdir(parents=True) - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr() - assert out.out == "No peers paired yet.\n" - assert out.err == "" - - -def test_peers_root_file_warns_and_treats_empty( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - peers = journal / "peers" - peers.parent.mkdir(parents=True) - peers.write_text("not a directory", encoding="utf-8") - - assert list_cli.main(_args()) == 0 - - out = capsys.readouterr() - assert out.out == "No peers paired yet.\n" - assert out.err == ( - f"warning: peer bundle root {peers.resolve()} exists but is not a " - "directory; treating as empty\n" - ) - - -def test_json_outputs_flat_single_line_schema( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - bundle = journal / "peers" / "inst-1" - _write_bundle( - bundle, - label="alpha", - instance_id="inst-1", - home_label="home-a", - paired_at=NEW, - fingerprint="sha256:0123456789abcdef", - ) - expected = [ - { - "kind": "peer", - "label": "alpha", - "instance_id": "inst-1", - "home_label": "home-a", - "paired_at": NEW, - "fingerprint": "sha256:0123456789abcdef", - "bundle_dir": str(bundle.resolve()), - } - ] - - assert list_cli.main(_args(json=True)) == 0 - - out = capsys.readouterr() - assert out.out == json.dumps(expected) + "\n" - assert out.out.count("\n") == 1 - assert out.err == "" - - -def test_observers_flag_walks_peer_and_xdg_observer_roots( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, - capsys: pytest.CaptureFixture[str], -) -> None: - journal = _set_journal(tmp_path, monkeypatch) - xdg = tmp_path / "xdg" - monkeypatch.setenv("XDG_CONFIG_HOME", str(xdg)) - peer_bundle = journal / "peers" / "inst-1" - observer_bundle = xdg / "solstone-observer" / "spl" / "observer-a" - _write_bundle( - peer_bundle, - label="peer-a", - instance_id="inst-1", - paired_at=SAME, - ) - _write_bundle( - observer_bundle, - label="observer-a", - instance_id="obs-1", - paired_at=SAME, - ) - - assert list_cli.main(_args(observers=True)) == 0 - human = capsys.readouterr() - assert "Peers:\n" in human.out - assert " peer-a (inst-1)\n" in human.out - assert "\nObservers:\n" in human.out - assert " observer-a (obs-1)\n" in human.out - - assert list_cli.main(_args(observers=True, json=True)) == 0 - json_out = capsys.readouterr() - assert json_out.out.count("\n") == 1 - records = json.loads(json_out.out) - assert [record["kind"] for record in records] == ["peer", "observer"] - assert records[0]["bundle_dir"] == str(peer_bundle.resolve()) - assert records[1]["bundle_dir"] == str(observer_bundle.resolve()) diff --git a/tests/spl/test_service.py b/tests/spl/test_service.py index 3c329e356..55ddac728 100644 --- a/tests/spl/test_service.py +++ b/tests/spl/test_service.py @@ -289,6 +289,6 @@ def test_dispatch_surfaces_for_journal_spl_and_sol_link( assert (module_path, preset_args, surface) == ( "solstone.think.link", [], - "access", + "universal", ) assert link_module.main is link_cli.main diff --git a/tests/test_link_call_parity.py b/tests/test_link_call_parity.py index d31781adf..df369c18c 100644 --- a/tests/test_link_call_parity.py +++ b/tests/test_link_call_parity.py @@ -361,6 +361,31 @@ def test_list_uses_display_label_when_assigned_label_is_empty(runner, monkeypatc ) +def test_list_uses_server_composed_display_labels(runner, monkeypatch): + monkeypatch.setattr(link_call, "_now_utc", lambda: _parse_iso(FROZEN_NOW)) + _add_device( + "sha256:" + ("a" * 64), + "laptop", + paired_at=PAIRED_AT, + client_label="host-1", + ) + _add_device( + "sha256:" + ("b" * 64), + "phone", + paired_at=PAIRED_AT, + client_label="phone", + ) + + result = runner.invoke(link_call.app, ["list"]) + + assert result.exit_code == 0 + lines = [line for line in result.stdout.splitlines() if line.startswith("- ")] + assert len(lines) == 2 + assert lines[0] != lines[1] + assert any("laptop (host-1)" in line for line in lines) + assert any(line.startswith("- phone —") for line in lines) + + def test_list_legacy_observer_role_uses_linked_systems_heading(runner): _add_device("sha256:bbbbbbbbbbbbbbbb0000", "observer", role="observer") @@ -371,6 +396,52 @@ def test_list_legacy_observer_role_uses_linked_systems_heading(runner): assert "Peers:" not in result.stdout +def test_authorized_clients_flat_view(runner, monkeypatch): + monkeypatch.setattr(link_call, "_now_utc", lambda: _parse_iso(FROZEN_NOW)) + first_fingerprint = "sha256:" + ("a" * 64) + second_fingerprint = "sha256:" + ("b" * 64) + _add_device( + first_fingerprint, + "laptop", + paired_at=PAIRED_AT, + last_seen_at=LAST_SEEN_AT, + client_label="host-1", + ) + _add_device( + second_fingerprint, + "phone", + paired_at=PAIRED_AT, + client_label="phone", + ) + + result = runner.invoke(link_call.app, ["authorized-clients"]) + + assert result.exit_code == 0 + assert result.stdout == ( + f"{first_fingerprint} laptop (host-1) last seen 30 minutes ago\n" + f"{second_fingerprint} phone last seen never\n" + ) + + +def test_authorized_clients_empty_store(runner): + result = runner.invoke(link_call.app, ["authorized-clients"]) + + assert result.exit_code == 0 + assert result.stdout == "No authorized clients.\n" + + +def test_observer_pause_stub_does_not_require_service(runner, monkeypatch): + def fail_get_client(): + raise AssertionError("observer-pause must not make an HTTP request") + + monkeypatch.setattr(link_call, "get_client", fail_get_client) + + result = runner.invoke(link_call.app, ["observer-pause"]) + + assert result.exit_code == 0 + assert result.stdout == "observer-pause is not yet available.\n" + + def test_unpair_success_and_not_found_outputs(runner): _add_device("sha256:" + ("a" * 64), "phone")