import sqlite3 import time from pathlib import Path import pytest import requests from sqlalchemy import create_engine from sqlalchemy.orm import clear_mappers, sessionmaker from backend import config from backend.adapters.orm import metadata, start_mappers from backend.adapters.sqlite import create_db from tests.integration.test_repository_student_sqlite import populate_students from tests.integration.test_repository_tribe_sqlite import populate_tribes @pytest.fixture def in_memory_db(): engine = create_engine("sqlite:///:memory:") metadata.create_all(engine) return engine @pytest.fixture def session(in_memory_db): start_mappers() yield sessionmaker(bind=in_memory_db)() clear_mappers() @pytest.fixture def sqlite_conn(): conn = sqlite3.connect("sqlite.db") create_db(conn) yield conn conn.close() @pytest.fixture def clean_db(sqlite_conn): sqlite_conn.execute("""DROP TABLE tribes""") sqlite_conn.execute("""DROP TABLE students""") sqlite_conn.commit() create_db(sqlite_conn) @pytest.fixture def populate_db(sqlite_conn): _tribes = [] _students = [] def _populate_db(): tribes = populate_tribes(sqlite_conn) _tribes += tribes students = populate_students(sqlite_conn, tribes) _students += students return tribes, students yield _populate_db for student in _students: sqlite_conn.execute( """ DELETE FROM students WHERE id=:id """, {"id": student.id}, ) for tribe in _tribes: sqlite_conn.execute( """ DELETE FROM tribes WHERE name=:name """, {"name": tribe.name}, ) sqlite_conn.commit() def wait_for_webapp_to_come_up(): deadline = time.time() + 10 url = config.get_api_url() while time.time() < deadline: try: return requests.get(url) except ConnectionError: time.sleep(0.5) pytest.fail("API never came up") @pytest.fixture def restart_api(): (Path(__file__).parent.parent / "backend" / "api" / "main.py").touch() time.sleep(0.5) wait_for_webapp_to_come_up()