Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion grapharc/cli/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -161,8 +161,12 @@ def load(explicit: Path | None = None, *, cwd: Path | None = None) -> Settings:
return Settings()

try:
# `UnicodeDecodeError` is a `ValueError`, so it belongs in this tuple
# explicitly: without it a stray binary `grapharc.toml` in the working
# directory tracebacks out of every configurable command, because this
# file is picked up implicitly rather than named by the operator.
document = tomllib.loads(path.read_text(encoding="utf-8"))
except (OSError, tomllib.TOMLDecodeError) as exc:
except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc:
raise ConfigError(f"{path}: {exc}") from exc

table = document.get(TABLE, document)
Expand Down
8 changes: 7 additions & 1 deletion grapharc/cli/graphrun.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,13 @@ def load_topology(path: Path) -> dict[str, Any]:
"""
if not path.is_file():
raise PlanSetupError(f"no such graph file: {path}")
text = path.read_text(encoding="utf-8")
try:
text = path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError) as exc:
# A topology saved as UTF-16, or truncated in transit, is a file we
# cannot run — not a crash. `UnicodeDecodeError` is a `ValueError`, so
# neither decoder below would ever have caught it.
raise PlanSetupError(f"{path}: {exc}") from exc
try:
if path.suffix.lower() == ".toml":
return tomllib.loads(text)
Expand Down
14 changes: 14 additions & 0 deletions grapharc/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -242,9 +242,23 @@ def _existing_trace(path: Path, *, command: str, as_json: bool) -> TraceRecorder
Checked before constructing the recorder because `TraceRecorder.__init__`
creates the parent directory: a typo in a read-only command should not leave
a directory behind.

Existence is not enough. A directory, a file whose permissions forbid the
read, or any other `OSError` used to escape as a traceback with exit 1,
because the handlers below catch only `TraceReadError` — so the file is
opened here, where the failure is still reportable as the exit-2 document
the contract promises.
"""
if not path.exists():
return fail(f"no such trace file: {path}", as_json=as_json, command=command)
try:
path.open("rb").close()
except OSError as exc:
return fail(
f"unreadable trace file: {path}: {exc.strerror or exc}",
as_json=as_json,
command=command,
)
return TraceRecorder(path)


Expand Down
4 changes: 4 additions & 0 deletions grapharc/cli/plan.py
Original file line number Diff line number Diff line change
Expand Up @@ -251,6 +251,10 @@ def plan(
def _announce(message: str) -> None:
# Printed *and flushed* before the run parks: a terminal user (or a
# log tailer) must learn how to answer without waiting for the exit.
# Silent in JSON mode: stdout there carries exactly one document, and
# a notice printed ahead of it makes the whole output unparseable.
if as_json:
return
print(message, flush=True, file=sys.stdout)

approval = file_approval(
Expand Down
31 changes: 31 additions & 0 deletions tests/test_approval.py
Original file line number Diff line number Diff line change
Expand Up @@ -299,3 +299,34 @@ def test_handshake_files_are_written_atomically(tmp_path):
_write_atomically(target, {"fingerprint": "fp"})
assert json.loads(target.read_text()) == {"fingerprint": "fp"}
assert not list(tmp_path.glob("*.tmp"))


def test_plan_approve_in_json_mode_emits_one_document(tmp_path, capsys):
"""The park notice used to print ahead of the document, so nothing parsed.

`--approve` is the flag most likely to be driven unattended: a script starts
a gated plan, a human answers out of band, the script reads the result. That
is exactly the combination whose output could not be loaded.
"""
trace = tmp_path / "run" / "trace.jsonl"

code = main(
["plan", "ship it", "--approve", "--approval-timeout", "0.2",
"--trace", str(trace), "--json"]
)
captured = capsys.readouterr()

assert code == 1, "an unanswered gate is a negative answer, not a crash"
assert captured.err == ""
payload = json.loads(captured.out)
assert payload["ok"] is False
assert "not approved" in payload["detail"]


def test_plan_approve_in_text_mode_still_announces_how_to_answer(tmp_path, capsys):
"""Silencing the notice in JSON mode must not silence it for a human."""
trace = tmp_path / "run" / "trace.jsonl"

main(["plan", "ship it", "--approve", "--approval-timeout", "0.2", "--trace", str(trace)])

assert "grapharc approve" in capsys.readouterr().out
48 changes: 48 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -373,6 +373,37 @@ def test_a_malformed_trace_fails_as_one_json_document(argv, tmp_path, capsys):
assert err == ""


# A path that exists but cannot be read is the same class of failure as a
# malformed one, and used to escape as a traceback with exit 1: `_existing_trace`
# tested `exists()` and the handlers catch only `TraceReadError`, so every other
# `OSError` went straight past both.
@pytest.mark.parametrize("argv", READERS, ids=lambda argv: argv[0])
def test_a_directory_where_a_trace_belongs_is_a_report_not_a_traceback(
argv, tmp_path, capsys
):
directory = tmp_path / "adir"
directory.mkdir()
code, out, err = call([argv[0], str(directory), *argv[1:]], capsys)
assert code == 2
assert out == ""
assert err.startswith(f"error: unreadable trace file: {directory}: ")
assert "Traceback" not in err


@pytest.mark.parametrize("argv", READERS, ids=lambda argv: argv[0])
def test_a_directory_where_a_trace_belongs_fails_as_one_json_document(
argv, tmp_path, capsys
):
directory = tmp_path / "adir"
directory.mkdir()
code, payload, err = call_json([argv[0], str(directory), *argv[1:]], capsys)
assert code == 2
assert payload["ok"] is False
assert payload["command"] == argv[0]
assert payload["error"].startswith(f"unreadable trace file: {directory}: ")
assert err == ""


# -- models -------------------------------------------------------------------


Expand Down Expand Up @@ -1395,6 +1426,23 @@ def test_run_says_which_file_is_missing(tmp_path, capsys):
assert "no such graph file" in err


def test_run_reports_a_graph_file_that_is_not_utf8(tmp_path, capsys):
"""`UnicodeDecodeError` is a `ValueError`, so neither decoder caught it.

A topology saved as UTF-16 or truncated in transit used to exit 1 with a
traceback and an empty document.
"""
binary = tmp_path / "bin.json"
binary.write_bytes(b"\xff\xfe\x00binary")

code, payload, err = call_json(["run", str(binary)], capsys)

assert code == 2
assert payload["ok"] is False
assert "utf-8" in payload["error"]
assert err == ""


def test_a_policy_document_gates_a_hand_written_graph_too(tmp_path, capsys):
"""§12.2 on the deterministic path: the TOML file decides here as well."""
graph = _write_graph(tmp_path, _DENIED_GRAPH)
Expand Down
13 changes: 13 additions & 0 deletions tests/test_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -235,6 +235,19 @@ def test_malformed_toml_names_the_file(tmp_path):
load(path)


def test_a_config_that_is_not_utf8_names_the_file(tmp_path):
"""`UnicodeDecodeError` is a `ValueError`, so it was in neither except tuple.

This file is picked up implicitly from the working directory, so a stray
binary `grapharc.toml` used to traceback out of every configurable command.
"""
path = tmp_path / CONFIG_NAME
path.write_bytes(b"\xff\xfe")

with pytest.raises(ConfigError, match=CONFIG_NAME):
load(path)


# -- the resolver itself -----------------------------------------------------


Expand Down
Loading