bdlg2023-sms-back/app/tests/conftest.py

import asyncio
from collections.abc import AsyncGenerator
from datetime import datetime
from uuid import uuid4

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import delete, select
from sqlalchemy.orm import Session

from app.core import config, security
from app.core.session import engine
from app.core.session import session as session_maker
from app.main import app
from app.models import (
    Base,
    GlobalRole,
    Organization,
    OrgRole,
    Project,
    Slot,
    SlotTag,
    SlotTemplate,
    Sms,
    User,
    UserOrganization,
    Volunteer,
)

default_user_id = "b75365d9-7bf9-4f54-add5-aeab333a087b"
default_user_email = "geralt@wiedzmin.pl"
default_user_password = "geralt"
default_user_password_hash = security.get_password_hash(default_user_password)
default_user_access_token = security.create_jwt_token(
    str(default_user_id), 60 * 60 * 24, refresh=False
)[0]

default_project_id = "e233ac66-3a29-4ae5-991e-30ec7334c566"
default_project_name = "Default project"
default_volunteer_id = "5514f4ef-75ee-40b2-ad99-420927c5c9e5"
default_slot_id = "def04027-0048-48e2-8f47-f955fe080b31"
default_sms_id = "b05c2b38-edb1-4e7a-9689-a0cba904ef29"
default_tag_id = "47709be8-7edb-4fb1-9de0-8a529a91856d"
default_template_id = "dcfcf2e9-8b1d-4bbf-b44b-aa511a7fe274"

default_organization_id = "a1f2e3d4-5b6c-47d8-9e0f-1a2b3c4d5e6f"
default_organization_name = "Default Org"


@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop
    loop.close()


@pytest_asyncio.fixture(scope="session")
async def test_db_setup_sessionmaker():
    assert config.settings.ENVIRONMENT == "PYTEST"
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)


@pytest_asyncio.fixture(autouse=True)
async def session(test_db_setup_sessionmaker) -> AsyncGenerator[Session, None]:
    with session_maker() as db:
        yield db
        for name, table in Base.metadata.tables.items():
            db.execute(delete(table))
        db.commit()


@pytest_asyncio.fixture(scope="session")
async def client() -> AsyncGenerator[AsyncClient, None]:
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        yield client


@pytest.fixture
def default_user(test_db_setup_sessionmaker) -> User:
    with session_maker() as db:
        result = db.execute(select(User).where(User.email == default_user_email))
        user = result.scalars().first()
        if user is None:
            new_user = User(
                email=default_user_email,
                hashed_password=default_user_password_hash,
                name="Geralt",
            )
            new_user.id = default_user_id
            db.add(new_user)
            db.commit()
            db.refresh(new_user)
            return new_user
        return user


@pytest.fixture
def default_organization(test_db_setup_sessionmaker) -> Organization:
    """Uses session_maker() directly, like default_user/default_project, so
    it's safe to depend on from fixtures that also use session_maker()
    without cross-session visibility issues (same DB, autocommitted)."""
    with session_maker() as db:
        result = db.execute(select(Organization).where(Organization.id == default_organization_id))
        org = result.scalars().first()
        if org is None:
            new_org = Organization(id=default_organization_id, name=default_organization_name)
            db.add(new_org)
            db.commit()
            db.refresh(new_org)
            return new_org
        return org


@pytest.fixture
def other_org_user(session: Session):
    """A user who is org_admin of a DIFFERENT organization entirely --
    distinct from make_org_user(role=None), which has no membership at all.
    This confirms role checks are scoped per-organization, not just
    'has some role somewhere'."""
    other_org = Organization(id=uuid4(), name="Other Org")
    session.add(other_org)
    session.commit()

    from app.core import security

    user_id = uuid4()
    user = User(
        id=user_id,
        email=f"{user_id}@test.com",
        hashed_password="x",
        name="Outsider",
    )
    session.add(user)
    session.commit()
    session.add(
        UserOrganization(user_id=user.id, organization_id=other_org.id, role=OrgRole.ORG_ADMIN)
    )
    session.commit()

    token = security.create_jwt_token(str(user_id), 60 * 60 * 24, refresh=False)[0]
    headers = {"Authorization": f"Bearer {token}"}
    return user, headers


