135 lines
3.2 KiB
Python
135 lines
3.2 KiB
Python
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()
|