From 9aa6b862a921003e2ba24ee9a1d5c8558d419488 Mon Sep 17 00:00:00 2001 From: bearsyankees Date: Thu, 27 Aug 2026 17:46:29 -0400 Subject: [PATCH] feat(cloud): improve human navigation and output --- strix/interface/cloud/__init__.py | 2 + strix/interface/cloud/http.py | 6 ++ strix/interface/cloud/render.py | 123 ++++++++++++++++++++++++-- strix/interface/cloud/runner.py | 14 ++- strix/interface/cloud/workspaces.py | 29 ++++++- tests/test_cloud_cli.py | 128 ++++++++++++++++++++++++++++ 6 files changed, 289 insertions(+), 13 deletions(-) diff --git a/strix/interface/cloud/__init__.py b/strix/interface/cloud/__init__.py index 2838e32b..7ce2ce81 100644 --- a/strix/interface/cloud/__init__.py +++ b/strix/interface/cloud/__init__.py @@ -42,6 +42,8 @@ def run_cloud(argv: list[str]) -> int: return 0 group, rest = argv[0], argv[1:] + if group == "workspace": + group = "workspaces" if group in ("login", "logout", "whoami"): return _run_session(console, group, rest) if group == "credits": diff --git a/strix/interface/cloud/http.py b/strix/interface/cloud/http.py index 4c735dc0..f4e78328 100644 --- a/strix/interface/cloud/http.py +++ b/strix/interface/cloud/http.py @@ -98,6 +98,12 @@ def parsed(response: requests.Response) -> Any: def check(response: requests.Response) -> Any: data = parsed(response) if response.ok: + content_type = response.headers.get("content-type", "").lower() + if "application/json" not in content_type: + raise CloudError( + "the server returned a non-JSON response. Check STRIX_APP_URL and preview " + "access, then retry." + ) return data detail = "" if isinstance(data, dict): diff --git a/strix/interface/cloud/render.py b/strix/interface/cloud/render.py index a1c22f8d..e8fb6245 100644 --- a/strix/interface/cloud/render.py +++ b/strix/interface/cloud/render.py @@ -6,6 +6,7 @@ import json import sys from typing import TYPE_CHECKING, Any +from rich.markup import escape from rich.table import Table @@ -15,22 +16,49 @@ if TYPE_CHECKING: _MAX_TABLE_COLUMNS = 8 _MAX_CELL_LENGTH = 60 +_NARROW_TABLE_WIDTH = 120 _PREFERRED_KEYS = ( - "id", "name", "title", + "repository_full_name", + "pr_number", + "pr_title", + "head_branch", + "base_branch", + "verdict", "domain", + "target", + "default_branch", + "branch", + "display_number", "status", "state", "severity", + "cve", + "cvss", + "finding_type", + "findings_count", + "open_findings_count", "role", "email", + "firstName", + "lastName", "url", + "provider", + "secret_prefix", "scan_type", "engagement_type", + "estimated_credits", + "cron_expression", + "timezone", + "next_run_at", + "is_active", "created_at", "updated_at", + "expires_at", + "last_used_at", + "id", ) @@ -39,13 +67,27 @@ def json_mode(*, flag: bool) -> bool: return flag or not sys.stdout.isatty() -def emit(console: Console, data: Any, *, as_json: bool) -> None: +def emit( + console: Console, + data: Any, + *, + as_json: bool, + row_numbers: bool = False, + omit_columns: frozenset[str] = frozenset(), + hint: str | None = None, +) -> None: if as_json: sys.stdout.write(json.dumps(data, indent=2, default=str) + "\n") return rows = _list_of_dicts(data) if rows is not None: - _print_table(console, rows) + _print_table( + console, + rows, + row_numbers=row_numbers, + omit_columns=omit_columns, + hint=hint, + ) return if isinstance(data, str): console.print(data) @@ -66,28 +108,93 @@ def _list_of_dicts(data: Any) -> list[dict[str, Any]] | None: return data -def _print_table(console: Console, rows: list[dict[str, Any]]) -> None: - columns: list[str] = [key for key in _PREFERRED_KEYS if any(key in row for row in rows)] +def _print_table( + console: Console, + rows: list[dict[str, Any]], + *, + row_numbers: bool = False, + omit_columns: frozenset[str] = frozenset(), + hint: str | None = None, +) -> None: + columns: list[str] = [ + key + for key in _PREFERRED_KEYS + if key not in omit_columns and any(key in row for row in rows) + ] for row in rows: for key in row: if ( key not in columns + and key not in omit_columns and len(columns) < _MAX_TABLE_COLUMNS and not isinstance(row[key], dict | list) ): columns.append(key) + columns = columns[:_MAX_TABLE_COLUMNS] + if console.width < _NARROW_TABLE_WIDTH: + _print_cards(console, rows, columns, row_numbers=row_numbers) + console.print(f"[dim]{len(rows)} item(s). Use --json for the full records.[/]") + if hint: + console.print(f"[dim]{hint}[/]") + return table = Table(show_lines=False) - for column in columns[:_MAX_TABLE_COLUMNS]: + if row_numbers: + table.add_column("#", justify="right", style="cyan", no_wrap=True) + for column in columns: table.add_column(column) - for row in rows: - table.add_row(*[_cell(row.get(column)) for column in columns[:_MAX_TABLE_COLUMNS]]) + for index, row in enumerate(rows, start=1): + cells = [_cell(row.get(column)) for column in columns] + if row_numbers: + cells.insert(0, str(index)) + table.add_row(*cells) console.print(table) console.print(f"[dim]{len(rows)} item(s). Use --json for the full records.[/]") + if hint: + console.print(f"[dim]{hint}[/]") + + +def _print_cards( + console: Console, + rows: list[dict[str, Any]], + columns: list[str], + *, + row_numbers: bool, +) -> None: + """Render list rows legibly when a terminal is too narrow for a table.""" + for index, row in enumerate(rows, start=1): + parts = [ + f"[bold]{escape(_human_label(column))}:[/] {escape(_cell(row.get(column)))}" + for column in columns + if row.get(column) is not None + ] + prefix = f"[cyan]{index}.[/] " if row_numbers else "[cyan]•[/] " + console.print(prefix + " [dim]·[/] ".join(parts), soft_wrap=False) + + +def _human_label(column: str) -> str: + labels = { + "repository_full_name": "repo", + "pr_number": "PR", + "pr_title": "title", + "head_branch": "head", + "base_branch": "base", + "findings_count": "findings", + "open_findings_count": "open", + "display_number": "finding", + "created_at": "created", + "updated_at": "updated", + "expires_at": "expires", + "last_used_at": "last used", + "secret_prefix": "prefix", + } + return labels.get(column, column.replace("_", " ")) def _cell(value: Any) -> str: if value is None: return "" + if isinstance(value, bool): + return "yes" if value else "no" text = str(value) if len(text) > _MAX_CELL_LENGTH: return text[: _MAX_CELL_LENGTH - 1] + "…" diff --git a/strix/interface/cloud/runner.py b/strix/interface/cloud/runner.py index 22af6f60..ad697613 100644 --- a/strix/interface/cloud/runner.py +++ b/strix/interface/cloud/runner.py @@ -129,7 +129,19 @@ def _execute( result = _wait(console, cmd, result, token=token, as_json=as_json) if cmd.link: return _handoff_link(console, cmd, args, result, as_json=as_json) - emit(console, result, as_json=as_json) + workspace_list = cmd.method == "GET" and cmd.path == "/workspaces" + emit( + console, + result, + as_json=as_json, + row_numbers=workspace_list, + omit_columns=frozenset({"id"}) if workspace_list else frozenset(), + hint=( + "Switch with `strix cloud workspaces use NUMBER`." + if workspace_list + else None + ), + ) return http.EXIT_OK diff --git a/strix/interface/cloud/workspaces.py b/strix/interface/cloud/workspaces.py index 0ff23fe5..1ba67602 100644 --- a/strix/interface/cloud/workspaces.py +++ b/strix/interface/cloud/workspaces.py @@ -25,7 +25,11 @@ def run_workspace_use(argv: list[str]) -> int: prog="strix cloud workspaces use", description="Switch the stored API token to another workspace.", ) - parser.add_argument("workspace", metavar="WORKSPACE", help="Workspace ID or exact name.") + parser.add_argument( + "workspace", + metavar="WORKSPACE", + help="Workspace number from `workspaces list`, ID, or exact name.", + ) parser.add_argument( "--scopes", nargs="+", @@ -119,6 +123,14 @@ def _find_workspace(selector: str, *, token: str | None) -> dict[str, Any]: if not workspaces: raise http.CloudError("no workspaces found for this account.") wanted = selector.strip() + if wanted.isdigit(): + index = int(wanted) + if 1 <= index <= len(workspaces): + return workspaces[index - 1] + raise http.CloudError( + f"workspace number must be between 1 and {len(workspaces)}. " + "Run `strix cloud workspaces` to see the numbered list." + ) by_id = [w for w in workspaces if w.get("id") == wanted] if by_id: return by_id[0] @@ -126,7 +138,16 @@ def _find_workspace(selector: str, *, token: str | None) -> dict[str, Any]: if len(by_name) == 1: return by_name[0] if len(by_name) > 1: - ids = ", ".join(str(w.get("id")) for w in by_name) - raise http.CloudError(f"multiple workspaces are named {wanted!r}. Use an ID: {ids}") - names = ", ".join(str(w.get("name")) for w in workspaces) + numbers = ", ".join( + str(index) + for index, workspace in enumerate(workspaces, start=1) + if workspace in by_name + ) + raise http.CloudError( + f"multiple workspaces are named {wanted!r}. Use its list number: {numbers}" + ) + names = ", ".join( + f"{index}: {workspace.get('name')}" + for index, workspace in enumerate(workspaces, start=1) + ) raise http.CloudError(f"no workspace matches {wanted!r}. Your workspaces: {names}") diff --git a/tests/test_cloud_cli.py b/tests/test_cloud_cli.py index d6dc8fc2..f420852f 100644 --- a/tests/test_cloud_cli.py +++ b/tests/test_cloud_cli.py @@ -60,6 +60,22 @@ def test_unknown_verb_returns_usage_error() -> None: assert cloud.run_cloud(["scans", "bogus"]) == 2 +def test_successful_html_response_is_reported_without_dumping_html( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr( + http, + "request", + lambda *_a, **_k: FakeResponse(text="preview gate"), + ) + + assert cloud.run_cloud(["workspaces", "list", "--json"]) == 1 + output = capsys.readouterr().out + assert "non-JSON response" in output + assert "STRIX_APP_URL" in output + assert " None: assert cloud.run_cloud(["scans"]) == 0 @@ -636,6 +652,118 @@ def test_group_help_lists_all_verbs_instead_of_default_verb_help(capsys: Any) -> assert "use" in output +def test_workspace_alias_routes_to_workspaces(monkeypatch: pytest.MonkeyPatch) -> None: + seen: dict[str, str] = {} + + def fake_request(method: str, path: str, **_kwargs: Any) -> FakeResponse: + seen.update(method=method, path=path) + return FakeResponse(payload={"workspaces": []}) + + monkeypatch.setattr(http, "request", fake_request) + assert cloud.run_cloud(["workspace", "list", "--json"]) == 0 + assert seen == {"method": "GET", "path": "/workspaces"} + + +def test_workspace_human_list_is_numbered_and_hides_ids( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr( + http, + "request", + lambda *_a, **_k: FakeResponse( + payload={ + "workspaces": [ + {"id": "org_secret", "name": "Team One", "role": "admin", "current": True} + ] + } + ), + ) + + assert cloud.run_cloud(["workspaces", "list"]) == 0 + output = capsys.readouterr().out + assert "1." in output + assert "Team One" in output + assert "yes" in output + assert "org_secret" not in output + assert "workspaces use NUMBER" in output + + +def test_pr_review_human_list_prioritizes_actionable_fields( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr( + http, + "request", + lambda *_a, **_k: FakeResponse( + payload={ + "items": [ + { + "id": "review-id", + "organization_id": "org-id", + "user_id": "user-id", + "installation_id": 42, + "repository_full_name": "usestrix/strix", + "pr_number": 1177, + "pr_title": "Improve cloud CLI", + "head_branch": "feature", + "base_branch": "main", + "verdict": "pass", + "status": "posted", + "findings_count": 0, + } + ], + "meta": {"total": 1}, + } + ), + ) + + assert cloud.run_cloud(["pr-reviews", "list"]) == 0 + output = capsys.readouterr().out + for value in ("usestrix/strix", "1177", "Improve cloud CLI", "feature", "main", "pass"): + assert value in output + for value in ("org-id", "user-id", "installation_id"): + assert value not in output + + +def test_workspace_use_accepts_list_number( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + auth_path = tmp_path / "platform-auth.json" + monkeypatch.setattr(platform_cli, "AUTH_PATH", auth_path) + monkeypatch.setattr(workspaces, "AUTH_PATH", auth_path) + platform_cli.save_record( + {"api_token": "old", "scopes": ["organizations:read", "tokens:write"]} + ) + called_paths: list[str] = [] + + def fake_request(_method: str, path: str, **_kwargs: Any) -> FakeResponse: + called_paths.append(path) + if path == "/workspaces": + return FakeResponse( + payload={ + "workspaces": [ + {"id": "org_1", "name": "One"}, + {"id": "org_2", "name": "Two"}, + ] + } + ) + return FakeResponse( + status_code=201, + payload={ + "api_token": "new", + "organization_id": "org_2", + "organization_name": "Two", + "scopes": ["organizations:read", "tokens:write"], + }, + ) + + monkeypatch.setattr(http, "request", fake_request) + assert cloud.run_cloud(["workspaces", "use", "2", "--json"]) == 0 + assert called_paths == ["/workspaces", "/workspaces/org_2/token"] + + def test_logout_help_does_not_remove_stored_auth( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: Any ) -> None: