mirror of
https://github.com/agentrhq/authsome.git
synced 2026-09-19 01:34:19 +08:00
Manoj Bajaj
99 lines
3.8 KiB
Python
99 lines
3.8 KiB
Python
"""Tests for `authsome connections inspect`."""
|
|
|
|
# ruff: noqa: PLR2004
|
|
|
|
import json
|
|
|
|
from authsome.cli.main import cli
|
|
from authsome.errors import ConnectionNotFoundError, ProviderNotFoundError
|
|
|
|
|
|
def _make_connection_record() -> dict:
|
|
return {
|
|
"schema_version": 2,
|
|
"provider": "openai",
|
|
"identity": "default",
|
|
"connection_name": "default",
|
|
"auth_type": "api_key",
|
|
"status": "connected",
|
|
"api_key": "sk-secret",
|
|
"access_token": None,
|
|
"refresh_token": None,
|
|
"expires_at": None,
|
|
"scopes": [],
|
|
"token_type": None,
|
|
"base_url": None,
|
|
"api_url": None,
|
|
}
|
|
|
|
|
|
class TestInspectConnectionsCommand:
|
|
"""Tests for `authsome connections inspect <provider>`."""
|
|
|
|
def test_json_output_contains_record_fields(self, runner, mock_client) -> None:
|
|
mock_client.get_provider.return_value = {"name": "openai"}
|
|
mock_client.get_connection.return_value = _make_connection_record()
|
|
|
|
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "openai"])
|
|
assert result.exit_code == 0, result.output
|
|
data = json.loads(result.output)
|
|
assert data["provider"] == "openai"
|
|
assert data["status"] == "connected"
|
|
|
|
def test_sensitive_fields_redacted_by_default(self, runner, mock_client) -> None:
|
|
mock_client.get_provider.return_value = {"name": "openai"}
|
|
mock_client.get_connection.return_value = _make_connection_record()
|
|
|
|
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "openai"])
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.output)
|
|
assert data["api_key"] == "***REDACTED***"
|
|
|
|
def test_field_extraction(self, runner, mock_client) -> None:
|
|
mock_client.get_provider.return_value = {"name": "openai"}
|
|
mock_client.get_connection.return_value = _make_connection_record()
|
|
|
|
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "openai", "--field", "status"])
|
|
assert result.exit_code == 0, result.output
|
|
data = json.loads(result.output)
|
|
assert data["status"] == "connected"
|
|
|
|
def test_field_extraction_json(self, runner, mock_client) -> None:
|
|
mock_client.get_provider.return_value = {"name": "openai"}
|
|
mock_client.get_connection.return_value = _make_connection_record()
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
["--log-file", "", "connections", "inspect", "openai", "--field", "provider"],
|
|
)
|
|
assert result.exit_code == 0
|
|
data = json.loads(result.output)
|
|
assert data == {"provider": "openai", "v": 1}
|
|
|
|
def test_unknown_field_exits_1(self, runner, mock_client) -> None:
|
|
mock_client.get_provider.return_value = {"name": "openai"}
|
|
mock_client.get_connection.return_value = _make_connection_record()
|
|
|
|
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "openai", "--field", "nonexistent"])
|
|
assert result.exit_code == 1
|
|
data = json.loads(result.output)
|
|
assert data["error"] == "FieldNotFound"
|
|
|
|
def test_provider_not_found_exits_4(self, runner, mock_client) -> None:
|
|
mock_client.get_provider.side_effect = ProviderNotFoundError("unknown")
|
|
|
|
result = runner.invoke(cli, ["--log-file", "", "connections", "inspect", "unknown"])
|
|
assert result.exit_code == 4
|
|
|
|
def test_connection_not_found_exits_3(self, runner, mock_client) -> None:
|
|
mock_client.get_provider.return_value = {"name": "openai"}
|
|
mock_client.get_connection.side_effect = ConnectionNotFoundError(
|
|
provider="openai", connection="missing", identity="default"
|
|
)
|
|
|
|
result = runner.invoke(
|
|
cli,
|
|
["--log-file", "", "connections", "inspect", "openai", "--connection", "missing"],
|
|
)
|
|
assert result.exit_code == 3
|