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