@pytest.fixture
def default_user_org_membership(
    default_user: User, default_organization: Organization
) -> UserOrganization:
    """default_user is org_admin of default_organization."""
    with session_maker() as db:
        result = db.execute(
            select(UserOrganization).where(
                UserOrganization.user_id == default_user.id,
                UserOrganization.organization_id == default_organization.id,
            )
        )
        membership = result.scalars().first()
        if membership is None:
            new_membership = UserOrganization(
                user_id=default_user.id,
                organization_id=default_organization.id,
                role=OrgRole.ORG_ADMIN,
            )
            db.add(new_membership)
            db.commit()
            db.refresh(new_membership)
            return new_membership
        return membership


@pytest.fixture
def make_org_user(default_organization: Organization):
    """
    Factory fixture for parametrized role tests.
    make_org_user(role=OrgRole.RESPO_BENEVOLE) -> (user, headers), member of default_organization.
    make_org_user(role=None)                   -> authenticated user with NO membership (tests 403 path).
    make_org_user(global_role=GlobalRole.SUPER_ADMIN) -> bypasses org role checks entirely.
    Always creates a brand-new user (unique email) so parametrized cases don't collide.
    """

    def _factory(
        role: OrgRole | None = None,
        global_role: GlobalRole = GlobalRole.USER,
    ) -> tuple[User, dict]:
        with session_maker() as db:
            user_id = uuid4()
            user = User(
                id=user_id,
                email=f"{user_id}@test.com",
                hashed_password=default_user_password_hash,
                name="Test User",
                global_role=global_role,
            )
            db.add(user)
            db.commit()
            db.refresh(user)

            if role is not None:
                db.add(
                    UserOrganization(
                        user_id=user.id, organization_id=default_organization.id, role=role
                    )
                )
                db.commit()
                db.refresh(user)

        token = security.create_jwt_token(str(user_id), 60 * 60 * 24, refresh=False)[0]
        headers = {"Authorization": f"Bearer {token}"}
        return user, headers

    return _factory


@pytest.fixture
def default_project(default_organization: Organization) -> Project:
    """An empty private project. Depends on default_organization so the FK
    is always satisfied regardless of which fixtures a test happens to request."""
    with session_maker() as db:
        result = db.execute(select(Project).where(Project.id == default_project_id))
        project = result.scalars().first()
        if project is None:
            new_project = Project(
                organization_id=default_organization.id,
                name=default_project_name,
                is_public=False,
            )
            new_project.id = default_project_id
            db.add(new_project)
            db.commit()
            db.refresh(new_project)
            return new_project
        return project


@pytest.fixture
def default_public_project(default_organization: Organization) -> Project:
    """A public project with 1 volunteer, 1 slot & 1 sms associated to."""
    with session_maker() as db:
        result = db.execute(select(Project).where(Project.id == default_project_id))
        project = result.scalars().first()
        if project is None:
            new_project = Project(
                organization_id=default_organization.id,
                name=default_project_name,
                is_public=True,
            )
            new_project.id = default_project_id
            db.add(new_project)

            volunteer = Volunteer(
                project_id=default_project_id,
                name="Arthur",
                surname="Pandragon",
                email="arthur.pandragon@kamelot.fr",
                phone_number="02 66 66 66 66 66",
                automatic_sms=True,
            )
            volunteer.id = default_volunteer_id
            db.add(volunteer)

            slot = Slot(
                project_id=default_project_id,
                title="être roi",
                starting_time=datetime(1600, 1, 1),
                ending_time=datetime(1900, 1, 1),
            )
            slot.id = default_slot_id
            slot.volunteers.append(volunteer)
            db.add(slot)

            tag = SlotTag(project_id=default_project_id, title="Royal")
            tag.id = default_tag_id
            db.add(tag)

            sms = Sms(
                project_id=default_project_id,
                content="Bonjour sir",
                phone_number="66 66 66 66 66",
            )
            sms.id = default_sms_id
            db.add(sms)

            tmp = SlotTemplate(project_id=default_project_id, title="basic template")
            tmp.id = default_template_id
            db.add(tmp)

            db.commit()
            db.refresh(new_project)
            return new_project
        return project


@pytest.fixture
def default_user_headers(default_user: User):
    return {"Authorization": f"Bearer {default_user_access_token}"}