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

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

import pytest
import pytest_asyncio
from httpx import 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, Project, Slot, SlotTag, SlotTemplate, Sms, User, 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"


@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 if we use TEST_DB URL for 100%
    assert config.settings.ENVIRONMENT == "PYTEST"

    # always drop and create test db tables between tests session
    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

        # delete all data from all tables after test
        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(app=app, base_url="http://test") as client:
        client.headers.update({"Host": "localhost"})
        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,
            )
            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_project(test_db_setup_sessionmaker) -> Project:
    """An empty private project"""
    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(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(test_db_setup_sessionmaker) -> 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(name=default_project_name, is_public=True)
            new_project.id = default_project_id
            db.add(new_project)
            # Create a volunteer
            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)
            # Create a slot
            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)
            # Create a sms
            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.commit()
            db.refresh(sms)
            db.refresh(slot)
            db.refresh(volunteer)
            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}"}