Compare commits

..

No commits in common. "a953631d19f41661c79cbb5b7a3ae236a547a07d" and "3b98a881e7cfefb5ee33eb9bd59d039151d79f3e" have entirely different histories.

9 changed files with 223 additions and 382 deletions

3
.gitignore vendored
View File

@ -122,6 +122,3 @@ dmypy.json
# Pyre type checker # Pyre type checker
.pyre/ .pyre/
# temporary database
sqlite.db

View File

@ -1,7 +1,7 @@
import sqlite3 import sqlite3
from fastapi import FastAPI, status from fastapi import FastAPI, status
from fastapi.responses import JSONResponse, RedirectResponse, Response from fastapi.responses import JSONResponse
from backend.adapters.sqlite import create_db from backend.adapters.sqlite import create_db
from backend.api.model import StudentModel, TribeModel from backend.api.model import StudentModel, TribeModel
@ -35,51 +35,21 @@ student_repo = StudentSQLiteRepository(conn)
app = FastAPI() app = FastAPI()
@app.post("/tribes", response_class=RedirectResponse, status_code=status.HTTP_302_FOUND) @app.post("/tribes", status_code=status.HTTP_201_CREATED, response_model=TribeModel)
async def post_tribe(item: TribeModel): async def post_tribe(item: TribeModel):
tribe = Tribe(**item.dict())
try: try:
tribe = services.add_tribe( services.add_tribe(
name=item.name, level=item.level, tribe_repo=tribe_repo, conn=conn name=item.name, level=item.level, tribe_repo=tribe_repo, conn=conn
) )
except TribeExists: except TribeExists:
return JSONResponse( return JSONResponse(
status_code=status.HTTP_409_CONFLICT, status_code=status.HTTP_409_CONFLICT,
content=f"The tribe {item.name} already exists", content=f"The tribe {tribe.name} already exists",
) )
return f"/tribes/{tribe.name}" return tribe.to_dict()
@app.put(
"/tribes/{name}", response_class=RedirectResponse, status_code=status.HTTP_302_FOUND
)
async def put_tribe(name: str, item: TribeModel):
try:
tribe = services.update_tribe(
name=item.name, level=item.level, tribe_repo=tribe_repo, conn=conn
)
except TribeDoesNotExist:
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content=f"The tribe {name} does not exists",
)
return f"/tribes/{tribe.name}"
@app.delete("/tribes/{name}")
async def delete_tribe(name: str):
try:
services.delete_tribe(name=name, tribe_repo=tribe_repo, conn=conn)
except TribeDoesNotExist:
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content=f"The tribe {name} does not exists",
)
return Response(
status_code=status.HTTP_204_NO_CONTENT,
)
@app.get("/tribes", response_model=list[TribeModel]) @app.get("/tribes", response_model=list[TribeModel])
@ -96,9 +66,7 @@ async def get_tribe(name: str):
return tribe.to_dict() return tribe.to_dict()
@app.post( @app.post("/students", status_code=status.HTTP_201_CREATED, response_model=StudentModel)
"/students", response_class=RedirectResponse, status_code=status.HTTP_302_FOUND
)
async def post_student(item: StudentModel): async def post_student(item: StudentModel):
if item.id is not None: if item.id is not None:
return JSONResponse( return JSONResponse(
@ -120,28 +88,13 @@ async def post_student(item: StudentModel):
content=f"The tribe {item.tribe_name} does not exists. You can't add a student in it.", content=f"The tribe {item.tribe_name} does not exists. You can't add a student in it.",
) )
return f"/students/{student.id}"
@app.get("/students/{id}", status_code=status.HTTP_200_OK, response_model=StudentModel)
async def get_student(id: str):
tribes = tribe_repo.list()
student = student_repo.get(id, tribes)
return student.to_dict() return student.to_dict()
@app.get("/students", status_code=status.HTTP_200_OK, response_model=list[StudentModel])
async def list_students():
tribes = tribe_repo.list()
students = student_repo.list(tribes)
return [t.to_dict() for t in students]
@app.put( @app.put(
"/students/{student_id}", "/students/{student_id}",
response_class=RedirectResponse, status_code=status.HTTP_200_OK,
status_code=status.HTTP_302_FOUND, response_model=StudentModel,
) )
async def put_student(student_id, item: StudentModel): async def put_student(student_id, item: StudentModel):
if student_id != item.id: if student_id != item.id:
@ -170,25 +123,4 @@ async def put_student(student_id, item: StudentModel):
content=f"The student {item.name} ({item.id=}) does not exists. You can't modify it.", content=f"The student {item.name} ({item.id=}) does not exists. You can't modify it.",
) )
return f"/students/{student.id}" return student.to_dict()
@app.delete(
"/students/{student_id}",
)
async def delete_student(student_id):
try:
student = services.delete_student(
id=student_id,
student_repo=student_repo,
conn=conn,
)
except StudentDoesExist:
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content=f"The student ({student_id=}) does not exists. You can't delete it.",
)
return Response(
status_code=status.HTTP_204_NO_CONTENT,
)

