conftest.py 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155
  1. import asyncio
  2. from collections.abc import AsyncGenerator
  3. from datetime import datetime
  4. import pytest
  5. import pytest_asyncio
  6. from httpx import AsyncClient
  7. from sqlalchemy import delete, select
  8. from sqlalchemy.orm import Session
  9. from app.core import config, security
  10. from app.core.session import engine
  11. from app.core.session import session as session_maker
  12. from app.main import app
  13. from app.models import Base, Project, Slot, Sms, User, Volunteer
  14. default_user_id = "b75365d9-7bf9-4f54-add5-aeab333a087b"
  15. default_user_email = "geralt@wiedzmin.pl"
  16. default_user_password = "geralt"
  17. default_user_password_hash = security.get_password_hash(default_user_password)
  18. default_user_access_token = security.create_jwt_token(
  19. str(default_user_id), 60 * 60 * 24, refresh=False
  20. )[0]
  21. default_project_id = "e233ac66-3a29-4ae5-991e-30ec7334c566"
  22. default_project_name = "Default project"
  23. default_volunteer_id = "5514f4ef-75ee-40b2-ad99-420927c5c9e5"
  24. default_slot_id = "def04027-0048-48e2-8f47-f955fe080b31"
  25. default_sms_id = "b05c2b38-edb1-4e7a-9689-a0cba904ef29"
  26. @pytest.fixture(scope="session")
  27. def event_loop():
  28. loop = asyncio.new_event_loop()
  29. asyncio.set_event_loop(loop)
  30. yield loop
  31. loop.close()
  32. @pytest_asyncio.fixture(scope="session")
  33. async def test_db_setup_sessionmaker():
  34. # assert if we use TEST_DB URL for 100%
  35. assert config.settings.ENVIRONMENT == "PYTEST"
  36. # always drop and create test db tables between tests session
  37. Base.metadata.drop_all(engine)
  38. Base.metadata.create_all(engine)
  39. @pytest_asyncio.fixture(autouse=True)
  40. async def session(test_db_setup_sessionmaker) -> AsyncGenerator[Session, None]:
  41. with session_maker() as db:
  42. yield db
  43. # delete all data from all tables after test
  44. for name, table in Base.metadata.tables.items():
  45. db.execute(delete(table))
  46. db.commit()
  47. @pytest_asyncio.fixture(scope="session")
  48. async def client() -> AsyncGenerator[AsyncClient, None]:
  49. async with AsyncClient(app=app, base_url="http://test") as client:
  50. client.headers.update({"Host": "localhost"})
  51. yield client
  52. @pytest.fixture
  53. def default_user(test_db_setup_sessionmaker) -> User:
  54. with session_maker() as db:
  55. result = db.execute(select(User).where(User.email == default_user_email))
  56. user = result.scalars().first()
  57. if user is None:
  58. new_user = User(
  59. email=default_user_email,
  60. hashed_password=default_user_password_hash,
  61. )
  62. new_user.id = default_user_id
  63. db.add(new_user)
  64. db.commit()
  65. db.refresh(new_user)
  66. return new_user
  67. return user
  68. @pytest.fixture
  69. def default_project(test_db_setup_sessionmaker) -> Project:
  70. """An empty private project"""
  71. with session_maker() as db:
  72. result = db.execute(select(Project).where(Project.id == default_project_id))
  73. project = result.scalars().first()
  74. if project is None:
  75. new_project = Project(name=default_project_name, is_public=False)
  76. new_project.id = default_project_id
  77. db.add(new_project)
  78. db.commit()
  79. db.refresh(new_project)
  80. return new_project
  81. return project
  82. @pytest.fixture
  83. def default_public_project(test_db_setup_sessionmaker) -> Project:
  84. """A public project with 1 volunteer, 1 slot & 1 sms associated to."""
  85. with session_maker() as db:
  86. result = db.execute(select(Project).where(Project.id == default_project_id))
  87. project = result.scalars().first()
  88. if project is None:
  89. new_project = Project(name=default_project_name, is_public=True)
  90. new_project.id = default_project_id
  91. db.add(new_project)
  92. # Create a volunteer
  93. volunteer = Volunteer(
  94. project_id=default_project_id,
  95. name="Arthur",
  96. surname="Pandragon",
  97. email="arthur.pandragon@kamelot.fr",
  98. phone_number="02 66 66 66 66 66",
  99. automatic_sms=True,
  100. )
  101. volunteer.id = default_volunteer_id
  102. db.add(volunteer)
  103. # Create a slot
  104. slot = Slot(
  105. project_id=default_project_id,
  106. title="être roi",
  107. starting_time=datetime(1600, 1, 1),
  108. ending_time=datetime(1900, 1, 1),
  109. )
  110. slot.id = default_slot_id
  111. slot.volunteers.append(volunteer)
  112. db.add(slot)
  113. # Create a sms
  114. sms = Sms(
  115. project_id=default_project_id,
  116. content="Bonjour sir",
  117. phone_number="66 66 66 66 66",
  118. )
  119. sms.id = default_sms_id
  120. db.add(sms)
  121. db.commit()
  122. db.refresh(sms)
  123. db.refresh(slot)
  124. db.refresh(volunteer)
  125. db.refresh(new_project)
  126. return new_project
  127. return project
  128. @pytest.fixture
  129. def default_user_headers(default_user: User):
  130. return {"Authorization": f"Bearer {default_user_access_token}"}