Compare commits
5 Commits
3b98a881e7
...
a953631d19
Author | SHA1 | Date | |
---|---|---|---|
a953631d19 | |||
ccb59975f7 | |||
0ebc24ff29 | |||
07595f1fd8 | |||
f3302e2132 |
3
.gitignore
vendored
3
.gitignore
vendored
@ -122,3 +122,6 @@ dmypy.json
|
|||||||
|
|
||||||
# Pyre type checker
|
# Pyre type checker
|
||||||
.pyre/
|
.pyre/
|
||||||
|
|
||||||
|
# temporary database
|
||||||
|
sqlite.db
|
||||||
|
@ -1,7 +1,7 @@
|
|||||||
import sqlite3
|
import sqlite3
|
||||||
|
|
||||||
from fastapi import FastAPI, status
|
from fastapi import FastAPI, status
|
||||||
from fastapi.responses import JSONResponse
|
from fastapi.responses import JSONResponse, RedirectResponse, Response
|
||||||
|
|
||||||
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,21 +35,51 @@ student_repo = StudentSQLiteRepository(conn)
|
|||||||
app = FastAPI()
|
app = FastAPI()
|
||||||
|
|
||||||
|
|
||||||
@app.post("/tribes", status_code=status.HTTP_201_CREATED, response_model=TribeModel)
|
@app.post("/tribes", response_class=RedirectResponse, status_code=status.HTTP_302_FOUND)
|
||||||
async def post_tribe(item: TribeModel):
|
async def post_tribe(item: TribeModel):
|
||||||
tribe = Tribe(**item.dict())
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
services.add_tribe(
|
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 {tribe.name} already exists",
|
content=f"The tribe {item.name} already exists",
|
||||||
)
|
)
|
||||||
|
|
||||||
return tribe.to_dict()
|
return f"/tribes/{tribe.name}"
|
||||||
|
|
||||||
|
|
||||||
|
@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])
|
||||||
@ -66,7 +96,9 @@ async def get_tribe(name: str):
|
|||||||
return tribe.to_dict()
|
return tribe.to_dict()
|
||||||
|
|
||||||
|
|
||||||
@app.post("/students", status_code=status.HTTP_201_CREATED, response_model=StudentModel)
|
@app.post(
|
||||||
|
"/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(
|
||||||
@ -88,13 +120,28 @@ 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}",
|
||||||
status_code=status.HTTP_200_OK,
|
response_class=RedirectResponse,
|
||||||
response_model=StudentModel,
|
status_code=status.HTTP_302_FOUND,
|
||||||
)
|
)
|
||||||
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:
|
||||||
@ -123,4 +170,25 @@ 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 student.to_dict()
|
return f"/students/{student.id}"
|
||||||
|
|
||||||
|
|
||||||
|
@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,
|
||||||
|
)
|
||||||
|
@ -77,12 +77,29 @@ 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 delete(self, student: Student) -> None:
|
def list_id(self):
|
||||||
|
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": student.id,
|
"id": id,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
@ -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, tribe: Tribe) -> None:
|
def delete(self, name: str) -> None:
|
||||||
tribes = self.list()
|
tribes = self.list()
|
||||||
if tribe.name not in map(lambda x: x.name, tribes):
|
if name not in map(lambda x: x.name, tribes):
|
||||||
raise TribeRepositoryError(
|
raise TribeRepositoryError(
|
||||||
f"The tribe {tribe.name} doesn't exists. Can't delete it."
|
f"The 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": tribe.name,
|
"name": name,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
@ -10,8 +10,9 @@ 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 tests.integration.test_repository_student_sqlite import populate_students
|
from backend.model.student import Student
|
||||||
from tests.integration.test_repository_tribe_sqlite import populate_tribes
|
from backend.model.tribe import Tribe
|
||||||
|
from tests.model.fakes import build_student, build_tribes
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
@ -29,7 +30,7 @@ def session(in_memory_db):
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def sqlite_conn():
|
def memory_sqlite_conn():
|
||||||
sqlite_db = ":memory:"
|
sqlite_db = ":memory:"
|
||||||
conn = sqlite3.connect(sqlite_db)
|
conn = sqlite3.connect(sqlite_db)
|
||||||
create_db(conn)
|
create_db(conn)
|
||||||
@ -49,35 +50,70 @@ 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(sqlite_conn):
|
def populate_db():
|
||||||
_tribes = []
|
class Student_tribe_context:
|
||||||
_students = []
|
_tribes = []
|
||||||
|
_students = []
|
||||||
|
|
||||||
def _populate_db():
|
def __init__(self, conn):
|
||||||
tribes = populate_tribes(sqlite_conn)
|
self.conn = conn
|
||||||
_tribes += tribes
|
|
||||||
students = populate_students(sqlite_conn, tribes)
|
|
||||||
_students += students
|
|
||||||
return tribes, students
|
|
||||||
|
|
||||||
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:
|
def __exit__(self, *args):
|
||||||
sqlite_conn.execute(
|
|
||||||
"""
|
for student in self._students:
|
||||||
DELETE FROM students WHERE id=:id
|
self.conn.execute(
|
||||||
""",
|
"""
|
||||||
{"id": student.id},
|
DELETE FROM students WHERE id=:id
|
||||||
)
|
""",
|
||||||
for tribe in _tribes:
|
{"id": student.id},
|
||||||
sqlite_conn.execute(
|
)
|
||||||
"""
|
for tribe in self._tribes:
|
||||||
DELETE FROM tribes WHERE name=:name
|
self.conn.execute(
|
||||||
""",
|
"""
|
||||||
{"name": tribe.name},
|
DELETE FROM tribes WHERE name=:name
|
||||||
)
|
""",
|
||||||
sqlite_conn.commit()
|
{"name": tribe.name},
|
||||||
|
)
|
||||||
|
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():
|
||||||
|
@ -2,7 +2,7 @@ import pytest
|
|||||||
import requests
|
import requests
|
||||||
|
|
||||||
from backend import config
|
from backend import config
|
||||||
from tests.model.fakes import build_tribes
|
from tests.model.fakes import build_student, build_tribes
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.usefixtures("restart_api")
|
@pytest.mark.usefixtures("restart_api")
|
||||||
@ -15,7 +15,10 @@ 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)
|
||||||
|
|
||||||
assert r.status_code == 201
|
post_request = r.history[0]
|
||||||
|
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"]
|
||||||
@ -72,7 +75,30 @@ 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() == []
|
||||||
|
@ -13,7 +13,10 @@ 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)
|
||||||
|
|
||||||
assert r.status_code == 201
|
post_request = r.history[0]
|
||||||
|
assert post_request.status_code == 302
|
||||||
|
|
||||||
|
assert r.status_code == 200
|
||||||
assert r.json() == {
|
assert r.json() == {
|
||||||
"assessments": [],
|
"assessments": [],
|
||||||
"level": "2nd",
|
"level": "2nd",
|
||||||
@ -37,13 +40,73 @@ def test_api_post_tribe_already_exists():
|
|||||||
|
|
||||||
@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_put_tribe():
|
||||||
tribe = build_tribes(1)[0]
|
tribe = build_tribes(1)[0]
|
||||||
|
|
||||||
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
|
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("clean_db")
|
||||||
|
def test_api_post_list_tribe():
|
||||||
|
tribe = build_tribes(1)[0]
|
||||||
|
|
||||||
|
url = config.get_api_url()
|
||||||
|
r = requests.post(f"{url}/tribes", json=tribe.to_dict())
|
||||||
|
|
||||||
r = requests.get(f"{url}/tribes")
|
r = requests.get(f"{url}/tribes")
|
||||||
assert r.json() == [
|
assert r.json() == [
|
||||||
|
@ -3,135 +3,105 @@ 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 populate_students(conn, tribes: list[Tribe]) -> list[Student]:
|
def test_get_student(memory_sqlite_conn, populate_db):
|
||||||
cursor = conn.cursor()
|
with populate_db(memory_sqlite_conn) as (prebuild_tribes, prebuild_students):
|
||||||
prebuild_students = build_student(tribes, 2)
|
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||||
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
|
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):
|
def test_get_student_not_exists(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||||
prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
|
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||||
|
with pytest.raises(ValueError):
|
||||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
student_repo.get("student0", prebuild_tribes)
|
||||||
|
|
||||||
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(sqlite_conn):
|
def test_list_students(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
with populate_db(memory_sqlite_conn) as (prebuild_tribes, prebuild_students):
|
||||||
prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
|
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||||
|
students = student_repo.list(prebuild_tribes)
|
||||||
|
|
||||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
assert prebuild_students == students
|
||||||
with pytest.raises(ValueError):
|
|
||||||
student_repo.get("student0", prebuild_tribes)
|
|
||||||
|
|
||||||
|
|
||||||
def test_list_students(sqlite_conn):
|
def test_add_student(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||||
prebuild_students = populate_students(sqlite_conn, prebuild_tribes)
|
student_repo = StudentSQLiteRepository(memory_sqlite_conn)
|
||||||
|
|
||||||
student_repo = StudentSQLiteRepository(sqlite_conn)
|
student_infos = {"name": "student1", "tribe": prebuild_tribes[0]}
|
||||||
students = student_repo.list(prebuild_tribes)
|
student = Student(**student_infos)
|
||||||
|
|
||||||
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_update_student(sqlite_conn):
|
def test_add_student_fail_exists(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||||
prebuild_students = populate_students(sqlite_conn, 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]
|
with pytest.raises(sqlite3.IntegrityError):
|
||||||
student.name = "Boby"
|
student_repo.add(student)
|
||||||
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_does_not_exists(sqlite_conn):
|
def test_update_student(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
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)
|
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):
|
def test_update_student_does_not_exists(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||||
prebuild_students = populate_students(sqlite_conn, 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()
|
with pytest.raises(StudentRepositoryError):
|
||||||
student_repo.delete(deleted_student)
|
student_repo.update(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
|
||||||
|
@ -1,5 +1,3 @@
|
|||||||
import sqlite3
|
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from backend.model.tribe import Tribe
|
from backend.model.tribe import Tribe
|
||||||
@ -7,60 +5,41 @@ from backend.repository.tribe_sqlite_repository import (
|
|||||||
TribeRepositoryError,
|
TribeRepositoryError,
|
||||||
TribeSQLiteRepository,
|
TribeSQLiteRepository,
|
||||||
)
|
)
|
||||||
from tests.model.fakes import build_tribes
|
|
||||||
|
|
||||||
|
|
||||||
def populate_tribes(conn) -> list[Tribe]:
|
def test_get_tribe(memory_sqlite_conn, populate_db):
|
||||||
cursor = conn.cursor()
|
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||||
tribes = build_tribes(3)
|
name = prebuild_tribes[0].name
|
||||||
cursor.executemany(
|
|
||||||
"""
|
|
||||||
INSERT INTO tribes(name, level) VALUES (?, ?)
|
|
||||||
""",
|
|
||||||
[t.to_tuple() for t in tribes],
|
|
||||||
)
|
|
||||||
conn.commit()
|
|
||||||
|
|
||||||
return tribes
|
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||||
|
tribes = tribe_repo.get(name)
|
||||||
|
|
||||||
|
assert prebuild_tribes[0] == tribes
|
||||||
|
|
||||||
|
|
||||||
def test_get_tribe(sqlite_conn):
|
def test_get_tribe_not_exists(memory_sqlite_conn):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
tribe_repo = TribeSQLiteRepository(memory_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(sqlite_conn):
|
def test_list_tribes(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
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)
|
assert prebuild_tribes == listed_tribes
|
||||||
tribes = tribe_repo.list()
|
|
||||||
|
|
||||||
assert prebuild_tribes == tribes
|
|
||||||
|
|
||||||
|
|
||||||
def test_add_tribe(sqlite_conn):
|
def test_add_tribe(memory_sqlite_conn):
|
||||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
tribe_repo = TribeSQLiteRepository(memory_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)
|
||||||
sqlite_conn.commit()
|
memory_sqlite_conn.commit()
|
||||||
|
|
||||||
cursor = sqlite_conn.cursor()
|
cursor = memory_sqlite_conn.cursor()
|
||||||
cursor.execute(
|
cursor.execute(
|
||||||
"""
|
"""
|
||||||
SELECT * FROM tribes WHERE name=?
|
SELECT * FROM tribes WHERE name=?
|
||||||
@ -72,46 +51,43 @@ def test_add_tribe(sqlite_conn):
|
|||||||
assert row == tribe_infos
|
assert row == tribe_infos
|
||||||
|
|
||||||
|
|
||||||
def test_add_tribe_fail_exists(sqlite_conn):
|
def test_add_tribe_fail_exists(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
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):
|
||||||
existing_tribe = prebuild_tribes[0]
|
tribe_repo.add(existing_tribe)
|
||||||
with pytest.raises(TribeRepositoryError):
|
|
||||||
tribe_repo.add(existing_tribe)
|
|
||||||
|
|
||||||
|
|
||||||
def test_update_tribe(sqlite_conn):
|
def test_update_tribe(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
|
||||||
|
|
||||||
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
|
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)
|
||||||
sqlite_conn.commit()
|
memory_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(sqlite_conn):
|
def test_update_tribe_not_exists(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
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")
|
||||||
name = prebuild_tribes[0].name
|
with pytest.raises(TribeRepositoryError):
|
||||||
new_tribe = Tribe("Tribe0", "Term")
|
tribe_repo.update("iouiou", new_tribe)
|
||||||
with pytest.raises(TribeRepositoryError):
|
|
||||||
tribe_repo.update("iouiou", new_tribe)
|
|
||||||
|
|
||||||
|
|
||||||
def test_delete_tribe(sqlite_conn):
|
def test_delete_tribe(memory_sqlite_conn, populate_db):
|
||||||
prebuild_tribes = populate_tribes(sqlite_conn)
|
with populate_db(memory_sqlite_conn) as (prebuild_tribes, _):
|
||||||
|
tribe_repo = TribeSQLiteRepository(memory_sqlite_conn)
|
||||||
tribe_repo = TribeSQLiteRepository(sqlite_conn)
|
deleted_tribe = prebuild_tribes.pop()
|
||||||
deleted_tribe = prebuild_tribes.pop()
|
deleted_tribe.name = "iouiou"
|
||||||
deleted_tribe.name = "iouiou"
|
with pytest.raises(TribeRepositoryError):
|
||||||
with pytest.raises(TribeRepositoryError):
|
tribe_repo.delete(deleted_tribe)
|
||||||
tribe_repo.delete(deleted_tribe)
|
|
||||||
|
Loading…
Reference in New Issue
Block a user