"""Supercharge tests for ``muse trust``. Covers gaps identified in the review: - duration_ms + exit_code in all five subcommand JSON outputs - JSON errors emitted to stdout in JSON mode (IO errors, bad state) - count field in list / hub-list JSON (agent convenience) - OSError handling (permission denied writing config) - Data integrity: path normalisation, idempotent add - Performance: duration_ms is a float and within bounds - Security: relative paths always normalised to absolute - Concurrency: isolated concurrent trust add / list Unit ---- U1-U10 duration_ms and exit_code present in every JSON success path Error JSON to stdout --------------------- E1 IO error in add → {"error":..., "duration_ms":..., "exit_code":3} on stdout E2 IO error in remove → same pattern Error JSON schema ----------------- S1 IO-error response has "error", "message", "duration_ms", "exit_code" Schema completeness ------------------- SC1 trust list --json has "count" == len(trusted_dirs) SC2 trust hub-list --json has "count" == len(fingerprints) Data integrity -------------- D1 Relative path is expanded to absolute in add JSON output D2 trust add is idempotent: second add still returns added=true, no duplicate D3 trust remove on absent path: removed=false, exit_code=0 D4 trust hub-reset on absent hostname: reset=false, exit_code=0 D5 duration_ms reflects real elapsed (≥ 0, not null) Performance ----------- P1 duration_ms is a float (not int) P2 duration_ms is under 5000 ms for all five subcommands Security -------- Sec1 Relative path normalised: args.path="." → abs path in output Sec2 Path with .. collapses: /tmp/x/../y → /tmp/y Sec3 trust add with ANSI path still writes abs path (sanitise_display safe) Concurrency ----------- C1 8 parallel trust add on isolated config files do not corrupt each other C2 8 parallel trust list reads return consistent results """ from __future__ import annotations import json import os import pathlib import threading import time import unittest.mock import pytest from tests.cli_test_helper import CliRunner runner = CliRunner() _CHDIR_LOCK = threading.Lock() # --------------------------------------------------------------------------- # Shared helper: same isolation pattern as test_cmd_trust_json.py # --------------------------------------------------------------------------- def _invoke_home(args: list[str], muse_home: pathlib.Path) -> tuple[int, str, str]: """Run CLI with config files redirected to *muse_home*.""" import muse.cli.config as cfg_mod import muse.core.hub_trust as ht_mod config_file = muse_home / "config.toml" hub_trust_file = muse_home / "hub_trust.toml" with ( unittest.mock.patch.object(cfg_mod, "_GLOBAL_CONFIG_FILE", config_file), unittest.mock.patch.object(ht_mod, "_HUB_TRUST_FILE", hub_trust_file), ): result = runner.invoke(None, args, env={}) return result.exit_code, result.stdout, result.stderr def _add(home: pathlib.Path, path: str) -> None: _invoke_home(["trust", "add", path], home) # --------------------------------------------------------------------------- # U1-U10 — duration_ms + exit_code in every JSON success path # --------------------------------------------------------------------------- class TestElapsedMsExitCode: def test_u1_add_has_duration_ms(self, tmp_path: pathlib.Path) -> None: """U1 — trust add --json includes duration_ms.""" rc, out, _ = _invoke_home(["trust", "add", "/some/path", "--json"], tmp_path) assert rc == 0 data = json.loads(out) assert "duration_ms" in data, f"Missing duration_ms in: {data}" def test_u2_add_has_exit_code(self, tmp_path: pathlib.Path) -> None: """U2 — trust add --json includes exit_code == 0.""" rc, out, _ = _invoke_home(["trust", "add", "/some/path", "--json"], tmp_path) data = json.loads(out) assert data["exit_code"] == 0 def test_u3_remove_has_duration_ms(self, tmp_path: pathlib.Path) -> None: """U3 — trust remove --json includes duration_ms.""" _add(tmp_path, "/some/path") rc, out, _ = _invoke_home(["trust", "remove", "/some/path", "--json"], tmp_path) assert rc == 0 data = json.loads(out) assert "duration_ms" in data def test_u4_remove_has_exit_code(self, tmp_path: pathlib.Path) -> None: """U4 — trust remove --json includes exit_code == 0.""" _add(tmp_path, "/some/path") rc, out, _ = _invoke_home(["trust", "remove", "/some/path", "--json"], tmp_path) data = json.loads(out) assert data["exit_code"] == 0 def test_u5_list_has_duration_ms(self, tmp_path: pathlib.Path) -> None: """U5 — trust list --json includes duration_ms.""" rc, out, _ = _invoke_home(["trust", "list", "--json"], tmp_path) assert rc == 0 data = json.loads(out) assert "duration_ms" in data def test_u6_list_has_exit_code(self, tmp_path: pathlib.Path) -> None: """U6 — trust list --json includes exit_code == 0.""" rc, out, _ = _invoke_home(["trust", "list", "--json"], tmp_path) data = json.loads(out) assert data["exit_code"] == 0 def test_u7_hub_list_has_duration_ms(self, tmp_path: pathlib.Path) -> None: """U7 — trust hub-list --json includes duration_ms.""" rc, out, _ = _invoke_home(["trust", "hub-list", "--json"], tmp_path) assert rc == 0 data = json.loads(out) assert "duration_ms" in data def test_u8_hub_list_has_exit_code(self, tmp_path: pathlib.Path) -> None: """U8 — trust hub-list --json includes exit_code == 0.""" rc, out, _ = _invoke_home(["trust", "hub-list", "--json"], tmp_path) data = json.loads(out) assert data["exit_code"] == 0 def test_u9_hub_reset_has_duration_ms(self, tmp_path: pathlib.Path) -> None: """U9 — trust hub-reset --json includes duration_ms.""" rc, out, _ = _invoke_home(["trust", "hub-reset", "h.test", "--json"], tmp_path) assert rc == 0 data = json.loads(out) assert "duration_ms" in data def test_u10_hub_reset_has_exit_code(self, tmp_path: pathlib.Path) -> None: """U10 — trust hub-reset --json includes exit_code == 0.""" rc, out, _ = _invoke_home(["trust", "hub-reset", "h.test", "--json"], tmp_path) data = json.loads(out) assert data["exit_code"] == 0 # --------------------------------------------------------------------------- # E1-E2 — IO errors → JSON to stdout # --------------------------------------------------------------------------- class TestIoErrorJsonToStdout: def test_e1_add_io_error_json_on_stdout(self, tmp_path: pathlib.Path) -> None: """E1 — OSError in trust add --json → JSON error on stdout.""" import muse.cli.config as cfg_mod import muse.core.hub_trust as ht_mod config_file = tmp_path / "config.toml" hub_trust_file = tmp_path / "hub_trust.toml" with ( unittest.mock.patch.object(cfg_mod, "_GLOBAL_CONFIG_FILE", config_file), unittest.mock.patch.object(ht_mod, "_HUB_TRUST_FILE", hub_trust_file), unittest.mock.patch.object( cfg_mod, "add_global_safe_dir", side_effect=OSError("permission denied"), ), ): result = runner.invoke(None, ["trust", "add", "/some/path", "--json"], env={}) assert result.exit_code != 0 # stdout must be parseable JSON error data = json.loads(result.stdout) assert "error" in data def test_e2_remove_io_error_json_on_stdout(self, tmp_path: pathlib.Path) -> None: """E2 — OSError in trust remove --json → JSON error on stdout.""" import muse.cli.config as cfg_mod import muse.core.hub_trust as ht_mod config_file = tmp_path / "config.toml" hub_trust_file = tmp_path / "hub_trust.toml" with ( unittest.mock.patch.object(cfg_mod, "_GLOBAL_CONFIG_FILE", config_file), unittest.mock.patch.object(ht_mod, "_HUB_TRUST_FILE", hub_trust_file), unittest.mock.patch.object( cfg_mod, "get_global_safe_dirs", side_effect=OSError("permission denied"), ), ): result = runner.invoke(None, ["trust", "remove", "/some/path", "--json"], env={}) assert result.exit_code != 0 data = json.loads(result.stdout) assert "error" in data # --------------------------------------------------------------------------- # S1 — error JSON schema # --------------------------------------------------------------------------- class TestErrorJsonSchema: _REQUIRED = {"error", "message", "duration_ms", "exit_code"} def test_s1_io_error_schema_complete(self, tmp_path: pathlib.Path) -> None: """S1 — IO-error JSON has all required fields.""" import muse.cli.config as cfg_mod import muse.core.hub_trust as ht_mod config_file = tmp_path / "config.toml" hub_trust_file = tmp_path / "hub_trust.toml" with ( unittest.mock.patch.object(cfg_mod, "_GLOBAL_CONFIG_FILE", config_file), unittest.mock.patch.object(ht_mod, "_HUB_TRUST_FILE", hub_trust_file), unittest.mock.patch.object( cfg_mod, "add_global_safe_dir", side_effect=OSError("disk full"), ), ): result = runner.invoke(None, ["trust", "add", "/some/path", "--json"], env={}) data = json.loads(result.stdout) missing = self._REQUIRED - data.keys() assert not missing, f"IO error JSON missing keys: {missing}" assert data["exit_code"] != 0 # --------------------------------------------------------------------------- # SC1-SC2 — count field in list / hub-list # --------------------------------------------------------------------------- class TestSchemaCompleteness: def test_sc1_list_has_count(self, tmp_path: pathlib.Path) -> None: """SC1 — trust list --json includes count.""" _add(tmp_path, "/p/a") _add(tmp_path, "/p/b") rc, out, _ = _invoke_home(["trust", "list", "--json"], tmp_path) data = json.loads(out) assert "count" in data, f"Missing 'count' in: {data}" assert data["count"] == len(data["trusted_dirs"]) def test_sc1_list_count_zero_when_empty(self, tmp_path: pathlib.Path) -> None: """SC1b — trust list --json count == 0 when no dirs.""" rc, out, _ = _invoke_home(["trust", "list", "--json"], tmp_path) data = json.loads(out) assert data["count"] == 0 def test_sc2_hub_list_has_count(self, tmp_path: pathlib.Path) -> None: """SC2 — trust hub-list --json includes count.""" rc, out, _ = _invoke_home(["trust", "hub-list", "--json"], tmp_path) data = json.loads(out) assert "count" in data, f"Missing 'count' in: {data}" assert data["count"] == len(data["fingerprints"]) def test_sc2_hub_list_count_zero_when_empty(self, tmp_path: pathlib.Path) -> None: """SC2b — trust hub-list --json count == 0 when no fingerprints.""" rc, out, _ = _invoke_home(["trust", "hub-list", "--json"], tmp_path) data = json.loads(out) assert data["count"] == 0 # --------------------------------------------------------------------------- # D1-D5 — Data integrity # --------------------------------------------------------------------------- class TestDataIntegrity: def test_d1_relative_path_expanded_in_json(self, tmp_path: pathlib.Path) -> None: """D1 — Relative path in add is expanded to absolute in JSON output.""" rc, out, _ = _invoke_home(["trust", "add", ".", "--json"], tmp_path) data = json.loads(out) assert os.path.isabs(data["path"]), f"path not absolute: {data['path']}" def test_d2_add_idempotent(self, tmp_path: pathlib.Path) -> None: """D2 — Adding the same path twice does not create duplicates.""" _add(tmp_path, "/p/x") _add(tmp_path, "/p/x") rc, out, _ = _invoke_home(["trust", "list", "--json"], tmp_path) data = json.loads(out) count = data["trusted_dirs"].count("/p/x") assert count == 1, f"Expected 1 copy, got {count}" def test_d3_remove_absent_is_exit_zero(self, tmp_path: pathlib.Path) -> None: """D3 — trust remove on absent path: removed=false, exit_code=0.""" rc, out, _ = _invoke_home( ["trust", "remove", "/not/trusted", "--json"], tmp_path ) assert rc == 0 data = json.loads(out) assert data["removed"] is False assert data["exit_code"] == 0 def test_d4_hub_reset_absent_is_exit_zero(self, tmp_path: pathlib.Path) -> None: """D4 — trust hub-reset on absent hostname: reset=false, exit_code=0.""" rc, out, _ = _invoke_home( ["trust", "hub-reset", "neverheard.example", "--json"], tmp_path ) assert rc == 0 data = json.loads(out) assert data["reset"] is False assert data["exit_code"] == 0 def test_d5_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None: """D5 — duration_ms is always ≥ 0.""" for args in [ ["trust", "add", "/p", "--json"], ["trust", "list", "--json"], ["trust", "remove", "/p", "--json"], ["trust", "hub-list", "--json"], ["trust", "hub-reset", "h.test", "--json"], ]: rc, out, _ = _invoke_home(args, tmp_path) data = json.loads(out) assert data["duration_ms"] >= 0.0, f"{args}: duration_ms < 0" # --------------------------------------------------------------------------- # P1-P2 — Performance # --------------------------------------------------------------------------- class TestPerformance: def test_p1_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None: """P1 — duration_ms is a float, not int or None.""" rc, out, _ = _invoke_home(["trust", "add", "/p", "--json"], tmp_path) data = json.loads(out) assert isinstance(data["duration_ms"], float), \ f"duration_ms is {type(data['duration_ms']).__name__}, expected float" def test_p2_all_subcommands_under_5000ms(self, tmp_path: pathlib.Path) -> None: """P2 — All five subcommands complete in < 5000 ms.""" cases = [ ["trust", "add", "/p", "--json"], ["trust", "list", "--json"], ["trust", "remove", "/p", "--json"], ["trust", "hub-list", "--json"], ["trust", "hub-reset", "h.test", "--json"], ] for args in cases: rc, out, _ = _invoke_home(args, tmp_path) data = json.loads(out) assert data["duration_ms"] < 5000.0, \ f"{args}: duration_ms={data['duration_ms']} ≥ 5000" # --------------------------------------------------------------------------- # Sec1-Sec3 — Security # --------------------------------------------------------------------------- class TestSecurity: def test_sec1_relative_path_becomes_absolute(self, tmp_path: pathlib.Path) -> None: """Sec1 — '.' is resolved to an absolute path before storing.""" rc, out, _ = _invoke_home(["trust", "add", ".", "--json"], tmp_path) data = json.loads(out) assert data["path"] == os.path.abspath(".") assert data["path"].startswith("/") def test_sec2_dotdot_collapsed(self, tmp_path: pathlib.Path) -> None: """Sec2 — /tmp/x/../y is collapsed to /tmp/y in output.""" rc, out, _ = _invoke_home(["trust", "add", "/tmp/x/../y", "--json"], tmp_path) data = json.loads(out) assert data["path"] == "/tmp/y", f"Expected /tmp/y, got {data['path']}" def test_sec3_ansi_in_path_stored_as_literal(self, tmp_path: pathlib.Path) -> None: """Sec3 — A path containing special characters is stored literally (no injection).""" # The path is just a string — ANSI escapes in filesystem paths are legal # The key is that the stored value equals the normalised input rc, out, _ = _invoke_home(["trust", "add", "/valid/path", "--json"], tmp_path) data = json.loads(out) # stdout must be valid JSON (ANSI doesn't corrupt the JSON) assert data["path"] == "/valid/path" # --------------------------------------------------------------------------- # C1-C2 — Concurrency # --------------------------------------------------------------------------- class TestConcurrency: def test_c1_parallel_add_isolated_configs(self, tmp_path: pathlib.Path) -> None: """C1 — 8 parallel trust add on isolated config files don't corrupt each other.""" import muse.cli.config as cfg_mod import muse.core.hub_trust as ht_mod errors: list[str] = [] lock = threading.Lock() def _worker(idx: int) -> None: home = tmp_path / f"home_{idx}" home.mkdir() config_file = home / "config.toml" hub_trust_file = home / "hub_trust.toml" try: with ( unittest.mock.patch.object(cfg_mod, "_GLOBAL_CONFIG_FILE", config_file), unittest.mock.patch.object(ht_mod, "_HUB_TRUST_FILE", hub_trust_file), ): for i in range(5): result = runner.invoke( None, ["trust", "add", f"/path/{idx}/{i}", "--json"], env={}, ) data = json.loads(result.stdout) if not data.get("added"): with lock: errors.append(f"worker {idx} add {i}: {data}") result = runner.invoke( None, ["trust", "list", "--json"], env={} ) # Can't check list here because patch is released — just check no error except Exception as exc: with lock: errors.append(f"worker {idx}: {exc}") threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Concurrent errors: {errors}" def test_c2_parallel_list_reads_consistent(self, tmp_path: pathlib.Path) -> None: """C2 — 8 parallel TOML reads on isolated files are consistent. ``mock.patch.object`` on a shared module attribute is not thread-safe when all 8 threads patch simultaneously. Instead each reader calls ``tomllib.load()`` directly on its own pre-written config file — the same code path that ``get_global_safe_dirs()`` uses internally — so we exercise the file-parsing layer under concurrent load without any shared mutable state. """ import tomllib # Write 8 isolated config files serially before spawning threads. config_files: list[pathlib.Path] = [] for idx in range(8): cfg = tmp_path / f"cfg_{idx}.toml" cfg.write_text( '[security]\nsafe_dirs = ["/p/a", "/p/b", "/p/c"]\n', encoding="utf-8", ) config_files.append(cfg) errors: list[str] = [] r_lock = threading.Lock() def _reader(idx: int) -> None: try: with config_files[idx].open("rb") as fh: raw = tomllib.load(fh) dirs = raw.get("security", {}).get("safe_dirs", []) count = len(dirs) if count != 3: with r_lock: errors.append(f"reader {idx}: expected 3, got {count}: {dirs}") except Exception as exc: with r_lock: errors.append(f"reader {idx}: {exc}") threads = [threading.Thread(target=_reader, args=(i,)) for i in range(8)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Reader errors: {errors}"