refactor(vault): trim the malformed-state coverage and drop the load_config shape guard
load_config() always returns a dict, so the isinstance re-init was defense-in-depth. Coverage folded to one invariant per bug and surface: every VaultStore reader raises VaultError on a corrupt file, `hermes vault` prints a clean error line, and both the CLI sources toggle and the TUI vault.source.set RPC persist the opt-out over a scalar section.
This commit is contained in:
@@ -152,8 +152,6 @@ def _cmd_sources(args) -> None:
|
||||
c.print(f"[red]Unknown password manager {name!r}[/] (expected one of {', '.join(classes)})")
|
||||
return
|
||||
cfg = load_config()
|
||||
if not isinstance(cfg, dict):
|
||||
cfg = {}
|
||||
section = _ensure_dict(_ensure_dict(cfg, "vault"), name)
|
||||
if args.enable:
|
||||
section.pop("enabled", None) # detected managers are on by default; drop the opt-out
|
||||
|
||||
@@ -24,27 +24,6 @@ def _store(tmp_path) -> VaultStore:
|
||||
return store
|
||||
|
||||
|
||||
def test_malformed_json_vault_file_raises_vault_error(tmp_path):
|
||||
store = _store(tmp_path)
|
||||
_corrupt(store, b"{not json")
|
||||
with pytest.raises(VaultError, match="corrupted"):
|
||||
store.list_items()
|
||||
|
||||
|
||||
def test_non_dict_json_vault_file_raises_vault_error(tmp_path):
|
||||
store = _store(tmp_path)
|
||||
_corrupt(store, b"[1, 2]")
|
||||
with pytest.raises(VaultError, match="corrupted"):
|
||||
store.list_items()
|
||||
|
||||
|
||||
def test_non_utf8_vault_file_raises_vault_error(tmp_path):
|
||||
store = _store(tmp_path)
|
||||
_corrupt(store, b"\xff\xfe\x00binary")
|
||||
with pytest.raises(VaultError, match="corrupted"):
|
||||
store.list_items()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"op",
|
||||
["list_items", "get_meta", "remove_item", "resolve_secret", "add_item"],
|
||||
|
||||
@@ -8,7 +8,7 @@ from types import SimpleNamespace
|
||||
import yaml
|
||||
|
||||
from agent.vault_store import VaultStore
|
||||
from hermes_cli.vault import _cmd_rm, _cmd_sources, vault_command
|
||||
from hermes_cli.vault import _cmd_sources, vault_command
|
||||
|
||||
|
||||
def _corrupt_vault(home):
|
||||
@@ -47,16 +47,3 @@ def test_vault_command_surfaces_corrupt_store_as_clean_error(tmp_path, monkeypat
|
||||
out = capsys.readouterr().out
|
||||
assert "Error" in out
|
||||
assert "corrupted" in out
|
||||
|
||||
|
||||
def test_vault_rm_surfaces_corrupt_store_as_clean_error(tmp_path, monkeypatch, capsys):
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
_corrupt_vault(home)
|
||||
|
||||
vault_command(SimpleNamespace(_vault_handler=_cmd_rm, handle="vault_x"))
|
||||
|
||||
out = capsys.readouterr().out
|
||||
assert "Error" in out
|
||||
assert "corrupted" in out
|
||||
|
||||
@@ -156,53 +156,6 @@ def test_source_set_tolerates_scalar_vault_section(home, monkeypatch):
|
||||
assert rows["bitwarden"]["enabled"] is False
|
||||
|
||||
|
||||
def test_source_set_tolerates_scalar_named_section(home, monkeypatch):
|
||||
"""``vault: {bitwarden: true}`` must not crash either: the named section is
|
||||
coerced before ``enabled`` is written."""
|
||||
monkeypatch.setattr(
|
||||
"agent.vault_backends.base.is_installed", lambda name: name == "bitwarden"
|
||||
)
|
||||
(home / "config.yaml").write_text("vault:\n bitwarden: true\n")
|
||||
_result(
|
||||
srv._methods["vault.source.set"](83, {"name": "bitwarden", "enabled": False})
|
||||
)
|
||||
rows = _sources_rows(home)
|
||||
assert rows["bitwarden"]["enabled"] is False
|
||||
|
||||
|
||||
def test_source_set_enable_tolerates_scalar_vault_section(home, monkeypatch):
|
||||
"""The enable arm (``section.pop``) must also survive a scalar ``vault:``
|
||||
section: coercion happens before either write shape."""
|
||||
monkeypatch.setattr(
|
||||
"agent.vault_backends.base.is_installed", lambda name: name == "bitwarden"
|
||||
)
|
||||
(home / "config.yaml").write_text("vault: true\n")
|
||||
_result(
|
||||
srv._methods["vault.source.set"](84, {"name": "bitwarden", "enabled": True})
|
||||
)
|
||||
rows = _sources_rows(home)
|
||||
assert rows["bitwarden"]["enabled"] is True
|
||||
|
||||
|
||||
def test_vault_list_reports_corrupt_store_cleanly(home):
|
||||
"""A decryptable-but-malformed vault file degrades to the 5095 envelope
|
||||
carrying the VaultError text, never a traceback."""
|
||||
from agent.vault_store import VaultStore
|
||||
|
||||
store = VaultStore(home / "vault")
|
||||
store.add_item(
|
||||
"login",
|
||||
"site",
|
||||
{"identifier_type": "email", "identifier": "u", "password": "p"},
|
||||
origin="https://x.example",
|
||||
)
|
||||
store._vault_path.write_bytes(store._fernet().encrypt(b"{not json"))
|
||||
|
||||
err = _error(srv._methods["vault.list"](95, {}))
|
||||
assert err["code"] == 5095
|
||||
assert "corrupted" in err["message"]
|
||||
|
||||
|
||||
def test_remove_is_idempotent(home):
|
||||
item_id = _result(srv._methods["vault.add"](1, dict(_LOGIN_PARAMS)))["id"]
|
||||
assert _result(srv._methods["vault.remove"](2, {"id": item_id}))["removed"] is True
|
||||
|
||||
@@ -88,8 +88,6 @@ def _(rid, params: dict) -> dict:
|
||||
return _err(rid, 5095, f"unknown vault source: {name}")
|
||||
enabled = bool(params.get("enabled"))
|
||||
cfg = load_config()
|
||||
if not isinstance(cfg, dict):
|
||||
cfg = {}
|
||||
section = _ensure_dict(_ensure_dict(cfg, "vault"), name)
|
||||
if enabled:
|
||||
section.pop("enabled", None) # detected managers are on by default; this removes the opt-out
|
||||
|
||||
Reference in New Issue
Block a user