diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py new file mode 100644 index 0000000..04b9c33 --- /dev/null +++ b/backend/tests/test_config.py @@ -0,0 +1,111 @@ +import pytest +from unittest.mock import patch +from string import Template + +from app.config import Settings, settings + + +def test_settings_is_singleton_like(): + """The module-level `settings` object is reused across imports.""" + from app.config import settings as s2 + assert settings is s2 + + +def test_cors_origins_defaults(): + s = Settings( + SECRET_KEY="test-key", + SQLALCHEMY_DATABASE_URI="sqlite+aiosqlite:///:memory:", + ) + assert "http://localhost" in s.CORS_ORIGINS + assert "http://localhost:5173" in s.CORS_ORIGINS + assert len(s.CORS_ORIGINS) == 2 + + +def test_cors_origins_are_strings(): + s = Settings( + SECRET_KEY="test-key", + SQLALCHEMY_DATABASE_URI="sqlite+aiosqlite:///:memory:", + ) + for origin in s.CORS_ORIGINS: + assert isinstance(origin, str) + + +def test_role_request_notification_message_has_placeholders(): + s = Settings( + SECRET_KEY="test-key", + SQLALCHEMY_DATABASE_URI="sqlite+aiosqlite:///:memory:", + ) + rendered = Template(s.ROLE_REQUEST_NOTIFICATION_MESSAGE).substitute( + user="alice", role="admin" + ) + assert "alice" in rendered + assert "admin" in rendered + + +def test_role_request_approved_message_has_role_placeholder(): + s = Settings( + SECRET_KEY="test-key", + SQLALCHEMY_DATABASE_URI="sqlite+aiosqlite:///:memory:", + ) + rendered = Template(s.ROLE_REQUEST_APPROVED_MESSAGE).substitute(role="jury") + assert "jury" in rendered + + +def test_role_request_rejected_message_has_role_placeholder(): + s = Settings( + SECRET_KEY="test-key", + SQLALCHEMY_DATABASE_URI="sqlite+aiosqlite:///:memory:", + ) + rendered = Template(s.ROLE_REQUEST_REJECTED_MESSAGE).substitute(role="moderator") + assert "moderator" in rendered + + +def test_settings_secret_key_from_constructor(): + s = Settings( + SECRET_KEY="my-secret", + SQLALCHEMY_DATABASE_URI="sqlite+aiosqlite:///:memory:", + ) + assert s.SECRET_KEY == "my-secret" + + +def test_settings_database_uri_from_constructor(): + uri = "postgresql+asyncpg://user:pass@localhost/testdb" + s = Settings( + SECRET_KEY="key", + SQLALCHEMY_DATABASE_URI=uri, + ) + assert s.SQLALCHEMY_DATABASE_URI == uri + + +def test_settings_secret_key_from_env(monkeypatch): + monkeypatch.setenv("SECRET_KEY", "env-secret") + monkeypatch.setenv("SQLALCHEMY_DATABASE_URI", "sqlite+aiosqlite:///:memory:") + s = Settings() + assert s.SECRET_KEY == "env-secret" + + +def test_settings_database_uri_from_env(monkeypatch): + monkeypatch.setenv("SECRET_KEY", "key") + uri = "postgresql+asyncpg://u:p@host/db" + monkeypatch.setenv("SQLALCHEMY_DATABASE_URI", uri) + s = Settings() + assert s.SQLALCHEMY_DATABASE_URI == uri + + +def test_role_request_notification_message_missing_placeholder_raises(): + """Substituting with missing kwargs should raise KeyError/ValueError.""" + s = Settings( + SECRET_KEY="key", + SQLALCHEMY_DATABASE_URI="sqlite+aiosqlite:///:memory:", + ) + # Message needs both $user and $role; omitting one should fail + with pytest.raises((KeyError, ValueError)): + Template(s.ROLE_REQUEST_NOTIFICATION_MESSAGE).substitute(user="alice") + + +def test_settings_cors_origins_can_be_overridden(monkeypatch): + monkeypatch.setenv("SECRET_KEY", "key") + monkeypatch.setenv("SQLALCHEMY_DATABASE_URI", "sqlite+aiosqlite:///:memory:") + monkeypatch.setenv("CORS_ORIGINS", '["http://example.com"]') + s = Settings() + assert "http://example.com" in s.CORS_ORIGINS \ No newline at end of file diff --git a/backend/tests/test_db.py b/backend/tests/test_db.py new file mode 100644 index 0000000..dd1b858 --- /dev/null +++ b/backend/tests/test_db.py @@ -0,0 +1,36 @@ +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db import get_session + + +async def test_get_session_yields_async_session(): + """get_session should yield an AsyncSession instance.""" + gen = get_session() + session = await gen.__anext__() + assert isinstance(session, AsyncSession) + # Close the generator cleanly + try: + await gen.aclose() + except Exception: + pass + + +async def test_get_session_closes_after_use(): + """After the generator is exhausted, the session is no longer active.""" + sessions = [] + async for sess in get_session(): + sessions.append(sess) + assert len(sessions) == 1 + assert isinstance(sessions[0], AsyncSession) + + +async def test_get_session_provides_new_session_each_call(): + """Each call to get_session should produce an independent session.""" + session1 = None + session2 = None + async for s in get_session(): + session1 = s + async for s in get_session(): + session2 = s + assert session1 is not session2 \ No newline at end of file diff --git a/backend/tests/test_dependencies.py b/backend/tests/test_dependencies.py new file mode 100644 index 0000000..1f7afe8 --- /dev/null +++ b/backend/tests/test_dependencies.py @@ -0,0 +1,233 @@ +"""Tests for backend/app/dependencies/current_user.py""" +import pytest +from unittest.mock import MagicMock, AsyncMock, patch +from fastapi import HTTPException, status +from fastapi.security import HTTPAuthorizationCredentials +from firebase_admin.auth import ( + InvalidIdTokenError, + ExpiredIdTokenError, + RevokedIdTokenError, + CertificateFetchError, + UserDisabledError, +) +from sqlalchemy import select +from sqlalchemy.orm import selectinload + +from app.dependencies.current_user import get_current_user +from app.models import User +from tests.factories import UserFactory + + +def _make_bearer_token(credentials: str = "valid-token"): + """Create a mock HTTPAuthorizationCredentials object (HTTP path).""" + token = MagicMock(spec=HTTPAuthorizationCredentials) + token.credentials = credentials + return token + + +def _make_firebase_user(uid: str, email: str, display_name: str | None = "Test User"): + """Create a mock firebase_admin.auth.UserRecord.""" + user_record = MagicMock() + user_record.uid = uid + user_record.email = email + user_record.display_name = display_name + return user_record + + +async def test_get_current_user_returns_existing_user(db_session, create): + """Valid token for a user already in the DB returns that user.""" + user = await create(UserFactory) + # Reload with relationships to avoid lazy-loading issues + stmt = select(User).where(User.id == user.id).options(selectinload(User.roles)) + user = (await db_session.execute(stmt)).unique().scalar_one() + + firebase_token_data = {"email": user.email} + firebase_user_record = _make_firebase_user(uid=user.firebase_uid, email=user.email) + bearer_token = _make_bearer_token("valid-token") + + with ( + patch("app.dependencies.current_user.auth.verify_id_token", return_value=firebase_token_data), + patch("app.dependencies.current_user.auth.get_user_by_email", return_value=firebase_user_record), + ): + result = await get_current_user(session=db_session, token=bearer_token) + + assert result.id == user.id + assert result.email == user.email + + +async def test_get_current_user_creates_new_user_when_not_found(db_session, create): + """Valid token for a user not yet in DB creates and returns a new User record.""" + uid = "new-firebase-uid-123" + email = "newuser@example.com" + display_name = "New User" + + firebase_token_data = {"email": email} + firebase_user_record = _make_firebase_user(uid=uid, email=email, display_name=display_name) + bearer_token = _make_bearer_token("valid-token") + + with ( + patch("app.dependencies.current_user.auth.verify_id_token", return_value=firebase_token_data), + patch("app.dependencies.current_user.auth.get_user_by_email", return_value=firebase_user_record), + ): + result = await get_current_user(session=db_session, token=bearer_token) + + assert result.firebase_uid == uid + assert result.email == email + assert result.full_name == display_name + assert result.id is not None + + # Verify it's actually in the DB + stmt = select(User).where(User.firebase_uid == uid) + db_user = (await db_session.execute(stmt)).scalar_one_or_none() + assert db_user is not None + + +async def test_get_current_user_raises_400_when_display_name_null(db_session): + """If Firebase user has no display_name and they aren't in the DB, raise 400.""" + uid = "no-name-uid" + email = "nonameuser@example.com" + + firebase_token_data = {"email": email} + firebase_user_record = _make_firebase_user(uid=uid, email=email, display_name=None) + bearer_token = _make_bearer_token("valid-token") + + with ( + patch("app.dependencies.current_user.auth.verify_id_token", return_value=firebase_token_data), + patch("app.dependencies.current_user.auth.get_user_by_email", return_value=firebase_user_record), + ): + with pytest.raises(HTTPException) as exc_info: + await get_current_user(session=db_session, token=bearer_token) + + assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST + assert "Full name" in exc_info.value.detail + + +async def test_get_current_user_raises_401_on_invalid_token(db_session): + """InvalidIdTokenError from Firebase raises HTTP 401.""" + bearer_token = _make_bearer_token("bad-token") + + with patch( + "app.dependencies.current_user.auth.verify_id_token", + side_effect=InvalidIdTokenError("bad token"), + ): + with pytest.raises(HTTPException) as exc_info: + await get_current_user(session=db_session, token=bearer_token) + + assert exc_info.value.status_code == status.HTTP_401_UNAUTHORIZED + assert "Invalid id token" in exc_info.value.detail + + +async def test_get_current_user_raises_401_on_expired_token(db_session): + """ExpiredIdTokenError from Firebase raises HTTP 401.""" + bearer_token = _make_bearer_token("expired-token") + + with patch( + "app.dependencies.current_user.auth.verify_id_token", + side_effect=ExpiredIdTokenError("expired", cause=None), + ): + with pytest.raises(HTTPException) as exc_info: + await get_current_user(session=db_session, token=bearer_token) + + assert exc_info.value.status_code == status.HTTP_401_UNAUTHORIZED + + +async def test_get_current_user_raises_401_on_revoked_token(db_session): + """RevokedIdTokenError from Firebase raises HTTP 401.""" + bearer_token = _make_bearer_token("revoked-token") + + with patch( + "app.dependencies.current_user.auth.verify_id_token", + side_effect=RevokedIdTokenError("revoked"), + ): + with pytest.raises(HTTPException) as exc_info: + await get_current_user(session=db_session, token=bearer_token) + + assert exc_info.value.status_code == status.HTTP_401_UNAUTHORIZED + + +async def test_get_current_user_raises_401_on_value_error(db_session): + """ValueError from verify_id_token raises HTTP 401.""" + bearer_token = _make_bearer_token("garbage-token") + + with patch( + "app.dependencies.current_user.auth.verify_id_token", + side_effect=ValueError("not a token"), + ): + with pytest.raises(HTTPException) as exc_info: + await get_current_user(session=db_session, token=bearer_token) + + assert exc_info.value.status_code == status.HTTP_401_UNAUTHORIZED + + +async def test_get_current_user_raises_401_on_user_disabled(db_session): + """UserDisabledError from Firebase raises HTTP 401.""" + bearer_token = _make_bearer_token("disabled-token") + + with patch( + "app.dependencies.current_user.auth.verify_id_token", + side_effect=UserDisabledError("disabled"), + ): + with pytest.raises(HTTPException) as exc_info: + await get_current_user(session=db_session, token=bearer_token) + + assert exc_info.value.status_code == status.HTTP_401_UNAUTHORIZED + + +async def test_get_current_user_websocket_path_uses_string_token(db_session, create): + """When token is a plain string (WebSocket), it is passed directly to verify_id_token.""" + user = await create(UserFactory) + stmt = select(User).where(User.id == user.id).options(selectinload(User.roles)) + user = (await db_session.execute(stmt)).unique().scalar_one() + + firebase_token_data = {"email": user.email} + firebase_user_record = _make_firebase_user(uid=user.firebase_uid, email=user.email) + # Plain string — no 'credentials' attribute + string_token = "plain-string-token" + + with ( + patch("app.dependencies.current_user.auth.verify_id_token", return_value=firebase_token_data) as mock_verify, + patch("app.dependencies.current_user.auth.get_user_by_email", return_value=firebase_user_record), + ): + result = await get_current_user(session=db_session, token=string_token) + + # verify_id_token should be called with the raw string, not with .credentials + mock_verify.assert_called_once() + call_args = mock_verify.call_args[0] + assert call_args[0] == string_token + assert result.id == user.id + + +async def test_get_current_user_http_path_uses_credentials_attribute(db_session, create): + """When token has a 'credentials' attr (HTTP), the .credentials value is passed to verify_id_token.""" + user = await create(UserFactory) + stmt = select(User).where(User.id == user.id).options(selectinload(User.roles)) + user = (await db_session.execute(stmt)).unique().scalar_one() + + firebase_token_data = {"email": user.email} + firebase_user_record = _make_firebase_user(uid=user.firebase_uid, email=user.email) + bearer_token = _make_bearer_token("http-cred-value") + + with ( + patch("app.dependencies.current_user.auth.verify_id_token", return_value=firebase_token_data) as mock_verify, + patch("app.dependencies.current_user.auth.get_user_by_email", return_value=firebase_user_record), + ): + result = await get_current_user(session=db_session, token=bearer_token) + + mock_verify.assert_called_once() + call_args = mock_verify.call_args[0] + assert call_args[0] == "http-cred-value" + assert result.id == user.id + + +async def test_get_current_user_raises_401_on_certificate_fetch_error(db_session): + """CertificateFetchError from Firebase raises HTTP 401.""" + bearer_token = _make_bearer_token("token") + + with patch( + "app.dependencies.current_user.auth.verify_id_token", + side_effect=CertificateFetchError(cause=Exception("fetch failed"), message="cert error"), + ): + with pytest.raises(HTTPException) as exc_info: + await get_current_user(session=db_session, token=bearer_token) + + assert exc_info.value.status_code == status.HTTP_401_UNAUTHORIZED