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 backend.model.student import Student
from backend.model.tribe import Tribe
from tests.model.fakes import build_student, build_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 memory_sqlite_conn():
    sqlite_db = ":memory:"
    conn = sqlite3.connect(sqlite_db)
    create_db(conn)
    yield conn
    conn.close()


@pytest.fixture
def clean_db():
    sqlite_db = "sqlite.db"
    conn = sqlite3.connect(sqlite_db)
    create_db(conn)
    yield
    cursor = conn.cursor()
    cursor.execute("""DROP TABLE tribes""")
    cursor.execute("""DROP TABLE students""")
    conn.commit()


def populate_tribes(conn) -> list[Tribe]:
    cursor = conn.cursor()
    tribes = build_tribes(3)
    cursor.executemany(
        """
        INSERT INTO tribes(name, level) VALUES (?, ?)
    """,
        [t.to_tuple() for t in tribes],
    )
    conn.commit()

    return tribes


def populate_students(conn, tribes: list[Tribe]) -> list[Student]:
    cursor = conn.cursor()
    prebuild_students = build_student(tribes, 2)
    cursor.executemany(
        """
        INSERT INTO students(id, name, tribe_name) VALUES (:id, :name, :tribe_name)
    """,
        [s.to_dict() for s in prebuild_students],
    )
    conn.commit()

    return prebuild_students


@pytest.fixture
def populate_db():
    class Student_tribe_context:
        _tribes = []
        _students = []

        def __init__(self, conn):
            self.conn = conn

        def __enter__(self):
            self._tribes += populate_tribes(self.conn)
            self._students += populate_students(self.conn, self._tribes)
            return self._tribes, self._students

        def __exit__(self, *args):

            for student in self._students:
                self.conn.execute(
                    """
                    DELETE FROM students WHERE id=:id
                    """,
                    {"id": student.id},
                )
            for tribe in self._tribes:
                self.conn.execute(
                    """
                    DELETE FROM tribes WHERE name=:name
                    """,
                    {"name": tribe.name},
                )
            self.conn.commit()

    def fixture(conn):
        return Student_tribe_context(conn)

    yield fixture


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