01e3382aaf
- Use Annotated[str, Form()] for FastAPI dependencies (FAST002) - Add missing type annotations across src/ and tests/ (ANN001/003/201/202) - Reduce function arguments via request.form() reads (PLR0913) - Combine return paths to reduce return statements (PLR0911) - Use anyio.Path for async-safe filesystem operations (ASYNC240) - Extract constants, helpers, and dict comprehensions for clarity - Move inline imports to top-level (PLC0415) - Use raw strings for regex match patterns (RUF043) - Fix redundant get_session_user call in delete_user (not-iterable) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import aiosqlite
|
|
|
|
from porchlight.models import User
|
|
from porchlight.store.protocols import ConsentRepository
|
|
from porchlight.store.sqlite.repositories import SQLiteConsentRepository, SQLiteUserRepository
|
|
|
|
|
|
async def _create_user(db: aiosqlite.Connection) -> User:
|
|
"""Helper to create a test user."""
|
|
user_repo = SQLiteUserRepository(db)
|
|
user = User(
|
|
userid="test-user-id",
|
|
username="testuser",
|
|
email="test@example.com",
|
|
created_at=datetime.now(UTC),
|
|
updated_at=datetime.now(UTC),
|
|
)
|
|
return await user_repo.create(user)
|
|
|
|
|
|
async def test_implements_protocol(db: aiosqlite.Connection) -> None:
|
|
repo = SQLiteConsentRepository(db)
|
|
assert isinstance(repo, ConsentRepository)
|
|
|
|
|
|
async def test_set_and_get_consent(db: aiosqlite.Connection) -> None:
|
|
user = await _create_user(db)
|
|
repo = SQLiteConsentRepository(db)
|
|
await repo.set_consent(user.userid, "test-rp", ["openid", "profile"])
|
|
|
|
consent = await repo.get_consent(user.userid, "test-rp")
|
|
assert consent is not None
|
|
assert consent.userid == user.userid
|
|
assert consent.client_id == "test-rp"
|
|
assert consent.scopes == ["openid", "profile"]
|
|
assert isinstance(consent.created_at, datetime)
|
|
assert isinstance(consent.updated_at, datetime)
|
|
|
|
|
|
async def test_get_consent_not_found(db: aiosqlite.Connection) -> None:
|
|
user = await _create_user(db)
|
|
repo = SQLiteConsentRepository(db)
|
|
consent = await repo.get_consent(user.userid, "nonexistent")
|
|
assert consent is None
|
|
|
|
|
|
async def test_set_consent_upserts(db: aiosqlite.Connection) -> None:
|
|
user = await _create_user(db)
|
|
repo = SQLiteConsentRepository(db)
|
|
await repo.set_consent(user.userid, "test-rp", ["openid"])
|
|
|
|
original = await repo.get_consent(user.userid, "test-rp")
|
|
assert original is not None
|
|
|
|
await repo.set_consent(user.userid, "test-rp", ["openid", "profile", "email"])
|
|
|
|
consent = await repo.get_consent(user.userid, "test-rp")
|
|
assert consent is not None
|
|
assert consent.scopes == ["openid", "profile", "email"]
|
|
assert consent.created_at == original.created_at
|
|
assert consent.updated_at >= original.updated_at
|
|
|
|
|
|
async def test_delete_consent(db: aiosqlite.Connection) -> None:
|
|
user = await _create_user(db)
|
|
repo = SQLiteConsentRepository(db)
|
|
await repo.set_consent(user.userid, "test-rp", ["openid"])
|
|
|
|
result = await repo.delete_consent(user.userid, "test-rp")
|
|
assert result is True
|
|
|
|
consent = await repo.get_consent(user.userid, "test-rp")
|
|
assert consent is None
|
|
|
|
|
|
async def test_delete_consent_not_found(db: aiosqlite.Connection) -> None:
|
|
user = await _create_user(db)
|
|
repo = SQLiteConsentRepository(db)
|
|
result = await repo.delete_consent(user.userid, "nonexistent")
|
|
assert result is False
|
|
|
|
|
|
async def test_list_consents(db: aiosqlite.Connection) -> None:
|
|
user = await _create_user(db)
|
|
repo = SQLiteConsentRepository(db)
|
|
await repo.set_consent(user.userid, "rp-a", ["openid"])
|
|
await repo.set_consent(user.userid, "rp-b", ["openid", "profile"])
|
|
|
|
consents = await repo.list_consents(user.userid)
|
|
assert len(consents) == 2
|
|
client_ids = {c.client_id for c in consents}
|
|
assert client_ids == {"rp-a", "rp-b"}
|
|
|
|
|
|
async def test_list_consents_empty(db: aiosqlite.Connection) -> None:
|
|
user = await _create_user(db)
|
|
repo = SQLiteConsentRepository(db)
|
|
consents = await repo.list_consents(user.userid)
|
|
assert consents == []
|
|
|
|
|
|
async def test_consent_deleted_on_user_cascade(db: aiosqlite.Connection) -> None:
|
|
"""Consent records are deleted when the user is deleted (CASCADE)."""
|
|
user = await _create_user(db)
|
|
user_repo = SQLiteUserRepository(db)
|
|
repo = SQLiteConsentRepository(db)
|
|
|
|
await repo.set_consent(user.userid, "test-rp", ["openid"])
|
|
await user_repo.delete(user.userid)
|
|
|
|
consent = await repo.get_consent(user.userid, "test-rp")
|
|
assert consent is None
|