wehub-resource-sync
4b6817381b
Benchmark image — build + push to ECR (any adapter) / build + push (push) Waiting to run
CI / quality (ubuntu-latest) (push) Waiting to run
CI / test (tools-runtime) (push) Waiting to run
CI / test (e2e-general) (push) Waiting to run
CI / test (cli-runtime) (push) Waiting to run
CI / test (e2e-provider-and-openclaw) (push) Waiting to run
CI / test (integrations-and-misc) (push) Waiting to run
CI / coverage-report (push) Blocked by required conditions
CI / test-kubernetes (push) Waiting to run
CI / should-run-thorough (push) Waiting to run
CI / test-thorough (cloudwatch-demo) (push) Blocked by required conditions
CI / test-thorough (flink-ecs) (push) Blocked by required conditions
CI / test-thorough (upstream-lambda) (push) Blocked by required conditions
CI / test-thorough (prefect-ecs-fargate) (push) Blocked by required conditions
CodeQL / Analyze (python) (push) Waiting to run
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Blocked by required conditions
Release / publish-release (push) Blocked by required conditions
Release / publish-main-release (push) Blocked by required conditions
Release / prepare (push) Waiting to run
Release / verify (push) Blocked by required conditions
Release / build-python-dist (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Blocked by required conditions
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Blocked by required conditions
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Waiting to run
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Waiting to run
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Waiting to run
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
601 lines
23 KiB
Python
601 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import click
|
|
import pytest
|
|
|
|
from config.repl_config import ReplConfig
|
|
from platform.analytics import provider
|
|
from platform.analytics.events import Event
|
|
from surfaces.cli.__main__ import _sentry_entrypoint_for_invocation, main
|
|
|
|
|
|
class _EmptyCatalog:
|
|
def filter(self, *, category: str, search: str) -> list[object]:
|
|
_ = (category, search)
|
|
return []
|
|
|
|
|
|
def _stub_analytics_httpx(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, object]]:
|
|
posted_payloads: list[dict[str, object]] = []
|
|
|
|
class _StubResponse:
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
class _StubClient:
|
|
def __init__(self, *_args, **_kwargs) -> None:
|
|
pass
|
|
|
|
def __enter__(self) -> _StubClient:
|
|
return self
|
|
|
|
def __exit__(self, _exc_type, _exc, _tb) -> None:
|
|
return None
|
|
|
|
def post(self, url: str, json: dict[str, object]) -> _StubResponse:
|
|
posted_payloads.append({"url": url, "json": json})
|
|
return _StubResponse()
|
|
|
|
monkeypatch.setattr(provider.httpx, "Client", _StubClient)
|
|
return posted_payloads
|
|
|
|
|
|
def test_main_runs_health_command(monkeypatch) -> None:
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
|
|
with (
|
|
patch("integrations.verify.verify_integrations") as mock_verify,
|
|
patch("integrations.verify.format_verification_results") as mock_format,
|
|
):
|
|
mock_verify.return_value = [
|
|
{
|
|
"service": "aws",
|
|
"source": "local store",
|
|
"status": "passed",
|
|
"detail": "ok",
|
|
}
|
|
]
|
|
mock_format.return_value = (
|
|
"\n"
|
|
" SERVICE SOURCE STATUS DETAIL\n"
|
|
" aws local store passed ok\n"
|
|
)
|
|
|
|
exit_code = main(["health"])
|
|
|
|
assert exit_code == 0
|
|
|
|
|
|
def test_main_does_not_capture_expected_usage_errors_to_sentry(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: list[BaseException] = []
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
monkeypatch.setattr(
|
|
"surfaces.interactive_shell.utils.error_handling.exception_reporting.capture_exception",
|
|
lambda exc, **_kwargs: captured.append(exc),
|
|
)
|
|
|
|
exit_code = main(["integrations", "show", "nonexistent"])
|
|
|
|
assert exit_code != 0
|
|
assert captured == []
|
|
|
|
|
|
def test_main_treats_onboard_abort_as_clean_cancel(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.init_sentry", lambda **_kw: None)
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.wizard.flow.run_wizard",
|
|
lambda: (_ for _ in ()).throw(click.Abort()),
|
|
)
|
|
|
|
exit_code = main(["onboard"])
|
|
|
|
assert exit_code == 0
|
|
|
|
|
|
def test_main_allows_update_when_sentry_sdk_missing(monkeypatch, capsys) -> None:
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
|
|
def _raise_missing_sentry(**_kwargs: object) -> None:
|
|
raise ModuleNotFoundError("No module named 'sentry_sdk'", name="sentry_sdk")
|
|
|
|
monkeypatch.setattr("surfaces.cli.__main__.init_sentry", _raise_missing_sentry)
|
|
monkeypatch.setattr("surfaces.cli.lifecycle.update._fetch_latest_version", lambda: "9999.0.0")
|
|
monkeypatch.setattr("surfaces.cli.lifecycle.update._is_update_available", lambda _c, _l: False)
|
|
|
|
exit_code = main(["update", "--check"])
|
|
|
|
assert exit_code == 0
|
|
assert "already up to date" in capsys.readouterr().out
|
|
|
|
|
|
def test_main_non_update_still_raises_when_sentry_sdk_missing(monkeypatch) -> None:
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
|
|
def _raise_missing_sentry(**_kwargs: object) -> None:
|
|
raise ModuleNotFoundError("No module named 'sentry_sdk'", name="sentry_sdk")
|
|
|
|
monkeypatch.setattr("surfaces.cli.__main__.init_sentry", _raise_missing_sentry)
|
|
|
|
with pytest.raises(ModuleNotFoundError):
|
|
main(["health"])
|
|
|
|
|
|
def test_main_does_not_capture_analytics_for_help(monkeypatch, capsys) -> None:
|
|
captured: list[str] = []
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_first_run_if_needed", lambda: captured.append("install")
|
|
)
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_cli_invoked", lambda *_args: captured.append("cli")
|
|
)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
|
|
exit_code = main(["--help"])
|
|
|
|
assert exit_code == 0
|
|
assert "Usage:" in capsys.readouterr().out
|
|
assert captured == []
|
|
|
|
|
|
def test_main_does_not_capture_unknown_command_to_sentry(monkeypatch, capsys) -> None:
|
|
captured: list[str] = []
|
|
captured_errors: list[BaseException] = []
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_first_run_if_needed", lambda: captured.append("install")
|
|
)
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_cli_invoked", lambda *_args: captured.append("cli")
|
|
)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr(
|
|
"surfaces.interactive_shell.utils.error_handling.exception_reporting.capture_exception",
|
|
lambda exc, **_kwargs: captured_errors.append(exc),
|
|
)
|
|
|
|
exit_code = main(["not-a-command"])
|
|
|
|
assert exit_code != 0
|
|
assert "No such command" in capsys.readouterr().err
|
|
assert captured == []
|
|
assert captured_errors == []
|
|
|
|
|
|
def test_main_does_not_capture_invalid_option_parse_error(monkeypatch, capsys) -> None:
|
|
captured: list[str] = []
|
|
captured_errors: list[BaseException] = []
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_first_run_if_needed", lambda: captured.append("install")
|
|
)
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_cli_invoked", lambda *_args: captured.append("cli")
|
|
)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr(
|
|
"surfaces.interactive_shell.utils.error_handling.exception_reporting.capture_exception",
|
|
lambda exc, **_kwargs: captured_errors.append(exc),
|
|
)
|
|
|
|
exit_code = main(["--definitely-wrong-option"])
|
|
|
|
assert exit_code == 2
|
|
stderr = capsys.readouterr().err
|
|
assert "--definitely-wrong-option" in stderr and "No such option" in stderr
|
|
assert captured == []
|
|
assert captured_errors == []
|
|
|
|
|
|
def test_main_captures_analytics_once_for_accepted_command(monkeypatch, capsys) -> None:
|
|
captured: list[str] = []
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_first_run_if_needed", lambda: captured.append("install")
|
|
)
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_cli_invoked", lambda *_args: captured.append("cli")
|
|
)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("integrations.verify.verify_integrations", lambda: [])
|
|
|
|
exit_code = main(["health"])
|
|
|
|
assert exit_code == 0
|
|
assert "OpenSRE Health" in capsys.readouterr().out
|
|
assert captured == ["install", "cli"]
|
|
|
|
|
|
def test_main_fast_version_command_skips_runtime_bootstrap(monkeypatch, capsys) -> None:
|
|
captured: list[dict[str, object] | None] = []
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_first_run_if_needed",
|
|
lambda: captured.append({"unexpected": "install"}),
|
|
)
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_cli_invoked",
|
|
lambda properties=None: captured.append(properties),
|
|
)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
|
|
exit_code = main(["version"])
|
|
|
|
assert exit_code == 0
|
|
assert "opensre" in capsys.readouterr().out
|
|
assert captured == []
|
|
|
|
|
|
def test_main_debug_sentry_sends_synthetic_event(monkeypatch, capsys) -> None:
|
|
debug_module = importlib.import_module("surfaces.cli.commands.debug")
|
|
captured: list[tuple[tuple[object, ...], dict[str, object]]] = []
|
|
root_init_entrypoints: list[str | None] = []
|
|
flush_calls: list[int] = []
|
|
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.init_sentry",
|
|
lambda entrypoint=None: root_init_entrypoints.append(entrypoint),
|
|
)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr(debug_module, "sentry_transport_enabled", lambda: True)
|
|
monkeypatch.setattr(debug_module, "resolved_sentry_dsn_host", lambda: "sentry.example.test")
|
|
|
|
def capture_stub(*args: object, **kwargs: object) -> str:
|
|
captured.append((args, kwargs))
|
|
return "event-123"
|
|
|
|
monkeypatch.setattr(debug_module, "capture_exception", capture_stub)
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"sentry_sdk",
|
|
SimpleNamespace(flush=lambda timeout: flush_calls.append(timeout)),
|
|
)
|
|
|
|
exit_code = main(["debug", "sentry"])
|
|
|
|
assert exit_code == 0
|
|
output = capsys.readouterr().out
|
|
assert "Sentry DSN host: sentry.example.test" in output
|
|
assert "Sentry event ID: event-123" in output
|
|
assert "Sentry flush sent: yes" in output
|
|
assert root_init_entrypoints == ["debug"]
|
|
assert flush_calls == [5]
|
|
assert captured[0][1]["context"] == "debug.sentry"
|
|
assert captured[0][1]["tags"] == {"debug": "true", "surface": "debug"}
|
|
|
|
|
|
def test_sentry_entrypoint_uses_debug_for_debug_group_invocations() -> None:
|
|
assert _sentry_entrypoint_for_invocation(["debug", "future-check"]) == "debug"
|
|
|
|
|
|
def test_main_debug_sentry_exits_nonzero_when_disabled(monkeypatch, capsys) -> None:
|
|
debug_module = importlib.import_module("surfaces.cli.commands.debug")
|
|
monkeypatch.setattr("surfaces.cli.__main__.init_sentry", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr(debug_module, "sentry_transport_enabled", lambda: False)
|
|
monkeypatch.setattr(debug_module, "resolved_sentry_dsn_host", lambda: "")
|
|
|
|
exit_code = main(["debug", "sentry"])
|
|
|
|
assert exit_code == 1
|
|
assert "Sentry is disabled or no DSN is configured." in capsys.readouterr().err
|
|
|
|
|
|
def test_main_debug_sentry_exits_nonzero_when_flush_fails(monkeypatch, capsys) -> None:
|
|
debug_module = importlib.import_module("surfaces.cli.commands.debug")
|
|
monkeypatch.setattr("surfaces.cli.__main__.init_sentry", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr(debug_module, "sentry_transport_enabled", lambda: True)
|
|
monkeypatch.setattr(debug_module, "resolved_sentry_dsn_host", lambda: "sentry.example.test")
|
|
monkeypatch.setattr(debug_module, "capture_exception", lambda *_args, **_kw: "event-123")
|
|
|
|
def flush_stub(*, timeout: int) -> bool:
|
|
assert timeout == 5
|
|
return False
|
|
|
|
monkeypatch.setitem(
|
|
sys.modules,
|
|
"sentry_sdk",
|
|
SimpleNamespace(flush=flush_stub),
|
|
)
|
|
|
|
exit_code = main(["debug", "sentry"])
|
|
|
|
assert exit_code == 1
|
|
output = capsys.readouterr().out
|
|
assert "Sentry event ID: event-123" in output
|
|
assert "Sentry flush sent: no" in output
|
|
|
|
|
|
def test_main_emits_first_run_install_before_cli_invoked(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
|
|
) -> None:
|
|
# This test validates analytics event ordering only; avoid real Sentry init
|
|
# side effects (e.g. sdk integration hooks) that are unrelated to the
|
|
# install/cli-invoked event contract.
|
|
monkeypatch.setattr("surfaces.cli.__main__.init_sentry", lambda **_kw: None)
|
|
provider.shutdown_analytics(flush=False)
|
|
provider._instance = None
|
|
provider._cached_anonymous_id = None
|
|
provider._cached_identity_persistence = "unknown"
|
|
provider._first_run_marker_created_this_process = False
|
|
provider._pending_user_id_load_failures.clear()
|
|
monkeypatch.delenv("OPENSRE_NO_TELEMETRY", raising=False)
|
|
monkeypatch.delenv("OPENSRE_ANALYTICS_DISABLED", raising=False)
|
|
monkeypatch.delenv("OPENSRE_SENTRY_DISABLED", raising=False)
|
|
monkeypatch.delenv("DO_NOT_TRACK", raising=False)
|
|
monkeypatch.setattr(provider, "_CONFIG_DIR", tmp_path)
|
|
monkeypatch.setattr(provider, "_ANONYMOUS_ID_PATH", tmp_path / "anonymous_id")
|
|
monkeypatch.setattr(provider, "_FIRST_RUN_PATH", tmp_path / "installed")
|
|
monkeypatch.setattr(provider, "_event_log_state", provider._EventLogState())
|
|
monkeypatch.setattr(provider.atexit, "register", lambda _func: None)
|
|
posted_payloads = _stub_analytics_httpx(monkeypatch)
|
|
|
|
monkeypatch.setattr("surfaces.cli.__main__.render_landing", lambda _group: None)
|
|
|
|
exit_code = main(["--no-interactive"])
|
|
|
|
assert exit_code == 0
|
|
assert "OpenSRE is in Public Beta" in capsys.readouterr().err
|
|
# CLI exit is non-blocking; wait for the daemon worker to finish posts.
|
|
analytics = provider._instance
|
|
if analytics is not None and analytics._worker is not None:
|
|
analytics._worker.join(timeout=2.0)
|
|
assert [payload["json"]["event"] for payload in posted_payloads] == [
|
|
Event.INSTALL_DETECTED.value,
|
|
Event.CLI_INVOKED.value,
|
|
]
|
|
provider.shutdown_analytics(flush=False)
|
|
provider._instance = None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("argv", "subcommand_event", "setup"),
|
|
[
|
|
(
|
|
["onboard"],
|
|
"onboard_started",
|
|
"surfaces.cli.wizard.flow.run_wizard",
|
|
),
|
|
(
|
|
["integrations", "list"],
|
|
"integrations_listed",
|
|
"integrations.cli.cmd_list",
|
|
),
|
|
(
|
|
["tests", "list"],
|
|
"tests_listed",
|
|
"surfaces.cli.tests.discover.load_test_catalog",
|
|
),
|
|
],
|
|
)
|
|
def test_main_captures_cli_invoked_before_reported_subcommand_families(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
argv: list[str],
|
|
subcommand_event: str,
|
|
setup: str,
|
|
) -> None:
|
|
captured: list[str] = []
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_first_run_if_needed", lambda: captured.append("install")
|
|
)
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.capture_cli_invoked", lambda *_args: captured.append("cli")
|
|
)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
|
|
if setup == "surfaces.cli.wizard.flow.run_wizard":
|
|
onboard_module = importlib.import_module("surfaces.cli.commands.onboard")
|
|
monkeypatch.setattr(setup, lambda: 0)
|
|
monkeypatch.setattr(
|
|
onboard_module,
|
|
"capture_onboard_started",
|
|
lambda: captured.append(subcommand_event),
|
|
)
|
|
monkeypatch.setattr(onboard_module, "capture_onboard_completed", lambda _cfg: None)
|
|
elif setup == "integrations.cli.cmd_list":
|
|
integrations_module = importlib.import_module("surfaces.cli.commands.integrations")
|
|
monkeypatch.setattr(setup, lambda: None)
|
|
monkeypatch.setattr(
|
|
integrations_module,
|
|
"capture_integrations_listed",
|
|
lambda: captured.append(subcommand_event),
|
|
)
|
|
else:
|
|
tests_module = importlib.import_module("surfaces.cli.commands.tests")
|
|
monkeypatch.setattr(setup, _EmptyCatalog)
|
|
|
|
def _capture_tests_listed(_category: str, *, search: bool) -> None:
|
|
_ = (_category, search)
|
|
captured.append(subcommand_event)
|
|
|
|
monkeypatch.setattr(
|
|
tests_module,
|
|
"capture_tests_listed",
|
|
_capture_tests_listed,
|
|
)
|
|
|
|
exit_code = main(argv)
|
|
|
|
assert exit_code == 0
|
|
assert captured[:3] == ["install", "cli", subcommand_event]
|
|
|
|
|
|
def test_no_interactive_falls_through_to_landing_page(monkeypatch) -> None:
|
|
"""Regression for Greptile P1 (PR #591): --no-interactive previously ran
|
|
`raise SystemExit(run_repl(...))` unconditionally on a TTY, returning 0 but
|
|
never reaching render_landing(). The fix guards the SystemExit on
|
|
`config.enabled`, so disabled mode falls through to render_landing().
|
|
"""
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
|
|
# Force the TTY branch so the regression path is actually exercised.
|
|
monkeypatch.setattr("surfaces.cli.__main__.sys.stdin.isatty", lambda: True)
|
|
monkeypatch.setattr("surfaces.cli.__main__.sys.stdout.isatty", lambda: True)
|
|
|
|
# Force disabled interactive config via the loader. Return a disabled config
|
|
# regardless of how the CLI resolved the flag.
|
|
monkeypatch.setattr(
|
|
"config.repl_config.ReplConfig.load",
|
|
classmethod(lambda _cls, **_kw: ReplConfig(enabled=False, layout="classic")),
|
|
)
|
|
|
|
landing_calls: list[int] = []
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.render_landing",
|
|
lambda _group: landing_calls.append(1),
|
|
)
|
|
|
|
# run_repl must NOT be invoked when config.enabled is False.
|
|
def _fail_if_called(**_kw: object) -> int:
|
|
raise AssertionError("run_repl must not run when config.enabled=False")
|
|
|
|
with patch("surfaces.interactive_shell.run_repl", side_effect=_fail_if_called):
|
|
exit_code = main(["--no-interactive"])
|
|
|
|
assert exit_code == 0
|
|
assert landing_calls == [1], "render_landing should be called exactly once"
|
|
|
|
|
|
def test_default_no_args_enters_repl(monkeypatch) -> None:
|
|
"""Regression: the default invocation `opensre` (no args, TTY) must enter
|
|
the REPL. A previous Click misconfiguration (is_flag + flag_value=False)
|
|
made the `interactive` kwarg resolve to False even with no flag, so every
|
|
local run silently rendered the landing page. Assert the CLI passes
|
|
cli_enabled=True into ReplConfig.load and actually calls run_repl.
|
|
"""
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.sys.stdin.isatty", lambda: True)
|
|
monkeypatch.setattr("surfaces.cli.__main__.sys.stdout.isatty", lambda: True)
|
|
|
|
load_calls: list[dict] = []
|
|
orig_load = ReplConfig.load
|
|
|
|
@classmethod # type: ignore[misc]
|
|
def spy_load(cls, **kw): # type: ignore[no-untyped-def]
|
|
load_calls.append(kw)
|
|
return orig_load(**kw)
|
|
|
|
monkeypatch.setattr("config.repl_config.ReplConfig.load", spy_load)
|
|
|
|
landing_calls: list[int] = []
|
|
monkeypatch.setattr(
|
|
"surfaces.cli.__main__.render_landing",
|
|
lambda _group: landing_calls.append(1),
|
|
)
|
|
|
|
with (
|
|
patch("surfaces.interactive_shell.run_repl", return_value=0),
|
|
patch("surfaces.interactive_shell.main.run_repl", return_value=0),
|
|
):
|
|
exit_code = main([])
|
|
|
|
assert exit_code == 0
|
|
assert len(load_calls) >= 1
|
|
repl_load = load_calls[-1]
|
|
assert repl_load.get("cli_enabled") is True, (
|
|
f"default no-args run must pass cli_enabled=True, got {repl_load}"
|
|
)
|
|
assert repl_load.get("cli_layout") is None
|
|
assert repl_load.get("cli_theme") is None, (
|
|
f"default no-args run must leave theme env/config overridable, got {repl_load}"
|
|
)
|
|
assert landing_calls == [], "REPL should run, not landing page"
|
|
|
|
|
|
def test_resume_flag_enters_repl_with_session_id(monkeypatch) -> None:
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.sys.stdin.isatty", lambda: True)
|
|
monkeypatch.setattr("surfaces.cli.__main__.sys.stdout.isatty", lambda: True)
|
|
|
|
load_calls: list[dict] = []
|
|
orig_load = ReplConfig.load
|
|
|
|
@classmethod # type: ignore[misc]
|
|
def spy_load(cls, **kw): # type: ignore[no-untyped-def]
|
|
load_calls.append(kw)
|
|
return orig_load(**kw)
|
|
|
|
monkeypatch.setattr("config.repl_config.ReplConfig.load", spy_load)
|
|
|
|
run_repl_calls: list[dict[str, object]] = []
|
|
|
|
def _capture_run_repl(**kwargs: object) -> int:
|
|
run_repl_calls.append(kwargs)
|
|
return 0
|
|
|
|
with patch("surfaces.interactive_shell.run_repl", side_effect=_capture_run_repl):
|
|
exit_code = main(["--resume", "8988e743"])
|
|
|
|
assert exit_code == 0
|
|
assert load_calls[-1]["cli_enabled"] is True
|
|
assert len(run_repl_calls) == 1
|
|
assert run_repl_calls[0]["resume_session_id"] == "8988e743"
|
|
assert run_repl_calls[0]["config"].enabled is True
|
|
|
|
|
|
def test_invalid_theme_flag_returns_usage_error(monkeypatch, capsys) -> None:
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
|
|
exit_code = main(["--theme", "chartreuse"])
|
|
|
|
assert exit_code == 2
|
|
err = capsys.readouterr().err
|
|
assert "Invalid value for '--theme'" in err
|
|
assert "chartreuse" in err
|
|
|
|
|
|
def test_valid_theme_flag_passes_normalized_value(monkeypatch) -> None:
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_first_run_if_needed", lambda: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.shutdown_analytics", lambda **_kw: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.capture_cli_invoked", lambda *_args: None)
|
|
monkeypatch.setattr("surfaces.cli.__main__.sys.stdin.isatty", lambda: True)
|
|
monkeypatch.setattr("surfaces.cli.__main__.sys.stdout.isatty", lambda: True)
|
|
|
|
load_calls: list[dict] = []
|
|
|
|
@classmethod # type: ignore[misc]
|
|
def spy_load(_cls, **kw): # type: ignore[no-untyped-def]
|
|
load_calls.append(kw)
|
|
return ReplConfig(enabled=True, layout="classic", theme="blue")
|
|
|
|
monkeypatch.setattr("config.repl_config.ReplConfig.load", spy_load)
|
|
|
|
with (
|
|
patch("surfaces.interactive_shell.run_repl", return_value=0),
|
|
patch("surfaces.interactive_shell.main.run_repl", return_value=0),
|
|
):
|
|
exit_code = main(["--theme", "BLUE"])
|
|
|
|
assert exit_code == 0
|
|
assert len(load_calls) >= 1
|
|
assert all(call.get("cli_theme") == "blue" for call in load_calls)
|