mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
phernandez
8d21a38588
`bm cloud login` only caught SubscriptionRequiredError after the post-login `/proxy/health` subscription check. OAuth succeeds and tokens are saved, but if that check returns anything else — a 5xx while the tenant instance is still provisioning, a 403/401 whose body doesn't match the subscription_required shape, or a transport error — make_api_request raises a generic CloudAPIError that escaped uncaught, dumping a raw httpx.raise_for_status traceback. Users read this as "login failed" even though authentication actually worked. Add a CloudAPIError handler that prints a clean, actionable message and exits non-zero. make_api_request wraps every httpx error (status and transport) in CloudAPIError, so the single handler covers them all. Fixes #863. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: phernandez <paul@basicmachines.co>
304 lines
11 KiB
Python
304 lines
11 KiB
Python
"""Tests for cloud authentication and subscription validation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import asynccontextmanager
|
|
from typing import Any, cast
|
|
|
|
import httpx
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from basic_memory.cli.app import app
|
|
from basic_memory.cli.commands.cloud.api_client import (
|
|
CloudAPIError,
|
|
SubscriptionRequiredError,
|
|
make_api_request,
|
|
)
|
|
|
|
|
|
class _StubAuth:
|
|
def __init__(self, token: str = "test-token", login_ok: bool = True):
|
|
self._token = token
|
|
self._login_ok = login_ok
|
|
|
|
async def get_valid_token(self) -> str:
|
|
return self._token
|
|
|
|
async def login(self) -> bool:
|
|
return self._login_ok
|
|
|
|
|
|
def _auth(auth: _StubAuth) -> Any:
|
|
return cast(Any, auth)
|
|
|
|
|
|
def _make_http_client_factory(handler):
|
|
@asynccontextmanager
|
|
async def _factory():
|
|
transport = httpx.MockTransport(handler)
|
|
async with httpx.AsyncClient(transport=transport) as client:
|
|
yield client
|
|
|
|
return _factory
|
|
|
|
|
|
class TestAPIClientErrorHandling:
|
|
"""Tests for API client error handling."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_subscription_required_error(self):
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
403,
|
|
json={
|
|
"detail": {
|
|
"error": "subscription_required",
|
|
"message": "Active subscription required for CLI access",
|
|
"subscribe_url": "https://basicmemory.com/subscribe",
|
|
}
|
|
},
|
|
request=request,
|
|
)
|
|
|
|
auth = _StubAuth()
|
|
with pytest.raises(SubscriptionRequiredError) as exc_info:
|
|
await make_api_request(
|
|
"GET",
|
|
"https://test.com/api/endpoint",
|
|
auth=_auth(auth),
|
|
http_client_factory=_make_http_client_factory(handler),
|
|
)
|
|
|
|
err = exc_info.value
|
|
assert err.status_code == 403
|
|
assert err.subscribe_url == "https://basicmemory.com/subscribe"
|
|
assert "Active subscription required" in str(err)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_subscription_required_error_flat_format(self):
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
403,
|
|
json={
|
|
"error": "subscription_required",
|
|
"message": "Active subscription required",
|
|
"subscribe_url": "https://basicmemory.com/subscribe",
|
|
},
|
|
request=request,
|
|
)
|
|
|
|
auth = _StubAuth()
|
|
with pytest.raises(SubscriptionRequiredError) as exc_info:
|
|
await make_api_request(
|
|
"GET",
|
|
"https://test.com/api/endpoint",
|
|
auth=_auth(auth),
|
|
http_client_factory=_make_http_client_factory(handler),
|
|
)
|
|
|
|
err = exc_info.value
|
|
assert err.status_code == 403
|
|
assert err.subscribe_url == "https://basicmemory.com/subscribe"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_generic_403_error(self):
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
403,
|
|
json={"error": "forbidden", "message": "Access denied"},
|
|
request=request,
|
|
)
|
|
|
|
auth = _StubAuth()
|
|
with pytest.raises(CloudAPIError) as exc_info:
|
|
await make_api_request(
|
|
"GET",
|
|
"https://test.com/api/endpoint",
|
|
auth=_auth(auth),
|
|
http_client_factory=_make_http_client_factory(handler),
|
|
)
|
|
|
|
err = exc_info.value
|
|
assert not isinstance(err, SubscriptionRequiredError)
|
|
assert err.status_code == 403
|
|
|
|
|
|
class TestLoginCommand:
|
|
"""Tests for cloud login command with subscription validation."""
|
|
|
|
def test_login_without_subscription_shows_error(self, monkeypatch):
|
|
runner = CliRunner()
|
|
|
|
# Stub auth object returned by CLIAuth(...)
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.CLIAuth",
|
|
lambda **_kwargs: _StubAuth(login_ok=True),
|
|
)
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.get_cloud_config",
|
|
lambda: ("client_id", "domain", "https://cloud.example.com"),
|
|
)
|
|
|
|
async def fake_make_api_request(*_args, **_kwargs):
|
|
raise SubscriptionRequiredError(
|
|
message="Active subscription required for CLI access",
|
|
subscribe_url="https://basicmemory.com/subscribe",
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.make_api_request",
|
|
fake_make_api_request,
|
|
)
|
|
|
|
result = runner.invoke(app, ["cloud", "login"])
|
|
assert result.exit_code == 1
|
|
assert "Subscription Required" in result.stdout
|
|
assert "Active subscription required" in result.stdout
|
|
assert "https://basicmemory.com/subscribe" in result.stdout
|
|
assert "bm cloud login" in result.stdout
|
|
|
|
def test_login_with_subscription_succeeds(self, monkeypatch):
|
|
runner = CliRunner()
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.CLIAuth",
|
|
lambda **_kwargs: _StubAuth(login_ok=True),
|
|
)
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.get_cloud_config",
|
|
lambda: ("client_id", "domain", "https://cloud.example.com"),
|
|
)
|
|
|
|
async def fake_make_api_request(*_args, **_kwargs):
|
|
# Response is only used for status validation in login().
|
|
return httpx.Response(200, json={"status": "healthy"})
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.make_api_request",
|
|
fake_make_api_request,
|
|
)
|
|
|
|
result = runner.invoke(app, ["cloud", "login"])
|
|
assert result.exit_code == 0
|
|
assert "Cloud authentication successful" in result.stdout
|
|
assert "Cloud host ready: https://cloud.example.com" in result.stdout
|
|
|
|
def test_login_health_check_error_shows_clean_message(self, monkeypatch):
|
|
"""Regression for #863: a non-subscription error from the post-login
|
|
/proxy/health check must produce a clean message, not a raw traceback.
|
|
|
|
OAuth has already succeeded at this point; the tenant instance may still
|
|
be provisioning (5xx) or return some other non-subscription_required error.
|
|
"""
|
|
runner = CliRunner()
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.CLIAuth",
|
|
lambda **_kwargs: _StubAuth(login_ok=True),
|
|
)
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.get_cloud_config",
|
|
lambda: ("client_id", "domain", "https://cloud.example.com"),
|
|
)
|
|
|
|
async def fake_make_api_request(*_args, **_kwargs):
|
|
# e.g. tenant instance not ready yet -> proxy returns 503
|
|
raise CloudAPIError("API request failed: 503 Service Unavailable", status_code=503)
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.make_api_request",
|
|
fake_make_api_request,
|
|
)
|
|
|
|
result = runner.invoke(app, ["cloud", "login"])
|
|
# Clean exit, no traceback leaking the exception class.
|
|
assert result.exit_code == 1
|
|
assert result.exception is None or isinstance(result.exception, SystemExit)
|
|
# Collapse Rich's line-wrapping before matching multi-word phrases.
|
|
output = " ".join(result.stdout.split())
|
|
assert "couldn't verify cloud access" in output
|
|
assert "bm cloud status" in output
|
|
assert "Traceback" not in output
|
|
|
|
def test_login_authentication_failure(self, monkeypatch):
|
|
runner = CliRunner()
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.CLIAuth",
|
|
lambda **_kwargs: _StubAuth(login_ok=False),
|
|
)
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.get_cloud_config",
|
|
lambda: ("client_id", "domain", "https://cloud.example.com"),
|
|
)
|
|
|
|
result = runner.invoke(app, ["cloud", "login"])
|
|
assert result.exit_code == 1
|
|
assert "Login failed" in result.stdout
|
|
|
|
|
|
class TestLogoutCommand:
|
|
"""Tests for `bm cloud logout`."""
|
|
|
|
@staticmethod
|
|
def _patch(monkeypatch, default_workspace):
|
|
class FakeConfig:
|
|
cloud_client_id = "cid"
|
|
cloud_domain = "https://auth.example.com"
|
|
|
|
def __init__(self):
|
|
self.default_workspace = default_workspace
|
|
|
|
saved: list[FakeConfig] = []
|
|
config_instance = FakeConfig()
|
|
logout_called = {"value": False}
|
|
|
|
class FakeConfigManager:
|
|
config = config_instance
|
|
|
|
def save_config(self, cfg):
|
|
saved.append(cfg)
|
|
|
|
class FakeAuth:
|
|
def __init__(self, **_kwargs):
|
|
pass
|
|
|
|
def logout(self):
|
|
logout_called["value"] = True
|
|
|
|
monkeypatch.setattr(
|
|
"basic_memory.cli.commands.cloud.core_commands.ConfigManager", FakeConfigManager
|
|
)
|
|
monkeypatch.setattr("basic_memory.cli.commands.cloud.core_commands.CLIAuth", FakeAuth)
|
|
return config_instance, saved, logout_called
|
|
|
|
def test_logout_clears_default_workspace(self, monkeypatch):
|
|
"""Regression for #755: logout must invalidate the cached workspace."""
|
|
config_instance, saved, logout_called = self._patch(
|
|
monkeypatch, default_workspace="tenant-org-123"
|
|
)
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(app, ["cloud", "logout"])
|
|
|
|
assert result.exit_code == 0
|
|
assert logout_called["value"] is True
|
|
assert config_instance.default_workspace is None
|
|
# Save was called once because a non-None value needed clearing.
|
|
assert len(saved) == 1
|
|
assert saved[0].default_workspace is None
|
|
|
|
def test_logout_skips_save_when_no_default_workspace(self, monkeypatch):
|
|
"""If nothing was cached, logout shouldn't rewrite the config file."""
|
|
config_instance, saved, logout_called = self._patch(monkeypatch, default_workspace=None)
|
|
runner = CliRunner()
|
|
|
|
result = runner.invoke(app, ["cloud", "logout"])
|
|
|
|
assert result.exit_code == 0
|
|
assert logout_called["value"] is True
|
|
assert config_instance.default_workspace is None
|
|
# No save: avoid touching the file when there's nothing to clear.
|
|
assert saved == []
|