wehub-resource-sync
4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
2372 lines
87 KiB
Python
2372 lines
87 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from integrations.llm_cli.codex_oauth import CodexOAuthResult
|
|
from surfaces.cli.wizard import _integration_configurators, _ui, flow
|
|
from surfaces.cli.wizard import store as wizard_store
|
|
from surfaces.cli.wizard.env_sync import sync_provider_env
|
|
from surfaces.cli.wizard.probes import ProbeResult
|
|
from tests.integrations.llm_cli.testing_helpers import write_fake_runnable_cli_bin
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _stub_managed_llm_secret_persistence(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Wizard flow tests should not touch the developer's real keychain."""
|
|
monkeypatch.setattr(_ui, "save_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
|
|
def test_run_wizard_advanced_remote_falls_back_to_local(monkeypatch, tmp_path, capsys) -> None:
|
|
# advanced -> falls back to local -> change provider? Yes -> pick anthropic -> skip integrations
|
|
select_responses = iter(
|
|
["advanced", "remote", "anthropic", "api_key", "claude-opus-4-7", "skip"]
|
|
)
|
|
confirm_responses = iter([True, True]) # "use local instead?" and "Change provider?"
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_confirm(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(confirm_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = "secret-key"
|
|
return m
|
|
|
|
saved: dict[str, object] = {}
|
|
saved_llm_keys: list[tuple[str, str]] = []
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "confirm", _mock_confirm)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
flow, "probe_remote_target", lambda: ProbeResult("remote", True, "remote ok")
|
|
)
|
|
|
|
def _save_local_config(**kwargs):
|
|
saved.update(kwargs)
|
|
return tmp_path / "opensre.json"
|
|
|
|
monkeypatch.setattr(flow, "save_local_config", _save_local_config)
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"save_api_key",
|
|
lambda provider, value, **_kwargs: saved_llm_keys.append((provider, value)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved["wizard_mode"] == "advanced"
|
|
assert saved["provider"] == "anthropic"
|
|
assert "api_key" not in saved
|
|
assert saved_llm_keys == [("anthropic", "secret-key")]
|
|
|
|
output = capsys.readouterr().out
|
|
assert "Summary" in output
|
|
assert "4/4" in output
|
|
assert "next" in output
|
|
assert "Done." in output
|
|
assert output.index("Summary") < output.index("Done.")
|
|
|
|
|
|
def test_prompt_value_can_signal_wizard_back(monkeypatch) -> None:
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = None
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui.questionary, "password", _mock_password)
|
|
|
|
with pytest.raises(_ui.WizardBack):
|
|
_ui._prompt_value("OpenAI API key", secret=True, back_on_cancel=True)
|
|
|
|
|
|
def test_run_wizard_no_saved_provider_shows_selection(monkeypatch, tmp_path) -> None:
|
|
"""With no saved config the provider list is shown immediately (no confirm prompt)."""
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "skip"])
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = "secret-key"
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
exit_code = flow.run_wizard()
|
|
assert exit_code == 0
|
|
|
|
|
|
def test_run_wizard_shows_keyring_fix_steps_when_secure_storage_is_unavailable(
|
|
monkeypatch, tmp_path, capsys
|
|
) -> None:
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7"])
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = "secret-key"
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"save_api_key",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(
|
|
RuntimeError("Secure local credential storage is unavailable on this machine.")
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"get_keyring_setup_instructions",
|
|
lambda _env_var: (
|
|
"Current keyring backend: keyring.backends.fail.Keyring.",
|
|
"Install it first: sudo apt update && sudo apt install -y gnome-keyring dbus-user-session",
|
|
"Start a D-Bus shell: dbus-run-session -- sh",
|
|
),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 1
|
|
output = capsys.readouterr().out
|
|
assert "OpenSRE could not save your API key to the local system keychain." in output
|
|
assert "Install it first: sudo apt update && sudo apt install -y gnome-keyring" in output
|
|
assert "dbus-user-session" in output
|
|
assert "Start a D-Bus shell: dbus-run-session -- sh" in output
|
|
|
|
|
|
def test_run_wizard_configures_optional_integrations(monkeypatch, tmp_path, capsys) -> None:
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "grafana"])
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
password_responses = iter(
|
|
[
|
|
"llm-secret",
|
|
"grafana-token",
|
|
]
|
|
)
|
|
text_responses = iter(["https://grafana.example.com"])
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_grafana_integration",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="Grafana ok"),
|
|
)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_slack_webhook",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="Slack ok"),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved_integrations == [
|
|
(
|
|
"grafana",
|
|
{
|
|
"credentials": {
|
|
"endpoint": "https://grafana.example.com",
|
|
"api_key": "grafana-token",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
assert synced_env_values == [
|
|
{
|
|
"GRAFANA_INSTANCE_URL": "https://grafana.example.com",
|
|
},
|
|
]
|
|
output = capsys.readouterr().out
|
|
assert "Grafana" in output
|
|
|
|
|
|
def test_run_wizard_configures_honeycomb(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "honeycomb"])
|
|
password_responses = iter(["llm-secret", "hny_test"])
|
|
text_responses = iter(["prod-api", "https://api.honeycomb.io"])
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_honeycomb_integration",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="Honeycomb ok"),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved_integrations == [
|
|
(
|
|
"honeycomb",
|
|
{
|
|
"credentials": {
|
|
"api_key": "hny_test",
|
|
"dataset": "prod-api",
|
|
"base_url": "https://api.honeycomb.io",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
assert synced_env_values == [
|
|
{
|
|
"HONEYCOMB_DATASET": "prod-api",
|
|
"HONEYCOMB_API_URL": "https://api.honeycomb.io",
|
|
}
|
|
]
|
|
|
|
|
|
def test_run_wizard_configures_coralogix(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "coralogix"])
|
|
password_responses = iter(["llm-secret", "cx_test"])
|
|
text_responses = iter(
|
|
[
|
|
"https://api.coralogix.com",
|
|
"payments",
|
|
"worker",
|
|
]
|
|
)
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_coralogix_integration",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="Coralogix ok"),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved_integrations == [
|
|
(
|
|
"coralogix",
|
|
{
|
|
"credentials": {
|
|
"api_key": "cx_test",
|
|
"base_url": "https://api.coralogix.com",
|
|
"application_name": "payments",
|
|
"subsystem_name": "worker",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
assert synced_env_values == [
|
|
{
|
|
"CORALOGIX_API_URL": "https://api.coralogix.com",
|
|
"CORALOGIX_APPLICATION_NAME": "payments",
|
|
"CORALOGIX_SUBSYSTEM_NAME": "worker",
|
|
}
|
|
]
|
|
|
|
|
|
def test_run_wizard_configures_dagster(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "dagster"])
|
|
password_responses = iter(["llm-secret", "dag_test_token"])
|
|
text_responses = iter(["http://localhost:3000"])
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
synced_secrets: list[tuple[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_dagster_integration",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="Dagster ok"),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"sync_env_secret",
|
|
lambda key, value: synced_secrets.append((key, value)),
|
|
)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved_integrations == [
|
|
(
|
|
"dagster",
|
|
{
|
|
"credentials": {
|
|
"endpoint": "http://localhost:3000",
|
|
"api_token": "dag_test_token",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
assert synced_env_values == [{"DAGSTER_ENDPOINT": "http://localhost:3000"}]
|
|
assert synced_secrets == [("DAGSTER_API_TOKEN", "dag_test_token")]
|
|
|
|
|
|
def test_run_wizard_configures_dagster_oss_skips_secret(monkeypatch, tmp_path) -> None:
|
|
"""OSS path: empty api_token must not call sync_env_secret."""
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "dagster"])
|
|
password_responses = iter(["llm-secret", ""])
|
|
text_responses = iter(["http://localhost:3000"])
|
|
synced_env_values: list[dict[str, str]] = []
|
|
synced_secrets: list[tuple[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_dagster_integration",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="Dagster ok"),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"sync_env_values",
|
|
lambda values, **_kwargs: synced_env_values.append(values) or (tmp_path / ".env"),
|
|
)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"sync_env_secret",
|
|
lambda key, value: synced_secrets.append((key, value)),
|
|
)
|
|
monkeypatch.setattr(
|
|
_integration_configurators, "upsert_integration", lambda *_args, **_kwargs: None
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert synced_env_values == [{"DAGSTER_ENDPOINT": "http://localhost:3000"}]
|
|
assert synced_secrets == [] # OSS path: no token
|
|
|
|
|
|
def test_run_wizard_configures_slack_persists_webhook(monkeypatch, tmp_path) -> None:
|
|
"""Regression: the wizard must upsert the Slack webhook to the store.
|
|
|
|
Previously `_configure_slack` validated the webhook and reported "Slack" in
|
|
the success summary but never called `upsert_integration`, silently
|
|
discarding the webhook (the catalog reads Slack store-first, so nothing was
|
|
readable afterwards). The webhook is a secret, so it belongs in the store,
|
|
not `.env` — `sync_env_values` is called with an empty mapping.
|
|
"""
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "slack"])
|
|
# webhook_url is prompted with secret=True, so it comes from the password mock.
|
|
password_responses = iter(["llm-secret", "https://hooks.slack.com/services/T0/B0/XXXXX"])
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
# Slack has no plain-text prompt; guard against an unexpected one.
|
|
m = MagicMock()
|
|
m.ask.return_value = ""
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_slack_webhook",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="Slack ok"),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved_integrations == [
|
|
(
|
|
"slack",
|
|
{"credentials": {"webhook_url": "https://hooks.slack.com/services/T0/B0/XXXXX"}},
|
|
)
|
|
]
|
|
# Webhook is a secret: it goes to the store, not `.env`.
|
|
assert synced_env_values == [{}]
|
|
|
|
|
|
def test_run_wizard_dagster_retries_on_validation_failure(monkeypatch, tmp_path) -> None:
|
|
"""Two consecutive Dagster validation failures, then success.
|
|
|
|
Proves the retry loop recovers from N consecutive failures (not just one),
|
|
and that only the final successful attempt reaches the persistence layer.
|
|
"""
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "dagster"])
|
|
# Two wrong tokens, then the correct one.
|
|
password_responses = iter(["llm-secret", "bad_token_1", "bad_token_2", "dag_good"])
|
|
# endpoint is prompted on each retry (three attempts total).
|
|
text_responses = iter(
|
|
["http://localhost:3000", "http://localhost:3000", "http://localhost:3000"]
|
|
)
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
synced_secrets: list[tuple[str, str]] = []
|
|
validation_call_count = 0
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
def _validate_dagster(**_kwargs):
|
|
nonlocal validation_call_count
|
|
validation_call_count += 1
|
|
if validation_call_count < 3:
|
|
return flow.IntegrationHealthResult(
|
|
ok=False, detail="Dagster GraphQL probe failed: HTTP 401"
|
|
)
|
|
return flow.IntegrationHealthResult(ok=True, detail="Dagster ok")
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators, "validate_dagster_integration", _validate_dagster
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"sync_env_secret",
|
|
lambda key, value: synced_secrets.append((key, value)),
|
|
)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert validation_call_count == 3 # two failures, then success
|
|
# Only the successful attempt should reach the persistence layer:
|
|
assert saved_integrations == [
|
|
(
|
|
"dagster",
|
|
{
|
|
"credentials": {
|
|
"endpoint": "http://localhost:3000",
|
|
"api_token": "dag_good",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
assert synced_env_values == [{"DAGSTER_ENDPOINT": "http://localhost:3000"}]
|
|
assert synced_secrets == [("DAGSTER_API_TOKEN", "dag_good")]
|
|
|
|
|
|
def test_run_wizard_configures_github_mcp_and_sentry(monkeypatch, tmp_path, capsys) -> None:
|
|
select_responses = iter(
|
|
[
|
|
"quickstart",
|
|
"anthropic",
|
|
"api_key",
|
|
"claude-opus-4-7",
|
|
"github",
|
|
flow.DEFAULT_GITHUB_MCP_MODE,
|
|
"token",
|
|
"auto",
|
|
"any",
|
|
"summary",
|
|
]
|
|
)
|
|
text_responses = iter(
|
|
[
|
|
flow.DEFAULT_GITHUB_MCP_URL,
|
|
"repos,issues,pull_requests,actions,search",
|
|
]
|
|
)
|
|
password_responses = iter(
|
|
[
|
|
"llm-secret",
|
|
"ghp_test",
|
|
]
|
|
)
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_github_mcp_integration",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="GitHub MCP ok"),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"sync_env_values",
|
|
_sync_env_values,
|
|
)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved_integrations == [
|
|
(
|
|
"github",
|
|
{
|
|
"credentials": {
|
|
"url": flow.DEFAULT_GITHUB_MCP_URL,
|
|
"mode": flow.DEFAULT_GITHUB_MCP_MODE,
|
|
"auth_token": "ghp_test",
|
|
"command": "",
|
|
"args": [],
|
|
"toolsets": ["repos", "issues", "pull_requests", "actions", "search"],
|
|
}
|
|
},
|
|
),
|
|
]
|
|
assert synced_env_values == [
|
|
{
|
|
"GITHUB_MCP_URL": flow.DEFAULT_GITHUB_MCP_URL,
|
|
"GITHUB_MCP_MODE": flow.DEFAULT_GITHUB_MCP_MODE,
|
|
"GITHUB_MCP_COMMAND": "",
|
|
"GITHUB_MCP_ARGS": "",
|
|
"GITHUB_MCP_TOOLSETS": "repos,issues,pull_requests,actions,search",
|
|
},
|
|
]
|
|
|
|
output = capsys.readouterr().out
|
|
assert "GitHub MCP" in output
|
|
|
|
|
|
def test_run_wizard_reuses_saved_defaults_when_user_keeps_provider(monkeypatch, tmp_path) -> None:
|
|
"""When provider is already set and user declines to change, saved values are reused."""
|
|
saved: dict[str, object] = {}
|
|
saved_llm_keys: list[tuple[str, str]] = []
|
|
|
|
def _mock_select(*_args, choices=None, default=None, **_kwargs):
|
|
m = MagicMock()
|
|
prompt = str(_args[0]) if _args else ""
|
|
selected_value = "skip" if "integration" in prompt.lower() else default
|
|
if choices is not None:
|
|
for choice in choices:
|
|
if getattr(choice, "title", None) == selected_value:
|
|
selected_value = choice.value
|
|
break
|
|
m.ask.return_value = selected_value
|
|
return m
|
|
|
|
def _mock_confirm(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
# "Change provider?" -> No (keep saved openai)
|
|
m.ask.return_value = False
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "confirm", _mock_confirm)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"load_local_config",
|
|
lambda _path: {
|
|
"wizard": {"mode": "quickstart"},
|
|
"targets": {
|
|
"local": {
|
|
"provider": "openai",
|
|
"model": "gpt-5.4-mini",
|
|
"api_key_env": "OPENAI_API_KEY",
|
|
"api_key": "saved-secret",
|
|
}
|
|
},
|
|
},
|
|
)
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(_ui, "has_llm_api_key", lambda _env: False)
|
|
|
|
def _save_local_config(**kwargs):
|
|
saved.update(kwargs)
|
|
return tmp_path / "opensre.json"
|
|
|
|
monkeypatch.setattr(flow, "save_local_config", _save_local_config)
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"save_api_key",
|
|
lambda provider, value, **_kwargs: saved_llm_keys.append((provider, value)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved["wizard_mode"] == "quickstart"
|
|
assert saved["provider"] == "openai"
|
|
assert saved["model"] == "gpt-5.4-mini"
|
|
assert "api_key" not in saved
|
|
assert saved_llm_keys == [("openai", "saved-secret")]
|
|
|
|
|
|
def test_run_wizard_changes_model_when_user_keeps_provider(monkeypatch, tmp_path) -> None:
|
|
"""When provider is kept but user opts to change model, the new choice is saved."""
|
|
saved: dict[str, object] = {}
|
|
confirm_prompts: list[str] = []
|
|
|
|
def _mock_select(*_args, choices=None, default=None, **_kwargs):
|
|
m = MagicMock()
|
|
prompt = str(_args[0]) if _args else ""
|
|
if "Choose your LLM provider" in prompt:
|
|
m.ask.return_value = "openai"
|
|
elif "Choose OpenAI model" in prompt:
|
|
m.ask.return_value = "gpt-5.4-mini"
|
|
else:
|
|
m.ask.return_value = default
|
|
return m
|
|
|
|
def _mock_confirm(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
prompt = str(_args[0]) if _args else ""
|
|
confirm_prompts.append(prompt)
|
|
if "Change provider?" in prompt:
|
|
m.ask.return_value = False
|
|
elif "Change model?" in prompt:
|
|
m.ask.return_value = True
|
|
else:
|
|
m.ask.return_value = False
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "confirm", _mock_confirm)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"load_local_config",
|
|
lambda _path: {
|
|
"wizard": {"mode": "quickstart"},
|
|
"targets": {
|
|
"local": {
|
|
"provider": "openai",
|
|
"model": "gpt-5.4-mini",
|
|
"api_key_env": "OPENAI_API_KEY",
|
|
"api_key": "saved-secret",
|
|
}
|
|
},
|
|
},
|
|
)
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(_ui, "has_llm_api_key", lambda _env: True)
|
|
|
|
def _save_local_config(**kwargs):
|
|
saved.update(kwargs)
|
|
return tmp_path / "opensre.json"
|
|
|
|
monkeypatch.setattr(flow, "save_local_config", _save_local_config)
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved["provider"] == "openai"
|
|
assert saved["model"] == "gpt-5.4-mini"
|
|
assert "Change model?" in confirm_prompts
|
|
|
|
|
|
def test_run_wizard_persists_matching_local_config_and_env(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "openai", "api_key", "gpt-5.4-mini", "skip"])
|
|
saved_llm_keys: list[tuple[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = "openai-secret"
|
|
return m
|
|
|
|
store_path = tmp_path / "opensre.json"
|
|
env_path = tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: store_path)
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"save_local_config",
|
|
lambda **kwargs: wizard_store.save_local_config(path=store_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"sync_provider_env",
|
|
lambda **kwargs: sync_provider_env(env_path=env_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"save_api_key",
|
|
lambda provider, value, **_kwargs: saved_llm_keys.append((provider, value)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
|
|
payload = json.loads(store_path.read_text(encoding="utf-8"))
|
|
env_values = env_path.read_text(encoding="utf-8")
|
|
|
|
assert payload["wizard"]["mode"] == "quickstart"
|
|
assert payload["targets"]["local"]["provider"] == "openai"
|
|
assert payload["targets"]["local"]["api_key_env"] == "OPENAI_API_KEY"
|
|
assert payload["targets"]["local"]["model_env"] == "OPENAI_REASONING_MODEL"
|
|
assert "api_key" not in payload["targets"]["local"]
|
|
|
|
assert "LLM_PROVIDER=openai\n" in env_values
|
|
assert "OPENAI_API_KEY=" not in env_values
|
|
assert saved_llm_keys == [("openai", "openai-secret")]
|
|
|
|
|
|
def test_run_wizard_codex_skips_api_key_and_runs_cli_onboarding(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "codex", "", "skip"])
|
|
saved_llm_keys: list[tuple[str, str]] = []
|
|
cli_onboarding_providers: list[str] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _cli_onboarding(provider, **_kwargs):
|
|
cli_onboarding_providers.append(provider.value)
|
|
return "ok"
|
|
|
|
store_path = tmp_path / "opensre.json"
|
|
env_path = tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: store_path)
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(flow, "_run_cli_llm_onboarding", _cli_onboarding)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"save_local_config",
|
|
lambda **kwargs: wizard_store.save_local_config(path=store_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"sync_provider_env",
|
|
lambda **kwargs: sync_provider_env(env_path=env_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"save_api_key",
|
|
lambda provider, value, **_kwargs: saved_llm_keys.append((provider, value)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert cli_onboarding_providers == ["codex"]
|
|
assert saved_llm_keys == []
|
|
|
|
payload = json.loads(store_path.read_text(encoding="utf-8"))
|
|
env_values = env_path.read_text(encoding="utf-8")
|
|
assert payload["targets"]["local"]["provider"] == "codex"
|
|
assert payload["targets"]["local"]["api_key_env"] == ""
|
|
assert payload["targets"]["local"]["model_env"] == "CODEX_MODEL"
|
|
assert "LLM_PROVIDER=codex\n" in env_values
|
|
assert "CODEX_MODEL=\n" in env_values
|
|
assert "LLM_AUTH_METHOD=oauth\n" in env_values
|
|
|
|
|
|
def test_run_wizard_openai_oauth_is_onboarding_auth_method(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "openai", "oauth", "gpt-5.5", "skip"])
|
|
saved_llm_keys: list[tuple[str, str]] = []
|
|
cli_onboarding_providers: list[str] = []
|
|
saved_summary: dict[str, object] = {}
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _cli_onboarding(provider, **kwargs):
|
|
cli_onboarding_providers.append(provider.value)
|
|
assert kwargs["display_label"] == "OpenAI OAuth"
|
|
return "ok"
|
|
|
|
store_path = tmp_path / "opensre.json"
|
|
env_path = tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: store_path)
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(flow, "_run_cli_llm_onboarding", _cli_onboarding)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"save_local_config",
|
|
lambda **kwargs: wizard_store.save_local_config(path=store_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"sync_provider_env",
|
|
lambda **kwargs: sync_provider_env(env_path=env_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"save_api_key",
|
|
lambda provider, value, **_kwargs: saved_llm_keys.append((provider, value)),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow, "_render_saved_summary", lambda **kwargs: saved_summary.update(kwargs)
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert cli_onboarding_providers == ["codex"]
|
|
assert saved_llm_keys == []
|
|
|
|
payload = json.loads(store_path.read_text(encoding="utf-8"))
|
|
env_values = env_path.read_text(encoding="utf-8")
|
|
assert payload["targets"]["local"]["provider"] == "openai"
|
|
assert payload["targets"]["local"]["auth_method"] == "oauth"
|
|
assert payload["targets"]["local"]["api_key_env"] == "OPENAI_API_KEY"
|
|
assert payload["targets"]["local"]["model_env"] == "CODEX_MODEL"
|
|
assert payload["targets"]["local"]["model"] == "gpt-5.5"
|
|
assert "LLM_PROVIDER=openai\n" in env_values
|
|
assert "LLM_AUTH_METHOD=oauth\n" in env_values
|
|
assert "CODEX_MODEL=gpt-5.5\n" in env_values
|
|
assert "OPENAI_API_KEY=" not in env_values
|
|
assert saved_summary["provider_label"] == "OpenAI OAuth"
|
|
assert saved_summary["credential_line"] == "OpenAI OAuth tokens (Codex CLI)"
|
|
|
|
|
|
def test_run_wizard_anthropic_oauth_is_onboarding_auth_method(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "anthropic", "oauth", "", "skip"])
|
|
cli_onboarding_providers: list[str] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _cli_onboarding(provider, **kwargs):
|
|
cli_onboarding_providers.append(provider.value)
|
|
assert kwargs["display_label"] == "Anthropic OAuth"
|
|
return "ok"
|
|
|
|
store_path = tmp_path / "opensre.json"
|
|
env_path = tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: store_path)
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(flow, "_run_cli_llm_onboarding", _cli_onboarding)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"save_local_config",
|
|
lambda **kwargs: wizard_store.save_local_config(path=store_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"sync_provider_env",
|
|
lambda **kwargs: sync_provider_env(env_path=env_path, **kwargs),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert cli_onboarding_providers == ["claude-code"]
|
|
|
|
payload = json.loads(store_path.read_text(encoding="utf-8"))
|
|
env_values = env_path.read_text(encoding="utf-8")
|
|
assert payload["targets"]["local"]["provider"] == "anthropic"
|
|
assert payload["targets"]["local"]["auth_method"] == "oauth"
|
|
assert payload["targets"]["local"]["api_key_env"] == "ANTHROPIC_API_KEY"
|
|
assert payload["targets"]["local"]["model_env"] == "CLAUDE_CODE_MODEL"
|
|
assert "LLM_PROVIDER=anthropic\n" in env_values
|
|
assert "LLM_AUTH_METHOD=oauth\n" in env_values
|
|
assert "CLAUDE_CODE_MODEL=\n" in env_values
|
|
|
|
|
|
def test_run_wizard_claude_code_skips_api_key_and_runs_cli_onboarding(
|
|
monkeypatch, tmp_path
|
|
) -> None:
|
|
select_responses = iter(["quickstart", "claude-code", "", "skip"])
|
|
saved_llm_keys: list[tuple[str, str]] = []
|
|
cli_onboarding_providers: list[str] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _cli_onboarding(provider, **_kwargs):
|
|
cli_onboarding_providers.append(provider.value)
|
|
return "ok"
|
|
|
|
store_path = tmp_path / "opensre.json"
|
|
env_path = tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: store_path)
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(flow, "_run_cli_llm_onboarding", _cli_onboarding)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"save_local_config",
|
|
lambda **kwargs: wizard_store.save_local_config(path=store_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"sync_provider_env",
|
|
lambda **kwargs: sync_provider_env(env_path=env_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"save_api_key",
|
|
lambda provider, value, **_kwargs: saved_llm_keys.append((provider, value)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert cli_onboarding_providers == ["claude-code"]
|
|
assert saved_llm_keys == []
|
|
|
|
payload = json.loads(store_path.read_text(encoding="utf-8"))
|
|
env_values = env_path.read_text(encoding="utf-8")
|
|
assert payload["targets"]["local"]["provider"] == "claude-code"
|
|
assert payload["targets"]["local"]["api_key_env"] == ""
|
|
assert payload["targets"]["local"]["model_env"] == "CLAUDE_CODE_MODEL"
|
|
assert "LLM_PROVIDER=claude-code\n" in env_values
|
|
assert "CLAUDE_CODE_MODEL=\n" in env_values
|
|
|
|
|
|
def test_run_wizard_gemini_cli_skips_api_key_and_runs_cli_onboarding(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "gemini-cli", "", "skip"])
|
|
saved_llm_keys: list[tuple[str, str]] = []
|
|
cli_onboarding_providers: list[str] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _cli_onboarding(provider, **_kwargs):
|
|
cli_onboarding_providers.append(provider.value)
|
|
return "ok"
|
|
|
|
store_path = tmp_path / "opensre.json"
|
|
env_path = tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: store_path)
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(flow, "_run_cli_llm_onboarding", _cli_onboarding)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"save_local_config",
|
|
lambda **kwargs: wizard_store.save_local_config(path=store_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"sync_provider_env",
|
|
lambda **kwargs: sync_provider_env(env_path=env_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"save_api_key",
|
|
lambda provider, value, **_kwargs: saved_llm_keys.append((provider, value)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert cli_onboarding_providers == ["gemini-cli"]
|
|
assert saved_llm_keys == []
|
|
|
|
payload = json.loads(store_path.read_text(encoding="utf-8"))
|
|
env_values = env_path.read_text(encoding="utf-8")
|
|
assert payload["targets"]["local"]["provider"] == "gemini-cli"
|
|
assert payload["targets"]["local"]["api_key_env"] == ""
|
|
assert payload["targets"]["local"]["model_env"] == "GEMINI_CLI_MODEL"
|
|
assert "LLM_PROVIDER=gemini-cli\n" in env_values
|
|
assert "GEMINI_CLI_MODEL=\n" in env_values
|
|
|
|
|
|
def test_run_cli_llm_onboarding_ok_when_logged_in(monkeypatch) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
adapter.detect.return_value = MagicMock(
|
|
installed=True,
|
|
logged_in=True,
|
|
detail="Logged in.",
|
|
)
|
|
provider = MagicMock()
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
assert result == "ok"
|
|
adapter.detect.assert_called_once()
|
|
|
|
|
|
def test_run_cli_llm_onboarding_repick_when_not_logged_in(monkeypatch) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
adapter.detect.return_value = MagicMock(
|
|
installed=True,
|
|
logged_in=False,
|
|
detail="Not logged in.",
|
|
)
|
|
provider = MagicMock()
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
|
|
monkeypatch.setattr(flow, "_choose", lambda *_args, **_kwargs: "repick")
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
assert result == "repick"
|
|
|
|
|
|
def test_run_cli_llm_onboarding_ok_after_login_retry(monkeypatch) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
detect_calls: list[object] = []
|
|
|
|
def _detect():
|
|
detect_calls.append(None)
|
|
if len(detect_calls) == 1:
|
|
return MagicMock(
|
|
installed=True,
|
|
logged_in=False,
|
|
detail="Not logged in.",
|
|
)
|
|
return MagicMock(installed=True, logged_in=True, detail="Logged in.")
|
|
|
|
adapter.detect = _detect
|
|
provider = MagicMock()
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
|
|
monkeypatch.setattr(flow, "_choose", lambda *_args, **_kwargs: "retry")
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
assert result == "ok"
|
|
assert len(detect_calls) == 2
|
|
|
|
|
|
def test_run_cli_llm_onboarding_launches_managed_codex_oauth(monkeypatch, tmp_path) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
detect_calls: list[object] = []
|
|
|
|
def _detect():
|
|
detect_calls.append(None)
|
|
if len(detect_calls) == 1:
|
|
return MagicMock(
|
|
installed=True,
|
|
logged_in=False,
|
|
bin_path="/usr/local/bin/codex",
|
|
detail="Not logged in.",
|
|
)
|
|
return MagicMock(installed=True, logged_in=True, detail="Logged in.")
|
|
|
|
adapter.detect = _detect
|
|
provider = MagicMock()
|
|
provider.value = "codex"
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
oauth_calls: list[object] = []
|
|
|
|
def _fake_codex_oauth_login() -> CodexOAuthResult:
|
|
oauth_calls.append(None)
|
|
return CodexOAuthResult(
|
|
account_id="account-123",
|
|
auth_path=tmp_path / "codex-home" / "auth.json",
|
|
detail="OpenAI OAuth tokens stored for Codex.",
|
|
)
|
|
|
|
monkeypatch.setattr(flow, "_choose", lambda *_args, **_kwargs: "login")
|
|
monkeypatch.setattr(flow, "run_codex_oauth_login", _fake_codex_oauth_login)
|
|
monkeypatch.setenv("OPENSRE_LLM_AUTH_METADATA_PATH", str(tmp_path / "llm-auth.json"))
|
|
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
|
|
assert result == "ok"
|
|
assert oauth_calls == [None]
|
|
assert len(detect_calls) == 1
|
|
|
|
|
|
def test_run_cli_llm_onboarding_surfaces_managed_codex_oauth_error(monkeypatch) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
adapter.detect.return_value = MagicMock(
|
|
installed=True,
|
|
logged_in=None,
|
|
bin_path="/opt/homebrew/bin/codex",
|
|
detail="Auth status unknown.",
|
|
)
|
|
provider = MagicMock()
|
|
provider.value = "codex"
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
|
|
choices: list[str] = ["login", "repick"]
|
|
|
|
def _choose(*_args, **_kwargs):
|
|
return choices.pop(0)
|
|
|
|
monkeypatch.setattr(flow, "_choose", _choose)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"run_codex_oauth_login",
|
|
MagicMock(side_effect=flow.CodexOAuthError("Could not bind localhost:1455.")),
|
|
)
|
|
monkeypatch.setattr(flow, "_run_interactive_login_process", MagicMock())
|
|
|
|
result = flow._run_cli_llm_onboarding(provider, display_label="OpenAI OAuth")
|
|
|
|
assert result == "repick"
|
|
flow._run_interactive_login_process.assert_not_called()
|
|
|
|
|
|
def test_run_cli_llm_onboarding_launches_claude_browser_login(monkeypatch) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "claude-code"
|
|
adapter.binary_env_key = "CLAUDE_CODE_BIN"
|
|
adapter.install_hint = "npm i -g @anthropic-ai/claude-code"
|
|
adapter.auth_hint = "Run: claude auth login"
|
|
detect_calls: list[object] = []
|
|
|
|
def _detect():
|
|
detect_calls.append(None)
|
|
if len(detect_calls) == 1:
|
|
return MagicMock(
|
|
installed=True,
|
|
logged_in=False,
|
|
bin_path="/opt/homebrew/bin/claude",
|
|
detail="Not authenticated.",
|
|
)
|
|
return MagicMock(installed=True, logged_in=True, detail="Authenticated.")
|
|
|
|
adapter.detect = _detect
|
|
provider = MagicMock()
|
|
provider.value = "claude-code"
|
|
provider.label = "Anthropic Claude Code CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
login_commands: list[list[str]] = []
|
|
|
|
def _fake_login(command: list[str]) -> flow._LoginProcessResult:
|
|
login_commands.append(command)
|
|
return flow._LoginProcessResult(returncode=0)
|
|
|
|
monkeypatch.setattr(flow, "_choose", lambda *_args, **_kwargs: "login")
|
|
monkeypatch.setattr(flow, "_run_interactive_login_process", _fake_login)
|
|
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
|
|
assert result == "ok"
|
|
assert login_commands == [["/opt/homebrew/bin/claude", "auth", "login"]]
|
|
assert len(detect_calls) == 1
|
|
|
|
|
|
def test_run_cli_llm_onboarding_repick_when_auth_status_unclear(monkeypatch) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
adapter.detect.return_value = MagicMock(
|
|
installed=True,
|
|
logged_in=None,
|
|
detail="Auth status unknown.",
|
|
)
|
|
provider = MagicMock()
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
|
|
monkeypatch.setattr(flow, "_choose", lambda *_args, **_kwargs: "repick")
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
assert result == "repick"
|
|
|
|
|
|
def test_run_cli_llm_onboarding_ok_after_unclear_auth_retry(monkeypatch) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
detect_calls: list[object] = []
|
|
|
|
def _detect():
|
|
detect_calls.append(None)
|
|
if len(detect_calls) == 1:
|
|
return MagicMock(
|
|
installed=True,
|
|
logged_in=None,
|
|
detail="Auth status unknown.",
|
|
)
|
|
return MagicMock(installed=True, logged_in=True, detail="Logged in.")
|
|
|
|
adapter.detect = _detect
|
|
provider = MagicMock()
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
|
|
monkeypatch.setattr(flow, "_choose", lambda *_args, **_kwargs: "retry")
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
assert result == "ok"
|
|
assert len(detect_calls) == 2
|
|
|
|
|
|
def test_run_cli_llm_onboarding_repick_when_user_chooses_repick(monkeypatch) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
adapter.detect.return_value = MagicMock(
|
|
installed=False,
|
|
logged_in=None,
|
|
detail="Not found.",
|
|
)
|
|
provider = MagicMock()
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
|
|
monkeypatch.setattr(flow, "_choose", lambda *_args, **_kwargs: "repick")
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
assert result == "repick"
|
|
|
|
|
|
def test_run_cli_llm_onboarding_path_override_then_ok(monkeypatch, tmp_path) -> None:
|
|
fake_bin = write_fake_runnable_cli_bin(tmp_path, "codex")
|
|
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
|
|
detect_calls: list[object] = []
|
|
|
|
def _detect():
|
|
detect_calls.append(None)
|
|
if len(detect_calls) == 1:
|
|
return MagicMock(installed=False, logged_in=None, detail="Not found.")
|
|
return MagicMock(installed=True, logged_in=True, detail="Logged in.")
|
|
|
|
adapter.detect = _detect
|
|
provider = MagicMock()
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
|
|
monkeypatch.setattr(flow, "_choose", lambda *_args, **_kwargs: "path")
|
|
monkeypatch.setattr(flow, "_prompt_value", lambda *_args, **_kwargs: str(fake_bin))
|
|
monkeypatch.setattr(flow, "sync_env_values", lambda *_args, **_kwargs: None)
|
|
|
|
original_codex_bin = os.environ.get("CODEX_BIN")
|
|
try:
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
|
|
assert result == "ok"
|
|
assert len(detect_calls) == 2
|
|
# os.environ must be updated in-process so the next detect() call in the
|
|
# retry loop resolves the new binary without a process restart.
|
|
assert os.environ.get("CODEX_BIN") == str(fake_bin)
|
|
finally:
|
|
if original_codex_bin is None:
|
|
os.environ.pop("CODEX_BIN", None)
|
|
else:
|
|
os.environ["CODEX_BIN"] = original_codex_bin
|
|
|
|
|
|
def test_run_cli_llm_onboarding_abort_after_max_retries(monkeypatch) -> None:
|
|
adapter = MagicMock()
|
|
adapter.name = "codex"
|
|
adapter.binary_env_key = "CODEX_BIN"
|
|
adapter.install_hint = "npm i -g @openai/codex"
|
|
adapter.auth_hint = "Run: codex login"
|
|
adapter.detect.return_value = MagicMock(
|
|
installed=False,
|
|
logged_in=None,
|
|
detail="Not found.",
|
|
)
|
|
provider = MagicMock()
|
|
provider.label = "OpenAI Codex CLI"
|
|
provider.adapter_factory = lambda: adapter
|
|
|
|
monkeypatch.setattr(flow, "_choose", lambda *_args, **_kwargs: "retry")
|
|
result = flow._run_cli_llm_onboarding(provider)
|
|
assert result == "abort"
|
|
assert adapter.detect.call_count == 10
|
|
|
|
|
|
def test_credential_line_for_saved_summary_cli_codex() -> None:
|
|
from surfaces.cli.wizard import config as wizard_config
|
|
|
|
codex = next(p for p in wizard_config.SUPPORTED_PROVIDERS if p.value == "codex")
|
|
assert flow._credential_line_for_saved_summary(codex) == "OpenAI Codex CLI (Run: codex login)"
|
|
|
|
|
|
def test_credential_line_for_saved_summary_cli_claude_code() -> None:
|
|
from surfaces.cli.wizard import config as wizard_config
|
|
|
|
claude_code = next(p for p in wizard_config.SUPPORTED_PROVIDERS if p.value == "claude-code")
|
|
assert flow._credential_line_for_saved_summary(claude_code) == (
|
|
"Anthropic Claude Code CLI (Run: claude auth login or set ANTHROPIC_API_KEY)"
|
|
)
|
|
|
|
|
|
def test_credential_line_for_saved_summary_cli_gemini_cli() -> None:
|
|
from surfaces.cli.wizard import config as wizard_config
|
|
|
|
gemini_cli = next(p for p in wizard_config.SUPPORTED_PROVIDERS if p.value == "gemini-cli")
|
|
assert flow._credential_line_for_saved_summary(gemini_cli) == (
|
|
"Google Gemini CLI (Run: gemini (interactive login) or set GEMINI_API_KEY)"
|
|
)
|
|
|
|
|
|
def test_credential_line_for_saved_summary_cli_copilot() -> None:
|
|
from surfaces.cli.wizard import config as wizard_config
|
|
|
|
copilot = next(p for p in wizard_config.SUPPORTED_PROVIDERS if p.value == "copilot")
|
|
line = flow._credential_line_for_saved_summary(copilot)
|
|
# PR #1533: hint surfaces both CLI paths (`copilot login`, `gh auth login`)
|
|
# before the env-var bypass, matching the new CLI-first probe order.
|
|
assert line.startswith("GitHub Copilot CLI (Run `copilot login`")
|
|
assert "gh auth login" in line
|
|
assert "COPILOT_GITHUB_TOKEN" in line
|
|
|
|
|
|
def test_credential_line_for_saved_summary_anthropic() -> None:
|
|
from surfaces.cli.wizard import config as wizard_config
|
|
|
|
anthropic = next(p for p in wizard_config.SUPPORTED_PROVIDERS if p.value == "anthropic")
|
|
assert flow._credential_line_for_saved_summary(anthropic) == "system keychain"
|
|
|
|
|
|
def test_credential_line_for_saved_summary_cli_without_factory() -> None:
|
|
from surfaces.cli.wizard.config import ModelOption, ProviderOption
|
|
|
|
p = ProviderOption(
|
|
value="codex",
|
|
label="Fake CLI",
|
|
group="Local CLI providers",
|
|
api_key_env="",
|
|
model_env="CODEX_MODEL",
|
|
default_model="",
|
|
models=(ModelOption(value="", label="default"),),
|
|
credential_kind="cli",
|
|
adapter_factory=None,
|
|
allow_custom_models=True,
|
|
)
|
|
assert flow._credential_line_for_saved_summary(p) == "Fake CLI (CLI)"
|
|
|
|
|
|
def test_run_wizard_configures_gitlab(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "gitlab"])
|
|
password_responses = iter(["llm-secret", "glpat_test"])
|
|
text_responses = iter(["https://gitlab.example.com/api/v4"])
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
synced_env_secrets: list[tuple[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_gitlab_integration",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="GitLab ok"),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
def _sync_env_secret(key: str, value: str) -> None:
|
|
synced_env_secrets.append((key, value))
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_secret", _sync_env_secret)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved_integrations == [
|
|
(
|
|
"gitlab",
|
|
{
|
|
"credentials": {
|
|
"base_url": "https://gitlab.example.com/api/v4",
|
|
"auth_token": "glpat_test",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
assert synced_env_secrets == [("GITLAB_ACCESS_TOKEN", "glpat_test")]
|
|
assert synced_env_values == [
|
|
{
|
|
"GITLAB_BASE_URL": "https://gitlab.example.com/api/v4",
|
|
}
|
|
]
|
|
|
|
|
|
def test_run_wizard_gitlab_retries_on_validation_failure(monkeypatch, tmp_path) -> None:
|
|
"""When GitLab validation fails the first time, the wizard retries and succeeds."""
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "gitlab"])
|
|
# First attempt: wrong token; second attempt: correct token
|
|
password_responses = iter(["llm-secret", "bad_token", "glpat_good"])
|
|
# base_url is prompted on each retry
|
|
text_responses = iter(
|
|
[
|
|
"https://gitlab.com/api/v4",
|
|
"https://gitlab.com/api/v4",
|
|
]
|
|
)
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
synced_env_secrets: list[tuple[str, str]] = []
|
|
validation_call_count = 0
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
def _validate_gitlab(**_kwargs):
|
|
nonlocal validation_call_count
|
|
validation_call_count += 1
|
|
if validation_call_count == 1:
|
|
return flow.IntegrationHealthResult(ok=False, detail="Unauthorized")
|
|
return flow.IntegrationHealthResult(ok=True, detail="GitLab ok")
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(_integration_configurators, "validate_gitlab_integration", _validate_gitlab)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
def _sync_env_secret(key: str, value: str) -> None:
|
|
synced_env_secrets.append((key, value))
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_secret", _sync_env_secret)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert validation_call_count == 2
|
|
assert saved_integrations == [
|
|
(
|
|
"gitlab",
|
|
{
|
|
"credentials": {
|
|
"base_url": "https://gitlab.com/api/v4",
|
|
"auth_token": "glpat_good",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
assert synced_env_secrets == [("GITLAB_ACCESS_TOKEN", "glpat_good")]
|
|
assert synced_env_values == [
|
|
{
|
|
"GITLAB_BASE_URL": "https://gitlab.com/api/v4",
|
|
}
|
|
]
|
|
|
|
|
|
def test_run_wizard_switches_provider_and_keeps_store_and_env_in_sync(
|
|
monkeypatch, tmp_path
|
|
) -> None:
|
|
# Saved: anthropic. User says yes to "Change provider?" and picks openai.
|
|
select_responses = iter(["quickstart", "openai", "api_key", "gpt-5.4-mini", "skip"])
|
|
confirm_responses = iter([True]) # "Change provider?" -> Yes
|
|
saved_llm_keys: list[tuple[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_confirm(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(confirm_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = "fresh-openai-key"
|
|
return m
|
|
|
|
store_path = tmp_path / "opensre.json"
|
|
env_path = tmp_path / ".env"
|
|
wizard_store.save_local_config(
|
|
wizard_mode="quickstart",
|
|
provider="anthropic",
|
|
model="claude-opus-4-5",
|
|
api_key_env="ANTHROPIC_API_KEY",
|
|
model_env="ANTHROPIC_MODEL",
|
|
probes={
|
|
"local": {"target": "local", "reachable": True, "detail": "ok"},
|
|
"remote": {"target": "remote", "reachable": False, "detail": "down"},
|
|
},
|
|
path=store_path,
|
|
)
|
|
env_path.write_text(
|
|
"LLM_PROVIDER=anthropic\n"
|
|
"ANTHROPIC_API_KEY=saved-anthropic-key\n"
|
|
"ANTHROPIC_MODEL=claude-opus-4-5\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "confirm", _mock_confirm)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: store_path)
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"save_local_config",
|
|
lambda **kwargs: wizard_store.save_local_config(path=store_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"sync_provider_env",
|
|
lambda **kwargs: sync_provider_env(env_path=env_path, **kwargs),
|
|
)
|
|
monkeypatch.setattr(
|
|
_ui,
|
|
"save_api_key",
|
|
lambda provider, value, **_kwargs: saved_llm_keys.append((provider, value)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
|
|
payload = json.loads(store_path.read_text(encoding="utf-8"))
|
|
env_values = env_path.read_text(encoding="utf-8")
|
|
|
|
assert payload["targets"]["local"]["provider"] == "openai"
|
|
assert payload["targets"]["local"]["api_key_env"] == "OPENAI_API_KEY"
|
|
assert payload["targets"]["local"]["model_env"] == "OPENAI_REASONING_MODEL"
|
|
assert "api_key" not in payload["targets"]["local"]
|
|
|
|
assert "LLM_PROVIDER=openai\n" in env_values
|
|
assert "OPENAI_API_KEY=" not in env_values
|
|
assert "ANTHROPIC_API_KEY=" not in env_values
|
|
assert "OPENAI_REASONING_MODEL=" in env_values
|
|
assert saved_llm_keys == [("openai", "fresh-openai-key")]
|
|
|
|
|
|
def test_run_wizard_configures_opensearch(monkeypatch, tmp_path) -> None:
|
|
"""Happy path: user picks opensearch, enters URL + basic auth, all gets persisted."""
|
|
select_responses = iter(
|
|
["quickstart", "anthropic", "api_key", "claude-opus-4-7", "opensearch", "basic"]
|
|
)
|
|
password_responses = iter(["llm-secret", "secret-pass"])
|
|
text_responses = iter(["https://my-cluster.example.com", "admin"])
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
synced_env_secrets: list[tuple[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_opensearch_integration",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(ok=True, detail="OpenSearch ok"),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
def _sync_env_secret(key: str, value: str) -> None:
|
|
synced_env_secrets.append((key, value))
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_secret", _sync_env_secret)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"build_demo_action_response",
|
|
lambda: {"success": True, "topics": [], "guidance": []},
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved_integrations == [
|
|
(
|
|
"opensearch",
|
|
{
|
|
"credentials": {
|
|
"url": "https://my-cluster.example.com",
|
|
"username": "admin",
|
|
"password": "secret-pass",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
assert synced_env_secrets == [("OPENSEARCH_PASSWORD", "secret-pass")]
|
|
assert synced_env_values == [
|
|
{
|
|
"OPENSEARCH_URL": "https://my-cluster.example.com",
|
|
"OPENSEARCH_USERNAME": "admin",
|
|
}
|
|
]
|
|
|
|
|
|
def test_run_wizard_opensearch_retries_on_validation_failure(monkeypatch, tmp_path) -> None:
|
|
"""When OpenSearch validation fails the first time, the wizard retries and succeeds."""
|
|
select_responses = iter(
|
|
["quickstart", "anthropic", "api_key", "claude-opus-4-7", "opensearch", "basic", "basic"]
|
|
)
|
|
password_responses = iter(["llm-secret", "wrong-pass", "correct-pass"])
|
|
text_responses = iter(
|
|
[
|
|
"https://my-cluster.example.com",
|
|
"admin",
|
|
"https://my-cluster.example.com",
|
|
"admin",
|
|
]
|
|
)
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
validation_call_count = 0
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
def _validate_opensearch(**_kwargs):
|
|
nonlocal validation_call_count
|
|
validation_call_count += 1
|
|
if validation_call_count == 1:
|
|
return flow.IntegrationHealthResult(ok=False, detail="HTTP 401: unauthorized")
|
|
return flow.IntegrationHealthResult(ok=True, detail="OpenSearch ok")
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators, "validate_opensearch_integration", _validate_opensearch
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"build_demo_action_response",
|
|
lambda: {"success": True, "topics": [], "guidance": []},
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert validation_call_count == 2
|
|
assert saved_integrations == [
|
|
(
|
|
"opensearch",
|
|
{
|
|
"credentials": {
|
|
"url": "https://my-cluster.example.com",
|
|
"username": "admin",
|
|
"password": "correct-pass",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
def test_run_wizard_opensearch_rejects_empty_api_key(monkeypatch, tmp_path) -> None:
|
|
"""When user picks api_key auth but enters an empty key, the wizard rejects it.
|
|
|
|
Regression for the silent-credential-drop bug: on a cluster with security
|
|
disabled the validator probe would return 200, result.ok would be True,
|
|
and the integration would persist as URL-only — discarding the user's
|
|
chosen auth method without any visible error.
|
|
|
|
The wizard now guards against empty api_key explicitly (before the probe
|
|
runs) and re-prompts with a clear error message. Verified by checking
|
|
that validation is only called once the user supplies a non-empty key.
|
|
"""
|
|
# User picks: opensearch -> api_key auth -> (rejected, empty) -> api_key auth retry
|
|
select_responses = iter(
|
|
[
|
|
"quickstart",
|
|
"anthropic",
|
|
"api_key",
|
|
"claude-opus-4-7",
|
|
"opensearch",
|
|
"api_key",
|
|
"api_key",
|
|
]
|
|
)
|
|
# First api_key prompt: empty (rejected). Second: valid key.
|
|
password_responses = iter(["llm-secret", "", "valid-api-key"])
|
|
text_responses = iter(
|
|
[
|
|
"https://my-cluster.example.com",
|
|
"https://my-cluster.example.com",
|
|
]
|
|
)
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
validation_call_count = 0
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
def _validate_opensearch(**_kwargs):
|
|
nonlocal validation_call_count
|
|
validation_call_count += 1
|
|
return flow.IntegrationHealthResult(ok=True, detail="OpenSearch ok")
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators, "validate_opensearch_integration", _validate_opensearch
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"build_demo_action_response",
|
|
lambda: {"success": True, "topics": [], "guidance": []},
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
# Validator should only be called once — after the user supplies a real key.
|
|
# The empty-key attempt must be blocked before reaching the probe.
|
|
assert validation_call_count == 1
|
|
assert saved_integrations == [
|
|
(
|
|
"opensearch",
|
|
{
|
|
"credentials": {
|
|
"url": "https://my-cluster.example.com",
|
|
"api_key": "valid-api-key",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
def test_run_wizard_opensearch_rejects_empty_basic_password(monkeypatch, tmp_path) -> None:
|
|
"""When user picks basic auth but leaves password empty, the wizard rejects it.
|
|
|
|
Companion regression for the half-credential bug: ElasticsearchConfig.headers
|
|
silently drops the Authorization header when either half of Basic Auth is
|
|
empty, so the agent would send unauthenticated requests against a
|
|
security-enabled cluster and fail with a confusing 401.
|
|
|
|
The wizard now guards against half-populated Basic Auth before the probe
|
|
runs and re-prompts with a clear error message. Verified by checking that
|
|
validation is only called once the user supplies both halves.
|
|
"""
|
|
# User picks: opensearch -> basic auth -> (rejected, empty pass) -> basic auth retry
|
|
select_responses = iter(
|
|
["quickstart", "anthropic", "api_key", "claude-opus-4-7", "opensearch", "basic", "basic"]
|
|
)
|
|
# First password prompt: empty (rejected). Second attempt: valid password.
|
|
password_responses = iter(["llm-secret", "", "real-pass"])
|
|
text_responses = iter(
|
|
[
|
|
"https://my-cluster.example.com",
|
|
"admin",
|
|
"https://my-cluster.example.com",
|
|
"admin",
|
|
]
|
|
)
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
validation_call_count = 0
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
def _validate_opensearch(**_kwargs):
|
|
nonlocal validation_call_count
|
|
validation_call_count += 1
|
|
return flow.IntegrationHealthResult(ok=True, detail="OpenSearch ok")
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators, "validate_opensearch_integration", _validate_opensearch
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
monkeypatch.setattr(
|
|
flow,
|
|
"build_demo_action_response",
|
|
lambda: {"success": True, "topics": [], "guidance": []},
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
# Validator should only be called once — after the user supplies both halves.
|
|
# The empty-password attempt must be blocked before reaching the probe.
|
|
assert validation_call_count == 1
|
|
assert saved_integrations == [
|
|
(
|
|
"opensearch",
|
|
{
|
|
"credentials": {
|
|
"url": "https://my-cluster.example.com",
|
|
"username": "admin",
|
|
"password": "real-pass",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
def test_run_wizard_configures_telegram(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "telegram"])
|
|
password_responses = iter(["llm-secret", "123:ABC"])
|
|
text_responses = iter(["-1001234567890"])
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
synced_env_values: list[dict[str, str]] = []
|
|
synced_env_secrets: list[tuple[str, str]] = []
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"validate_telegram_bot",
|
|
lambda **_kwargs: flow.IntegrationHealthResult(
|
|
ok=True, detail="Connected to Telegram bot @opensre_bot."
|
|
),
|
|
)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
|
|
def _sync_env_values(values: dict[str, str], **_kwargs):
|
|
synced_env_values.append(values)
|
|
return tmp_path / ".env"
|
|
|
|
def _sync_env_secret(key: str, value: str) -> None:
|
|
synced_env_secrets.append((key, value))
|
|
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_values", _sync_env_values)
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_secret", _sync_env_secret)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert saved_integrations == [
|
|
(
|
|
"telegram",
|
|
{
|
|
"credentials": {
|
|
"bot_token": "123:ABC",
|
|
"default_chat_id": "-1001234567890",
|
|
}
|
|
},
|
|
)
|
|
]
|
|
assert synced_env_secrets == [("TELEGRAM_BOT_TOKEN", "123:ABC")]
|
|
assert synced_env_values == [{"TELEGRAM_DEFAULT_CHAT_ID": "-1001234567890"}]
|
|
|
|
|
|
def test_run_wizard_telegram_retries_on_validation_failure(monkeypatch, tmp_path) -> None:
|
|
select_responses = iter(["quickstart", "anthropic", "api_key", "claude-opus-4-7", "telegram"])
|
|
password_responses = iter(["llm-secret", "bad-token", "123:GOOD"])
|
|
text_responses = iter(["-1001", "-1001"])
|
|
saved_integrations: list[tuple[str, dict]] = []
|
|
validation_call_count = 0
|
|
|
|
def _mock_select(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(select_responses)
|
|
return m
|
|
|
|
def _mock_password(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(password_responses)
|
|
return m
|
|
|
|
def _mock_text(*_args, **_kwargs):
|
|
m = MagicMock()
|
|
m.ask.return_value = next(text_responses)
|
|
return m
|
|
|
|
def _validate(**_kwargs):
|
|
nonlocal validation_call_count
|
|
validation_call_count += 1
|
|
if validation_call_count == 1:
|
|
return flow.IntegrationHealthResult(ok=False, detail="Telegram API check failed.")
|
|
return flow.IntegrationHealthResult(ok=True, detail="Connected to Telegram bot @bot.")
|
|
|
|
monkeypatch.setattr(_ui, "select_prompt", _mock_select)
|
|
monkeypatch.setattr(flow.questionary, "password", _mock_password)
|
|
monkeypatch.setattr(flow.questionary, "text", _mock_text)
|
|
monkeypatch.setattr(_ui, "get_store_path", lambda: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "probe_local_target", lambda _path: ProbeResult("local", True, "ok"))
|
|
monkeypatch.setattr(_integration_configurators, "validate_telegram_bot", _validate)
|
|
monkeypatch.setattr(flow, "save_local_config", lambda **_kwargs: tmp_path / "opensre.json")
|
|
monkeypatch.setattr(flow, "sync_provider_env", lambda **_kwargs: tmp_path / ".env")
|
|
monkeypatch.setattr(_ui, "save_llm_api_key", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(
|
|
_integration_configurators, "sync_env_values", lambda *_a, **_kw: tmp_path / ".env"
|
|
)
|
|
monkeypatch.setattr(_integration_configurators, "sync_env_secret", lambda *_a, **_kw: None)
|
|
monkeypatch.setattr(
|
|
_integration_configurators,
|
|
"upsert_integration",
|
|
lambda service, payload: saved_integrations.append((service, payload)),
|
|
)
|
|
|
|
exit_code = flow.run_wizard()
|
|
|
|
assert exit_code == 0
|
|
assert validation_call_count == 2
|
|
assert saved_integrations[0][0] == "telegram"
|
|
assert saved_integrations[0][1]["credentials"]["bot_token"] == "123:GOOD"
|