recopytex/tests/conftest.py

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()