Tests

.github/workflows/lint.yml

@@ -47,6 +47,12 @@ jobs:
   pytest:
     name: Unit Testing
     runs-on: self-hosted
+    env:
+      DATABASE_URI: "${{ secrets.POSTGRES_CONNECTION }}"
+      FRONTEND_URL: "https://localhost:8080"
+      CAS_SERVER_URL: "https://login.ugent.be"
+      SECRET_KEY: "test" 
+      ALGORITHM: "HS256" 
     steps:
     - uses: actions/checkout@v3
     - name: Set up Python 3.12
@@ -61,11 +67,11 @@ jobs:
     - name: Initialize Database
       working-directory: backend
       run: |
-        psql -U selab -p 2002 -d selab < selab_script.sql
+        psql -U selab -p 2002 -d selabtest < selab_script.sql
     - name: Test with pytest
       run: |
         pip install pytest pytest-cov pytest-html pytest-sugar pytest-json-report
-        DATABASE_URI='postgresql://[email protected]:2002/selab' py.test -v --cov --html=reports/pytest/report.html
+        py.test -v --cov --html=reports/pytest/report.html
     - name: Archive pytest coverage results
       uses: actions/upload-artifact@v1
       with:

backend/README.md

@@ -60,3 +60,13 @@ authorize each request, add the token in the `Authorization` header.
 ```sh
 autopep8 -ri .
 ```
+
+## Testing
+
+You can add tests by creating `test_*` files and `test_*` functions under `tests` directory.
+
+### Run the tests (in the virtual environment):
+
+```sh
+pytest -v
+```

backend/requirements.txt

@@ -12,12 +12,17 @@ exceptiongroup==1.1.2
 fastapi==0.109.2
 greenlet==3.0.3
 h11==0.14.0
+httpcore==1.0.4
 httptools==0.6.1
+httpx==0.27.0
 idna==3.6
+iniconfig==2.0.0
 itsdangerous==2.1.2
 lsprotocol==2023.0.0a2
 lxml==5.1.0
 nodeenv==1.8.0
+packaging==24.0
+pluggy==1.4.0
 psycopg2-binary==2.9.9
 pycodestyle==2.11.1
 pycparser==2.21
@@ -26,10 +31,12 @@ pydantic_core==2.16.2
 pygls==1.0.1
 PyJWT==2.8.0
 pyright==1.1.352
+pytest==8.1.1
 python-cas==1.6.0
 python-dotenv==1.0.1
 PyYAML==6.0.1
 requests==2.31.0
+setuptools==69.1.1
 six==1.16.0
 sniffio==1.3.0
 SQLAlchemy==2.0.27

backend/src/auth/dependencies.py

@@ -23,8 +23,7 @@ def verify_jwt_token(
             credentials.credentials,
             CONFIG.secret_key,
             algorithms=[CONFIG.algorithm],
-            verify_signature=True,
-            options={"require": ["exp", "sub"]},
+            options={"require": ["exp", "sub"], "verify_signature": True},
         )
         user_id = payload["sub"]
         return user_id

backend/src/database.py

@@ -1,5 +1,5 @@
-from sqlalchemy import create_engine
-from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy import MetaData, create_engine
+from sqlalchemy.orm import declarative_base
 from sqlalchemy.orm import sessionmaker
 from src import config
 

backend/src/subject/dependencies.py

@@ -32,5 +32,5 @@ async def user_permission_validation(
 ):
     if not user.is_admin:
         teachers = await service.get_teachers(db, subject_id)
-        if not list(filter(lambda teacher: teacher.id == user.uid, teachers)):
+        if not list(filter(lambda teacher: teacher.uid == user.uid, teachers)):
             raise NotAuthorized()

backend/src/subject/router.py

@@ -9,8 +9,8 @@
 from .schemas import Subject, SubjectCreate, SubjectList
 
 router = APIRouter(
-    prefix="/api/subject",
-    tags=["subject"],
+    prefix="/api/subjects",
+    tags=["subjects"],
     responses={404: {"description": "Not found"}},
 )
 

backend/src/subject/schemas.py

@@ -1,6 +1,6 @@
 from enum import Enum
 from typing import Sequence
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ConfigDict
 
 
 class SubjectBase(BaseModel):
@@ -12,15 +12,13 @@ class SubjectCreate(SubjectBase):
 
 
 class Subject(SubjectBase):
-    id: int
+    model_config = ConfigDict(from_attributes=True)
 
