Refact: rewrite populate db
This commit is contained in:
@@ -10,8 +10,9 @@ 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
|
||||
from backend.model.student import Student
|
||||
from backend.model.tribe import Tribe
|
||||
from tests.model.fakes import build_student, build_tribes
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -29,7 +30,7 @@ def session(in_memory_db):
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sqlite_conn():
|
||||
def memory_sqlite_conn():
|
||||
sqlite_db = ":memory:"
|
||||
conn = sqlite3.connect(sqlite_db)
|
||||
create_db(conn)
|
||||
@@ -49,35 +50,70 @@ def clean_db():
|
||||
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(sqlite_conn):
|
||||
_tribes = []
|
||||
_students = []
|
||||
def populate_db():
|
||||
class Student_tribe_context:
|
||||
_tribes = []
|
||||
_students = []
|
||||
|
||||
def _populate_db():
|
||||
tribes = populate_tribes(sqlite_conn)
|
||||
_tribes += tribes
|
||||
students = populate_students(sqlite_conn, tribes)
|
||||
_students += students
|
||||
return tribes, students
|
||||
def __init__(self, conn):
|
||||
self.conn = conn
|
||||
|
||||
yield _populate_db
|
||||
def __enter__(self):
|
||||
self._tribes += populate_tribes(self.conn)
|
||||
self._students += populate_students(self.conn, self._tribes)
|
||||
return self._tribes, self._students
|
||||
|
||||
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 __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():
|
||||
|
||||
@@ -3,135 +3,105 @@ import sqlite3
|
||||
import pytest
|
||||
|
||||
from backend.model.student import Student
|
||||
from backend.model.tribe import Tribe
|
||||
from backend.repository.student_sqlite_repository import (
|
||||
StudentRepositoryError,
|
||||
StudentSQLiteRepository,
|
||||
)
|
||||
from tests.integration.test_repository_tribe_sqlite import populate_tribes
|
||||
from tests.model.fakes import build_student
|
||||
|
||||
|
||||
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()
|
||||
def test_get_student(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, prebuild_students):
|
||||
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
return prebuild_students
|
||||
student_id = prebuild_students[0].id
|
||||
student = student_repo.get(student_id, prebuild_tribes)
|
||||
|
||||
assert prebuild_students[0] == student
|
||||
|
||||
|
||||
def test_get_student(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
|
||||
|
||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
||||
|
||||
student_id = prebuild_students[0].id
|
||||
student = student_repo.get(student_id, prebuild_tribes)
|
||||
|
||||
assert prebuild_students[0] == student
|
||||
def test_get_student_not_exists(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||
with pytest.raises(ValueError):
|
||||
student_repo.get("student0", prebuild_tribes)
|
||||
|
||||
|
||||
def test_get_student_not_exists(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
|
||||
def test_list_students(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, prebuild_students):
|
||||
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||
students = student_repo.list(prebuild_tribes)
|
||||
|
||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
||||
with pytest.raises(ValueError):
|
||||
student_repo.get("student0", prebuild_tribes)
|
||||
assert prebuild_students == students
|
||||
|
||||
|
||||
def test_list_students(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
|
||||
def test_add_student(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
||||
students = student_repo.list(prebuild_tribes)
|
||||
|
||||
assert prebuild_students == students
|
||||
|
||||
|
||||
def test_add_student(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
|
||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
||||
|
||||
student_infos = {"name": "student1", "tribe": prebuild_tribes[0]}
|
||||
student = Student(**student_infos)
|
||||
student_repo.add(student)
|
||||
sqlite_conn.commit()
|
||||
|
||||
cursor = sqlite_conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT id, name, tribe_name FROM students WHERE id=?
|
||||
""",
|
||||
(student.id,),
|
||||
)
|
||||
|
||||
row = cursor.fetchone()
|
||||
assert row == student.to_tuple()
|
||||
|
||||
|
||||
def test_add_student_fail_exists(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
|
||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
||||
|
||||
student_infos = {"name": "student1", "tribe": prebuild_tribes[0]}
|
||||
student = Student(**student_infos)
|
||||
student_repo.add(student)
|
||||
sqlite_conn.commit()
|
||||
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
student_infos = {"name": "student1", "tribe": prebuild_tribes[0]}
|
||||
student = Student(**student_infos)
|
||||
student_repo.add(student)
|
||||
memory_sqlite_conn.commit()
|
||||
|
||||
cursor = memory_sqlite_conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT id, name, tribe_name FROM students WHERE id=?
|
||||
""",
|
||||
(student.id,),
|
||||
)
|
||||
|
||||
row = cursor.fetchone()
|
||||
assert row == student.to_tuple()
|
||||
|
||||
|
||||
def test_update_student(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
|
||||
def test_add_student_fail_exists(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
||||
student_infos = {"name": "student1", "tribe": prebuild_tribes[0]}
|
||||
student = Student(**student_infos)
|
||||
student_repo.add(student)
|
||||
memory_sqlite_conn.commit()
|
||||
|
||||
student = prebuild_students[0]
|
||||
student.name = "Boby"
|
||||
student.tribe = prebuild_tribes[-1]
|
||||
|
||||
student_repo.update(student)
|
||||
sqlite_conn.commit()
|
||||
|
||||
student_list = student_repo.list(prebuild_tribes)
|
||||
assert set(student_list) == set(prebuild_students)
|
||||
|
||||
moded_student = next(filter(lambda s: s.id == student.id, student_list))
|
||||
assert moded_student == student
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
student_repo.add(student)
|
||||
|
||||
|
||||
def test_update_student_does_not_exists(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
def test_update_student(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, prebuild_students):
|
||||
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
||||
student = prebuild_students[0]
|
||||
student.name = "Boby"
|
||||
student.tribe = prebuild_tribes[-1]
|
||||
|
||||
student = Student(name="jkl", tribe=prebuild_tribes[0])
|
||||
|
||||
with pytest.raises(StudentRepositoryError):
|
||||
student_repo.update(student)
|
||||
memory_sqlite_conn.commit()
|
||||
|
||||
student_list = student_repo.list(prebuild_tribes)
|
||||
assert set(student_list) == set(prebuild_students)
|
||||
|
||||
moded_student = next(filter(lambda s: s.id == student.id, student_list))
|
||||
assert moded_student == student
|
||||
|
||||
|
||||
def test_delete_student(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
|
||||
def test_update_student_does_not_exists(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
||||
student = Student(name="jkl", tribe=prebuild_tribes[0])
|
||||
|
||||
deleted_student = prebuild_students.pop()
|
||||
student_repo.delete(deleted_student)
|
||||
sqlite_conn.commit()
|
||||
with pytest.raises(StudentRepositoryError):
|
||||
student_repo.update(student)
|
||||
|
||||
assert student_repo.list(prebuild_tribes) == prebuild_students
|
||||
|
||||
def test_delete_student(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, prebuild_students):
|
||||
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
deleted_student = prebuild_students.pop()
|
||||
student_repo.delete(deleted_student.id)
|
||||
memory_sqlite_conn.commit()
|
||||
|
||||
assert student_repo.list(prebuild_tribes) == prebuild_students
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
|
||||
from backend.model.tribe import Tribe
|
||||
@@ -7,60 +5,41 @@ from backend.repository.tribe_sqlite_repository import (
|
||||
TribeRepositoryError,
|
||||
TribeSQLiteRepository,
|
||||
)
|
||||
from tests.model.fakes import build_tribes
|
||||
|
||||
|
||||
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()
|
||||
def test_get_tribe(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
name = prebuild_tribes[0].name
|
||||
|
||||
return tribes
|
||||
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||
tribes = tribe_repo.get(name)
|
||||
|
||||
assert prebuild_tribes[0] == tribes
|
||||
|
||||
|
||||
def test_get_tribe(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
|
||||
name = prebuild_tribes[0].name
|
||||
|
||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
||||
tribes = tribe_repo.get(name)
|
||||
|
||||
assert prebuild_tribes[0] == tribes
|
||||
|
||||
|
||||
def test_get_tribe_not_exists(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
|
||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
||||
def test_get_tribe_not_exists(memory_sqlite_conn):
|
||||
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||
with pytest.raises(TribeRepositoryError):
|
||||
tribe_repo.get("Tribe0")
|
||||
|
||||
|
||||
def test_list_tribes(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
def test_list_tribes(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||
listed_tribes = tribe_repo.list()
|
||||
|
||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
||||
tribes = tribe_repo.list()
|
||||
|
||||
assert prebuild_tribes == tribes
|
||||
assert prebuild_tribes == listed_tribes
|
||||
|
||||
|
||||
def test_add_tribe(sqlite_conn):
|
||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
||||
def test_add_tribe(memory_sqlite_conn):
|
||||
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
tribe_infos = ("tribe1", "2nd")
|
||||
tribe = Tribe(*tribe_infos)
|
||||
tribe_repo.add(tribe)
|
||||
sqlite_conn.commit()
|
||||
memory_sqlite_conn.commit()
|
||||
|
||||
cursor = sqlite_conn.cursor()
|
||||
cursor = memory_sqlite_conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT * FROM tribes WHERE name=?
|
||||
@@ -72,46 +51,43 @@ def test_add_tribe(sqlite_conn):
|
||||
assert row == tribe_infos
|
||||
|
||||
|
||||
def test_add_tribe_fail_exists(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
def test_add_tribe_fail_exists(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
||||
|
||||
existing_tribe = prebuild_tribes[0]
|
||||
with pytest.raises(TribeRepositoryError):
|
||||
tribe_repo.add(existing_tribe)
|
||||
existing_tribe = prebuild_tribes[0]
|
||||
with pytest.raises(TribeRepositoryError):
|
||||
tribe_repo.add(existing_tribe)
|
||||
|
||||
|
||||
def test_update_tribe(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
def test_update_tribe(memory_sqlite_conn, populate_db):
|
||||
|
||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
name = prebuild_tribes[0].name
|
||||
new_tribe = Tribe("Tribe0", "Term")
|
||||
tribe_repo.update(name, new_tribe)
|
||||
sqlite_conn.commit()
|
||||
name = prebuild_tribes[0].name
|
||||
new_tribe = Tribe("Tribe0", "Term")
|
||||
tribe_repo.update(name, new_tribe)
|
||||
memory_sqlite_conn.commit()
|
||||
|
||||
prebuild_tribes[0] = new_tribe
|
||||
assert tribe_repo.list() == prebuild_tribes
|
||||
prebuild_tribes[0] = new_tribe
|
||||
assert tribe_repo.list() == prebuild_tribes
|
||||
|
||||
|
||||
def test_update_tribe_not_exists(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
def test_update_tribe_not_exists(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||
|
||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
||||
|
||||
name = prebuild_tribes[0].name
|
||||
new_tribe = Tribe("Tribe0", "Term")
|
||||
with pytest.raises(TribeRepositoryError):
|
||||
tribe_repo.update("iouiou", new_tribe)
|
||||
name = prebuild_tribes[0].name
|
||||
new_tribe = Tribe("Tribe0", "Term")
|
||||
with pytest.raises(TribeRepositoryError):
|
||||
tribe_repo.update("iouiou", new_tribe)
|
||||
|
||||
|
||||
def test_delete_tribe(sqlite_conn):
|
||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
||||
|
||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
||||
deleted_tribe = prebuild_tribes.pop()
|
||||
deleted_tribe.name = "iouiou"
|
||||
with pytest.raises(TribeRepositoryError):
|
||||
tribe_repo.delete(deleted_tribe)
|
||||
def test_delete_tribe(memory_sqlite_conn, populate_db):
|
||||
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||
deleted_tribe = prebuild_tribes.pop()
|
||||
deleted_tribe.name = "iouiou"
|
||||
with pytest.raises(TribeRepositoryError):
|
||||
tribe_repo.delete(deleted_tribe)
|
||||
|
||||
Reference in New Issue
Block a user