View File

@ -77,29 +77,12 @@ class StudentSQLiteRepository(AbstractRepository):
rows = cursor.fetchall() rows = cursor.fetchall()
return [self._rebuild_student(r, tribes) for r in rows] return [self._rebuild_student(r, tribes) for r in rows]
def list_id(self): def delete(self, student: Student) -> None:
cursor = self.conn.cursor()
cursor.execute(
"""
SELECT id FROM students
"""
)
rows = cursor.fetchall()
return [r[0] for r in rows]
def delete(self, id: str) -> None:
students_id = self.list_id()
if id not in students_id:
raise StudentRepositoryError(
f"The student {id} doesn't exists. Can't delete it."
)
self.conn.execute( self.conn.execute(
""" """
DELETE FROM students WHERE id=:id DELETE FROM students WHERE id=:id
""", """,
{ {
"id": id, "id": student.id,
}, },
) )

View File

@ -72,11 +72,11 @@ class TribeSQLiteRepository(AbstractRepository):
rows = cursor.fetchall() rows = cursor.fetchall()
return [Tribe(*r) for r in rows] return [Tribe(*r) for r in rows]
def delete(self, name: str) -> None: def delete(self, tribe: Tribe) -> None:
tribes = self.list() tribes = self.list()
if name not in map(lambda x: x.name, tribes): if tribe.name not in map(lambda x: x.name, tribes):
raise TribeRepositoryError( raise TribeRepositoryError(
f"The tribe {name} doesn't exists. Can't delete it." f"The tribe {tribe.name} doesn't exists. Can't delete it."
) )
self.conn.execute( self.conn.execute(
@ -84,6 +84,6 @@ class TribeSQLiteRepository(AbstractRepository):
DELETE FROM tribes WHERE name=:name DELETE FROM tribes WHERE name=:name
""", """,
{ {
"name": name, "name": tribe.name,
}, },
) )

View File

