feat: add oauth provider methods and connected accounts to both facades

- create_oauth_provider links OAuth provider to a user
- list_oauth_providers and get_oauth_provider for provider lookup
- list_connected_accounts as alias for list_oauth_providers
- Sync and async versions on both facade classes with tests
This commit is contained in:
Ejiro Asiuwhu 2026-03-25 01:37:25 +01:00
parent 7e41fc8f95
commit eae4842ab1
2 changed files with 354 additions and 0 deletions

View File

@ -711,6 +711,59 @@ class StackServerApp:
return None
return data
# -- OAuth providers -----------------------------------------------------
def create_oauth_provider(
self,
user_id: str,
*,
account_id: str,
provider_config_id: str,
email: str,
allow_sign_in: bool,
allow_connected_accounts: bool,
) -> OAuthProvider:
"""Link an OAuth provider to a user."""
body = _build_params(
account_id=account_id,
provider_config_id=provider_config_id,
email=email,
allow_sign_in=allow_sign_in,
allow_connected_accounts=allow_connected_accounts,
)
data = self._client.request(
"POST", f"/users/{user_id}/oauth-providers", body=body
)
return OAuthProvider.model_validate(data)
def list_oauth_providers(self, user_id: str) -> list[OAuthProvider]:
"""List OAuth providers linked to a user."""
data = self._client.request(
"GET", f"/users/{user_id}/oauth-providers"
)
return [
OAuthProvider.model_validate(item)
for item in (data or {}).get("items", [])
]
def get_oauth_provider(
self, user_id: str, provider_id: str
) -> OAuthProvider | None:
"""Get a specific OAuth provider for a user.
Fetches all providers and filters by *provider_id*.
Returns ``None`` if not found.
"""
providers = self.list_oauth_providers(user_id)
return next((p for p in providers if p.id == provider_id), None)
def list_connected_accounts(self, user_id: str) -> list[OAuthProvider]:
"""List connected accounts for a user.
This is an alias for :meth:`list_oauth_providers`.
"""
return self.list_oauth_providers(user_id)
# ---------------------------------------------------------------------------
# AsyncStackServerApp (async)
@ -1328,3 +1381,60 @@ class AsyncStackServerApp:
if data is None:
return None
return data
# -- OAuth providers -----------------------------------------------------
async def create_oauth_provider(
self,
user_id: str,
*,
account_id: str,
provider_config_id: str,
email: str,
allow_sign_in: bool,
allow_connected_accounts: bool,
) -> OAuthProvider:
"""Link an OAuth provider to a user."""
body = _build_params(
account_id=account_id,
provider_config_id=provider_config_id,
email=email,
allow_sign_in=allow_sign_in,
allow_connected_accounts=allow_connected_accounts,
)
data = await self._client.request(
"POST", f"/users/{user_id}/oauth-providers", body=body
)
return OAuthProvider.model_validate(data)
async def list_oauth_providers(
self, user_id: str
) -> list[OAuthProvider]:
"""List OAuth providers linked to a user."""
data = await self._client.request(
"GET", f"/users/{user_id}/oauth-providers"
)
return [
OAuthProvider.model_validate(item)
for item in (data or {}).get("items", [])
]
async def get_oauth_provider(
self, user_id: str, provider_id: str
) -> OAuthProvider | None:
"""Get a specific OAuth provider for a user.
Fetches all providers and filters by *provider_id*.
Returns ``None`` if not found.
"""
providers = await self.list_oauth_providers(user_id)
return next((p for p in providers if p.id == provider_id), None)
async def list_connected_accounts(
self, user_id: str
) -> list[OAuthProvider]:
"""List connected accounts for a user.
This is an alias for :meth:`list_oauth_providers`.
"""
return await self.list_oauth_providers(user_id)

View File

