Fix/tests (#38)

* refactor: unify API key flow logic, update authentication tests to use bridge, and add CI workflow badge.

* refactor: expand test coverage and add Codecov badge to README

* refactor: remove unused TokenExpiredError alias from client tests

* chore: remove ty type checking from CI workflow

* feat: add type checking, improve token refresh error handling, and enable HTTP server port reuse in tests

* refactor: add whitespace to conftest for improved readability

* fix: ensure HTTPServer cleanup via finally blocks and improved test fixture teardown

* style: reorder imports in conftest.py to follow PEP 8 standards

* refactor: clean up whitespace and formatting in test server cleanup fixture

* refactor: enable address reuse directly in PKCE flow handlers and remove global test configuration

* feat: make callback port configurable in PKCE and DCR PKCE flows and update tests to use dynamic ports

* test: dynamic callback port assignment in PKCE and DCR PKCE flow tests
This commit is contained in:
Tejas
2026-04-22 16:14:39 +05:30
committed by GitHub
parent b29201776b
commit 676608f733
23 changed files with 2699 additions and 284 deletions
+44
View File
@@ -0,0 +1,44 @@
name: Tests
on:
push:
branches:
- '**'
pull_request:
branches:
- main
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.13"
cache: "pip"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e ".[dev]"
- name: Lint with ruff
run: |
ruff check src/ tests/
ruff format --check src/ tests/
- name: Type check with ty
run: ty check src/
- name: Run tests with coverage
run: pytest --cov=authsome --cov-report=xml --cov-report=term -p no:xdist
- name: Upload coverage reports to Codecov
uses: codecov/codecov-action@v4
with:
token: ${{ secrets.CODECOV_TOKEN }}
file: ./coverage.xml
fail_ci_if_error: false
+2
View File
@@ -4,6 +4,8 @@
[![Python 3.13+](https://img.shields.io/pypi/pyversions/authsome.svg)](https://pypi.org/project/authsome/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![PyPI downloads](https://img.shields.io/pypi/dm/authsome.svg)](https://pypi.org/project/authsome/)
[![Tests](https://github.com/manojbajaj95/authsome/actions/workflows/test.yml/badge.svg)](https://github.com/manojbajaj95/authsome/actions/workflows/test.yml)
[![codecov](https://codecov.io/gh/manojbajaj95/authsome/branch/main/graph/badge.svg)](https://codecov.io/gh/manojbajaj95/authsome)
```text
__ __
+1
View File
@@ -36,6 +36,7 @@ dev = [
"pytest>=7.0",
"pytest-cov>=4.0",
"ruff>=0.9",
"ty",
]
[project.scripts]
+30 -18
View File
@@ -28,7 +28,6 @@ from authsome.crypto.base import CryptoBackend
from authsome.crypto.keyring_crypto import KeyringCryptoBackend
from authsome.crypto.local_file_crypto import LocalFileCryptoBackend
from authsome.errors import (
ProviderNotFoundError,
AuthsomeError,
ConnectionNotFoundError,
CredentialMissingError,
@@ -342,9 +341,11 @@ class AuthClient:
# Fetch client credentials
client_record = self.get_provider_client_credentials(provider, profile_name)
flow_client_id = client_record.client_id if client_record else None
flow_client_secret = self.crypto.decrypt(client_record.client_secret) if client_record and client_record.client_secret else None
flow_client_secret = (
self.crypto.decrypt(client_record.client_secret) if client_record and client_record.client_secret else None
)
flow_api_key = None
# Secure Bridge Prompts for missing interactive inputs
@@ -358,19 +359,27 @@ class AuthClient:
}
)
missing_fields.append({"name": "client_id", "label": "Client ID", "type": "text"})
missing_fields.append({"name": "client_secret", "label": "Client Secret (Optional)", "type": "password", "required": False})
missing_fields.append(
{
"name": "client_secret",
"label": "Client Secret (Optional)",
"type": "password",
"required": False,
}
)
elif flow_type == FlowType.API_KEY and not flow_api_key:
missing_fields.append({"name": "api_key", "label": "API Key", "type": "password"})
if missing_fields:
from authsome.flows.bridge import secure_input_bridge
title = f"{definition.display_name} Credentials"
inputs = secure_input_bridge(title, missing_fields)
if flow_type in (FlowType.PKCE, FlowType.DEVICE_CODE):
flow_client_id = inputs.get("client_id")
secret_input = inputs.get("client_secret")
if client_record is None:
client_record = ProviderClientRecord(
profile=profile_name,
@@ -403,14 +412,15 @@ class AuthClient:
provider=provider,
)
client_record.client_id = record.metadata.pop("_dcr_client_id")
dcr_secret_dict = record.metadata.pop("_dcr_client_secret", None)
if dcr_secret_dict:
from authsome.crypto.base import EncryptedField
client_record.client_secret = EncryptedField(**dcr_secret_dict)
else:
client_record.client_secret = None
self._save_provider_client_credentials(client_record)
# Persist the connection record
@@ -565,10 +575,10 @@ class AuthClient:
"""
profile_name = profile or self.active_profile
store = self._get_store(profile_name)
# Ensure provider exists
self.get_provider(provider)
# 1. Log out of all connections for this provider to remotely revoke them
meta_key = build_store_key(
profile=profile_name,
@@ -581,10 +591,10 @@ class AuthClient:
# Make a copy of the list as logout will modify it
for conn_name in list(metadata.connection_names):
self.logout(provider, connection=conn_name, profile=profile_name)
# 2. Delete ProviderMetadataRecord
store.delete(meta_key)
# 3. Delete ProviderClientRecord
client_key = build_store_key(
profile=profile_name,
@@ -592,7 +602,7 @@ class AuthClient:
record_type="client",
)
store.delete(client_key)
logger.info("Revoked all credentials for provider=%s in profile=%s", provider, profile_name)
def remove(
@@ -605,10 +615,10 @@ class AuthClient:
Removes all local credential state and deletes the provider JSON file if it exists locally.
"""
profile_name = profile or self.active_profile
# Revoke all credentials (cleans up connections and client secrets)
self.revoke(provider, profile=profile_name)
# Delete the provider definition JSON if it's a local/custom provider
local_path = self._home / "providers" / f"{provider}.json"
if local_path.exists():
@@ -687,7 +697,7 @@ class AuthClient:
if "=" in line:
key, value = line.split("=", 1)
env[key] = value
def _dquote(s: str) -> str:
"""Double-quote a token so $VAR references are expanded by the shell."""
s = s.replace("\\", "\\\\")
@@ -964,7 +974,9 @@ class AuthClient:
if record.refresh_token:
try:
refreshed = self._refresh_token(record, provider)
return self.crypto.decrypt(refreshed.access_token) # type: ignore[arg-type]
if refreshed.access_token is None:
raise RefreshFailedError("Refreshed record missing access token", provider=provider)
return self.crypto.decrypt(refreshed.access_token)
except RefreshFailedError:
# Check if the token hasn't actually expired yet
if now < record.expires_at:
@@ -1008,7 +1020,7 @@ class AuthClient:
client_record = self.get_provider_client_credentials(provider_name, record.profile)
client_id = None
client_secret = None
if client_record:
client_id = client_record.client_id
if client_record.client_secret:
-3
View File
@@ -6,7 +6,6 @@ Spec §13.5: API Key Environment Import — read from environment variable, stor
from __future__ import annotations
import logging
from authsome.crypto.base import CryptoBackend
@@ -82,5 +81,3 @@ class ApiKeyFlow(AuthFlow):
account=AccountInfo(),
metadata={},
)
+26 -22
View File
@@ -5,17 +5,18 @@ web browser instead of the terminal. This prevents secrets from being exposed
in environments where agents or scripts might intercept standard I/O.
"""
from html import escape
import http.server
import logging
import socket
import threading
import urllib.parse
import webbrowser
from html import escape
from typing import Any
logger = logging.getLogger(__name__)
def _find_free_port() -> int:
"""Find a free TCP port on localhost."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
@@ -25,11 +26,11 @@ def _find_free_port() -> int:
class _BridgeHandler(http.server.BaseHTTPRequestHandler):
"""HTTP handler that renders a form and captures input."""
title: str = "Secure Input"
fields: list[dict[str, Any]] = []
result: dict[str, str] | None = None
def do_GET(self) -> None:
"""Serve the HTML form."""
html = [
@@ -38,18 +39,20 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler):
"<style>",
"body { font-family: system-ui, sans-serif; max-width: 400px; margin: 40px auto; padding: 20px; }",
"label { display: block; margin-bottom: 8px; font-weight: bold; }",
"input { width: 100%; padding: 8px; margin-bottom: 16px; border: 1px solid #ccc; border-radius: 4px; box-sizing: border-box; }",
"input { width: 100%; padding: 8px; margin-bottom: 16px; border: 1px solid #ccc; ",
"border-radius: 4px; box-sizing: border-box; }",
".static-wrap { display: flex; gap: 8px; margin-bottom: 16px; align-items: center; }",
".static-wrap input[readonly] { margin-bottom: 0; flex: 1; background: #f5f5f5; cursor: default; }",
"button { width: 100%; padding: 10px; background-color: #0066cc; color: white; border: none; border-radius: 4px; cursor: pointer; font-size: 16px; }",
"button { width: 100%; padding: 10px; background-color: #0066cc; color: white; border: none; ",
"border-radius: 4px; cursor: pointer; font-size: 16px; }",
"button:hover { background-color: #0052a3; }",
"button.copybtn { width: auto; padding: 8px 12px; font-size: 14px; flex-shrink: 0; }",
"</style>",
"</head><body>",
f"<h2>{self.title}</h2>",
"<form method='POST'>"
"<form method='POST'>",
]
for field in self.fields:
label = field["label"]
if field.get("type") == "static":
@@ -60,7 +63,7 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler):
"<div class='static-wrap'>"
f"<input type='text' readonly value='{val_esc}' aria-readonly='true'>"
"<button type='button' class='copybtn' "
"onclick=\"navigator.clipboard.writeText(this.previousElementSibling.value)\">"
'onclick="navigator.clipboard.writeText(this.previousElementSibling.value)">'
"Copy</button></div>"
)
continue
@@ -69,10 +72,10 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler):
required = "required" if field.get("required", True) else ""
html.append(f"<label for='{name}'>{label}</label>")
html.append(f"<input type='{input_type}' id='{name}' name='{name}' {required}>")
html.append("<button type='submit'>Submit Securely</button>")
html.append("</form></body></html>")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
@@ -83,10 +86,10 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler):
content_length = int(self.headers.get("Content-Length", 0))
post_data = self.rfile.read(content_length).decode("utf-8")
parsed = urllib.parse.parse_qs(post_data)
# Flatten the parse_qs output (which returns lists)
_BridgeHandler.result = {k: v[0] for k, v in parsed.items() if v}
# Send success page
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
@@ -98,10 +101,11 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler):
"</body></html>"
)
self.wfile.write(success_html.encode("utf-8"))
# Notify that we are done
def kill_server():
self.server.shutdown()
threading.Thread(target=kill_server, daemon=True).start()
def log_message(self, format: str, *args: Any) -> None:
@@ -111,35 +115,35 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler):
def secure_input_bridge(title: str, fields: list[dict[str, Any]]) -> dict[str, str]:
"""
Start a local server and open the browser to collect sensitive inputs securely.
Args:
title: The heading to display on the form.
fields: A list of dicts, each representing an input field.
Expected keys: 'name', 'label', 'type' (default: text), 'required' (default: True).
Returns:
A dictionary mapping field names to user input values.
"""
port = _find_free_port()
# Reset state
_BridgeHandler.title = title
_BridgeHandler.fields = fields
_BridgeHandler.result = None
server = http.server.HTTPServer(("127.0.0.1", port), _BridgeHandler)
server_thread = threading.Thread(target=server.serve_forever, daemon=True)
server_thread.start()
url = f"http://127.0.0.1:{port}"
print(f"\nRequires secure input for: {title}")
print(f"Opening browser for secure input...\nIf the browser doesn't open, visit:\n{url}\n")
webbrowser.open(url)
# Wait for the server to shutdown (triggered by do_POST)
server_thread.join(timeout=300) # 5 minute timeout
server_thread.join(timeout=300) # 5 minute timeout
if _BridgeHandler.result is None:
raise RuntimeError("Secure input bridge timed out or was cancelled.")
return _BridgeHandler.result
+26 -22
View File
@@ -109,6 +109,8 @@ class DcrPkceFlow(AuthFlow):
DCR registration endpoint is discovered via .well-known if not provided.
"""
callback_port: int = 7999
def authenticate(
self,
provider: ProviderDefinition,
@@ -139,7 +141,7 @@ class DcrPkceFlow(AuthFlow):
code_verifier, code_challenge = _generate_pkce()
# Start local callback server
port = 7999
port = self.callback_port
redirect_uri = f"http://127.0.0.1:{port}/callback"
# Reset handler state
@@ -151,30 +153,32 @@ class DcrPkceFlow(AuthFlow):
server_thread = threading.Thread(target=server.handle_request, daemon=True)
server_thread.start()
# Build authorization URL
state = secrets.token_urlsafe(32)
auth_params: dict[str, str] = {
"response_type": "code",
"client_id": client_id,
"redirect_uri": redirect_uri,
"state": state,
"code_challenge": code_challenge,
"code_challenge_method": "S256",
}
if effective_scopes:
auth_params["scope"] = " ".join(effective_scopes)
try:
# Build authorization URL
state = secrets.token_urlsafe(32)
auth_params: dict[str, str] = {
"response_type": "code",
"client_id": client_id,
"redirect_uri": redirect_uri,
"state": state,
"code_challenge": code_challenge,
"code_challenge_method": "S256",
}
if effective_scopes:
auth_params["scope"] = " ".join(effective_scopes)
auth_url = f"{provider.oauth.authorization_url}?{urllib.parse.urlencode(auth_params)}"
auth_url = f"{provider.oauth.authorization_url}?{urllib.parse.urlencode(auth_params)}"
logger.info("Opening browser for authorization...")
logger.debug("Authorization URL: %s", auth_url)
print(f"\nOpening browser for {provider.display_name} authorization...")
print(f"If the browser doesn't open, visit:\n{auth_url}\n")
webbrowser.open(auth_url)
logger.info("Opening browser for authorization...")
logger.debug("Authorization URL: %s", auth_url)
print(f"\nOpening browser for {provider.display_name} authorization...")
print(f"If the browser doesn't open, visit:\n{auth_url}\n")
webbrowser.open(auth_url)
# Wait for callback
server_thread.join(timeout=_CALLBACK_TIMEOUT_SECONDS)
server.server_close()
# Wait for callback
server_thread.join(timeout=_CALLBACK_TIMEOUT_SECONDS)
finally:
server.server_close()
if _CallbackHandler.error:
raise AuthenticationFailedError(
+25 -21
View File
@@ -104,6 +104,8 @@ class PkceFlow(AuthFlow):
`client` section (either literal or env:-prefixed).
"""
callback_port: int = 7999
def authenticate(
self,
provider: ProviderDefinition,
@@ -135,7 +137,7 @@ class PkceFlow(AuthFlow):
code_verifier, code_challenge = _generate_pkce()
# Start local callback server
port = 7999
port = self.callback_port
redirect_uri = f"http://127.0.0.1:{port}/callback"
# Reset handler state
@@ -147,29 +149,31 @@ class PkceFlow(AuthFlow):
server_thread = threading.Thread(target=server.handle_request, daemon=True)
server_thread.start()
# Build authorization URL
state = secrets.token_urlsafe(32)
auth_params: dict[str, str] = {
"response_type": "code",
"client_id": client_id,
"redirect_uri": redirect_uri,
"state": state,
"code_challenge": code_challenge,
"code_challenge_method": "S256",
}
if effective_scopes:
auth_params["scope"] = " ".join(effective_scopes)
try:
# Build authorization URL
state = secrets.token_urlsafe(32)
auth_params: dict[str, str] = {
"response_type": "code",
"client_id": client_id,
"redirect_uri": redirect_uri,
"state": state,
"code_challenge": code_challenge,
"code_challenge_method": "S256",
}
if effective_scopes:
auth_params["scope"] = " ".join(effective_scopes)
auth_url = f"{provider.oauth.authorization_url}?{urllib.parse.urlencode(auth_params)}"
auth_url = f"{provider.oauth.authorization_url}?{urllib.parse.urlencode(auth_params)}"
logger.info("Opening browser for authorization...")
print(f"\nOpening browser for {provider.display_name} authorization...")
print(f"If the browser doesn't open, visit:\n{auth_url}\n")
webbrowser.open(auth_url)
logger.info("Opening browser for authorization...")
print(f"\nOpening browser for {provider.display_name} authorization...")
print(f"If the browser doesn't open, visit:\n{auth_url}\n")
webbrowser.open(auth_url)
# Wait for callback
server_thread.join(timeout=_CALLBACK_TIMEOUT_SECONDS)
server.server_close()
# Wait for callback
server_thread.join(timeout=_CALLBACK_TIMEOUT_SECONDS)
finally:
server.server_close()
if _CallbackHandler.error:
raise AuthenticationFailedError(
-1
View File
@@ -117,4 +117,3 @@ class ProviderClientRecord(BaseModel):
metadata: dict[str, Any] = Field(default_factory=dict)
model_config = {"extra": "allow"}
+65
View File
@@ -0,0 +1,65 @@
"""Tests for the secure browser bridge."""
import urllib.parse
import urllib.request
from unittest.mock import patch
import pytest
from authsome.flows.bridge import _BridgeHandler, _find_free_port, secure_input_bridge
def test_find_free_port():
port = _find_free_port()
assert isinstance(port, int)
assert port > 0
def test_secure_input_bridge_success():
fields = [
{"name": "api_key", "label": "API Key", "type": "password"},
{"name": "username", "label": "Username", "required": False},
]
def mock_open(url):
# 1. Test GET
req = urllib.request.Request(url)
with urllib.request.urlopen(req) as response:
assert response.status == 200
html = response.read().decode("utf-8")
assert "Test Auth" in html
assert "API Key" in html
assert "Username" in html
# 2. Test POST
data = urllib.parse.urlencode({"api_key": "secret123", "username": "testuser"}).encode("utf-8")
req_post = urllib.request.Request(url, data=data, method="POST")
with urllib.request.urlopen(req_post) as response:
assert response.status == 200
html = response.read().decode("utf-8")
assert "Success!" in html
with patch("authsome.flows.bridge.webbrowser.open", side_effect=mock_open):
res = secure_input_bridge("Test Auth", fields)
assert res == {"api_key": "secret123", "username": "testuser"}
def test_secure_input_bridge_timeout():
fields = [{"name": "key", "label": "Key"}]
def mock_open(url):
# do nothing, simulate user ignoring the browser
pass
with patch("authsome.flows.bridge.webbrowser.open", side_effect=mock_open):
# To avoid waiting 300s, we mock Thread.join
with patch("threading.Thread.join"):
with pytest.raises(RuntimeError, match="timed out or was cancelled"):
secure_input_bridge("Test Auth", fields)
def test_log_message():
# Test _BridgeHandler.log_message does not crash
handler = _BridgeHandler.__new__(_BridgeHandler)
handler.log_message("test %s", "arg")
+404
View File
@@ -0,0 +1,404 @@
"""Tests for the Authsome CLI."""
import json
from unittest.mock import MagicMock, patch
import pytest
from click.testing import CliRunner
from authsome.cli import cli
from authsome.errors import AuthsomeError, ProviderNotFoundError, StoreUnavailableError
@pytest.fixture
def runner():
return CliRunner()
@pytest.fixture
def mock_client():
with patch("authsome.cli.AuthClient") as mock:
client = MagicMock()
mock.return_value = client
yield client
def test_init_command(runner, mock_client):
mock_client.home = "/mock/home"
result = runner.invoke(cli, ["init"])
assert result.exit_code == 0
assert "Initialized authsome at /mock/home" in result.output
mock_client.init.assert_called_once()
def test_init_json(runner, mock_client):
mock_client.home = "/mock/home"
result = runner.invoke(cli, ["init", "--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert data["status"] == "initialized"
def test_list_command(runner, mock_client):
mock_client.list_connections.return_value = [
{
"name": "openai",
"connections": [
{
"connection_name": "default",
"status": "connected",
"auth_type": "api_key",
"scopes": ["read"],
"expires_at": "2030",
}
],
}
]
mock_provider = MagicMock()
mock_provider.name = "openai"
mock_provider.display_name = "OpenAI"
mock_provider.auth_type.value = "api_key"
mock_client.list_providers_by_source.return_value = {"bundled": [mock_provider], "custom": []}
result = runner.invoke(cli, ["list"])
assert result.exit_code == 0
assert "Bundled Providers:" in result.output
assert "OpenAI" in result.output
assert "connected" in result.output
def test_list_command_no_connections_and_missing_scopes(runner, mock_client):
mock_client.list_connections.return_value = [
{
"name": "openai",
"connections": [{"connection_name": "default", "status": "connected", "auth_type": "api_key"}],
}
]
mock_provider = MagicMock()
mock_provider.name = "openai"
mock_provider.display_name = "OpenAI"
mock_provider.auth_type.value = "api_key"
mock_provider_empty = MagicMock()
mock_provider_empty.name = "github"
mock_provider_empty.display_name = "GitHub"
mock_provider_empty.auth_type.value = "oauth2"
mock_client.list_providers_by_source.return_value = {"bundled": [mock_provider, mock_provider_empty], "custom": []}
result = runner.invoke(cli, ["list"])
assert result.exit_code == 0
assert "(no connections)" in result.output
def test_list_json(runner, mock_client):
mock_client.list_connections.return_value = []
mock_client.list_providers_by_source.return_value = {"bundled": [], "custom": []}
result = runner.invoke(cli, ["list", "--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert "bundled" in data
def test_login_command(runner, mock_client):
mock_record = MagicMock()
mock_record.status.value = "connected"
mock_client.login.return_value = mock_record
result = runner.invoke(cli, ["login", "openai"])
assert result.exit_code == 0
assert "Successfully logged in to openai" in result.output
def test_login_json(runner, mock_client):
mock_record = MagicMock()
mock_record.status.value = "connected"
mock_client.login.return_value = mock_record
result = runner.invoke(cli, ["login", "openai", "--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert data["status"] == "success"
def test_login_error(runner, mock_client):
mock_client.login.side_effect = AuthsomeError("Already exists")
result = runner.invoke(cli, ["login", "openai"])
assert result.exit_code == 1
assert "Error: Already exists" in result.output
def test_error_mapping(runner, mock_client):
from authsome.errors import AuthenticationFailedError, CredentialMissingError, RefreshFailedError
mock_client.login.side_effect = ProviderNotFoundError("test")
result = runner.invoke(cli, ["login", "test"])
assert result.exit_code == 3
mock_client.login.side_effect = StoreUnavailableError("locked")
result = runner.invoke(cli, ["login", "test"])
assert result.exit_code == 7
mock_client.login.side_effect = AuthenticationFailedError("fail")
result = runner.invoke(cli, ["login", "test"])
assert result.exit_code == 4
mock_client.login.side_effect = CredentialMissingError("missing", provider="test")
result = runner.invoke(cli, ["login", "test"])
assert result.exit_code == 5
mock_client.login.side_effect = RefreshFailedError("refresh fail", provider="test")
result = runner.invoke(cli, ["login", "test"])
assert result.exit_code == 6
mock_client.login.side_effect = Exception("unknown")
result = runner.invoke(cli, ["login", "test"])
assert result.exit_code == 1
def test_logout(runner, mock_client):
result = runner.invoke(cli, ["logout", "openai"])
assert result.exit_code == 0
mock_client.logout.assert_called_with("openai", "default")
def test_logout_json(runner, mock_client):
result = runner.invoke(cli, ["logout", "openai", "--json"])
assert result.exit_code == 0
def test_revoke(runner, mock_client):
result = runner.invoke(cli, ["revoke", "openai"])
assert result.exit_code == 0
mock_client.revoke.assert_called_with("openai")
def test_revoke_json(runner, mock_client):
result = runner.invoke(cli, ["revoke", "openai", "--json"])
assert result.exit_code == 0
def test_remove(runner, mock_client):
result = runner.invoke(cli, ["remove", "openai"])
assert result.exit_code == 0
mock_client.remove.assert_called_with("openai")
def test_remove_json(runner, mock_client):
result = runner.invoke(cli, ["remove", "openai", "--json"])
assert result.exit_code == 0
def test_get(runner, mock_client):
mock_record = MagicMock()
mock_record.model_dump.return_value = {"status": "connected", "access_token": "secret"}
mock_client.crypto.decrypt.return_value = "decrypted_secret"
mock_client.get_connection.return_value = mock_record
result = runner.invoke(cli, ["get", "openai"])
assert result.exit_code == 0
assert "***REDACTED***" in result.output
def test_get_show_secret(runner, mock_client):
mock_record = MagicMock()
mock_record.model_dump.return_value = {"status": "connected", "access_token": "secret"}
mock_record.access_token = "encrypted_secret"
mock_client.crypto.decrypt.return_value = "decrypted_secret"
mock_client.get_connection.return_value = mock_record
result = runner.invoke(cli, ["get", "openai", "--show-secret"])
assert result.exit_code == 0
assert "decrypted_secret" in result.output
def test_get_field(runner, mock_client):
mock_record = MagicMock()
mock_record.model_dump.return_value = {"status": "connected"}
mock_client.get_connection.return_value = mock_record
result = runner.invoke(cli, ["get", "openai", "--field", "status"])
assert result.exit_code == 0
assert "connected" in result.output
def test_get_missing_field(runner, mock_client):
mock_record = MagicMock()
mock_record.model_dump.return_value = {"status": "connected"}
mock_client.get_connection.return_value = mock_record
result = runner.invoke(cli, ["get", "openai", "--field", "missing"])
assert result.exit_code == 1
def test_get_json(runner, mock_client):
mock_record = MagicMock()
mock_record.model_dump.return_value = {"status": "connected"}
mock_client.get_connection.return_value = mock_record
result = runner.invoke(cli, ["get", "openai", "--json"])
assert result.exit_code == 0
def test_get_field_json(runner, mock_client):
mock_record = MagicMock()
mock_record.model_dump.return_value = {"status": "connected"}
mock_client.get_connection.return_value = mock_record
result = runner.invoke(cli, ["get", "openai", "--json", "--field", "status"])
assert result.exit_code == 0
def test_inspect(runner, mock_client):
mock_def = MagicMock()
mock_def.model_dump.return_value = {"name": "openai"}
mock_client.get_provider.return_value = mock_def
result = runner.invoke(cli, ["inspect", "openai"])
assert result.exit_code == 0
assert '"name": "openai"' in result.output
def test_inspect_json(runner, mock_client):
mock_def = MagicMock()
mock_def.model_dump.return_value = {"name": "openai"}
mock_client.get_provider.return_value = mock_def
result = runner.invoke(cli, ["inspect", "openai", "--json"])
assert result.exit_code == 0
def test_export(runner, mock_client):
mock_client.export.return_value = "export VAR=1"
result = runner.invoke(cli, ["export", "openai", "--format", "shell"])
assert result.exit_code == 0
assert "export VAR=1" in result.output
# test empty output
mock_client.export.return_value = ""
result2 = runner.invoke(cli, ["export", "openai", "--format", "shell"])
assert result2.exit_code == 0
assert result2.output == ""
def test_run(runner, mock_client):
mock_run_result = MagicMock()
mock_run_result.returncode = 0
mock_client.run.return_value = mock_run_result
result = runner.invoke(cli, ["run", "--provider", "openai", "echo", "hello"])
assert result.exit_code == 0
mock_client.run.assert_called_with(["echo", "hello"], providers=["openai"])
def test_register_file_not_found(runner):
result = runner.invoke(cli, ["register", "nonexistent.json"])
assert result.exit_code == 1
assert "File not found" in result.output
def test_register_success(runner, mock_client, tmp_path):
f = tmp_path / "test.json"
f.write_text(
json.dumps(
{
"name": "test",
"display_name": "Test",
"auth_type": "api_key",
"flow": "api_key",
"api_key": {"header_name": "Auth"},
}
)
)
result = runner.invoke(cli, ["register", str(f)])
assert result.exit_code == 0
mock_client.register_provider.assert_called_once()
def test_register_success_json(runner, mock_client, tmp_path):
f = tmp_path / "test.json"
f.write_text(
json.dumps(
{
"name": "test",
"display_name": "Test",
"auth_type": "api_key",
"flow": "api_key",
"api_key": {"header_name": "Auth"},
}
)
)
result = runner.invoke(cli, ["register", str(f), "--json"])
assert result.exit_code == 0
def test_register_bad_json(runner, mock_client, tmp_path):
f = tmp_path / "bad.json"
f.write_text("{bad")
result = runner.invoke(cli, ["register", str(f)])
assert result.exit_code == 1
def test_whoami(runner, mock_client):
mock_client.home = "/mock/home"
mock_client.config.encryption.mode = "keyring"
result = runner.invoke(cli, ["whoami"])
assert result.exit_code == 0
assert "/mock/home" in result.output
def test_whoami_no_encryption_config(runner, mock_client):
mock_client.home = "/mock/home"
mock_client.config.encryption = None
result = runner.invoke(cli, ["whoami", "--json"])
assert result.exit_code == 0
def test_doctor(runner, mock_client):
mock_client.doctor.return_value = {
"home_exists": True,
"encryption": False,
"issues": ["error"],
"providers_count": 0,
}
result = runner.invoke(cli, ["doctor"])
assert result.exit_code == 1
assert "FAIL" in result.output
def test_doctor_json(runner, mock_client):
mock_client.doctor.return_value = {"home_exists": True, "encryption": True, "issues": []}
result = runner.invoke(cli, ["doctor", "--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert data["home_exists"] is True
def test_doctor_all_ok(runner, mock_client):
mock_client.doctor.return_value = {"home_exists": True, "encryption": True, "issues": [], "providers_count": 1}
result = runner.invoke(cli, ["doctor"])
assert result.exit_code == 0
assert "OK" in result.output
def test_common_options_error_handling_json(runner, mock_client):
mock_client.login.side_effect = AuthsomeError("Error")
result = runner.invoke(cli, ["login", "openai", "--json"])
assert result.exit_code == 1
data = json.loads(result.output)
assert data["error"] == "AuthsomeError"
def test_echo_no_color(runner, mock_client):
mock_client.logout.return_value = None
result = runner.invoke(cli, ["logout", "openai", "--no-color"])
assert result.exit_code == 0
def test_echo_quiet(runner, mock_client):
mock_client.logout.return_value = None
result = runner.invoke(cli, ["logout", "openai", "--quiet"])
assert result.exit_code == 0
assert "Logged out" not in result.output
+695 -93
View File
@@ -1,19 +1,39 @@
"""Tests for the AuthClient core."""
import json
from datetime import timedelta
from pathlib import Path
from unittest.mock import patch
from unittest.mock import MagicMock, patch
import pytest
import requests
from authsome.client import AuthClient
from authsome.errors import (
AuthsomeError,
ConnectionNotFoundError,
CredentialMissingError,
ProfileNotFoundError,
ProviderNotFoundError,
RefreshFailedError,
TokenExpiredError,
UnsupportedFlowError,
)
from authsome.models.connection import (
ConnectionRecord,
ProviderClientRecord,
)
from authsome.models.enums import AuthType, ConnectionStatus, ExportFormat, FlowType
from authsome.models.provider import ApiKeyConfig, ProviderDefinition
from authsome.models.provider import ApiKeyConfig, OAuthConfig, ProviderDefinition
from authsome.utils import utc_now
@pytest.fixture
def client(tmp_path: Path) -> AuthClient:
home = tmp_path / ".authsome"
with AuthClient(home=home) as c:
c.init()
yield c
class TestAuthClientInit:
@@ -66,17 +86,32 @@ class TestAuthClientInit:
client.init() # Should not fail
assert (home / "version").exists()
def test_authclient_default_home(self, monkeypatch):
monkeypatch.delenv("AUTHSOME_HOME", raising=False)
with patch("authsome.client.Path.home", return_value=Path("/mock/home")):
with AuthClient() as c:
assert str(c.home) == "/mock/home/.authsome"
def test_authclient_keyring_crypto(self, tmp_path, monkeypatch):
home = tmp_path / ".authsome"
home.mkdir()
(home / "config.json").write_text('{"encryption": {"mode": "keyring"}}')
with patch("authsome.client.KeyringCryptoBackend") as mock_backend:
with AuthClient(home=home) as c:
_ = c.crypto
mock_backend.assert_called_once()
def test_load_config_bad_json(self, tmp_path):
home = tmp_path / ".authsome"
home.mkdir()
(home / "config.json").write_text("{bad")
with AuthClient(home=home) as c:
assert c.config.default_profile == "default"
class TestAuthClientProviders:
"""Provider operations tests."""
@pytest.fixture
def client(self, tmp_path: Path) -> AuthClient:
home = tmp_path / ".authsome"
c = AuthClient(home=home)
c.init()
return c
def test_list_providers_includes_bundled(self, client: AuthClient) -> None:
providers = client.list_providers()
names = [p.name for p in providers]
@@ -97,24 +132,31 @@ class TestAuthClientProviders:
name="custom",
display_name="Custom Provider",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY_PROMPT,
flow=FlowType.API_KEY,
api_key=ApiKeyConfig(env_var="CUSTOM_KEY"),
)
client.register_provider(custom)
loaded = client.get_provider("custom")
assert loaded.display_name == "Custom Provider"
def test_list_providers_by_source(self, client: AuthClient):
provider = ProviderDefinition(
name="testlocal",
display_name="Test Local",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY,
api_key={"header_name": "Authorization"},
)
client.register_provider(provider)
sources = client.list_providers_by_source()
assert "bundled" in sources
assert "custom" in sources
assert len(sources["custom"]) > 0
class TestAuthClientProfiles:
"""Profile management tests."""
@pytest.fixture
def client(self, tmp_path: Path) -> AuthClient:
home = tmp_path / ".authsome"
c = AuthClient(home=home)
c.init()
return c
def test_default_profile_created(self, client: AuthClient) -> None:
profiles = client.list_profiles()
names = [p.name for p in profiles]
@@ -135,123 +177,647 @@ class TestAuthClientProfiles:
with pytest.raises(ProfileNotFoundError):
client.set_default_profile("nonexistent")
def test_list_profiles_issues(self, client: AuthClient):
profiles_dir = client.home / "profiles"
bad_dir = profiles_dir / "bad"
bad_dir.mkdir()
(bad_dir / "metadata.json").write_text("{bad json")
class TestAuthClientApiKeyLogin:
"""API key login integration tests."""
profiles = client.list_profiles()
names = [p.name for p in profiles]
assert "default" in names
assert "bad" not in names
@pytest.fixture
def client(self, tmp_path: Path) -> AuthClient:
home = tmp_path / ".authsome"
c = AuthClient(home=home)
c.init()
return c
def test_list_profiles_no_dir(self, client: AuthClient):
import shutil
profiles_dir = client.home / "profiles"
shutil.rmtree(profiles_dir)
assert client.list_profiles() == []
class TestAuthClientLogin:
"""Authentication flow integration tests."""
def test_api_key_login_and_get(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-test-123"):
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-test-123"}):
record = client.login("openai")
assert record.status == ConnectionStatus.CONNECTED
assert record.auth_type == AuthType.API_KEY
# Get connection
conn = client.get_connection("openai")
assert conn.status == ConnectionStatus.CONNECTED
def test_login_connection_exists(self, client: AuthClient):
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-1"}):
client.login("openai", "default")
with pytest.raises(AuthsomeError, match="already exists"):
client.login("openai", "default", force=False)
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-2"}):
client.login("openai", "default", force=True)
def test_login_unsupported_flow(self, client: AuthClient):
def mock_get_provider(name):
mock_def = MagicMock()
mock_def.flow = MagicMock(value="invalid_flow")
return mock_def
with patch.object(client, "get_provider", mock_get_provider):
with pytest.raises(UnsupportedFlowError):
client.login("test")
def test_login_oauth_bridge_prompt(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://auth", token_url="http://token"),
)
client.register_provider(provider)
mock_record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("access"),
)
with patch("authsome.client._FLOW_HANDLERS") as handlers:
mock_handler = MagicMock()
mock_handler.authenticate.return_value = mock_record
handlers.get.return_value = lambda: mock_handler
with patch(
"authsome.flows.bridge.secure_input_bridge",
return_value={"client_id": "cid", "client_secret": "csec"},
) as mock_bridge:
client.login("testoauth")
mock_bridge.assert_called_once()
creds = client.get_provider_client_credentials("testoauth", "default")
assert creds.client_id == "cid"
assert client.crypto.decrypt(creds.client_secret) == "csec"
def test_login_dcr_metadata_extraction(self, client: AuthClient):
provider = ProviderDefinition(
name="testdcr",
display_name="Test DCR",
auth_type=AuthType.OAUTH2,
flow=FlowType.DCR_PKCE,
oauth=OAuthConfig(authorization_url="http://auth", token_url="http://token"),
)
client.register_provider(provider)
encrypted_secret = client.crypto.encrypt("csec")
mock_record = ConnectionRecord(
schema_version=1,
provider="testdcr",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("access"),
metadata={"_dcr_client_id": "cid", "_dcr_client_secret": encrypted_secret.model_dump()},
)
with patch("authsome.client._FLOW_HANDLERS") as handlers:
mock_handler = MagicMock()
mock_handler.authenticate.return_value = mock_record
handlers.get.return_value = lambda: mock_handler
client.login("testdcr")
creds = client.get_provider_client_credentials("testdcr", "default")
assert creds.client_id == "cid"
assert client.crypto.decrypt(creds.client_secret) == "csec"
def test_login_dcr_metadata_extraction_no_secret(self, client: AuthClient):
provider = ProviderDefinition(
name="testdcr2",
display_name="Test DCR 2",
auth_type=AuthType.OAUTH2,
flow=FlowType.DCR_PKCE,
oauth=OAuthConfig(authorization_url="http://auth", token_url="http://token"),
)
client.register_provider(provider)
mock_record = ConnectionRecord(
schema_version=1,
provider="testdcr2",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("access"),
metadata={"_dcr_client_id": "cid"},
)
with patch("authsome.client._FLOW_HANDLERS") as handlers:
mock_handler = MagicMock()
mock_handler.authenticate.return_value = mock_record
handlers.get.return_value = lambda: mock_handler
client.login("testdcr2")
creds = client.get_provider_client_credentials("testdcr2", "default")
assert creds.client_id == "cid"
assert creds.client_secret is None
class TestAuthClientCredentials:
"""Credential retrieval tests."""
def test_api_key_get_access_token(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-test-456"):
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-test-456"}):
client.login("openai")
token = client.get_access_token("openai")
assert token == "sk-test-456"
def test_api_key_get_auth_headers(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-test-789"):
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-test-789"}):
client.login("openai")
headers = client.get_auth_headers("openai")
assert "Authorization" in headers
assert headers["Authorization"] == "Bearer sk-test-789"
def test_api_key_multiple_connections(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="key-1"):
client.login("openai", connection_name="personal")
def test_get_access_token_oauth(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
client.register_provider(provider)
with patch("authsome.flows.api_key.getpass.getpass", return_value="key-2"):
client.login("openai", connection_name="work")
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("token123"),
)
client._save_connection(record)
assert client.get_access_token("openai", connection="personal") == "key-1"
assert client.get_access_token("openai", connection="work") == "key-2"
assert client.get_access_token("testoauth") == "token123"
def test_get_nonexistent_connection(self, client: AuthClient) -> None:
def test_get_access_token_unsupported(self, client: AuthClient):
mock_record = MagicMock()
mock_record.auth_type = "UNKNOWN"
with patch.object(client, "get_connection", return_value=mock_record):
with pytest.raises(CredentialMissingError, match="Unsupported auth type"):
client.get_access_token("test")
def test_get_auth_headers_unsupported(self, client: AuthClient):
mock_record = MagicMock()
mock_record.auth_type = "UNKNOWN"
with patch.object(client, "get_provider"):
with patch.object(client, "get_connection", return_value=mock_record):
with pytest.raises(CredentialMissingError, match="Cannot build headers"):
client.get_auth_headers("test")
def test_get_auth_headers_api_key_custom(self, client: AuthClient):
provider = ProviderDefinition(
name="testapi",
display_name="Test API",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY,
api_key={"header_name": "X-API-KEY", "header_prefix": ""},
)
client.register_provider(provider)
record = ConnectionRecord(
schema_version=1,
provider="testapi",
profile="default",
connection_name="default",
auth_type=AuthType.API_KEY,
status=ConnectionStatus.CONNECTED,
api_key=client.crypto.encrypt("key123"),
)
client._save_connection(record)
headers = client.get_auth_headers("testapi")
assert headers["X-API-KEY"] == "key123"
def test_get_auth_headers_oauth(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
client.register_provider(provider)
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("oauth123"),
)
client._save_connection(record)
headers = client.get_auth_headers("testoauth")
assert headers["Authorization"] == "Bearer oauth123"
def test_get_auth_headers_api_key_no_config(self, client: AuthClient):
provider = ProviderDefinition(
name="testapi",
display_name="Test API",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY,
)
record = ConnectionRecord(
schema_version=1,
provider="testapi",
profile="default",
connection_name="default",
auth_type=AuthType.API_KEY,
status=ConnectionStatus.CONNECTED,
api_key=client.crypto.encrypt("key123"),
)
client._save_connection(record)
with patch.object(client, "get_provider", return_value=provider):
headers = client.get_auth_headers("testapi")
assert headers["Authorization"] == "Bearer key123"
def test_get_api_key_missing(self, client: AuthClient):
provider = ProviderDefinition(
name="test",
display_name="Test",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY,
api_key={"header_name": "Authorization"},
)
client.register_provider(provider)
record = ConnectionRecord(
schema_version=1,
provider="test",
profile="default",
connection_name="default",
auth_type=AuthType.API_KEY,
status=ConnectionStatus.CONNECTED,
)
client._save_connection(record)
with pytest.raises(CredentialMissingError, match="No API key stored"):
client._get_api_key(record)
class TestAuthClientTokenRefresh:
"""OAuth token refresh logic tests."""
def test_oauth_token_refresh(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
client.register_provider(provider)
now = utc_now()
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("old_acc"),
refresh_token=client.crypto.encrypt("ref"),
expires_at=now - timedelta(seconds=10), # expired
)
client._save_connection(record)
with pytest.raises(RefreshFailedError, match="No client_id"):
client.get_access_token("testoauth")
client._save_provider_client_credentials(
ProviderClientRecord(
profile="default",
provider="testoauth",
client_id="cid",
client_secret=client.crypto.encrypt("sec"),
)
)
mock_token_resp = MagicMock()
mock_token_resp.json.return_value = {"access_token": "new_acc", "refresh_token": "new_ref", "expires_in": 3600}
with patch("authsome.client.http_client.post", return_value=mock_token_resp):
assert client.get_access_token("testoauth") == "new_acc"
record.expires_at = now - timedelta(seconds=10)
client._save_connection(record)
with patch("authsome.client.http_client.post", side_effect=requests.RequestException("boom")):
with pytest.raises(RefreshFailedError):
client.get_access_token("testoauth")
def test_oauth_token_no_refresh_expired(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
client.register_provider(provider)
now = utc_now()
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("old_acc"),
expires_at=now - timedelta(seconds=10),
)
client._save_connection(record)
with pytest.raises(TokenExpiredError):
client.get_access_token("testoauth")
def test_oauth_token_refresh_failed_but_still_valid(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
client.register_provider(provider)
now = utc_now()
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("valid_acc"),
refresh_token=client.crypto.encrypt("ref"),
expires_at=now + timedelta(seconds=100),
)
client._save_connection(record)
client._save_provider_client_credentials(
ProviderClientRecord(profile="default", provider="testoauth", client_id="cid")
)
with patch("authsome.client.http_client.post", side_effect=requests.RequestException("boom")):
assert client.get_access_token("testoauth") == "valid_acc"
def test_oauth_token_missing_access_token(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
client.register_provider(provider)
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
)
client._save_connection(record)
with pytest.raises(CredentialMissingError, match="No access token stored"):
client.get_access_token("testoauth")
def test_get_access_token_valid_not_near_expiry(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
client.register_provider(provider)
now = utc_now()
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("valid_token"),
expires_at=now + timedelta(seconds=1000),
)
client._save_connection(record)
assert client.get_access_token("testoauth") == "valid_token"
def test_get_access_token_valid_no_refresh(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
client.register_provider(provider)
now = utc_now()
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("valid_token_no_ref"),
expires_at=now + timedelta(seconds=100),
)
client._save_connection(record)
assert client.get_access_token("testoauth") == "valid_token_no_ref"
def test_refresh_token_no_oauth_config(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth", display_name="Test OAuth", auth_type=AuthType.OAUTH2, flow=FlowType.PKCE, oauth=None
)
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
)
with patch.object(client, "get_provider", return_value=provider):
with pytest.raises(RefreshFailedError, match="No OAuth config"):
client._refresh_token(record, "testoauth")
def test_refresh_token_no_refresh_token(self, client: AuthClient):
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
refresh_token=None,
)
with patch.object(client, "get_provider", return_value=provider):
with pytest.raises(RefreshFailedError, match="No refresh token available"):
client._refresh_token(record, "testoauth")
class TestAuthClientLifecycle:
"""Connection lifecycle tests (logout, remove, revoke)."""
def test_logout_errors_and_revocation(self, client: AuthClient):
client.logout("openai", "nonexistent")
provider = ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t", revocation_url="http://revoke"),
)
client.register_provider(provider)
record = ConnectionRecord(
schema_version=1,
provider="testoauth",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("token123"),
)
client._save_connection(record)
client._update_provider_metadata("default", "testoauth", "default")
with patch("authsome.client.http_client.post") as mock_post:
client.logout("testoauth")
mock_post.assert_called_once_with("http://revoke", data={"token": "token123"}, timeout=15)
client._save_connection(record)
with patch("authsome.client.http_client.post", side_effect=requests.RequestException("boom")):
client.logout("testoauth")
def test_remove_connection(self, client: AuthClient) -> None:
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "key"}):
client.login("openai")
client.remove("openai")
with pytest.raises(ConnectionNotFoundError):
client.get_connection("openai", connection="nonexistent")
client.get_connection("openai")
def test_remove_nonexistent(self, client: AuthClient) -> None:
client.remove("openai")
def test_revoke_connection(self, client: AuthClient) -> None:
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "key"}):
client.login("openai")
client.revoke("openai")
with pytest.raises(ConnectionNotFoundError):
client.get_connection("openai")
def test_remove_bundled_provider(self, client: AuthClient):
client.remove("openai")
assert not (client.home / "providers" / "openai.json").exists()
class TestAuthClientExport:
"""Export operations tests."""
@pytest.fixture
def client(self, tmp_path: Path) -> AuthClient:
home = tmp_path / ".authsome"
c = AuthClient(home=home)
c.init()
return c
def test_export_env_format(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-export"):
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-export"}):
client.login("openai")
output = client.export("openai", format=ExportFormat.ENV)
assert "OPENAI_API_KEY=sk-export" in output
def test_export_shell_format(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-shell"):
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-shell"}):
client.login("openai")
output = client.export("openai", format=ExportFormat.SHELL)
assert "export OPENAI_API_KEY=sk-shell" in output
def test_export_json_format(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-json"):
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-json"}):
client.login("openai")
output = client.export("openai", format=ExportFormat.JSON)
data = json.loads(output)
assert data["OPENAI_API_KEY"] == "sk-json"
def test_export_oauth_and_unknown(self, client: AuthClient):
provider = ProviderDefinition(
name="testexport",
display_name="Test Export",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(authorization_url="http://a", token_url="http://t"),
)
client.register_provider(provider)
class TestAuthClientRemoveRevoke:
"""Remove and revoke operations tests."""
record = ConnectionRecord(
schema_version=1,
provider="testexport",
profile="default",
connection_name="default",
auth_type=AuthType.OAUTH2,
status=ConnectionStatus.CONNECTED,
access_token=client.crypto.encrypt("acc"),
refresh_token=client.crypto.encrypt("ref"),
)
client._save_connection(record)
@pytest.fixture
def client(self, tmp_path: Path) -> AuthClient:
home = tmp_path / ".authsome"
c = AuthClient(home=home)
c.init()
return c
env_out = client.export("testexport", format=ExportFormat.ENV)
assert "TESTEXPORT_ACCESS_TOKEN=acc" in env_out
assert "TESTEXPORT_REFRESH_TOKEN=ref" in env_out
assert client.export("testexport", format=MagicMock()) == ""
def test_remove_connection(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="key"):
class TestAuthClientRun:
"""Run command tests."""
def test_run_command(self, client: AuthClient):
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "sk-run"}):
client.login("openai")
client.remove("openai")
with pytest.raises(ConnectionNotFoundError):
client.get_connection("openai")
def test_remove_nonexistent(self, client: AuthClient) -> None:
with pytest.raises(ConnectionNotFoundError):
client.remove("openai")
def test_revoke_connection(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="key"):
client.login("openai")
client.revoke("openai")
conn = client.get_connection("openai")
assert conn.status == ConnectionStatus.REVOKED
assert conn.api_key is None
with patch("authsome.client.subprocess.run") as mock_run:
client.run(["echo", "hello", "world"], providers=["openai"])
mock_run.assert_called_once()
kwargs = mock_run.call_args[1]
assert kwargs["env"]["OPENAI_API_KEY"] == "sk-run"
assert kwargs["shell"] is True
assert "echo" in mock_run.call_args[0][0]
class TestAuthClientDoctor:
@@ -264,35 +830,71 @@ class TestAuthClientDoctor:
results = client.doctor()
assert results["home_exists"] is True
assert results["version_file"] is True
assert results["config_file"] is True
assert results["encryption"] is True
assert results["store"] is True
assert results["providers_count"] > 0
assert results["profiles_count"] > 0
assert results["issues"] == []
def test_doctor_issues(self, client: AuthClient, monkeypatch):
with patch.object(client.crypto, "encrypt", side_effect=Exception("crypto boom")):
res = client.doctor()
assert not res["encryption"]
assert "Encryption: crypto boom" in res["issues"]
class TestAuthClientListConnections:
"""List connections tests."""
store = client._get_store("default")
with patch.object(store, "set", side_effect=Exception("store boom")):
res = client.doctor()
assert not res["store"]
assert "Store: store boom" in res["issues"]
@pytest.fixture
def client(self, tmp_path: Path) -> AuthClient:
home = tmp_path / ".authsome"
c = AuthClient(home=home)
c.init()
return c
with patch.object(client, "list_providers", side_effect=Exception("prov boom")):
res = client.doctor()
assert "Providers: prov boom" in res["issues"]
with patch.object(client, "list_profiles", side_effect=Exception("prof boom")):
res = client.doctor()
assert "Profiles: prof boom" in res["issues"]
class TestAuthClientConnections:
"""Connection management tests."""
def test_list_connections_empty(self, client: AuthClient) -> None:
connections = client.list_connections()
assert connections == []
def test_list_connections_after_login(self, client: AuthClient) -> None:
with patch("authsome.flows.api_key.getpass.getpass", return_value="key"):
with patch("authsome.flows.bridge.secure_input_bridge", return_value={"api_key": "key"}):
client.login("openai")
connections = client.list_connections()
assert len(connections) == 1
assert connections[0]["name"] == "openai"
assert len(connections[0]["connections"]) == 1
assert connections[0]["connections"][0]["status"] == "connected"
def test_list_connections_edge_cases(self, client: AuthClient):
store = client._get_store("default")
store.set("profile:default:openai:junk:default", '{"junk": true}')
store.set("profile:default:openai:connection", '{"junk": true}')
store.set("profile:default:openai:connection:test", "")
connections = client.list_connections()
assert connections == []
def test_get_nonexistent_connection(self, client: AuthClient) -> None:
with pytest.raises(ConnectionNotFoundError):
client.get_connection("openai", connection="nonexistent")
def test_get_store_missing_profile(self, client: AuthClient):
with pytest.raises(ProfileNotFoundError):
client._get_store("missing")
def test_metadata_cleanup(self, client: AuthClient):
client._update_provider_metadata("default", "openai", "conn1")
client._update_provider_metadata("default", "openai", "conn2")
client._remove_from_provider_metadata("default", "openai", "conn1")
store = client._get_store("default")
meta = store.get("profile:default:openai:metadata")
assert "conn1" not in meta
assert "conn2" in meta
client._remove_from_provider_metadata("default", "openai", "conn2")
meta = json.loads(store.get("profile:default:openai:metadata"))
assert meta["last_used_connection"] is None
+137
View File
@@ -72,6 +72,58 @@ class TestLocalFileCryptoBackend:
decrypted = crypto.decrypt(encrypted)
assert decrypted == original
def test_json_load_error(self, tmp_path: Path) -> None:
from authsome.errors import EncryptionUnavailableError
key_file = tmp_path / "master.key"
key_file.write_text("invalid json")
with pytest.raises(EncryptionUnavailableError, match="Failed to read local key file"):
LocalFileCryptoBackend(tmp_path)
def test_chmod_error(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
import os
def mock_chmod(path, mode):
raise OSError("Mock error")
monkeypatch.setattr(os, "chmod", mock_chmod)
# Should not raise
_ = LocalFileCryptoBackend(tmp_path)
def test_encrypt_not_initialized(self, crypto: LocalFileCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
crypto._aesgcm = None
with pytest.raises(EncryptionUnavailableError, match="Master key not initialized"):
crypto.encrypt("test")
def test_decrypt_not_initialized(self, crypto: LocalFileCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
crypto._aesgcm = None
with pytest.raises(EncryptionUnavailableError, match="Master key not initialized"):
crypto.decrypt(EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="a", ciphertext="b", tag="c"))
def test_decrypt_unsupported_alg(self, crypto: LocalFileCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError, match="Unsupported algorithm"):
crypto.decrypt(EncryptedField(enc=1, alg="UNSUPPORTED", kid="local", nonce="a", ciphertext="b", tag="c"))
def test_decrypt_base64_decode_error(self, crypto: LocalFileCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError, match="Failed to decode envelope"):
crypto.decrypt(EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="!@#", ciphertext="b", tag="c"))
def test_decrypt_aesgcm_error(self, crypto: LocalFileCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError, match="Decryption failed"):
crypto.decrypt(
EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="abcd", ciphertext="abcd", tag="abcd")
)
class TestKeyringCryptoBackend:
"""OS Keyring crypto backend tests.
@@ -100,6 +152,91 @@ class TestKeyringCryptoBackend:
assert field.alg == "AES-256-GCM"
assert field.kid == "local"
def test_import_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
import sys
monkeypatch.setitem(sys.modules, "keyring", None)
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError, match="The 'keyring' package is required"):
KeyringCryptoBackend()
def test_get_password_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
import keyring
def mock_get(*args, **kwargs):
raise Exception("Mock error")
monkeypatch.setattr(keyring, "get_password", mock_get)
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError, match="Failed to access OS keyring"):
KeyringCryptoBackend()
def test_generate_new_keyring_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
import keyring
def mock_get(*args, **kwargs):
return None
def mock_set(*args, **kwargs):
pass
monkeypatch.setattr(keyring, "get_password", mock_get)
monkeypatch.setattr(keyring, "set_password", mock_set)
backend = KeyringCryptoBackend()
assert backend._master_key is not None
def test_set_password_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
import keyring
def mock_get(*args, **kwargs):
return None
def mock_set(*args, **kwargs):
raise Exception("Mock error")
monkeypatch.setattr(keyring, "get_password", mock_get)
monkeypatch.setattr(keyring, "set_password", mock_set)
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError, match="Failed to store master key"):
KeyringCryptoBackend()
def test_encrypt_not_initialized(self, crypto: KeyringCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
crypto._aesgcm = None
with pytest.raises(EncryptionUnavailableError, match="Master key not initialized"):
crypto.encrypt("test")
def test_decrypt_not_initialized(self, crypto: KeyringCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
crypto._aesgcm = None
with pytest.raises(EncryptionUnavailableError, match="Master key not initialized"):
crypto.decrypt(EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="a", ciphertext="b", tag="c"))
def test_decrypt_unsupported_alg(self, crypto: KeyringCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError, match="Unsupported algorithm"):
crypto.decrypt(EncryptedField(enc=1, alg="UNSUPPORTED", kid="local", nonce="a", ciphertext="b", tag="c"))
def test_decrypt_base64_decode_error(self, crypto: KeyringCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError, match="Failed to decode envelope"):
crypto.decrypt(EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="!@#", ciphertext="b", tag="c"))
def test_decrypt_aesgcm_error(self, crypto: KeyringCryptoBackend) -> None:
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError, match="Decryption failed"):
crypto.decrypt(
EncryptedField(enc=1, alg="AES-256-GCM", kid="local", nonce="abcd", ciphertext="abcd", tag="abcd")
)
class TestCrossBackendCompatibility:
"""Verify that both backends produce compatible envelope formats."""
+94
View File
@@ -0,0 +1,94 @@
"""Tests for errors.py."""
from authsome.errors import (
AuthenticationFailedError,
AuthsomeError,
ConnectionNotFoundError,
CredentialMissingError,
DiscoveryError,
EncryptionUnavailableError,
InvalidProviderSchemaError,
ProfileNotFoundError,
ProviderNotFoundError,
RefreshFailedError,
StoreUnavailableError,
TokenExpiredError,
UnsupportedAuthTypeError,
UnsupportedFlowError,
)
def test_authsome_error_formatting():
# Test operation present
err = AuthsomeError("message", operation="test_op")
assert str(err) == "(test_op) message"
# Test provider present
err2 = AuthsomeError("message", provider="github")
assert str(err2) == "[github] message"
def test_unsupported_auth_type_error():
err = UnsupportedAuthTypeError("magic", provider="github")
assert "Unsupported auth type: magic" in str(err)
assert "[github]" in str(err)
def test_unsupported_flow_error():
err = UnsupportedFlowError("unknown", provider="github")
assert "Unsupported flow: unknown" in str(err)
def test_credential_missing_error():
err = CredentialMissingError(provider="github")
assert "Credential not found" in str(err)
def test_token_expired_error():
err = TokenExpiredError(provider="github")
assert "Access token expired" in str(err)
def test_refresh_failed_error():
err = RefreshFailedError(provider="github")
assert "Token refresh failed: Unknown error" in str(err)
def test_encryption_unavailable_error():
err = EncryptionUnavailableError()
assert "Encryption backend unavailable" in str(err)
def test_store_unavailable_error():
err = StoreUnavailableError()
assert "Credential store unavailable" in str(err)
def test_discovery_error():
err = DiscoveryError("timeout", provider="github")
assert "Discovery failed: timeout" in str(err)
def test_provider_not_found_error():
err = ProviderNotFoundError("missing")
assert "Provider 'missing' not found" in str(err)
def test_invalid_provider_schema_error():
err = InvalidProviderSchemaError("bad json", provider="github")
assert "Invalid provider schema: bad json" in str(err)
def test_profile_not_found_error():
err = ProfileNotFoundError("work")
assert "Profile 'work' not found" in str(err)
def test_connection_not_found_error():
err = ConnectionNotFoundError(provider="github", connection="work", profile="default")
assert "Connection 'work' not found for provider 'github' in profile 'default'" in str(err)
def test_authentication_failed_error():
err = AuthenticationFailedError("invalid credentials", provider="github")
assert "Authentication failed: invalid credentials" in str(err)
+38 -80
View File
@@ -1,13 +1,12 @@
"""Tests for authentication flows."""
from pathlib import Path
from unittest.mock import patch
import pytest
from authsome.crypto.local_file_crypto import LocalFileCryptoBackend
from authsome.errors import AuthenticationFailedError, CredentialMissingError
from authsome.flows.api_key import ApiKeyEnvFlow, ApiKeyPromptFlow
from authsome.errors import AuthenticationFailedError
from authsome.flows.api_key import ApiKeyFlow
from authsome.models.enums import AuthType, ConnectionStatus, FlowType
from authsome.models.provider import ApiKeyConfig, ProviderDefinition
@@ -17,33 +16,32 @@ def _make_api_key_provider() -> ProviderDefinition:
name="testapi",
display_name="Test API",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY_PROMPT,
flow=FlowType.API_KEY,
api_key=ApiKeyConfig(
header_name="Authorization",
header_prefix="Bearer",
env_var="TEST_API_KEY",
),
)
class TestApiKeyPromptFlow:
"""API key prompt flow tests."""
class TestApiKeyFlow:
"""API key flow tests."""
@pytest.fixture
def crypto(self, tmp_path: Path) -> LocalFileCryptoBackend:
return LocalFileCryptoBackend(tmp_path)
def test_successful_login(self, crypto: LocalFileCryptoBackend) -> None:
flow = ApiKeyPromptFlow()
flow = ApiKeyFlow()
provider = _make_api_key_provider()
with patch("authsome.flows.api_key.getpass.getpass", return_value="sk-test-key-123"):
record = flow.authenticate(
provider=provider,
crypto=crypto,
profile="default",
connection_name="default",
)
record = flow.authenticate(
provider=provider,
crypto=crypto,
profile="default",
connection_name="default",
api_key="sk-test-key-123",
)
assert record.provider == "testapi"
assert record.profile == "default"
@@ -56,38 +54,38 @@ class TestApiKeyPromptFlow:
assert decrypted == "sk-test-key-123"
def test_empty_key_rejected(self, crypto: LocalFileCryptoBackend) -> None:
flow = ApiKeyPromptFlow()
flow = ApiKeyFlow()
provider = _make_api_key_provider()
with patch("authsome.flows.api_key.getpass.getpass", return_value=""):
with pytest.raises(AuthenticationFailedError, match="cannot be empty"):
flow.authenticate(
provider=provider,
crypto=crypto,
profile="default",
connection_name="default",
)
with pytest.raises(AuthenticationFailedError, match="API key was not provided"):
flow.authenticate(
provider=provider,
crypto=crypto,
profile="default",
connection_name="default",
api_key="",
)
def test_whitespace_only_rejected(self, crypto: LocalFileCryptoBackend) -> None:
flow = ApiKeyPromptFlow()
flow = ApiKeyFlow()
provider = _make_api_key_provider()
with patch("authsome.flows.api_key.getpass.getpass", return_value=" "):
with pytest.raises(AuthenticationFailedError, match="cannot be empty"):
flow.authenticate(
provider=provider,
crypto=crypto,
profile="default",
connection_name="default",
)
with pytest.raises(AuthenticationFailedError, match="cannot be empty"):
flow.authenticate(
provider=provider,
crypto=crypto,
profile="default",
connection_name="default",
api_key=" ",
)
def test_missing_api_key_config(self, crypto: LocalFileCryptoBackend) -> None:
flow = ApiKeyPromptFlow()
flow = ApiKeyFlow()
provider = ProviderDefinition(
name="noconfig",
display_name="No Config",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY_PROMPT,
flow=FlowType.API_KEY,
)
with pytest.raises(AuthenticationFailedError, match="missing 'api_key'"):
flow.authenticate(
@@ -95,58 +93,18 @@ class TestApiKeyPromptFlow:
crypto=crypto,
profile="default",
connection_name="default",
api_key="sk-test-key-123",
)
class TestApiKeyEnvFlow:
"""API key env import flow tests."""
@pytest.fixture
def crypto(self, tmp_path: Path) -> LocalFileCryptoBackend:
return LocalFileCryptoBackend(tmp_path)
def test_successful_env_import(self, crypto: LocalFileCryptoBackend, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("TEST_API_KEY", "env-key-value")
flow = ApiKeyEnvFlow()
def test_missing_api_key_parameter(self, crypto: LocalFileCryptoBackend) -> None:
flow = ApiKeyFlow()
provider = _make_api_key_provider()
record = flow.authenticate(
provider=provider,
crypto=crypto,
profile="default",
connection_name="default",
)
assert record.status == ConnectionStatus.CONNECTED
assert record.api_key is not None
assert crypto.decrypt(record.api_key) == "env-key-value"
def test_missing_env_var(self, crypto: LocalFileCryptoBackend, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("TEST_API_KEY", raising=False)
flow = ApiKeyEnvFlow()
provider = _make_api_key_provider()
with pytest.raises(CredentialMissingError, match="not set or empty"):
flow.authenticate(
provider=provider,
crypto=crypto,
profile="default",
connection_name="default",
)
def test_no_env_var_defined(self, crypto: LocalFileCryptoBackend) -> None:
flow = ApiKeyEnvFlow()
provider = ProviderDefinition(
name="noenv",
display_name="No Env",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY_ENV,
api_key=ApiKeyConfig(env_var=None),
)
with pytest.raises(AuthenticationFailedError, match="does not define an env_var"):
with pytest.raises(AuthenticationFailedError, match="API key was not provided"):
flow.authenticate(
provider=provider,
crypto=crypto,
profile="default",
connection_name="default",
api_key=None,
)
+351
View File
@@ -0,0 +1,351 @@
"""Tests for the DCR PKCE OAuth flow."""
import json
import urllib.parse
import urllib.request
from unittest.mock import MagicMock, patch
import pytest
import requests
from authsome.crypto.local_file_crypto import LocalFileCryptoBackend
from authsome.errors import AuthenticationFailedError, DiscoveryError
from authsome.flows.dcr_pkce import DcrPkceFlow, _CallbackHandler, _find_free_port, _generate_pkce
from authsome.models.connection import ConnectionStatus
from authsome.models.enums import AuthType, FlowType
from authsome.models.provider import OAuthConfig, ProviderDefinition
def _make_provider() -> ProviderDefinition:
return ProviderDefinition(
name="testdcr",
display_name="Test DCR",
auth_type=AuthType.OAUTH2,
flow=FlowType.DCR_PKCE,
oauth=OAuthConfig(
authorization_url="https://auth.example.com/auth",
token_url="https://auth.example.com/token",
),
)
def test_find_free_port():
port = _find_free_port()
assert isinstance(port, int)
assert port > 0
def test_generate_pkce():
verifier, challenge = _generate_pkce()
assert len(verifier) >= 43
assert len(challenge) >= 43
def test_callback_handler_log():
handler = _CallbackHandler.__new__(_CallbackHandler)
handler.log_message("test %s", "msg")
def test_missing_oauth(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
provider.oauth = None
flow = DcrPkceFlow()
with pytest.raises(AuthenticationFailedError, match="missing 'oauth' configuration"):
flow.authenticate(provider, crypto, "default", "default")
def test_discover_registration_endpoint_missing_oauth():
flow = DcrPkceFlow()
provider = _make_provider()
provider.oauth = None
with pytest.raises(DiscoveryError, match="No OAuth config"):
flow._discover_registration_endpoint(provider)
def test_discover_registration_endpoint_success():
flow = DcrPkceFlow()
provider = _make_provider()
mock_resp1 = MagicMock()
mock_resp1.status_code = 200
mock_resp1.json.return_value = {}
mock_resp2 = MagicMock()
mock_resp2.status_code = 200
mock_resp2.json.return_value = {"registration_endpoint": "https://auth.example.com/register"}
with patch("authsome.flows.dcr_pkce.http_client.get", side_effect=[mock_resp1, mock_resp2]):
endpoint = flow._discover_registration_endpoint(provider)
assert endpoint == "https://auth.example.com/register"
def test_discover_registration_endpoint_failure():
flow = DcrPkceFlow()
provider = _make_provider()
mock_resp = MagicMock()
mock_resp.status_code = 404
with patch("authsome.flows.dcr_pkce.http_client.get", side_effect=[requests.RequestException("boom"), mock_resp]):
with pytest.raises(DiscoveryError, match="Could not discover registration_endpoint"):
flow._discover_registration_endpoint(provider)
def test_register_client_missing_oauth():
flow = DcrPkceFlow()
provider = _make_provider()
provider.oauth = None
with pytest.raises(AuthenticationFailedError, match="No OAuth config"):
flow._register_client(provider, [])
def test_register_client_success():
flow = DcrPkceFlow()
provider = _make_provider()
provider.oauth.registration_endpoint = "https://auth.example.com/register"
mock_resp = MagicMock()
mock_resp.json.return_value = {"client_id": "new_cid", "client_secret": "new_sec"}
with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp):
cid, sec = flow._register_client(provider, ["scope1"])
assert cid == "new_cid"
assert sec == "new_sec"
def test_register_client_http_error():
flow = DcrPkceFlow()
provider = _make_provider()
provider.oauth.registration_endpoint = "https://auth.example.com/register"
with patch("authsome.flows.dcr_pkce.http_client.post", side_effect=requests.RequestException("boom")):
with pytest.raises(AuthenticationFailedError, match="Registration failed"):
flow._register_client(provider, [])
def test_register_client_invalid_json():
flow = DcrPkceFlow()
provider = _make_provider()
provider.oauth.registration_endpoint = "https://auth.example.com/register"
mock_resp = MagicMock()
mock_resp.json.side_effect = json.JSONDecodeError("msg", "doc", 0)
with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp):
with pytest.raises(AuthenticationFailedError, match="not valid JSON"):
flow._register_client(provider, [])
def test_register_client_missing_client_id():
flow = DcrPkceFlow()
provider = _make_provider()
provider.oauth.registration_endpoint = "https://auth.example.com/register"
mock_resp = MagicMock()
mock_resp.json.return_value = {"client_secret": "new_sec"}
with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp):
with pytest.raises(AuthenticationFailedError, match="missing client_id"):
flow._register_client(provider, [])
def test_dcr_pkce_flow_success(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DcrPkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
# Extract state from URL
parsed = urllib.parse.urlparse(url)
params = urllib.parse.parse_qs(parsed.query)
state = params["state"][0]
# Send callback
callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"
req = urllib.request.Request(callback_url)
with urllib.request.urlopen(req) as resp:
assert resp.status == 200
# Mock DCR
dcr_resp = MagicMock()
dcr_resp.json.return_value = {"client_id": "cid", "client_secret": "sec"}
# Mock Token Exchange
token_resp = MagicMock()
token_resp.json.return_value = {
"access_token": "mock_access",
"refresh_token": "mock_refresh",
"expires_in": 3600,
}
with patch("authsome.flows.dcr_pkce.http_client.post", side_effect=[dcr_resp, token_resp]):
with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.dcr_pkce.DcrPkceFlow._discover_registration_endpoint", return_value="url"):
record = flow.authenticate(provider, crypto, "default", "default", scopes=["test"])
assert record.status == ConnectionStatus.CONNECTED
assert record.metadata["_dcr_client_id"] == "cid"
assert record.metadata["_dcr_client_secret"] is not None
def test_dcr_pkce_flow_reuse_client(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DcrPkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
parsed = urllib.parse.urlparse(url)
params = urllib.parse.parse_qs(parsed.query)
assert "scope" not in params # No scopes passed
state = params["state"][0]
callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"
urllib.request.urlopen(urllib.request.Request(callback_url))
token_resp = MagicMock()
token_resp.json.return_value = {"access_token": "mock_access"}
with patch("authsome.flows.dcr_pkce.http_client.post", return_value=token_resp):
with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open):
# Pass client_id to skip DCR
record = flow.authenticate(provider, crypto, "default", "default", client_id="existing_cid")
assert record.status == ConnectionStatus.CONNECTED
assert record.metadata["_dcr_client_id"] == "existing_cid"
assert record.metadata["_dcr_client_secret"] is None
assert record.expires_at is None
def test_dcr_pkce_flow_callback_error(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DcrPkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
callback_url = f"http://127.0.0.1:{port}/callback?error=access_denied&error_description=User%20denied"
try:
urllib.request.urlopen(urllib.request.Request(callback_url))
except urllib.error.HTTPError as e:
assert e.code == 400
with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open):
with pytest.raises(AuthenticationFailedError, match="access_denied"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_dcr_pkce_flow_callback_invalid(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DcrPkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
callback_url = f"http://127.0.0.1:{port}/callback?other=123"
try:
urllib.request.urlopen(urllib.request.Request(callback_url))
except urllib.error.HTTPError as e:
assert e.code == 400
with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open):
with pytest.raises(AuthenticationFailedError, match="no code received"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_dcr_pkce_flow_state_mismatch(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DcrPkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state=wrong"
urllib.request.urlopen(urllib.request.Request(callback_url))
with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open):
with pytest.raises(AuthenticationFailedError, match="state mismatch"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_dcr_pkce_flow_timeout(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DcrPkceFlow()
def mock_open(url):
pass
with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.dcr_pkce._CALLBACK_TIMEOUT_SECONDS", 0.01):
with pytest.raises(AuthenticationFailedError, match="timed out"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_exchange_code_http_error(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DcrPkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
parsed = urllib.parse.urlparse(url)
state = urllib.parse.parse_qs(parsed.query)["state"][0]
urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"))
with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.dcr_pkce.http_client.post", side_effect=requests.RequestException("boom")):
with pytest.raises(AuthenticationFailedError, match="Token exchange failed"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_exchange_code_invalid_json(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DcrPkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
parsed = urllib.parse.urlparse(url)
params = urllib.parse.parse_qs(parsed.query)
state = params["state"][0]
urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"))
mock_resp = MagicMock()
mock_resp.json.side_effect = json.JSONDecodeError("msg", "doc", 0)
with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp):
with pytest.raises(AuthenticationFailedError, match="not valid JSON"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_exchange_code_missing_access_token(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DcrPkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
parsed = urllib.parse.urlparse(url)
params = urllib.parse.parse_qs(parsed.query)
state = params["state"][0]
urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"))
mock_resp = MagicMock()
mock_resp.json.return_value = {"error": "invalid_grant", "error_description": "bad code"}
with patch("authsome.flows.dcr_pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.dcr_pkce.http_client.post", return_value=mock_resp):
with pytest.raises(AuthenticationFailedError, match="bad code"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
+274
View File
@@ -0,0 +1,274 @@
"""Tests for the Device Code OAuth flow."""
import json
from unittest.mock import MagicMock, patch
import pytest
import requests
from authsome.crypto.local_file_crypto import LocalFileCryptoBackend
from authsome.errors import AuthenticationFailedError
from authsome.flows.device_code import DeviceCodeFlow
from authsome.models.connection import ConnectionStatus
from authsome.models.enums import AuthType, FlowType
from authsome.models.provider import OAuthConfig, ProviderDefinition
def _make_provider() -> ProviderDefinition:
return ProviderDefinition(
name="testdevice",
display_name="Test Device",
auth_type=AuthType.OAUTH2,
flow=FlowType.DEVICE_CODE,
oauth=OAuthConfig(
authorization_url="https://auth.example.com/auth",
device_authorization_url="https://auth.example.com/device",
token_url="https://auth.example.com/token",
),
)
def test_missing_oauth(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
provider.oauth = None
flow = DeviceCodeFlow()
with pytest.raises(AuthenticationFailedError, match="missing 'oauth' configuration"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_missing_device_url(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
provider.oauth.device_authorization_url = None
flow = DeviceCodeFlow()
with pytest.raises(AuthenticationFailedError, match="not have a device_authorization_url"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_missing_client_id(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
with pytest.raises(AuthenticationFailedError, match="requires a client_id"):
flow.authenticate(provider, crypto, "default", "default")
def test_request_device_code_http_error(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
with patch("authsome.flows.device_code.requests.post", side_effect=requests.RequestException("boom")):
with pytest.raises(AuthenticationFailedError, match="Device authorization request failed"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_request_device_code_invalid_json(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
mock_resp = MagicMock()
mock_resp.json.side_effect = json.JSONDecodeError("msg", "doc", 0)
with patch("authsome.flows.device_code.requests.post", return_value=mock_resp):
with pytest.raises(AuthenticationFailedError, match="not valid JSON"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_request_device_code_missing_fields(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
mock_resp = MagicMock()
# Missing user_code and verification_uri
mock_resp.json.return_value = {"device_code": "dc123"}
with patch("authsome.flows.device_code.requests.post", return_value=mock_resp):
with pytest.raises(AuthenticationFailedError, match="missing required fields"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_poll_for_token_timeout(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
device_resp = MagicMock()
device_resp.json.return_value = {
"device_code": "dc",
"user_code": "uc",
"verification_uri": "http://uri",
"expires_in": 1,
"interval": 1,
}
with patch("authsome.flows.device_code.requests.post", return_value=device_resp):
with patch("authsome.flows.device_code.time.sleep"):
with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 2]):
# loop runs once and then deadline is past
with pytest.raises(AuthenticationFailedError, match="Device authorization timed out"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_poll_for_token_success_and_errors(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
# We will simulate multiple polling attempts
device_resp = MagicMock()
device_resp.json.return_value = {
"device_code": "dc",
"user_code": "uc",
"verification_uri": "http://uri",
"verification_uri_complete": "http://uri?user_code=uc",
"expires_in": 300,
"interval": 1,
}
token_json_err = MagicMock()
token_json_err.json.side_effect = json.JSONDecodeError("msg", "doc", 0)
token_pending = MagicMock()
token_pending.status_code = 400
token_pending.json.return_value = {"error": "authorization_pending"}
token_slow_down = MagicMock()
token_slow_down.status_code = 400
token_slow_down.json.return_value = {"error": "slow_down"}
token_success = MagicMock()
token_success.status_code = 200
token_success.json.return_value = {"access_token": "acc", "refresh_token": "ref", "expires_in": 3600}
mock_post_responses = [
device_resp, # Phase 1 request
requests.RequestException("boom"), # Poll 1: request exception -> loops
token_json_err, # Poll 2: json decode error -> loops
token_pending, # Poll 3: pending -> loops
token_slow_down, # Poll 4: slow down -> interval increases
token_success, # Poll 5: success
]
with patch("authsome.flows.device_code.requests.post", side_effect=mock_post_responses):
with patch("authsome.flows.device_code.time.sleep") as mock_sleep:
with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0, 0, 0, 0, 0]):
record = flow.authenticate(
provider, crypto, "default", "default", scopes=["test_scope"], client_id="cid", client_secret="sec"
)
assert record.status == ConnectionStatus.CONNECTED
assert crypto.decrypt(record.access_token) == "acc"
assert "test_scope" in record.scopes
# Sleep is called 5 times
assert mock_sleep.call_count == 5
# The last sleep should be interval (1) + 5 = 6 due to slow_down
assert mock_sleep.call_args_list[-1][0][0] == 6
def test_poll_for_token_access_denied(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
device_resp = MagicMock()
device_resp.json.return_value = {
"device_code": "dc",
"user_code": "uc",
"verification_uri": "http://uri",
"expires_in": 300,
"interval": 1,
}
token_denied = MagicMock()
token_denied.status_code = 400
token_denied.json.return_value = {"error": "access_denied"}
with patch("authsome.flows.device_code.requests.post", side_effect=[device_resp, token_denied]):
with patch("authsome.flows.device_code.time.sleep"):
with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0]):
with pytest.raises(AuthenticationFailedError, match="User denied"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_poll_for_token_expired_token(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
device_resp = MagicMock()
device_resp.json.return_value = {
"device_code": "dc",
"user_code": "uc",
"verification_uri": "http://uri",
"expires_in": 300,
"interval": 1,
}
token_expired = MagicMock()
token_expired.status_code = 400
token_expired.json.return_value = {"error": "expired_token"}
with patch("authsome.flows.device_code.requests.post", side_effect=[device_resp, token_expired]):
with patch("authsome.flows.device_code.time.sleep"):
with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0]):
with pytest.raises(AuthenticationFailedError, match="Device code has expired"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_poll_for_token_unknown_error(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
device_resp = MagicMock()
device_resp.json.return_value = {
"device_code": "dc",
"user_code": "uc",
"verification_uri": "http://uri",
"expires_in": 300,
"interval": 1,
}
token_err = MagicMock()
token_err.status_code = 400
token_err.json.return_value = {"error": "unknown_error", "error_description": "weird"}
with patch("authsome.flows.device_code.requests.post", side_effect=[device_resp, token_err]):
with patch("authsome.flows.device_code.time.sleep"):
with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0]):
with pytest.raises(AuthenticationFailedError, match="weird"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_poll_for_token_success_no_expires_in(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = DeviceCodeFlow()
device_resp = MagicMock()
device_resp.json.return_value = {
"device_code": "dc",
"user_code": "uc",
"verification_uri": "http://uri",
"expires_in": 300,
"interval": 1,
}
token_success = MagicMock()
token_success.status_code = 200
token_success.json.return_value = {"access_token": "acc", "refresh_token": "ref"}
with patch("authsome.flows.device_code.requests.post", side_effect=[device_resp, token_success]):
with patch("authsome.flows.device_code.time.sleep"):
with patch("authsome.flows.device_code.time.monotonic", side_effect=[0, 0]):
record = flow.authenticate(provider, crypto, "default", "default", client_id="cid", client_secret="sec")
assert record.status == ConnectionStatus.CONNECTED
assert record.expires_at is None
+269
View File
@@ -0,0 +1,269 @@
"""Tests for the PKCE OAuth flow."""
import json
import urllib.parse
import urllib.request
from unittest.mock import MagicMock, patch
import pytest
import requests
from authsome.crypto.local_file_crypto import LocalFileCryptoBackend
from authsome.errors import AuthenticationFailedError
from authsome.flows.pkce import PkceFlow, _CallbackHandler, _find_free_port, _generate_pkce
from authsome.models.connection import ConnectionStatus
from authsome.models.enums import AuthType, FlowType
from authsome.models.provider import OAuthConfig, ProviderDefinition
def _make_provider() -> ProviderDefinition:
return ProviderDefinition(
name="testoauth",
display_name="Test OAuth",
auth_type=AuthType.OAUTH2,
flow=FlowType.PKCE,
oauth=OAuthConfig(
authorization_url="https://auth.example.com/auth",
token_url="https://auth.example.com/token",
),
)
def test_find_free_port():
port = _find_free_port()
assert isinstance(port, int)
assert port > 0
def test_generate_pkce():
verifier, challenge = _generate_pkce()
assert len(verifier) >= 43
assert len(challenge) >= 43
def test_missing_oauth(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
provider.oauth = None
flow = PkceFlow()
with pytest.raises(AuthenticationFailedError, match="missing 'oauth' configuration"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_missing_client_id(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = PkceFlow()
with pytest.raises(AuthenticationFailedError, match="requires a client_id"):
flow.authenticate(provider, crypto, "default", "default")
def test_callback_handler_log():
handler = _CallbackHandler.__new__(_CallbackHandler)
handler.log_message("test %s", "msg")
def test_pkce_flow_success(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = PkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
# Extract state from URL
parsed = urllib.parse.urlparse(url)
params = urllib.parse.parse_qs(parsed.query)
state = params["state"][0]
# Send callback
callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"
req = urllib.request.Request(callback_url)
with urllib.request.urlopen(req) as resp:
assert resp.status == 200
mock_resp = MagicMock()
mock_resp.json.return_value = {
"access_token": "mock_access",
"refresh_token": "mock_refresh",
"expires_in": 3600,
}
with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.pkce.http_client.post", return_value=mock_resp) as mock_post:
record = flow.authenticate(
provider, crypto, "default", "default", scopes=["read", "write"], client_id="cid", client_secret="sec"
)
assert record.status == ConnectionStatus.CONNECTED
assert "read" in record.scopes
assert crypto.decrypt(record.access_token) == "mock_access"
assert crypto.decrypt(record.refresh_token) == "mock_refresh"
assert record.expires_at is not None
mock_post.assert_called_once()
def test_pkce_flow_callback_error(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = PkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
callback_url = f"http://127.0.0.1:{port}/callback?error=access_denied&error_description=User%20denied"
req = urllib.request.Request(callback_url)
try:
urllib.request.urlopen(req)
except urllib.error.HTTPError as e:
assert e.code == 400
with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open):
with pytest.raises(AuthenticationFailedError, match="access_denied"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_pkce_flow_callback_invalid(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = PkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
# Missing code
callback_url = f"http://127.0.0.1:{port}/callback?other=123"
req = urllib.request.Request(callback_url)
try:
urllib.request.urlopen(req)
except urllib.error.HTTPError as e:
assert e.code == 400
with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open):
with pytest.raises(AuthenticationFailedError, match="no code received"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_pkce_flow_state_mismatch(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = PkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
# Wrong state
callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state=wrong_state"
req = urllib.request.Request(callback_url)
with urllib.request.urlopen(req) as _:
pass
with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open):
with pytest.raises(AuthenticationFailedError, match="state mismatch"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_pkce_exchange_http_error(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = PkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
parsed = urllib.parse.urlparse(url)
state = urllib.parse.parse_qs(parsed.query)["state"][0]
callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"
urllib.request.urlopen(urllib.request.Request(callback_url))
with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.pkce.http_client.post", side_effect=requests.RequestException("boom")):
with pytest.raises(AuthenticationFailedError, match="Token exchange failed: boom"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_pkce_exchange_invalid_json(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = PkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
parsed = urllib.parse.urlparse(url)
state = urllib.parse.parse_qs(parsed.query)["state"][0]
urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"))
mock_resp = MagicMock()
mock_resp.json.side_effect = json.JSONDecodeError("msg", "doc", 0)
with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.pkce.http_client.post", return_value=mock_resp):
with pytest.raises(AuthenticationFailedError, match="Token response was not valid JSON"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_pkce_exchange_missing_access_token(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = PkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
parsed = urllib.parse.urlparse(url)
state = urllib.parse.parse_qs(parsed.query)["state"][0]
urllib.request.urlopen(urllib.request.Request(f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"))
mock_resp = MagicMock()
mock_resp.json.return_value = {"error": "invalid_grant", "error_description": "bad code"}
with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.pkce.http_client.post", return_value=mock_resp):
with pytest.raises(AuthenticationFailedError, match="bad code"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_pkce_flow_no_scopes(tmp_path):
# Test missing scopes to hit line 161
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
# Ensure no scopes are set
provider.oauth.scopes = None
flow = PkceFlow()
flow.callback_port = _find_free_port()
port = flow.callback_port
def mock_open(url):
parsed = urllib.parse.urlparse(url)
params = urllib.parse.parse_qs(parsed.query)
assert "scope" not in params
state = params["state"][0]
callback_url = f"http://127.0.0.1:{port}/callback?code=mock_code&state={state}"
req = urllib.request.Request(callback_url)
urllib.request.urlopen(req)
mock_resp = MagicMock()
mock_resp.json.return_value = {"access_token": "mock"}
with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.pkce.http_client.post", return_value=mock_resp):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
def test_pkce_flow_timeout(tmp_path):
crypto = LocalFileCryptoBackend(tmp_path)
provider = _make_provider()
flow = PkceFlow()
flow.callback_port = _find_free_port()
def mock_open(url):
# Do nothing to simulate timeout
pass
with patch("authsome.flows.pkce.webbrowser.open", side_effect=mock_open):
with patch("authsome.flows.pkce._CALLBACK_TIMEOUT_SECONDS", 0.01):
with pytest.raises(AuthenticationFailedError, match="timed out"):
flow.authenticate(provider, crypto, "default", "default", client_id="cid")
+4 -20
View File
@@ -1,7 +1,5 @@
"""Tests for authsome data models."""
import pytest
from authsome.models.config import GlobalConfig
from authsome.models.connection import (
ConnectionRecord,
@@ -11,7 +9,7 @@ from authsome.models.connection import (
)
from authsome.models.enums import AuthType, ConnectionStatus, ExportFormat, FlowType
from authsome.models.profile import ProfileMetadata
from authsome.models.provider import ApiKeyConfig, ClientConfig, OAuthConfig, ProviderDefinition
from authsome.models.provider import ApiKeyConfig, OAuthConfig, ProviderDefinition
class TestEnums:
@@ -23,8 +21,7 @@ class TestEnums:
def test_flow_type_values(self) -> None:
assert FlowType.DCR_PKCE.value == "dcr_pkce"
assert FlowType.API_KEY_PROMPT.value == "api_key_prompt"
assert FlowType.API_KEY_ENV.value == "api_key_env"
assert FlowType.API_KEY.value == "api_key"
def test_connection_status_values(self) -> None:
assert ConnectionStatus.CONNECTED.value == "connected"
@@ -104,40 +101,27 @@ class TestProviderDefinition:
name="openai",
display_name="OpenAI",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY_PROMPT,
flow=FlowType.API_KEY,
api_key=ApiKeyConfig(
header_name="Authorization",
header_prefix="Bearer",
env_var="OPENAI_API_KEY",
),
)
assert provider.auth_type == AuthType.API_KEY
assert provider.api_key is not None
assert provider.api_key.env_var == "OPENAI_API_KEY"
def test_json_roundtrip(self) -> None:
provider = ProviderDefinition(
name="test",
display_name="Test",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY_PROMPT,
flow=FlowType.API_KEY,
api_key=ApiKeyConfig(),
)
json_str = provider.model_dump_json()
restored = ProviderDefinition.model_validate_json(json_str)
assert restored.name == "test"
def test_client_env_resolution(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("MY_CLIENT_ID", "resolved-id")
client = ClientConfig(client_id="env:MY_CLIENT_ID", client_secret="literal-secret")
assert client.resolve_client_id() == "resolved-id"
assert client.resolve_client_secret() == "literal-secret"
def test_client_env_missing(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("NONEXISTENT_VAR", raising=False)
client = ClientConfig(client_id="env:NONEXISTENT_VAR")
assert client.resolve_client_id() is None
class TestEncryptedField:
"""Encrypted field envelope model tests."""
+90 -3
View File
@@ -19,7 +19,7 @@ def _make_api_key_provider(name: str = "testprov") -> ProviderDefinition:
name=name,
display_name=f"Test {name}",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY_PROMPT,
flow=FlowType.API_KEY,
api_key=ApiKeyConfig(env_var=f"{name.upper()}_KEY"),
)
@@ -49,6 +49,8 @@ class TestProviderRegistry:
return ProviderRegistry(home)
def test_list_providers_empty(self, registry: ProviderRegistry) -> None:
# Also remove providers dir to hit line 222
registry._providers_dir.rmdir()
providers = registry.list_providers()
# Should have bundled providers
assert isinstance(providers, list)
@@ -60,6 +62,16 @@ class TestProviderRegistry:
assert "openai" in names
assert "github" in names
def test_list_providers_by_source(self, registry: ProviderRegistry) -> None:
provider = _make_api_key_provider("customprov")
registry.register_provider(provider)
sources = registry.list_providers_by_source()
assert "bundled" in sources
assert "custom" in sources
assert any(p.name == "customprov" for p in sources["custom"])
assert any(p.name == "openai" for p in sources["bundled"])
def test_get_bundled_provider(self, registry: ProviderRegistry) -> None:
provider = registry.get_provider("openai")
assert provider.name == "openai"
@@ -101,7 +113,7 @@ class TestProviderRegistry:
name="openai",
display_name="Custom OpenAI",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY_PROMPT,
flow=FlowType.API_KEY,
api_key=ApiKeyConfig(
header_name="X-Custom",
header_prefix="Key",
@@ -147,7 +159,7 @@ class TestProviderRegistry:
name="noapikey",
display_name="No API Key",
auth_type=AuthType.API_KEY,
flow=FlowType.API_KEY_PROMPT,
flow=FlowType.API_KEY,
# Missing api_key section
)
with pytest.raises(InvalidProviderSchemaError, match="requires an 'api_key'"):
@@ -166,3 +178,78 @@ class TestProviderRegistry:
)
with pytest.raises(InvalidProviderSchemaError, match="Invalid URL"):
registry.register_provider(provider)
def test_register_oauth_provider(self, registry: ProviderRegistry) -> None:
# This covers 204->exit and 195->exit
provider = _make_oauth_provider("goodoauth")
registry.register_provider(provider)
loaded = registry.get_provider("goodoauth")
assert loaded.auth_type == AuthType.OAUTH2
def test_validate_oauth_missing_optional_url(self, registry: ProviderRegistry) -> None:
# Test 197->195 where a URL field is explicitly empty string
provider = _make_oauth_provider("opturl")
provider.oauth.token_url = "" # type: ignore
# It won't fail validation on the empty URL, but it might fail on register or be fine
registry.register_provider(provider)
def test_list_providers_with_local(self, registry: ProviderRegistry) -> None:
# Test line 71: list_providers loading local providers
registry.register_provider(_make_api_key_provider("localprov"))
providers = registry.list_providers()
assert any(p.name == "localprov" for p in providers)
def test_unrecognized_auth_type(self, registry: ProviderRegistry) -> None:
provider = _make_api_key_provider()
# Bypass Pydantic validation to simulate invalid auth_type from storage
object.__setattr__(provider, "auth_type", "INVALID_TYPE")
with pytest.raises(InvalidProviderSchemaError, match="Unrecognized auth_type"):
registry._validate_provider(provider)
def test_load_provider_file_error(self, registry: ProviderRegistry) -> None:
# Test 215-216
registry._providers_dir.mkdir(parents=True, exist_ok=True)
bad_file = registry._providers_dir / "bad.json"
bad_file.write_text("invalid json")
with pytest.raises(InvalidProviderSchemaError, match="Failed to parse provider file"):
registry._load_provider_file(bad_file)
def test_load_local_providers_error_skipping(self, registry: ProviderRegistry) -> None:
# Test 225-229
registry._providers_dir.mkdir(parents=True, exist_ok=True)
bad_file = registry._providers_dir / "bad.json"
bad_file.write_text("invalid json")
providers = registry._load_local_providers()
assert "bad" not in providers
def test_load_bundled_providers_errors(self, registry: ProviderRegistry, monkeypatch: pytest.MonkeyPatch) -> None:
# Test 243-246
import importlib.resources
# Test ModuleNotFoundError
def mock_files_error(*args, **kwargs):
raise ModuleNotFoundError()
monkeypatch.setattr(importlib.resources, "files", mock_files_error)
assert registry._load_bundled_providers() == {}
# Test JSONDecodeError inside bundled providers
monkeypatch.undo()
class MockResource:
name = "bad.json"
def read_text(self, *args, **kwargs):
return "bad json"
class MockPkg:
def iterdir(self):
return [MockResource()]
def mock_files_success(*args, **kwargs):
return MockPkg()
monkeypatch.setattr(importlib.resources, "files", mock_files_success)
assert registry._load_bundled_providers() == {}
+52
View File
@@ -83,3 +83,55 @@ class TestSQLiteStore:
key = "profile:default:my-provider:connection:test_conn-1"
store.set(key, '{"ok": true}')
assert store.get(key) == '{"ok": true}'
def test_connect_error(self, tmp_path: Path) -> None:
profile_dir = tmp_path / "profiles" / "bad"
profile_dir.mkdir(parents=True)
# Create a directory where the db file should be to trigger an error
(profile_dir / "store.db").mkdir()
from authsome.errors import StoreUnavailableError
with pytest.raises(StoreUnavailableError):
SQLiteStore(profile_dir)
def test_ensure_connection_error(self, store: SQLiteStore) -> None:
from authsome.errors import StoreUnavailableError
store.close()
with pytest.raises(StoreUnavailableError, match="Store connection is closed"):
store.get("key1")
def test_lock_acquire_release_errors(self, store: SQLiteStore, monkeypatch: pytest.MonkeyPatch) -> None:
# Mock fcntl to raise OSError
import fcntl
def mock_flock(fd, operation):
raise OSError("Mock error")
monkeypatch.setattr(fcntl, "flock", mock_flock)
# lock acquire should catch OSError and just log warning
store._acquire_lock()
assert store._lock_fd is not None
# lock release should catch OSError
store._release_lock()
assert store._lock_fd is None
def test_double_acquire(self, store: SQLiteStore) -> None:
store._acquire_lock()
fd1 = store._lock_fd
store._acquire_lock() # should early return
assert store._lock_fd is fd1
store._release_lock()
def test_close_sqlite_error(self, store: SQLiteStore) -> None:
import sqlite3
class MockConn:
def close(self):
raise sqlite3.Error("Mock error")
store._conn = MockConn()
store.close() # should suppress error
assert store._conn is None
+71
View File
@@ -0,0 +1,71 @@
"""Tests for utils.py."""
from datetime import UTC, datetime
import pytest
from authsome.utils import build_store_key, is_filesystem_safe, parse_rfc3339, to_rfc3339, utc_now
def test_utc_now():
now = utc_now()
assert now.tzinfo == UTC
def test_to_rfc3339():
dt = datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)
assert to_rfc3339(dt) == "2023-01-01T12:00:00Z"
# Test dt without tzinfo
dt_naive = datetime(2023, 1, 1, 12, 0, 0)
assert to_rfc3339(dt_naive) == "2023-01-01T12:00:00Z"
def test_parse_rfc3339():
s = "2023-01-01T12:00:00Z"
dt = parse_rfc3339(s)
assert dt.tzinfo == UTC
s_offset = "2023-01-01T12:00:00+00:00"
dt_offset = parse_rfc3339(s_offset)
assert dt_offset.tzinfo.utcoffset(dt_offset).total_seconds() == 0
def test_is_filesystem_safe():
assert is_filesystem_safe("valid-name_1.2") is True
# Test empty name
assert is_filesystem_safe("") is False
assert is_filesystem_safe(None) is False
# Test path traversal
assert is_filesystem_safe("bad/name") is False
assert is_filesystem_safe("bad..name") is False
assert is_filesystem_safe("bad\\name") is False
assert is_filesystem_safe(".hidden") is False
def test_build_store_key():
# Test definition key
assert build_store_key(record_type="definition", provider="github") == "provider:github:definition"
# Test metadata key
assert (
build_store_key(profile="default", provider="github", record_type="metadata")
== "profile:default:github:metadata"
)
# Test state key
assert build_store_key(profile="default", provider="github", record_type="state") == "profile:default:github:state"
# Test connection key
assert (
build_store_key(profile="default", provider="github", record_type="connection", connection="personal")
== "profile:default:github:connection:personal"
)
# Test client key
assert (
build_store_key(profile="default", provider="github", record_type="client") == "profile:default:github:client"
)
# Test value error
with pytest.raises(ValueError):
build_store_key(profile="default", provider="github", record_type="unknown")
# Test missing provider with profile
with pytest.raises(ValueError):
build_store_key(profile="default", record_type="metadata")
Generated
+1 -1
View File
@@ -13,7 +13,7 @@ wheels = [
[[package]]
name = "authsome"
version = "0.1.7"
version = "0.1.9"
source = { editable = "." }
dependencies = [
{ name = "click" },