From 523c28fc4442aaf580ba6161ab71fe99cc855cfe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Goran=20Meki=C4=87?= Date: Sun, 21 Jun 2026 03:28:00 +0200 Subject: [PATCH] Basic LMS --- freenit/api/lms.py | 511 ++++++++++++++++++++++++ freenit/base_config.py | 2 + freenit/models/lms.py | 16 + freenit/models/sql/lms.py | 125 ++++++ freenit/modules.py | 6 + freenit/permissions.py | 2 +- freenit/project/project/base_config.py | 2 +- migrations/0006_add_lms.py | 488 +++++++++++++++++++++++ tests/factories.py | 52 +++ tests/test_lms.py | 527 +++++++++++++++++++++++++ 10 files changed, 1729 insertions(+), 2 deletions(-) create mode 100644 freenit/api/lms.py create mode 100644 freenit/models/lms.py create mode 100644 freenit/models/sql/lms.py create mode 100644 migrations/0006_add_lms.py create mode 100644 tests/test_lms.py diff --git a/freenit/api/lms.py b/freenit/api/lms.py new file mode 100644 index 0000000..5d3b6c6 --- /dev/null +++ b/freenit/api/lms.py @@ -0,0 +1,511 @@ +from datetime import datetime + +import oxyde +import pydantic +from fastapi import Depends, Header, HTTPException + +from freenit.api.router import route +from freenit.decorators import description +from freenit.models.lms import ( + Course, + CourseGroup, + CourseGroupOptional, + CourseGroupPermission, + CourseMember, + CourseOptional, + Lecture, + LectureOptional, + LectureState, +) +from freenit.models.pagination import Page, paginate +from freenit.models.user import User +from freenit.permissions import lms_perms + +tags = ["lms"] + + +# --------------------------------------------------------------------------- +# Request / response schemas +# --------------------------------------------------------------------------- + + +class CourseCreate(pydantic.BaseModel): + name: str + description: str | None = None + + +class CourseResponse(pydantic.BaseModel): + model_config = pydantic.ConfigDict(from_attributes=True) + + id: int + name: str + description: str | None = None + created_by_id: int | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + + +class LectureCreate(pydantic.BaseModel): + title: str + content: str | None = None + position: int | None = None + state: LectureState | None = None + + +class LectureResponse(pydantic.BaseModel): + model_config = pydantic.ConfigDict(from_attributes=True) + + id: int + course_id: int + title: str + content: str | None = None + position: int + state: str + created_at: datetime | None = None + updated_at: datetime | None = None + + +class CourseGroupCreate(pydantic.BaseModel): + name: str + description: str | None = None + permissions: list[str] = [] + + +class CourseGroupResponse(pydantic.BaseModel): + model_config = pydantic.ConfigDict(from_attributes=True) + + id: int + course_id: int + name: str + description: str | None = None + permissions: list[str] = [] + created_at: datetime | None = None + updated_at: datetime | None = None + + +class CourseMemberCreate(pydantic.BaseModel): + user_id: int + + +class CourseMemberResponse(pydantic.BaseModel): + model_config = pydantic.ConfigDict(from_attributes=True) + + id: int + group_id: int + user_id: int + created_at: datetime | None = None + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +async def _get_course(id: int) -> Course: + try: + return await Course.objects.get(id=id) + except oxyde.NotFoundError: + raise HTTPException(status_code=404, detail="No such course") + + +async def _get_lecture(id: int) -> Lecture: + try: + return await Lecture.objects.get(id=id) + except oxyde.NotFoundError: + raise HTTPException(status_code=404, detail="No such lecture") + + +async def _get_group(id: int) -> CourseGroup: + try: + return await CourseGroup.objects.get(id=id) + except oxyde.NotFoundError: + raise HTTPException(status_code=404, detail="No such course group") + + +async def _get_member(group_id: int, user_id: int) -> CourseMember: + try: + return await CourseMember.objects.get(group_id=group_id, user_id=user_id) + except oxyde.NotFoundError: + raise HTTPException(status_code=404, detail="No such course member") + + +async def _next_position(model, filter_kwargs: dict) -> int: + existing = await model.objects.filter(**filter_kwargs).all() + if not existing: + return 0 + return max((getattr(item, "position", 0) for item in existing), default=0) + 1 + + +async def _check_course_name_unique(name: str, exclude_id: int | None = None) -> None: + existing = await Course.objects.filter(name=name).all() + if any(item.id != exclude_id for item in existing): + raise HTTPException(status_code=409, detail="Course name already exists") + + +async def _check_group_name_unique( + course_id: int, name: str, exclude_id: int | None = None +) -> None: + existing = await CourseGroup.objects.filter(course_id=course_id, name=name).all() + if any(item.id != exclude_id for item in existing): + raise HTTPException( + status_code=409, detail="Group name already exists in this course" + ) + + +async def _check_lecture_title_unique( + course_id: int, title: str, exclude_id: int | None = None +) -> None: + existing = await Lecture.objects.filter(course_id=course_id, title=title).all() + if any(item.id != exclude_id for item in existing): + raise HTTPException( + status_code=409, detail="Lecture title already exists in this course" + ) + + +async def _validate_member(group_id: int, user_id: int) -> None: + try: + await User.objects.get(id=user_id) + except oxyde.NotFoundError: + raise HTTPException(status_code=404, detail="No such user") + + +async def _get_group_permissions(group_id: int) -> list[str]: + permissions = await CourseGroupPermission.objects.filter(group_id=group_id).all() + return [p.permission for p in permissions] + + +async def _set_group_permissions(group_id: int, permissions: list[str]) -> None: + existing = await CourseGroupPermission.objects.filter(group_id=group_id).all() + for perm in existing: + await perm.delete() + for permission in permissions: + await CourseGroupPermission.objects.create( + group_id=group_id, permission=permission + ) + + +async def _user_has_course_permission( + user: User, course_id: int, permission: str +) -> bool: + if user.admin: + return True + groups = await CourseGroup.objects.filter(course_id=course_id).all() + group_ids = [g.id for g in groups] + if not group_ids: + return False + members = await CourseMember.objects.filter( + group_id__in=group_ids, user_id=user.id + ).all() + if not members: + return False + member_group_ids = [m.group_id for m in members] + perms = await CourseGroupPermission.objects.filter( + group_id__in=member_group_ids, permission=permission + ).all() + return len(perms) > 0 + + +async def _require_course_permission( + user: User, course_id: int, permission: str +) -> None: + if not await _user_has_course_permission(user, course_id, permission): + raise HTTPException(status_code=403, detail="Forbidden") + + +async def _visible_lecture_states(user: User, course_id: int) -> list[str]: + states = [LectureState.PUBLISHED_PUBLIC] + if await _user_has_course_permission(user, course_id, "view"): + states.append(LectureState.PUBLISHED_PRIVATE) + if await _user_has_course_permission(user, course_id, "edit"): + states.append(LectureState.DRAFT) + return states + + +# --------------------------------------------------------------------------- +# Courses +# --------------------------------------------------------------------------- + + +@route("/courses", tags=tags) +class CourseListAPI: + @staticmethod + @description("Get courses") + async def get( + page: int = Header(default=1), + perpage: int = Header(default=10), + _: User = Depends(lms_perms), + ) -> Page[CourseResponse]: + return await paginate(Course.objects, page, perpage) + + @staticmethod + @description("Create course") + async def post( + data: CourseCreate, + user: User = Depends(lms_perms), + ) -> CourseResponse: + await _check_course_name_unique(data.name) + now = datetime.utcnow() + course = await Course.objects.create( + name=data.name, + description=data.description, + created_by_id=getattr(user, "id", None), + created_at=now, + updated_at=now, + ) + return CourseResponse.model_validate(course) + + +@route("/courses/{id}", tags=tags) +class CourseDetailAPI: + @staticmethod + async def get(id: int, _: User = Depends(lms_perms)) -> CourseResponse: + course = await _get_course(id) + return CourseResponse.model_validate(course) + + @staticmethod + async def patch( + id: int, + data: CourseOptional, + user: User = Depends(lms_perms), + ) -> CourseResponse: + course = await _get_course(id) + await _require_course_permission(user, id, "edit") + if data.name: + await _check_course_name_unique(data.name, exclude_id=id) + await course.patch(data) + return CourseResponse.model_validate(course) + + @staticmethod + async def delete(id: int, user: User = Depends(lms_perms)) -> CourseResponse: + course = await _get_course(id) + await _require_course_permission(user, id, "edit") + await course.delete() + return CourseResponse.model_validate(course) + + +# --------------------------------------------------------------------------- +# Lectures +# --------------------------------------------------------------------------- + + +@route("/courses/{course_id}/lectures", tags=tags) +class CourseLectureListAPI: + @staticmethod + @description("Get course lectures") + async def get( + course_id: int, + page: int = Header(default=1), + perpage: int = Header(default=10), + user: User = Depends(lms_perms), + ) -> Page[LectureResponse]: + await _get_course(course_id) + states = await _visible_lecture_states(user, course_id) + return await paginate( + Lecture.objects.filter(course_id=course_id, state__in=states).order_by( + "position" + ), + page, + perpage, + ) + + @staticmethod + @description("Create lecture") + async def post( + course_id: int, + data: LectureCreate, + user: User = Depends(lms_perms), + ) -> LectureResponse: + await _get_course(course_id) + await _require_course_permission(user, course_id, "edit") + await _check_lecture_title_unique(course_id, data.title) + position = data.position + if position is None: + position = await _next_position(Lecture, {"course_id": course_id}) + now = datetime.utcnow() + lecture = await Lecture.objects.create( + course_id=course_id, + title=data.title, + content=data.content, + position=position, + state=data.state or LectureState.DRAFT, + created_at=now, + updated_at=now, + ) + return LectureResponse.model_validate(lecture) + + +@route("/lectures/{id}", tags=tags) +class LectureDetailAPI: + @staticmethod + async def get(id: int, user: User = Depends(lms_perms)) -> LectureResponse: + lecture = await _get_lecture(id) + if lecture.state == LectureState.DRAFT: + await _require_course_permission(user, lecture.course_id, "edit") + elif lecture.state == LectureState.PUBLISHED_PRIVATE: + await _require_course_permission(user, lecture.course_id, "view") + return LectureResponse.model_validate(lecture) + + @staticmethod + async def patch( + id: int, + data: LectureOptional, + user: User = Depends(lms_perms), + ) -> LectureResponse: + lecture = await _get_lecture(id) + await _require_course_permission(user, lecture.course_id, "edit") + if data.title: + await _check_lecture_title_unique( + lecture.course_id, data.title, exclude_id=id + ) + await lecture.patch(data) + return LectureResponse.model_validate(lecture) + + @staticmethod + async def delete(id: int, user: User = Depends(lms_perms)) -> LectureResponse: + lecture = await _get_lecture(id) + await _require_course_permission(user, lecture.course_id, "edit") + await lecture.delete() + return LectureResponse.model_validate(lecture) + + +# --------------------------------------------------------------------------- +# Course Groups +# --------------------------------------------------------------------------- + + +@route("/courses/{course_id}/groups", tags=tags) +class CourseGroupListAPI: + @staticmethod + @description("Get course groups") + async def get( + course_id: int, + page: int = Header(default=1), + perpage: int = Header(default=10), + _: User = Depends(lms_perms), + ) -> Page[CourseGroupResponse]: + await _get_course(course_id) + result = await paginate( + CourseGroup.objects.filter(course_id=course_id), + page, + perpage, + ) + for item in result.data: + object.__setattr__( + item, "permissions", await _get_group_permissions(item.id) + ) + return result + + @staticmethod + @description("Create course group") + async def post( + course_id: int, + data: CourseGroupCreate, + user: User = Depends(lms_perms), + ) -> CourseGroupResponse: + await _get_course(course_id) + await _require_course_permission(user, course_id, "edit") + await _check_group_name_unique(course_id, data.name) + now = datetime.utcnow() + group = await CourseGroup.objects.create( + course_id=course_id, + name=data.name, + description=data.description, + created_at=now, + updated_at=now, + ) + await _set_group_permissions(group.id, data.permissions) + object.__setattr__(group, "permissions", await _get_group_permissions(group.id)) + return CourseGroupResponse.model_validate(group) + + +@route("/course-groups/{id}", tags=tags) +class CourseGroupDetailAPI: + @staticmethod + async def get(id: int, _: User = Depends(lms_perms)) -> CourseGroupResponse: + group = await _get_group(id) + object.__setattr__(group, "permissions", await _get_group_permissions(id)) + return CourseGroupResponse.model_validate(group) + + @staticmethod + async def patch( + id: int, + data: CourseGroupOptional, + user: User = Depends(lms_perms), + ) -> CourseGroupResponse: + group = await _get_group(id) + await _require_course_permission(user, group.course_id, "edit") + if data.name: + await _check_group_name_unique(group.course_id, data.name, exclude_id=id) + update = data.model_dump(exclude_none=True) + if "permissions" in update: + await _set_group_permissions(id, update.pop("permissions")) + if update: + await group.patch(CourseGroupOptional(**update)) + object.__setattr__(group, "permissions", await _get_group_permissions(id)) + return CourseGroupResponse.model_validate(group) + + @staticmethod + async def delete(id: int, user: User = Depends(lms_perms)) -> CourseGroupResponse: + group = await _get_group(id) + await _require_course_permission(user, group.course_id, "edit") + await group.delete() + return CourseGroupResponse.model_validate(group) + + +# --------------------------------------------------------------------------- +# Course Group Members +# --------------------------------------------------------------------------- + + +@route("/course-groups/{group_id}/members", tags=tags) +class CourseGroupMemberListAPI: + @staticmethod + @description("Get course group members") + async def get( + group_id: int, + page: int = Header(default=1), + perpage: int = Header(default=10), + _: User = Depends(lms_perms), + ) -> Page[CourseMemberResponse]: + await _get_group(group_id) + return await paginate( + CourseMember.objects.filter(group_id=group_id), + page, + perpage, + ) + + @staticmethod + @description("Add member to course group") + async def post( + group_id: int, + data: CourseMemberCreate, + user: User = Depends(lms_perms), + ) -> CourseMemberResponse: + group = await _get_group(group_id) + await _require_course_permission(user, group.course_id, "edit") + await _validate_member(group_id, data.user_id) + now = datetime.utcnow() + try: + member = await CourseMember.objects.create( + group_id=group_id, + user_id=data.user_id, + created_at=now, + ) + except oxyde.IntegrityError: + raise HTTPException(status_code=409, detail="User already in group") + return CourseMemberResponse.model_validate(member) + + +@route("/course-groups/{group_id}/members/{user_id}", tags=tags) +class CourseGroupMemberDetailAPI: + @staticmethod + async def delete( + group_id: int, + user_id: int, + user: User = Depends(lms_perms), + ) -> CourseMemberResponse: + group = await _get_group(group_id) + await _require_course_permission(user, group.course_id, "edit") + member = await _get_member(group_id, user_id) + await member.delete() + return CourseMemberResponse.model_validate(member) diff --git a/freenit/base_config.py b/freenit/base_config.py index e06e84b..38fcf2e 100644 --- a/freenit/base_config.py +++ b/freenit/base_config.py @@ -140,6 +140,7 @@ class BaseConfig: role = "freenit.models.sql.role" mailinglist = "freenit.models.sql.mailinglist" project = "freenit.models.sql.project" + lms = "freenit.models.sql.lms" modules = ["auth"] meta = None auth = Auth() @@ -199,6 +200,7 @@ class TestConfig(BaseConfig): "user", "role", "project", + "lms", "mailinglist", "domain", "dav", diff --git a/freenit/models/lms.py b/freenit/models/lms.py new file mode 100644 index 0000000..bdaae23 --- /dev/null +++ b/freenit/models/lms.py @@ -0,0 +1,16 @@ +from freenit.config import getConfig + +config = getConfig() +lms = config.get_model("lms") + +Course = lms.Course +CourseOptional = lms.CourseOptional +Lecture = lms.Lecture +LectureOptional = lms.LectureOptional +LectureState = lms.LectureState +CourseGroup = lms.CourseGroup +CourseGroupOptional = lms.CourseGroupOptional +CourseGroupPermission = lms.CourseGroupPermission +CourseMember = lms.CourseMember +NotFoundError = lms.NotFoundError +IntegrityError = lms.IntegrityError diff --git a/freenit/models/sql/lms.py b/freenit/models/sql/lms.py new file mode 100644 index 0000000..de58cf9 --- /dev/null +++ b/freenit/models/sql/lms.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum + +import oxyde +import pydantic + +from freenit.models.sql.base import OxydeBaseModel, User + +NotFoundError = oxyde.NotFoundError +IntegrityError = oxyde.IntegrityError + + +class LectureState(StrEnum): + DRAFT = "draft" + PUBLISHED_PUBLIC = "published_public" + PUBLISHED_PRIVATE = "published_private" + + +class Course(OxydeBaseModel): + id: int | None = oxyde.Field(default=None, db_pk=True) + name: str = oxyde.Field(db_unique=True) + description: str | None = oxyde.Field(default=None) + created_by: User | None = oxyde.Field( + default=None, db_fk="id", db_on_delete="SET NULL" + ) + created_at: datetime | None = oxyde.Field(default=None) + updated_at: datetime | None = oxyde.Field(default=None) + + class Meta: + is_table = True + table_name = "course" + + +class Lecture(OxydeBaseModel): + id: int | None = oxyde.Field(default=None, db_pk=True) + course: Course | None = oxyde.Field( + default=None, db_fk="id", db_on_delete="CASCADE" + ) + title: str = oxyde.Field() + content: str | None = oxyde.Field(default=None) + position: int = oxyde.Field(default=0) + state: LectureState = oxyde.Field(default=LectureState.DRAFT) + created_at: datetime | None = oxyde.Field(default=None) + updated_at: datetime | None = oxyde.Field(default=None) + + class Meta: + is_table = True + table_name = "lecture" + unique_together = [("course_id", "title")] + + +class CourseOptional(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra="forbid") + + name: str | None = None + description: str | None = None + + +class LectureOptional(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra="forbid") + + title: str | None = None + content: str | None = None + position: int | None = None + state: LectureState | None = None + + +class CourseGroup(OxydeBaseModel): + id: int | None = oxyde.Field(default=None, db_pk=True) + course: Course | None = oxyde.Field( + default=None, db_fk="id", db_on_delete="CASCADE" + ) + name: str = oxyde.Field() + description: str | None = oxyde.Field(default=None) + created_at: datetime | None = oxyde.Field(default=None) + updated_at: datetime | None = oxyde.Field(default=None) + + class Meta: + is_table = True + table_name = "course_group" + unique_together = [("course_id", "name")] + + +class CourseGroupOptional(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra="forbid") + + name: str | None = None + description: str | None = None + permissions: list[str] | None = None + + +class CourseGroupPermission(OxydeBaseModel): + id: int | None = oxyde.Field(default=None, db_pk=True) + group: CourseGroup | None = oxyde.Field( + default=None, db_fk="id", db_on_delete="CASCADE" + ) + permission: str = oxyde.Field() + + class Meta: + is_table = True + table_name = "course_group_permission" + unique_together = [("group_id", "permission")] + + +class CourseMember(OxydeBaseModel): + id: int | None = oxyde.Field(default=None, db_pk=True) + group: CourseGroup | None = oxyde.Field( + default=None, db_fk="id", db_on_delete="CASCADE" + ) + user: User | None = oxyde.Field(default=None, db_fk="id", db_on_delete="CASCADE") + created_at: datetime | None = oxyde.Field(default=None) + + class Meta: + is_table = True + table_name = "course_member" + unique_together = [("group_id", "user_id")] + + +Course.model_rebuild() +Lecture.model_rebuild() +CourseGroup.model_rebuild() +CourseGroupPermission.model_rebuild() +CourseMember.model_rebuild() diff --git a/freenit/modules.py b/freenit/modules.py index ccd4d0a..73d0251 100644 --- a/freenit/modules.py +++ b/freenit/modules.py @@ -36,6 +36,12 @@ def __hash__(self): models=["freenit.models.sql.project"], api="freenit.api.project", ), + "lms": Module( + name="lms", + dependencies=["user"], + models=["freenit.models.sql.lms"], + api="freenit.api.lms", + ), "mailinglist": Module( name="mailinglist", dependencies=["user"], diff --git a/freenit/permissions.py b/freenit/permissions.py index 5d09c0c..294d76a 100644 --- a/freenit/permissions.py +++ b/freenit/permissions.py @@ -2,9 +2,9 @@ domain_perms = permissions() group_perms = permissions() +lms_perms = permissions() mailinglist_perms = permissions() profile_perms = permissions() project_perms = permissions() role_perms = permissions() -theme_perms = permissions() user_perms = permissions() diff --git a/freenit/project/project/base_config.py b/freenit/project/project/base_config.py index 36c5b18..9ebe963 100644 --- a/freenit/project/project/base_config.py +++ b/freenit/project/project/base_config.py @@ -6,7 +6,7 @@ class BaseConfig(FreenitBaseConfig): version = "0.0.1" # Modules are inherited from FreenitBaseConfig (default: ["auth"]). # Add feature modules here, e.g.: - # modules = ["auth", "project"] + # modules = ["auth", "project", "lms"] stalwart_url = "http://stalwart.example.com" stalwart_admin = "%admin" stalwart_admin_pass = "" # nosec: B105 diff --git a/migrations/0006_add_lms.py b/migrations/0006_add_lms.py new file mode 100644 index 0000000..9940e31 --- /dev/null +++ b/migrations/0006_add_lms.py @@ -0,0 +1,488 @@ +"""Auto-generated migration. + +Created: 2026-06-20 23:45:00 +""" + +depends_on = "0005_add_project_groups" + + +def upgrade(ctx): + """Apply migration.""" + ctx.create_table( + "course", + fields=[ + { + "name": "id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": True, + "unique": True, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "name", + "python_type": "str", + "db_type": None, + "nullable": False, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "description", + "python_type": "str", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "created_at", + "python_type": "datetime", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "updated_at", + "python_type": "datetime", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "created_by_id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + ], + foreign_keys=[ + { + "name": "fk_course_created_by_id", + "columns": ["created_by_id"], + "ref_table": "user", + "ref_columns": ["id"], + "on_delete": "SET NULL", + "on_update": "CASCADE", + } + ], + indexes=[ + { + "name": "idx_course_name", + "fields": ["name"], + "unique": True, + } + ], + ) + ctx.create_table( + "lecture", + fields=[ + { + "name": "id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": True, + "unique": True, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "title", + "python_type": "str", + "db_type": None, + "nullable": False, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "content", + "python_type": "str", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "position", + "python_type": "int", + "db_type": None, + "nullable": False, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "state", + "python_type": "str", + "db_type": None, + "nullable": False, + "primary_key": False, + "unique": False, + "default": "draft", + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "created_at", + "python_type": "datetime", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "updated_at", + "python_type": "datetime", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "course_id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + ], + foreign_keys=[ + { + "name": "fk_lecture_course_id", + "columns": ["course_id"], + "ref_table": "course", + "ref_columns": ["id"], + "on_delete": "CASCADE", + "on_update": "CASCADE", + } + ], + indexes=[ + { + "name": "idx_lecture_course_id_title", + "fields": ["course_id", "title"], + "unique": True, + } + ], + ) + ctx.create_table( + "course_group", + fields=[ + { + "name": "id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": True, + "unique": True, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "name", + "python_type": "str", + "db_type": None, + "nullable": False, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "description", + "python_type": "str", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "created_at", + "python_type": "datetime", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "updated_at", + "python_type": "datetime", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "course_id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + ], + foreign_keys=[ + { + "name": "fk_course_group_course_id", + "columns": ["course_id"], + "ref_table": "course", + "ref_columns": ["id"], + "on_delete": "CASCADE", + "on_update": "CASCADE", + } + ], + indexes=[ + { + "name": "idx_course_group_course_id_name", + "fields": ["course_id", "name"], + "unique": True, + } + ], + ) + ctx.create_table( + "course_group_permission", + fields=[ + { + "name": "id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": True, + "unique": True, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "permission", + "python_type": "str", + "db_type": None, + "nullable": False, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "group_id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + ], + foreign_keys=[ + { + "name": "fk_course_group_permission_group_id", + "columns": ["group_id"], + "ref_table": "course_group", + "ref_columns": ["id"], + "on_delete": "CASCADE", + "on_update": "CASCADE", + } + ], + indexes=[ + { + "name": "idx_course_group_permission_group_id_permission", + "fields": ["group_id", "permission"], + "unique": True, + } + ], + ) + ctx.create_table( + "course_member", + fields=[ + { + "name": "id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": True, + "unique": True, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "created_at", + "python_type": "datetime", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "group_id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + { + "name": "user_id", + "python_type": "int", + "db_type": None, + "nullable": True, + "primary_key": False, + "unique": False, + "default": None, + "auto_increment": False, + "max_length": None, + "max_digits": None, + "decimal_places": None, + }, + ], + foreign_keys=[ + { + "name": "fk_course_member_group_id", + "columns": ["group_id"], + "ref_table": "course_group", + "ref_columns": ["id"], + "on_delete": "CASCADE", + "on_update": "CASCADE", + }, + { + "name": "fk_course_member_user_id", + "columns": ["user_id"], + "ref_table": "user", + "ref_columns": ["id"], + "on_delete": "CASCADE", + "on_update": "CASCADE", + }, + ], + indexes=[ + { + "name": "idx_course_member_group_id_user_id", + "fields": ["group_id", "user_id"], + "unique": True, + } + ], + ) + + +def downgrade(ctx): + """Revert migration.""" + ctx.drop_table("course_member") + ctx.drop_table("course_group_permission") + ctx.drop_table("course_group") + ctx.drop_table("lecture") + ctx.drop_table("course") diff --git a/tests/factories.py b/tests/factories.py index a51feaa..bd4d926 100644 --- a/tests/factories.py +++ b/tests/factories.py @@ -2,6 +2,13 @@ from passlib.hash import pbkdf2_sha256 from freenit.config import getConfig +from freenit.models.lms import ( + Course, + CourseGroup, + CourseGroupPermission, + CourseMember, + Lecture, +) from freenit.models.project import ( Board, Column, @@ -98,3 +105,48 @@ class Meta: group_id = None permission = factory.Faker("pystr") + + +class CourseFactory(factory.Factory): + class Meta: + model = Course + + name = factory.Faker("pystr") + description = factory.Faker("sentence") + created_by_id = None + + +class LectureFactory(factory.Factory): + class Meta: + model = Lecture + + title = factory.Faker("pystr") + content = factory.Faker("sentence") + position = factory.Faker("random_int") + state = "draft" + course_id = None + + +class CourseGroupFactory(factory.Factory): + class Meta: + model = CourseGroup + + name = factory.Faker("pystr") + description = factory.Faker("sentence") + course_id = None + + +class CourseMemberFactory(factory.Factory): + class Meta: + model = CourseMember + + group_id = None + user_id = None + + +class CourseGroupPermissionFactory(factory.Factory): + class Meta: + model = CourseGroupPermission + + group_id = None + permission = "edit" diff --git a/tests/test_lms.py b/tests/test_lms.py new file mode 100644 index 0000000..10d333b --- /dev/null +++ b/tests/test_lms.py @@ -0,0 +1,527 @@ +import pytest + +from . import factories + + +@pytest.mark.asyncio +class TestCourse: + async def test_create_course(self, client): + user = factories.User() + await user.save() + client.login(user=user) + data = {"name": "Test Course", "description": "A test course"} + response = client.post("/courses", data=data) + assert response.status_code == 200 + result = response.json() + assert result["name"] == data["name"] + assert result["description"] == data["description"] + assert result["created_by_id"] == user.id + + async def test_get_courses(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory(created_by_id=user.id) + await course.save() + response = client.get("/courses") + assert response.status_code == 200 + assert response.json()["total"] >= 1 + + async def test_get_course(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory(created_by_id=user.id) + await course.save() + response = client.get(f"/courses/{course.id}") + assert response.status_code == 200 + assert response.json()["id"] == course.id + + async def test_update_course(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + data = {"name": "Updated Course"} + response = client.patch(f"/courses/{course.id}", data=data) + assert response.status_code == 200 + assert response.json()["name"] == data["name"] + + async def test_delete_course(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + response = client.delete(f"/courses/{course.id}") + assert response.status_code == 200 + response = client.get(f"/courses/{course.id}") + assert response.status_code == 404 + + async def test_create_course_duplicate_name(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory(name="Unique Course") + await course.save() + response = client.post("/courses", data={"name": "Unique Course"}) + assert response.status_code == 409 + + async def test_update_course_duplicate_name(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course1 = factories.CourseFactory(name="Course One") + await course1.save() + course2 = factories.CourseFactory(name="Course Two") + await course2.save() + group = factories.CourseGroupFactory(course_id=course2.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + response = client.patch(f"/courses/{course2.id}", data={"name": "Course One"}) + assert response.status_code == 409 + + +@pytest.mark.asyncio +class TestLecture: + async def test_create_lecture(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + data = {"title": "Test Lecture", "content": "A test lecture"} + response = client.post(f"/courses/{course.id}/lectures", data=data) + assert response.status_code == 200 + result = response.json() + assert result["title"] == data["title"] + assert result["course_id"] == course.id + assert result["state"] == "draft" + + async def test_create_lecture_with_state(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + data = {"title": "Public Lecture", "state": "published_public"} + response = client.post(f"/courses/{course.id}/lectures", data=data) + assert response.status_code == 200 + assert response.json()["state"] == "published_public" + + async def test_get_lectures(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + lecture = factories.LectureFactory( + course_id=course.id, state="published_public" + ) + await lecture.save() + response = client.get(f"/courses/{course.id}/lectures") + assert response.status_code == 200 + assert response.json()["total"] >= 1 + + async def test_get_lecture(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + lecture = factories.LectureFactory( + course_id=course.id, state="published_public" + ) + await lecture.save() + response = client.get(f"/lectures/{lecture.id}") + assert response.status_code == 200 + assert response.json()["id"] == lecture.id + + async def test_draft_lecture_hidden_from_non_editor(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + lecture = factories.LectureFactory(course_id=course.id, state="draft") + await lecture.save() + response = client.get(f"/lectures/{lecture.id}") + assert response.status_code == 403 + response = client.get(f"/courses/{course.id}/lectures") + assert response.json()["total"] == 0 + + async def test_published_private_lecture_requires_view_permission(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + lecture = factories.LectureFactory( + course_id=course.id, state="published_private" + ) + await lecture.save() + response = client.get(f"/lectures/{lecture.id}") + assert response.status_code == 403 + + async def test_published_private_lecture_visible_with_view_permission(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + lecture = factories.LectureFactory( + course_id=course.id, state="published_private" + ) + await lecture.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="view" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + response = client.get(f"/lectures/{lecture.id}") + assert response.status_code == 200 + assert response.json()["id"] == lecture.id + + async def test_update_lecture(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + lecture = factories.LectureFactory(course_id=course.id) + await lecture.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + data = {"title": "Updated Lecture", "state": "published_public"} + response = client.patch(f"/lectures/{lecture.id}", data=data) + assert response.status_code == 200 + result = response.json() + assert result["title"] == data["title"] + assert result["state"] == "published_public" + + async def test_delete_lecture(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + lecture = factories.LectureFactory(course_id=course.id) + await lecture.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + response = client.delete(f"/lectures/{lecture.id}") + assert response.status_code == 200 + response = client.get(f"/lectures/{lecture.id}") + assert response.status_code == 404 + + async def test_create_lecture_duplicate_title(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + lecture = factories.LectureFactory(course_id=course.id, title="Lecture X") + await lecture.save() + response = client.post( + f"/courses/{course.id}/lectures", data={"title": "Lecture X"} + ) + assert response.status_code == 409 + + +@pytest.mark.asyncio +class TestCourseGroup: + async def test_create_course_group(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + data = {"name": "Test Group", "description": "A test group"} + response = client.post(f"/courses/{course.id}/groups", data=data) + assert response.status_code == 200 + result = response.json() + assert result["name"] == data["name"] + assert result["course_id"] == course.id + + async def test_get_course_groups(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + response = client.get(f"/courses/{course.id}/groups") + assert response.status_code == 200 + assert response.json()["total"] >= 1 + + async def test_get_course_group(self, client): + user = factories.User() + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + response = client.get(f"/course-groups/{group.id}") + assert response.status_code == 200 + assert response.json()["id"] == group.id + assert "edit" in response.json()["permissions"] + + async def test_update_course_group(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + data = {"name": "Updated Group"} + response = client.patch(f"/course-groups/{group.id}", data=data) + assert response.status_code == 200 + assert response.json()["name"] == data["name"] + + async def test_delete_course_group(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + response = client.delete(f"/course-groups/{group.id}") + assert response.status_code == 200 + response = client.get(f"/course-groups/{group.id}") + assert response.status_code == 404 + + async def test_create_course_group_duplicate_name(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id, name="Group X") + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + response = client.post(f"/courses/{course.id}/groups", data={"name": "Group X"}) + assert response.status_code == 409 + + +@pytest.mark.asyncio +class TestCourseGroupMember: + async def test_add_course_group_member(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + target = factories.User() + await target.save() + response = client.post( + f"/course-groups/{group.id}/members", data={"user_id": target.id} + ) + assert response.status_code == 200 + result = response.json() + assert result["group_id"] == group.id + assert result["user_id"] == target.id + + async def test_get_course_group_members(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + response = client.get(f"/course-groups/{group.id}/members") + assert response.status_code == 200 + assert response.json()["total"] >= 1 + + async def test_remove_course_group_member(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + response = client.delete(f"/course-groups/{group.id}/members/{user.id}") + assert response.status_code == 200 + response = client.get(f"/course-groups/{group.id}/members") + assert response.json()["total"] == 0 + + async def test_add_duplicate_course_group_member(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + response = client.post( + f"/course-groups/{group.id}/members", data={"user_id": user.id} + ) + assert response.status_code == 409 + + +@pytest.mark.asyncio +class TestCourseGroupPermissions: + async def test_create_course_group_with_permissions(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + data = { + "name": "Test Group", + "description": "A test group", + "permissions": ["edit"], + } + response = client.post(f"/courses/{course.id}/groups", data=data) + assert response.status_code == 200 + result = response.json() + assert result["permissions"] == data["permissions"] + + async def test_update_course_group_permissions(self, client): + user = factories.User(admin=True) + await user.save() + client.login(user=user) + course = factories.CourseFactory() + await course.save() + group = factories.CourseGroupFactory(course_id=course.id) + await group.save() + perm = factories.CourseGroupPermissionFactory( + group_id=group.id, permission="edit" + ) + await perm.save() + member = factories.CourseMemberFactory(group_id=group.id, user_id=user.id) + await member.save() + data = {"permissions": []} + response = client.patch(f"/course-groups/{group.id}", data=data) + assert response.status_code == 200 + result = response.json() + assert result["permissions"] == data["permissions"]