forked from cleveragents/cleveragents-core
freemo
92f7feaa32
Implement auth login/logout/status and team list/use CLI commands with stubbed responses when server is disabled. Includes token storage via OS keyring with encrypted file fallback. Key changes: - Add agents auth login/logout/status commands - Add agents team list/use commands - Add TokenStore with keyring + encrypted file fallback - Add auth_token, active_team, default_namespace settings - Wire AuthClient and ServerClient stubs - Add Behave BDD tests, Robot integration tests, ASV benchmarks ISSUES CLOSED: #340
447 lines
15 KiB
Python
447 lines
15 KiB
Python
"""Step definitions for auth and team CLI command scenarios."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
from behave import given, then, use_step_matcher, when
|
|
from behave.runner import Context
|
|
from typer.testing import CliRunner
|
|
|
|
use_step_matcher("re")
|
|
|
|
from cleveragents.cli.commands import auth as auth_mod # noqa: E402
|
|
from cleveragents.cli.commands import team as team_mod # noqa: E402
|
|
from cleveragents.cli.commands.auth import app as auth_app # noqa: E402
|
|
from cleveragents.cli.commands.team import app as team_app # noqa: E402
|
|
from cleveragents.infrastructure.auth.token_store import TokenStore # noqa: E402
|
|
from cleveragents.shared.redaction import REDACTED # noqa: E402
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
def _make_temp_store(context: Context) -> TokenStore:
|
|
"""Create a temp-backed TokenStore and store cleanup info."""
|
|
tmpdir = Path(tempfile.mkdtemp())
|
|
store = TokenStore(token_dir=tmpdir, use_keyring=False)
|
|
if not hasattr(context, "_tmp_dirs"):
|
|
context._tmp_dirs = []
|
|
context._tmp_dirs.append(tmpdir)
|
|
return store
|
|
|
|
|
|
def _cleanup_temps(context: Context) -> None:
|
|
"""Remove temp dirs created during the scenario."""
|
|
for d in getattr(context, "_tmp_dirs", []):
|
|
shutil.rmtree(str(d), ignore_errors=True)
|
|
|
|
|
|
def _run_auth(context: Context, args: list[str]) -> Any:
|
|
"""Run auth CLI command with patched token store."""
|
|
store: TokenStore = context.token_store
|
|
|
|
def _patched_store() -> TokenStore:
|
|
return store
|
|
|
|
def _patched_read_team() -> str | None:
|
|
data = getattr(context, "_config_data", {})
|
|
val = data.get("active_team")
|
|
if val is not None and str(val).strip():
|
|
return str(val).strip()
|
|
return None
|
|
|
|
def _patched_write_team(team: str | None) -> None:
|
|
if not hasattr(context, "_config_data"):
|
|
context._config_data = {}
|
|
if team is None:
|
|
context._config_data.pop("active_team", None)
|
|
else:
|
|
context._config_data["active_team"] = team
|
|
|
|
with (
|
|
patch.object(auth_mod, "_get_token_store", _patched_store),
|
|
patch.object(auth_mod, "_read_active_team", _patched_read_team),
|
|
patch.object(auth_mod, "_write_active_team", _patched_write_team),
|
|
):
|
|
result = runner.invoke(auth_app, args)
|
|
context.result = result
|
|
context.raw_output = result.output
|
|
return result
|
|
|
|
|
|
def _run_team(context: Context, args: list[str]) -> Any:
|
|
"""Run team CLI command with patched helpers."""
|
|
|
|
def _patched_read_team() -> str | None:
|
|
data = getattr(context, "_config_data", {})
|
|
val = data.get("active_team")
|
|
if val is not None and str(val).strip():
|
|
return str(val).strip()
|
|
return None
|
|
|
|
def _patched_write_team(team: str | None) -> None:
|
|
if not hasattr(context, "_config_data"):
|
|
context._config_data = {}
|
|
if team is None:
|
|
context._config_data.pop("active_team", None)
|
|
else:
|
|
context._config_data["active_team"] = team
|
|
|
|
with (
|
|
patch.object(team_mod, "_read_active_team", _patched_read_team),
|
|
patch.object(team_mod, "_write_active_team", _patched_write_team),
|
|
):
|
|
result = runner.invoke(team_app, args)
|
|
context.result = result
|
|
context.raw_output = result.output
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(r"a clean token store")
|
|
def step_clean_token_store(context: Context) -> None:
|
|
context.token_store = _make_temp_store(context)
|
|
context._config_data = {}
|
|
|
|
|
|
@given(r'a token store with token "(?P<token>[^"]*)"')
|
|
def step_token_store_with_token(context: Context, token: str) -> None:
|
|
store = _make_temp_store(context)
|
|
store.store_token(token)
|
|
context.token_store = store
|
|
context._config_data = {}
|
|
|
|
|
|
@given(r"a clean token store using file backend")
|
|
def step_clean_token_store_file(context: Context) -> None:
|
|
context.token_store = _make_temp_store(context)
|
|
context._config_data = {}
|
|
|
|
|
|
@given(r'the active team is set to "(?P<team>[^"]*)"')
|
|
def step_active_team_set(context: Context, team: str) -> None:
|
|
if not hasattr(context, "_config_data"):
|
|
context._config_data = {}
|
|
context._config_data["active_team"] = team
|
|
if not hasattr(context, "token_store"):
|
|
context.token_store = _make_temp_store(context)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(r'I run auth login with token "(?P<token>[^"]*)" and format "(?P<fmt>[^"]*)"')
|
|
def step_run_auth_login_fmt(context: Context, token: str, fmt: str) -> None:
|
|
_run_auth(context, ["login", "--token", token, "--format", fmt])
|
|
|
|
|
|
@when(r'I run auth login with token "(?P<token>[^"]*)"')
|
|
def step_run_auth_login(context: Context, token: str) -> None:
|
|
_run_auth(context, ["login", "--token", token])
|
|
|
|
|
|
@when(r"I run auth login with empty token")
|
|
def step_run_auth_login_empty(context: Context) -> None:
|
|
if not hasattr(context, "token_store"):
|
|
context.token_store = _make_temp_store(context)
|
|
_run_auth(context, ["login", "--token", ""])
|
|
|
|
|
|
@when(r"I run auth logout")
|
|
def step_run_auth_logout(context: Context) -> None:
|
|
_run_auth(context, ["logout"])
|
|
|
|
|
|
@when(r'I run auth logout with format "(?P<fmt>[^"]*)"')
|
|
def step_run_auth_logout_fmt(context: Context, fmt: str) -> None:
|
|
_run_auth(context, ["logout", "--format", fmt])
|
|
|
|
|
|
@when(r"I run auth status")
|
|
def step_run_auth_status(context: Context) -> None:
|
|
if not hasattr(context, "token_store"):
|
|
context.token_store = _make_temp_store(context)
|
|
_run_auth(context, ["status"])
|
|
|
|
|
|
@when(r'I run auth status with format "(?P<fmt>[^"]*)"')
|
|
def step_run_auth_status_fmt(context: Context, fmt: str) -> None:
|
|
if not hasattr(context, "token_store"):
|
|
context.token_store = _make_temp_store(context)
|
|
_run_auth(context, ["status", "--format", fmt])
|
|
|
|
|
|
@when(r"I run team list")
|
|
def step_run_team_list(context: Context) -> None:
|
|
if not hasattr(context, "token_store"):
|
|
context.token_store = _make_temp_store(context)
|
|
_run_team(context, ["list"])
|
|
|
|
|
|
@when(r'I run team list with format "(?P<fmt>[^"]*)"')
|
|
def step_run_team_list_fmt(context: Context, fmt: str) -> None:
|
|
if not hasattr(context, "token_store"):
|
|
context.token_store = _make_temp_store(context)
|
|
_run_team(context, ["list", "--format", fmt])
|
|
|
|
|
|
@when(r'I run team use with name "(?P<name>[^"]*)" and format "(?P<fmt>[^"]*)"')
|
|
def step_run_team_use_fmt(context: Context, name: str, fmt: str) -> None:
|
|
if not hasattr(context, "token_store"):
|
|
context.token_store = _make_temp_store(context)
|
|
_run_team(context, ["use", name, "--format", fmt])
|
|
|
|
|
|
@when(r'I run team use with name "(?P<name>[^"]*)"')
|
|
def step_run_team_use(context: Context, name: str) -> None:
|
|
if not hasattr(context, "token_store"):
|
|
context.token_store = _make_temp_store(context)
|
|
_run_team(context, ["use", name])
|
|
|
|
|
|
@when(r'I store token "(?P<token>[^"]*)"')
|
|
def step_store_token(context: Context, token: str) -> None:
|
|
context.token_store.store_token(token)
|
|
|
|
|
|
@when(r"I clear the token store")
|
|
def step_clear_token_store(context: Context) -> None:
|
|
context.token_store.clear_token()
|
|
|
|
|
|
@when(r"I try to store an empty token")
|
|
def step_try_store_empty(context: Context) -> None:
|
|
try:
|
|
context.token_store.store_token("")
|
|
context.token_error = None
|
|
except ValueError as exc:
|
|
context.token_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then(r'the auth result status should be "(?P<status>[^"]*)"')
|
|
def step_auth_result_status(context: Context, status: str) -> None:
|
|
assert context.result.exit_code == 0, (
|
|
f"Expected exit 0, got {context.result.exit_code}: {context.result.output}"
|
|
)
|
|
assert (
|
|
status in context.result.output.lower().replace("_", " ").replace("-", " ")
|
|
or status.replace("_", " ") in context.result.output.lower()
|
|
), f"Status '{status}' not found in output: {context.result.output}"
|
|
|
|
|
|
@then(r"the auth result should indicate server not validated")
|
|
def step_server_not_validated(context: Context) -> None:
|
|
out = context.result.output.lower()
|
|
assert "false" in out or "not validated" in out or "server mode" in out, (
|
|
f"Expected server-not-validated indicator in: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then(r"the auth result should contain a warning about server mode")
|
|
def step_server_warning(context: Context) -> None:
|
|
out = context.result.output
|
|
assert "Server mode not enabled" in out or "server" in out.lower(), (
|
|
f"Missing server mode warning in: {out}"
|
|
)
|
|
|
|
|
|
@then(r"the token should be stored")
|
|
def step_token_stored(context: Context) -> None:
|
|
assert context.token_store.has_token(), "Token should be stored"
|
|
|
|
|
|
@then(r"the token in output should be redacted")
|
|
def step_token_redacted(context: Context) -> None:
|
|
assert REDACTED in context.result.output, (
|
|
f"Expected {REDACTED} in output: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then(r"the auth command should fail with exit code 1")
|
|
def step_auth_fail(context: Context) -> None:
|
|
assert context.result.exit_code == 1, (
|
|
f"Expected exit 1, got {context.result.exit_code}"
|
|
)
|
|
|
|
|
|
@then(r"the auth result should be valid JSON")
|
|
def step_auth_valid_json(context: Context) -> None:
|
|
assert context.result.exit_code == 0, (
|
|
f"Expected exit 0, got {context.result.exit_code}: {context.result.output}"
|
|
)
|
|
context.json_result = json.loads(context.result.output)
|
|
|
|
|
|
@then(
|
|
r'the JSON result should contain key "(?P<key>[^"]*)" with value "(?P<value>[^"]*)"'
|
|
)
|
|
def step_json_key_value(context: Context, key: str, value: str) -> None:
|
|
data = context.json_result
|
|
assert key in data, f"Key '{key}' not in JSON: {data}"
|
|
assert str(data[key]) == value, f"Expected {key}={value!r}, got {data[key]!r}"
|
|
|
|
|
|
@then(r'the JSON result should contain key "(?P<key>[^"]*)"')
|
|
def step_json_key(context: Context, key: str) -> None:
|
|
data = context.json_result
|
|
assert key in data, f"Key '{key}' not in JSON: {data}"
|
|
|
|
|
|
@then(r"the auth result should indicate token was cleared")
|
|
def step_token_cleared(context: Context) -> None:
|
|
out = context.result.output.lower()
|
|
assert "true" in out or "cleared" in out, (
|
|
f"Expected token-cleared indicator: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then(r"the token should not be stored")
|
|
def step_token_not_stored(context: Context) -> None:
|
|
assert not context.token_store.has_token(), "Token should NOT be stored"
|
|
|
|
|
|
@then(r"the active team should be cleared")
|
|
def step_active_team_cleared(context: Context) -> None:
|
|
data = getattr(context, "_config_data", {})
|
|
assert data.get("active_team") is None, (
|
|
f"Active team should be cleared, got: {data.get('active_team')}"
|
|
)
|
|
|
|
|
|
@then(r"the auth result should indicate no token was cleared")
|
|
def step_no_token_cleared(context: Context) -> None:
|
|
out = context.result.output.lower()
|
|
assert "false" in out or "no token" in out or "cleared" in out, (
|
|
f"Expected no-token-cleared indicator: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then(r"the auth result should show not authenticated")
|
|
def step_not_authenticated(context: Context) -> None:
|
|
assert context.result.exit_code == 0
|
|
out = context.result.output.lower()
|
|
assert "not authenticated" in out or "false" in out, (
|
|
f"Expected not-authenticated: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then(r"the auth result should show no active team")
|
|
def step_no_active_team(context: Context) -> None:
|
|
out = context.result.output.lower()
|
|
assert "(none)" in out or "none" in out, (
|
|
f"Expected no active team: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then(r"the auth result should show authenticated")
|
|
def step_authenticated(context: Context) -> None:
|
|
assert context.result.exit_code == 0
|
|
out = context.result.output.lower()
|
|
assert "authenticated" in out, f"Expected authenticated: {context.result.output}"
|
|
|
|
|
|
@then(r"the team result should contain a server warning")
|
|
def step_team_server_warning(context: Context) -> None:
|
|
assert context.result.exit_code == 0
|
|
out = context.result.output
|
|
assert "Server mode not enabled" in out or "server" in out.lower(), (
|
|
f"Expected server warning: {out}"
|
|
)
|
|
|
|
|
|
@then(r"the team result should show no available teams")
|
|
def step_no_teams(context: Context) -> None:
|
|
out = context.result.output.lower()
|
|
assert "none" in out or "no" in out or "[]" in out, (
|
|
f"Expected no teams: {context.result.output}"
|
|
)
|
|
|
|
|
|
@then(r"the team result should be valid JSON")
|
|
def step_team_valid_json(context: Context) -> None:
|
|
assert context.result.exit_code == 0, (
|
|
f"Expected exit 0, got {context.result.exit_code}: {context.result.output}"
|
|
)
|
|
context.json_result = json.loads(context.result.output)
|
|
|
|
|
|
@then(r'the team result should show active team "(?P<name>[^"]*)"')
|
|
def step_team_active(context: Context, name: str) -> None:
|
|
assert context.result.exit_code == 0
|
|
out = context.result.output
|
|
assert name in out, f"Expected team '{name}' in: {out}"
|
|
|
|
|
|
@then(r"the team result should indicate server not validated")
|
|
def step_team_server_not_validated(context: Context) -> None:
|
|
out = context.result.output.lower()
|
|
assert "false" in out or "not validated" in out or "server mode" in out
|
|
|
|
|
|
@then(r'the team result should show previous team "(?P<name>[^"]*)"')
|
|
def step_team_previous(context: Context, name: str) -> None:
|
|
out = context.result.output
|
|
assert name in out, f"Expected previous team '{name}' in: {out}"
|
|
|
|
|
|
@then(r'retrieving the token should return "(?P<expected>[^"]*)"')
|
|
def step_retrieve_token(context: Context, expected: str) -> None:
|
|
token = context.token_store.get_token()
|
|
assert token == expected, f"Expected {expected!r}, got {token!r}"
|
|
|
|
|
|
@then(r"the token store should report file backend")
|
|
def step_file_backend(context: Context) -> None:
|
|
assert context.token_store.backend == "file"
|
|
|
|
|
|
@then(r"the token store should report no token")
|
|
def step_no_token(context: Context) -> None:
|
|
assert not context.token_store.has_token()
|
|
|
|
|
|
@then(r"a token validation error should be raised")
|
|
def step_token_validation_error(context: Context) -> None:
|
|
assert context.token_error is not None, "Expected ValueError"
|
|
assert isinstance(context.token_error, ValueError)
|
|
|
|
|
|
@then(r"the token store repr should contain REDACTED")
|
|
def step_repr_redacted(context: Context) -> None:
|
|
r = repr(context.token_store)
|
|
assert REDACTED in r, f"Expected REDACTED in repr: {r}"
|
|
|
|
|
|
@then(r'the token store repr should not contain "(?P<value>[^"]*)"')
|
|
def step_repr_no_value(context: Context, value: str) -> None:
|
|
r = repr(context.token_store)
|
|
assert value not in r, f"Value '{value}' should not appear in repr: {r}"
|
|
|
|
|
|
@then(r'the raw CLI output should not contain "(?P<text>[^"]*)"')
|
|
def step_raw_no_text(context: Context, text: str) -> None:
|
|
assert text not in context.raw_output, f"Text '{text}' should not appear in output"
|
|
|
|
|
|
@then(r'the raw CLI output should contain "(?P<text>[^"]*)"')
|
|
def step_raw_has_text(context: Context, text: str) -> None:
|
|
assert text in context.raw_output, (
|
|
f"Expected '{text}' in output: {context.raw_output}"
|
|
)
|