Skip to content

fix(cli): diagram in the canonical smoke matrix; scoped machine-output console redirect #1655

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Shashankss1205 merged 1 commit into from
Aug 19, 2026
Merged

fix(cli): diagram in the canonical smoke matrix; scoped machine-output console redirect #1655

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
43 changes: 30 additions & 13 deletions src/codegraphcontext/cli/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1178,14 +1178,22 @@ def diagram(
if level not in ("file", "call"):
console.print(f"[red]Unknown --level '{level}' (expected file or call)[/red]")
raise typer.Exit(code=2)
# stdout carries only the diagram; init chatter goes to stderr for the
# init window only (restored after — a permanent rebind leaks a closed
# stream into later in-process invocations).
import sys as _sys
from . import cli_helpers as _cli_helpers
# NB: the .file property GETTER materializes the current (possibly
# transient, CliRunner-captured) stream; the dynamic state lives in
# ._file, which is None while the console follows sys.stdout live.
_prev_console_file = _cli_helpers.console._file
if output is None:
# stdout carries only the diagram; init chatter goes to stderr.
import sys as _sys
from . import cli_helpers as _cli_helpers
_cli_helpers.console.file = _sys.stderr

_load_credentials()
services = _initialize_services(context)
try:
_load_credentials()
services = _initialize_services(context)
finally:
_cli_helpers.console._file = _prev_console_file
if not all(services[:3]):
raise typer.Exit(code=1)
db_manager, graph_builder, code_finder = services[:3]
Expand Down Expand Up @@ -3018,15 +3026,24 @@ def analyze_complexity(
if output_format not in ("text", "json", "csv"):
console.print(f"[red]Unknown --format '{output_format}' (expected text, json or csv)[/red]")
raise typer.Exit(code=2)
# stdout must carry ONLY the parseable document in machine formats: the
# service-init chatter from cli_helpers prints to stdout, so reroute that
# console to stderr FOR THE INIT WINDOW ONLY (`--format json | jq .`).
# The rebind must be restored — a permanent one leaks a closed stream into
# later in-process invocations (CliRunner-based tests).
import sys as _sys
from . import cli_helpers as _cli_helpers
# NB: the .file property GETTER materializes the current (possibly
# transient, CliRunner-captured) stream; the dynamic state lives in
# ._file, which is None while the console follows sys.stdout live.
_prev_console_file = _cli_helpers.console._file
if output_format in ("json", "csv"):
# stdout must carry ONLY the parseable document: the service-init
# chatter from cli_helpers prints to stdout, so reroute that console
# to stderr before initializing anything (`--format json | jq .`).
import sys as _sys
from . import cli_helpers as _cli_helpers
_cli_helpers.console.file = _sys.stderr
_load_credentials()
services = _initialize_services(context)
try:
_load_credentials()
services = _initialize_services(context)
finally:
_cli_helpers.console._file = _prev_console_file
if not all(services[:3]):
raise typer.Exit(code=1)
db_manager, graph_builder, code_finder = services[:3]
Expand Down
4 changes: 4 additions & 0 deletions tests/integration/cli/test_cli_commands.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,9 @@ def delete_repository_from_graph(self, _):


class _FakeCodeFinder:
def export_diagram_edges(self, *_args, **_kwargs):
return {"edges": [("repo/main.py", "json")], "total_count": 1, "truncated": False}

def find_by_function_name(self, *_args, **_kwargs):
return [{"name": "foo", "path": "repo/main.py", "line_number": 2, "is_dependency": False}]

Expand Down Expand Up @@ -472,6 +475,7 @@ def test_all_canonical_cli_commands_run_with_kuzudb(kuzudb_env, cli_test_stubs,
["update", "."],
["clean"],
["stats"],
["diagram"],
["delete", "."],
["visualize"],
["list"],
Expand Down
Loading