-    class Config:
-        from_attributes = True
+    id: int
 
 
 class SubjectList(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
     as_teacher: Sequence[Subject]
     as_student: Sequence[Subject]
-
-    class Config:
-        from_attributes = True

backend/src/subject/service.py

@@ -1,7 +1,7 @@
 from typing import Sequence
 from sqlalchemy.orm import Session
-from src.user.models import User
 from . import models, schemas
+from src.user.models import User
 
 
 async def get_subject(db: Session, subject_id: int) -> models.Subject:
@@ -19,8 +19,7 @@ async def get_subjects(db: Session, user_id: str) -> tuple[Sequence[models.Subje
 
 
 async def get_teachers(db: Session, subject_id: int) -> list[User]:
-    return db.query(User).join(models.TeacherSubject).filter(
-        models.TeacherSubject.c.subject_id == subject_id).all()
+    return db.query(User).join(models.TeacherSubject, models.TeacherSubject.c.subject_id == subject_id).all()
 
 
 async def create_subject(db: Session, subject: schemas.SubjectCreate) -> models.Subject:

backend/src/user/schemas.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ConfigDict
 
 
 class Userbase(BaseModel):
@@ -12,7 +12,6 @@ class UserCreate(Userbase):
 
 
 class User(Userbase):
-    is_admin: bool = Field(default=False)
+    model_config = ConfigDict(from_attributes=True)
 
-    class Config:
-        from_attributes = True
+    is_admin: bool = Field(default=False)

backend/src/user/service.py

@@ -12,3 +12,9 @@ async def create_user(db: Session, user: schemas.UserCreate) -> models.User:
     db.commit()
     db.refresh(db_user)
     return db_user
+
+
+async def set_admin(db: Session, user_id: str, value: bool):
+    user = await get_by_id(db, user_id)
+    user.is_admin = value
+    db.commit()

backend/tests/conftest.py

@@ -0,0 +1,57 @@
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.orm import Session, sessionmaker
+from src.auth.utils import create_jwt_token
+from src.database import engine
+from src.main import app
+from src.dependencies import get_db
+import pytest
+
+from src.user.schemas import UserCreate
+from src.user.service import create_user
+
+connection = engine.connect()
+trans = connection.begin()
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=connection)
+
+
+def get_db_override():
+    db = TestingSessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+app.dependency_overrides[get_db] = get_db_override
+
+
+@pytest.fixture
+def anyio_backend():
+    return 'asyncio'
+
+
+@pytest.fixture
+async def db():
+    global trans
+    db = TestingSessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+        trans.rollback()
+        trans = connection.begin()
+
+
+@pytest.fixture
+async def client(db: Session):
+    token = create_jwt_token("test")
+
+    await create_user(db, UserCreate(uid="test", given_name="tester", mail="[email protected]"))
+
+    transport = ASGITransport(app=app)  # type: ignore
+    async with AsyncClient(transport=transport, base_url="http://test", headers={"Authorization": f"Bearer {token.token}"}) as client:
+        yield client
+
+
+def pytest_sessionfinish():
+    connection.close()

backend/tests/test_subject.py

@@ -0,0 +1,76 @@
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.orm import Session
+from src.user.schemas import UserCreate
+
+from src.user.service import set_admin
+from src.user.service import create_user
+
+subject = {'name': 'test subject'}
+
+
+@pytest.fixture
+async def subject_id(client: AsyncClient, db: Session) -> int:
+    """Create new subject"""
+    await set_admin(db, "test", True)
+    response = await client.post("/api/subjects/", json=subject)
+    return response.json()["id"]
+
+
+@pytest.mark.anyio
+async def test_create_subject(client: AsyncClient, db: Session):
+
+    await set_admin(db, "test", False)
+    response = await client.post("/api/subjects/", json=subject)
+    assert response.status_code == 403  # Forbidden, not admin
+
+    await set_admin(db, "test", True)
+    response = await client.post("/api/subjects/", json=subject)
+    assert response.status_code == 201  # Created
+    assert response.json()["name"] == subject["name"]
+
+
+@pytest.mark.anyio
+async def test_get_subject(client: AsyncClient, subject_id: int):
+
+    response = await client.get(f"/api/subjects/{subject_id}")
+    assert response.status_code == 200
+    assert response.json()["name"] == subject["name"]
+
+
+@pytest.mark.anyio
+async def test_create_teacher(client: AsyncClient, db: Session, subject_id: int):
+
+    await set_admin(db, "test", False)
+    response2 = await client.post(f"/api/subjects/{subject_id}/teachers", params={'user_id': 'test'})
+    assert response2.status_code == 403  # Forbidden
+
+    await set_admin(db, "test", True)
+    response2 = await client.post(f"/api/subjects/{subject_id}/teachers", params={'user_id': 'test'})
+    assert response2.status_code == 201
+
+    await set_admin(db, "test", False)
+    await create_user(db, UserCreate(uid="test2", given_name="tester", mail="[email protected]"))
+    response2 = await client.post(f"/api/subjects/{subject_id}/teachers", params={'user_id': 'test2'})
+    assert response2.status_code == 201  # Success because we are teacher now
+
+
+@pytest.mark.anyio
+async def test_get_subjects(client: AsyncClient, subject_id: int):
+    await client.post(f"/api/subjects/{subject_id}/teachers", params={'user_id': 'test'})
+    response2 = await client.get("/api/subjects/")
+    assert response2.status_code == 200
+    assert len(response2.json()["as_teacher"]) == 1
+
+
+@pytest.mark.anyio
+async def test_delete_subject(client: AsyncClient, db: Session, subject_id: int):
+    await set_admin(db, "test", False)
+    response2 = await client.delete(f"/api/subjects/{subject_id}")
+    assert response2.status_code == 403  # Forbidden
+    await set_admin(db, "test", True)
+    response3 = await client.delete(f"/api/subjects/{subject_id}")
+    assert response3.status_code == 200
+
+    response4 = await client.get(f"/api/subjects/{subject_id}")
+    assert response4.status_code == 404  # Not Found