@ -10,9 +10,8 @@ from sqlalchemy.orm import clear_mappers, sessionmaker
from backend import config from backend import config
from backend.adapters.orm import metadata, start_mappers from backend.adapters.orm import metadata, start_mappers
from backend.adapters.sqlite import create_db from backend.adapters.sqlite import create_db
from backend.model.student import Student from tests.integration.test_repository_student_sqlite import populate_students
from backend.model.tribe import Tribe from tests.integration.test_repository_tribe_sqlite import populate_tribes
from tests.model.fakes import build_student, build_tribes
@pytest.fixture @pytest.fixture
@ -30,7 +29,7 @@ def session(in_memory_db):
@pytest.fixture @pytest.fixture
def memory_sqlite_conn(): def sqlite_conn():
sqlite_db = ":memory:" sqlite_db = ":memory:"
conn = sqlite3.connect(sqlite_db) conn = sqlite3.connect(sqlite_db)
create_db(conn) create_db(conn)
@ -50,70 +49,35 @@ def clean_db():
conn.commit() 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 @pytest.fixture
def populate_db(): def populate_db(sqlite_conn):
class Student_tribe_context: _tribes = []
_tribes = [] _students = []
_students = []
def __init__(self, conn): def _populate_db():
self.conn = conn tribes = populate_tribes(sqlite_conn)
_tribes += tribes
students = populate_students(sqlite_conn, tribes)
_students += students
return tribes, students
def __enter__(self): yield _populate_db
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 _students:
sqlite_conn.execute(
for student in self._students: """
self.conn.execute( DELETE FROM students WHERE id=:id
""" """,
DELETE FROM students WHERE id=:id {"id": student.id},
""", )
{"id": student.id}, for tribe in _tribes:
) sqlite_conn.execute(
for tribe in self._tribes: """
self.conn.execute( DELETE FROM tribes WHERE name=:name
""" """,
DELETE FROM tribes WHERE name=:name {"name": tribe.name},
""", )
{"name": tribe.name}, sqlite_conn.commit()
)
self.conn.commit()
def fixture(conn):
return Student_tribe_context(conn)
yield fixture
def wait_for_webapp_to_come_up(): def wait_for_webapp_to_come_up():

View File

@ -2,7 +2,7 @@ import pytest
import requests import requests
from backend import config from backend import config
from tests.model.fakes import build_student, build_tribes from tests.model.fakes import build_tribes
@pytest.mark.usefixtures("restart_api") @pytest.mark.usefixtures("restart_api")
@ -15,10 +15,7 @@ def test_api_post_student():
data = {"name": "zart", "tribe_name": tribe.name} data = {"name": "zart", "tribe_name": tribe.name}
r = requests.post(f"{url}/students", json=data) r = requests.post(f"{url}/students", json=data)
post_request = r.history[0] assert r.status_code == 201
assert post_request.status_code == 302
assert r.status_code == 200
assert r.json()["name"] == "zart" assert r.json()["name"] == "zart"
assert r.json()["tribe_name"] == tribe.name assert r.json()["tribe_name"] == tribe.name
assert r.json()["id"] assert r.json()["id"]
@ -75,30 +72,7 @@ def test_api_put_student():
r2 = requests.put(f"{url}/students/{student['id']}", json=student) r2 = requests.put(f"{url}/students/{student['id']}", json=student)
post_request = r2.history[0]
assert post_request.status_code == 302
assert r2.status_code == 200 assert r2.status_code == 200
assert r2.json()["name"] == "Choupinou" assert r2.json()["name"] == "Choupinou"
assert r2.json()["tribe_name"] == tribe.name assert r2.json()["tribe_name"] == tribe.name
assert r2.json()["id"] == r.json()["id"] assert r2.json()["id"] == r.json()["id"]
@pytest.mark.usefixtures("restart_api")
@pytest.mark.usefixtures("clean_db")
def test_api_delete_student():
url = config.get_api_url()
tribe = build_tribes(1)[0]
requests.post(f"{url}/tribes", json=tribe.to_dict())
student = build_student([tribe], 1)[0]
r = requests.post(
f"{url}/students", json={"name": student.name, "tribe_name": student.tribe.name}
)
student_id = r.json()["id"]
r = requests.delete(f"{url}/students/{student_id}")
assert r.status_code == 204
r = requests.get(f"{url}/students/")
assert r.json() == []

View File