@ -2169,3 +2169,247 @@ class TestAsyncCheckApiKey:
result = await app.check_api_key("invalid")
assert result is None
await app.aclose()
# ---------------------------------------------------------------------------
# OAuth provider test data
# ---------------------------------------------------------------------------
OAUTH_PROVIDER_JSON = {
"id": "provider-1",
"type": "google",
"userId": "user-1",
"accountId": "google-123",
"email": "a@b.com",
"allowSignIn": True,
"allowConnectedAccounts": False,
}
OAUTH_PROVIDER_JSON_2 = {
"id": "provider-2",
"type": "github",
"userId": "user-1",
"accountId": "gh-456",
"email": "a@github.com",
"allowSignIn": False,
"allowConnectedAccounts": True,
}
# ---------------------------------------------------------------------------
# StackServerApp - create_oauth_provider
# ---------------------------------------------------------------------------
class TestCreateOAuthProvider:
@respx.mock
def test_create_oauth_provider(self) -> None:
route = respx.post(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(200, json=OAUTH_PROVIDER_JSON)
)
app = StackServerApp(project_id="proj", secret_server_key="sk")
provider = app.create_oauth_provider(
"user-1",
account_id="google-123",
provider_config_id="google",
email="a@b.com",
allow_sign_in=True,
allow_connected_accounts=False,
)
assert provider.id == "provider-1"
assert provider.type == "google"
assert provider.user_id == "user-1"
assert provider.email == "a@b.com"
assert provider.allow_sign_in is True
assert provider.allow_connected_accounts is False
@respx.mock
def test_create_oauth_provider_sends_correct_body(self) -> None:
route = respx.post(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(200, json=OAUTH_PROVIDER_JSON)
)
app = StackServerApp(project_id="proj", secret_server_key="sk")
app.create_oauth_provider(
"user-1",
account_id="google-123",
provider_config_id="google",
email="a@b.com",
allow_sign_in=True,
allow_connected_accounts=False,
)
import json
body = json.loads(route.calls[0].request.content)
assert body["account_id"] == "google-123"
assert body["provider_config_id"] == "google"
assert body["email"] == "a@b.com"
assert body["allow_sign_in"] is True
assert body["allow_connected_accounts"] is False
# ---------------------------------------------------------------------------
# StackServerApp - list_oauth_providers
# ---------------------------------------------------------------------------
class TestListOAuthProviders:
@respx.mock
def test_list_oauth_providers(self) -> None:
respx.get(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(
200,
json={"items": [OAUTH_PROVIDER_JSON, OAUTH_PROVIDER_JSON_2]},
)
)
app = StackServerApp(project_id="proj", secret_server_key="sk")
providers = app.list_oauth_providers("user-1")
assert len(providers) == 2
assert providers[0].id == "provider-1"
assert providers[1].id == "provider-2"
@respx.mock
def test_list_oauth_providers_empty(self) -> None:
respx.get(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(200, json={"items": []})
)
app = StackServerApp(project_id="proj", secret_server_key="sk")
providers = app.list_oauth_providers("user-1")
assert providers == []
# ---------------------------------------------------------------------------
# StackServerApp - get_oauth_provider
# ---------------------------------------------------------------------------
class TestGetOAuthProvider:
@respx.mock
def test_get_oauth_provider_found(self) -> None:
respx.get(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(
200,
json={"items": [OAUTH_PROVIDER_JSON, OAUTH_PROVIDER_JSON_2]},
)
)
app = StackServerApp(project_id="proj", secret_server_key="sk")
provider = app.get_oauth_provider("user-1", "provider-2")
assert provider is not None
assert provider.id == "provider-2"
assert provider.type == "github"
@respx.mock
def test_get_oauth_provider_not_found(self) -> None:
respx.get(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(200, json={"items": []})
)
app = StackServerApp(project_id="proj", secret_server_key="sk")
provider = app.get_oauth_provider("user-1", "nonexistent")
assert provider is None
# ---------------------------------------------------------------------------
# StackServerApp - list_connected_accounts
# ---------------------------------------------------------------------------
class TestListConnectedAccounts:
@respx.mock
def test_list_connected_accounts(self) -> None:
respx.get(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(
200,
json={"items": [OAUTH_PROVIDER_JSON]},
)
)
app = StackServerApp(project_id="proj", secret_server_key="sk")
accounts = app.list_connected_accounts("user-1")
assert len(accounts) == 1
assert accounts[0].id == "provider-1"
# ---------------------------------------------------------------------------
# AsyncStackServerApp - OAuth provider methods
# ---------------------------------------------------------------------------
class TestAsyncCreateOAuthProvider:
@respx.mock
@pytest.mark.asyncio
async def test_create_oauth_provider(self) -> None:
respx.post(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(200, json=OAUTH_PROVIDER_JSON)
)
app = AsyncStackServerApp(project_id="proj", secret_server_key="sk")
provider = await app.create_oauth_provider(
"user-1",
account_id="google-123",
provider_config_id="google",
email="a@b.com",
allow_sign_in=True,
allow_connected_accounts=False,
)
assert provider.id == "provider-1"
assert provider.type == "google"
await app.aclose()
class TestAsyncListOAuthProviders:
@respx.mock
@pytest.mark.asyncio
async def test_list_oauth_providers(self) -> None:
respx.get(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(
200,
json={"items": [OAUTH_PROVIDER_JSON, OAUTH_PROVIDER_JSON_2]},
)
)
app = AsyncStackServerApp(project_id="proj", secret_server_key="sk")
providers = await app.list_oauth_providers("user-1")
assert len(providers) == 2
assert providers[0].id == "provider-1"
await app.aclose()
class TestAsyncGetOAuthProvider:
@respx.mock
@pytest.mark.asyncio
async def test_get_oauth_provider_found(self) -> None:
respx.get(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(
200,
json={"items": [OAUTH_PROVIDER_JSON, OAUTH_PROVIDER_JSON_2]},
)
)
app = AsyncStackServerApp(project_id="proj", secret_server_key="sk")
provider = await app.get_oauth_provider("user-1", "provider-2")
assert provider is not None
assert provider.id == "provider-2"
await app.aclose()
@respx.mock
@pytest.mark.asyncio
async def test_get_oauth_provider_not_found(self) -> None:
respx.get(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(200, json={"items": []})
)
app = AsyncStackServerApp(project_id="proj", secret_server_key="sk")
provider = await app.get_oauth_provider("user-1", "nonexistent")
assert provider is None
await app.aclose()
class TestAsyncListConnectedAccounts:
@respx.mock
@pytest.mark.asyncio
async def test_list_connected_accounts(self) -> None:
respx.get(f"{API_PREFIX}/users/user-1/oauth-providers").mock(
return_value=httpx.Response(
200,
json={"items": [OAUTH_PROVIDER_JSON]},
)
)
app = AsyncStackServerApp(project_id="proj", secret_server_key="sk")
accounts = await app.list_connected_accounts("user-1")
assert len(accounts) == 1
assert accounts[0].id == "provider-1"
await app.aclose()