@ -13,10 +13,7 @@ def test_api_post_tribe():
url = config.get_api_url() url = config.get_api_url()
r = requests.post(f"{url}/tribes", json=data) r = requests.post(f"{url}/tribes", json=data)
post_request = r.history[0] assert r.status_code == 201
assert post_request.status_code == 302
assert r.status_code == 200
assert r.json() == { assert r.json() == {
"assessments": [], "assessments": [],
"level": "2nd", "level": "2nd",
@ -38,68 +35,6 @@ def test_api_post_tribe_already_exists():
assert r.json() == f"The tribe {data['name']} already exists" assert r.json() == f"The tribe {data['name']} already exists"
@pytest.mark.usefixtures("restart_api")
@pytest.mark.usefixtures("clean_db")
def test_api_put_tribe():
tribe = build_tribes(1)[0]
url = config.get_api_url()
r = requests.post(f"{url}/tribes", json=tribe.to_dict())
mod_tribe = tribe
mod_tribe.level = "other level"
r = requests.put(f"{url}/tribes/{tribe.name}", json=mod_tribe.to_dict())
post_request = r.history[0]
assert post_request.status_code == 302
assert r.status_code == 200
r = requests.get(f"{url}/tribes")
assert [t["name"] for t in r.json()] == [mod_tribe.name]
assert [t["level"] for t in r.json()] == [mod_tribe.level]
@pytest.mark.usefixtures("restart_api")
@pytest.mark.usefixtures("clean_db")
def test_api_put_tribe_doesnt_exists():
tribe = build_tribes(1)[0]
url = config.get_api_url()
r = requests.put(f"{url}/tribes/{tribe.name}", json=tribe.to_dict())
assert r.status_code == 409
@pytest.mark.usefixtures("restart_api")
@pytest.mark.usefixtures("clean_db")
def test_api_delete_tribe():
tribe = build_tribes(1)[0]
url = config.get_api_url()
r = requests.post(f"{url}/tribes", json=tribe.to_dict())
r = requests.delete(f"{url}/tribes/{tribe.name}")
assert r.status_code == 204
r = requests.get(f"{url}/tribes")
assert r.json() == []
@pytest.mark.usefixtures("restart_api")
@pytest.mark.usefixtures("clean_db")
def test_api_delete_tribe_doesnt_exists():
tribe = build_tribes(1)[0]
url = config.get_api_url()
r = requests.post(f"{url}/tribes", json=tribe.to_dict())
r = requests.delete(f"{url}/tribes/notexisting")
assert r.status_code == 409
r = requests.get(f"{url}/tribes")
assert [t["name"] for t in r.json()] == [tribe.name]
assert [t["level"] for t in r.json()] == [tribe.level]
@pytest.mark.usefixtures("restart_api") @pytest.mark.usefixtures("restart_api")
@pytest.mark.usefixtures("clean_db") @pytest.mark.usefixtures("clean_db")
def test_api_post_list_tribe(): def test_api_post_list_tribe():
@ -108,6 +43,8 @@ def test_api_post_list_tribe():
url = config.get_api_url() url = config.get_api_url()
r = requests.post(f"{url}/tribes", json=tribe.to_dict()) r = requests.post(f"{url}/tribes", json=tribe.to_dict())
assert r.status_code == 201
r = requests.get(f"{url}/tribes") r = requests.get(f"{url}/tribes")
assert r.json() == [ assert r.json() == [
{ {

View File

@ -3,105 +3,135 @@ import sqlite3
import pytest import pytest
from backend.model.student import Student from backend.model.student import Student
from backend.model.tribe import Tribe
from backend.repository.student_sqlite_repository import ( from backend.repository.student_sqlite_repository import (
StudentRepositoryError, StudentRepositoryError,
StudentSQLiteRepository, StudentSQLiteRepository,
) )
from tests.integration.test_repository_tribe_sqlite import populate_tribes
from tests.model.fakes import build_student
def test_get_student(memory_sqlite_conn, populate_db): def populate_students(conn, tribes: list[Tribe]) -> list[Student]:
with populate_db(memory_sqlite_conn) as (prebuild_tribes, prebuild_students): cursor = conn.cursor()
student_repo = StudentSQLiteRepository(memory_sqlite_conn) 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()
student_id = prebuild_students[0].id return prebuild_students
student = student_repo.get(student_id, prebuild_tribes)
assert prebuild_students[0] == student
def test_get_student_not_exists(memory_sqlite_conn, populate_db): def test_get_student(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): prebuild_tribes = populate_tribes(sqlite_conn)
student_repo = StudentSQLiteRepository(memory_sqlite_conn) prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
with pytest.raises(ValueError):
student_repo.get("student0", 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_list_students(memory_sqlite_conn, populate_db): def test_get_student_not_exists(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, prebuild_students): prebuild_tribes = populate_tribes(sqlite_conn)
student_repo = StudentSQLiteRepository(memory_sqlite_conn) prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
students = student_repo.list(prebuild_tribes)
assert prebuild_students == students student_repo = StudentSQLiteRepository(sqlite_conn)
with pytest.raises(ValueError):
student_repo.get("student0", prebuild_tribes)
def test_add_student(memory_sqlite_conn, populate_db): def test_list_students(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): prebuild_tribes = populate_tribes(sqlite_conn)
student_repo = StudentSQLiteRepository(memory_sqlite_conn) prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
student_infos = {"name": "student1", "tribe": prebuild_tribes[0]} student_repo = StudentSQLiteRepository(sqlite_conn)
student = Student(**student_infos) 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_repo.add(student) 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_add_student_fail_exists(memory_sqlite_conn, populate_db): def test_update_student(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): prebuild_tribes = populate_tribes(sqlite_conn)
student_repo = StudentSQLiteRepository(memory_sqlite_conn) prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
student_infos = {"name": "student1", "tribe": prebuild_tribes[0]} student_repo = StudentSQLiteRepository(sqlite_conn)
student = Student(**student_infos)
student_repo.add(student)
memory_sqlite_conn.commit()
with pytest.raises(sqlite3.IntegrityError): student = prebuild_students[0]
student_repo.add(student) 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
def test_update_student(memory_sqlite_conn, populate_db): def test_update_student_does_not_exists(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, prebuild_students): prebuild_tribes = populate_tribes(sqlite_conn)
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
student = prebuild_students[0] student_repo = StudentSQLiteRepository(sqlite_conn)
student.name = "Boby"
student.tribe = prebuild_tribes[-1]
student = Student(name="jkl", tribe=prebuild_tribes[0])
with pytest.raises(StudentRepositoryError):
student_repo.update(student) 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_update_student_does_not_exists(memory_sqlite_conn, populate_db): def test_delete_student(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): prebuild_tribes = populate_tribes(sqlite_conn)
student_repo = StudentSQLiteRepository(memory_sqlite_conn) prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
student = Student(name="jkl", tribe=prebuild_tribes[0]) student_repo = StudentSQLiteRepository(sqlite_conn)
with pytest.raises(StudentRepositoryError): deleted_student = prebuild_students.pop()
student_repo.update(student) student_repo.delete(deleted_student)
sqlite_conn.commit()
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

View File

@ -1,3 +1,5 @@
import sqlite3
import pytest import pytest
from backend.model.tribe import Tribe from backend.model.tribe import Tribe
@ -5,41 +7,60 @@ from backend.repository.tribe_sqlite_repository import (
TribeRepositoryError, TribeRepositoryError,
TribeSQLiteRepository, TribeSQLiteRepository,
) )
from tests.model.fakes import build_tribes
def test_get_tribe(memory_sqlite_conn, populate_db): def populate_tribes(conn) -> list[Tribe]:
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): cursor = conn.cursor()
name = prebuild_tribes[0].name tribes = build_tribes(3)
cursor.executemany(
"""
INSERT INTO tribes(name, level) VALUES (?, ?)
""",
[t.to_tuple() for t in tribes],
)
conn.commit()
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn) return tribes
tribes = tribe_repo.get(name)
assert prebuild_tribes[0] == tribes
def test_get_tribe_not_exists(memory_sqlite_conn): def test_get_tribe(sqlite_conn):
tribe_repo = TribeSQLiteRepository(memory_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)
with pytest.raises(TribeRepositoryError): with pytest.raises(TribeRepositoryError):
tribe_repo.get("Tribe0") tribe_repo.get("Tribe0")
def test_list_tribes(memory_sqlite_conn, populate_db): def test_list_tribes(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): prebuild_tribes = populate_tribes(sqlite_conn)
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
listed_tribes = tribe_repo.list()
assert prebuild_tribes == listed_tribes tribe_repo = TribeSQLiteRepository(sqlite_conn)
tribes = tribe_repo.list()
assert prebuild_tribes == tribes
def test_add_tribe(memory_sqlite_conn): def test_add_tribe(sqlite_conn):
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn) tribe_repo = TribeSQLiteRepository(sqlite_conn)
tribe_infos = ("tribe1", "2nd") tribe_infos = ("tribe1", "2nd")
tribe = Tribe(*tribe_infos) tribe = Tribe(*tribe_infos)
tribe_repo.add(tribe) tribe_repo.add(tribe)
memory_sqlite_conn.commit() sqlite_conn.commit()
cursor = memory_sqlite_conn.cursor() cursor = sqlite_conn.cursor()
cursor.execute( cursor.execute(
""" """
SELECT * FROM tribes WHERE name=? SELECT * FROM tribes WHERE name=?
@ -51,43 +72,46 @@ def test_add_tribe(memory_sqlite_conn):
assert row == tribe_infos assert row == tribe_infos
def test_add_tribe_fail_exists(memory_sqlite_conn, populate_db): def test_add_tribe_fail_exists(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): prebuild_tribes = populate_tribes(sqlite_conn)
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
existing_tribe = prebuild_tribes[0] tribe_repo = TribeSQLiteRepository(sqlite_conn)
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(memory_sqlite_conn, populate_db): def test_update_tribe(sqlite_conn):
prebuild_tribes = populate_tribes(sqlite_conn)
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): tribe_repo = TribeSQLiteRepository(sqlite_conn)
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
name = prebuild_tribes[0].name name = prebuild_tribes[0].name
new_tribe = Tribe("Tribe0", "Term") new_tribe = Tribe("Tribe0", "Term")
tribe_repo.update(name, new_tribe) tribe_repo.update(name, new_tribe)
memory_sqlite_conn.commit() sqlite_conn.commit()
prebuild_tribes[0] = new_tribe prebuild_tribes[0] = new_tribe
assert tribe_repo.list() == prebuild_tribes assert tribe_repo.list() == prebuild_tribes
def test_update_tribe_not_exists(memory_sqlite_conn, populate_db): def test_update_tribe_not_exists(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): prebuild_tribes = populate_tribes(sqlite_conn)
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
name = prebuild_tribes[0].name tribe_repo = TribeSQLiteRepository(sqlite_conn)
new_tribe = Tribe("Tribe0", "Term")
with pytest.raises(TribeRepositoryError): name = prebuild_tribes[0].name
tribe_repo.update("iouiou", new_tribe) new_tribe = Tribe("Tribe0", "Term")
with pytest.raises(TribeRepositoryError):
tribe_repo.update("iouiou", new_tribe)
def test_delete_tribe(memory_sqlite_conn, populate_db): def test_delete_tribe(sqlite_conn):
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _): prebuild_tribes = populate_tribes(sqlite_conn)
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
deleted_tribe = prebuild_tribes.pop() tribe_repo = TribeSQLiteRepository(sqlite_conn)
deleted_tribe.name = "iouiou" deleted_tribe = prebuild_tribes.pop()
with pytest.raises(TribeRepositoryError): deleted_tribe.name = "iouiou"
tribe_repo.delete(deleted_tribe) with pytest.raises(TribeRepositoryError):
tribe_repo.delete(deleted_tribe)