diff --git a/controllers/endpoints/assignment_groups.py b/controllers/endpoints/assignment_groups.py index 86f8275a7..35813c565 100644 --- a/controllers/endpoints/assignment_groups.py +++ b/controllers/endpoints/assignment_groups.py @@ -15,7 +15,10 @@ from models.assignment import Assignment from models.assignment_group import AssignmentGroup from models.assignment_group_membership import AssignmentGroupMembership +from models.assignment_group_variation import AssignmentGroupVariation from models.data_formats.portation import export_bundle, import_bundle, export_zip, export_pdf_zip +from models.role import Role +from models.enums.roles import UserRoles blueprint_assignment_group = Blueprint('assignment_group', __name__, url_prefix='/assignment_group') @@ -387,3 +390,203 @@ def export_submissions(): filename = course.get_url_or_id() + '-' + assignment_group.get_filename(extension='.zip') return Response(bundle, mimetype='application/zip', headers={'Content-Disposition': 'attachment;filename={}'.format(filename)}) + + +# --------------------------------------------------------------------------- +# Question Variations Endpoints +# --------------------------------------------------------------------------- + +@blueprint_assignment_group.route('/get_variations', methods=['GET']) +@blueprint_assignment_group.route('/get_variations/', methods=['GET']) +@require_request_parameters('assignment_group_id') +@login_required +def get_variations(): + """ + Return variation configuration and, optionally, the assignments assigned to a + specific user in this group. + + Query parameters: + assignment_group_id – required + user_id – optional; if omitted, defaults to the current user + """ + if g.user.anonymous: + return jsonify(success=False, message="You must be logged in to view variation assignments.") + assignment_group_id = int(request.values.get('assignment_group_id')) + assignment_group = AssignmentGroup.by_id(assignment_group_id) + check_resource_exists(assignment_group, "Assignment Group", assignment_group_id) + + requested_user_id = maybe_int(request.values.get('user_id')) + if requested_user_id is None: + requested_user_id = g.user.id + # Instructors may view any student's variations; students may only view their own + if requested_user_id != g.user.id: + require_course_instructor(g.user, assignment_group.course_id) + + variation_groups = assignment_group.get_variation_groups() + assigned_ids = AssignmentGroupVariation.get_assignment_ids_for_user( + requested_user_id, assignment_group_id) + + return jsonify( + success=True, + assignment_group_id=assignment_group_id, + variation_count=assignment_group.variation_count, + variation_groups=variation_groups, + assigned_assignment_ids=assigned_ids, + user_id=requested_user_id, + ) + + +@blueprint_assignment_group.route('/assign_variation', methods=['POST']) +@blueprint_assignment_group.route('/assign_variation/', methods=['POST']) +@require_request_parameters('assignment_group_id', 'user_id', 'assignment_ids') +@login_required +def assign_variation(): + """ + Explicitly assign a set of variation assignments to a specific user. + + Form parameters: + assignment_group_id – required + user_id – required; the student being assigned + assignment_ids – required; JSON array of assignment IDs + """ + assignment_group_id = int(request.values.get('assignment_group_id')) + assignment_group = AssignmentGroup.by_id(assignment_group_id) + check_resource_exists(assignment_group, "Assignment Group", assignment_group_id) + require_course_instructor(g.user, assignment_group.course_id) + + target_user_id = int(request.values.get('user_id')) + try: + assignment_ids = json.loads(request.values.get('assignment_ids')) + except (ValueError, TypeError): + return jsonify(success=False, message="assignment_ids must be a JSON array of integers") + if not isinstance(assignment_ids, list): + return jsonify(success=False, message="assignment_ids must be a list of integers") + + # Validate all IDs belong to this group + valid_ids = {a.id for a in assignment_group.get_assignments()} + invalid = [aid for aid in assignment_ids if aid not in valid_ids] + if invalid: + return jsonify(success=False, + message="Some assignment IDs do not belong to this group: {}".format(invalid)) + + assignment_group.assign_variation_to_user(target_user_id, assignment_ids) + + return jsonify( + success=True, + assignment_group_id=assignment_group_id, + user_id=target_user_id, + assigned_assignment_ids=assignment_ids, + ) + + +@blueprint_assignment_group.route('/assign_variations_random', methods=['POST']) +@blueprint_assignment_group.route('/assign_variations_random/', methods=['POST']) +@require_request_parameters('assignment_group_id') +@login_required +def assign_variations_random(): + """ + Randomly assign variation assignments to all students enrolled in the group's course + who do not yet have a variation assignment for this group. + + Form parameters: + assignment_group_id – required + course_id – optional; defaults to the group's own course + overwrite – optional boolean; if true, re-assign even existing students + """ + assignment_group_id = int(request.values.get('assignment_group_id')) + assignment_group = AssignmentGroup.by_id(assignment_group_id) + check_resource_exists(assignment_group, "Assignment Group", assignment_group_id) + require_course_instructor(g.user, assignment_group.course_id) + + overwrite = maybe_bool(request.values.get('overwrite', 'false')) + course_id = maybe_int(request.values.get('course_id')) or assignment_group.course_id + + if assignment_group.variation_count == 0: + return jsonify(success=False, + message="This group has variation_count=0; set it > 0 to use variations.") + + if not assignment_group.get_variation_groups(): + return jsonify(success=False, + message="No variation groups defined. " + "Set variation_group on AssignmentGroupMembership records first.") + + # Find all students in the course + student_roles = Role.query.filter_by(course_id=course_id, name=UserRoles.LEARNER).all() + assigned_count = 0 + skipped_count = 0 + for role in student_roles: + user_id = role.user_id + existing = AssignmentGroupVariation.get_assignment_ids_for_user(user_id, assignment_group_id) + if existing and not overwrite: + skipped_count += 1 + continue + assignment_group.assign_random_variation(user_id) + assigned_count += 1 + + return jsonify( + success=True, + assignment_group_id=assignment_group_id, + course_id=course_id, + assigned_count=assigned_count, + skipped_count=skipped_count, + ) + + +@blueprint_assignment_group.route('/edit_variation_settings', methods=['GET', 'POST']) +@blueprint_assignment_group.route('/edit_variation_settings/', methods=['GET', 'POST']) +@require_request_parameters('assignment_group_id') +@login_required +def edit_variation_settings(): + """ + GET – return the current variation_count and membership variation_group values. + POST – update variation_count on the group and variation_group on individual memberships. + + Form parameters (POST): + assignment_group_id – required + variation_count – required integer; 0 = all assignments, N = student does N + variation_group[] – optional per-membership variation group tag + """ + assignment_group_id = int(request.values.get('assignment_group_id')) + assignment_group = AssignmentGroup.by_id(assignment_group_id) + check_resource_exists(assignment_group, "Assignment Group", assignment_group_id) + require_course_instructor(g.user, assignment_group.course_id) + + if request.method == 'POST': + variation_count = maybe_int(request.values.get('variation_count')) + if variation_count is None or variation_count < 0: + return jsonify(success=False, message="variation_count must be a non-negative integer") + assignment_group.variation_count = variation_count + + memberships = assignment_group.get_memberships() + for membership in memberships: + key = 'variation_group[{}]'.format(membership.id) + if key in request.values: + raw = request.values.get(key) + membership.variation_group = maybe_int(raw) + + from models.generics.models import db + db.session.commit() + + return jsonify( + success=True, + assignment_group_id=assignment_group_id, + variation_count=assignment_group.variation_count, + ) + else: + memberships = assignment_group.get_memberships() + membership_data = [ + { + 'id': m.id, + 'assignment_id': m.assignment_id, + 'position': m.position, + 'variation_group': m.variation_group, + } + for m in memberships + ] + return jsonify( + success=True, + assignment_group_id=assignment_group_id, + variation_count=assignment_group.variation_count, + memberships=membership_data, + ) + diff --git a/controllers/helpers.py b/controllers/helpers.py index 0c05de90b..f19cab40f 100644 --- a/controllers/helpers.py +++ b/controllers/helpers.py @@ -207,7 +207,7 @@ def parse_assignment_load(assignment_id_or_url=None): else: assignments = [] else: - assignments = assignment_group.get_assignments() + assignments = assignment_group.get_assignments(user_id=user_id) # No existing assignment, let's get the default if not assignments: # And no known course? Better get the default course! diff --git a/migrations/versions/b1c2d3e4f5a6_add_question_variations.py b/migrations/versions/b1c2d3e4f5a6_add_question_variations.py new file mode 100644 index 000000000..ac45f1d26 --- /dev/null +++ b/migrations/versions/b1c2d3e4f5a6_add_question_variations.py @@ -0,0 +1,61 @@ +"""Add question variations support in assignment groups + +Adds: + * assignment_group.variation_count – how many variation assignments a student completes + * assignment_group_membership.variation_group – pool tag (NULL = required, int = variation pool) + * assignment_group_variation table – per-student variation assignment records + +Revision ID: b1c2d3e4f5a6 +Revises: a62e70d564b3 +Create Date: 2025-09-01 00:00:00.000000 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'b1c2d3e4f5a6' +down_revision = 'a62e70d564b3' +branch_labels = None +depends_on = None + + +def upgrade(): + # Add variation_count to assignment_group (default 0 = all assignments required) + op.add_column('assignment_group', + sa.Column('variation_count', sa.Integer(), nullable=False, server_default='0')) + + # Add variation_group to assignment_group_membership (NULL = required for all) + op.add_column('assignment_group_membership', + sa.Column('variation_group', sa.Integer(), nullable=True)) + + # Create the assignment_group_variation table + op.create_table( + 'assignment_group_variation', + sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True), + sa.Column('date_created', sa.DateTime(), nullable=True), + sa.Column('date_modified', sa.DateTime(), nullable=True), + sa.Column('user_id', sa.Integer(), sa.ForeignKey('user.id'), nullable=False), + sa.Column('assignment_group_id', sa.Integer(), sa.ForeignKey('assignment_group.id'), nullable=False), + sa.Column('assignment_id', sa.Integer(), sa.ForeignKey('assignment.id'), nullable=False), + ) + op.create_unique_constraint( + 'uq_agv_user_group_assignment', + 'assignment_group_variation', + ['user_id', 'assignment_group_id', 'assignment_id'], + ) + op.create_index('agv_user_group_index', 'assignment_group_variation', + ['user_id', 'assignment_group_id']) + op.create_index('agv_group_index', 'assignment_group_variation', + ['assignment_group_id']) + + +def downgrade(): + op.drop_index('agv_group_index', 'assignment_group_variation') + op.drop_index('agv_user_group_index', 'assignment_group_variation') + op.drop_constraint('uq_agv_user_group_assignment', 'assignment_group_variation', + type_='unique') + op.drop_table('assignment_group_variation') + op.drop_column('assignment_group_membership', 'variation_group') + op.drop_column('assignment_group', 'variation_count') diff --git a/models/__init__.py b/models/__init__.py index abb85d431..18a29650b 100644 --- a/models/__init__.py +++ b/models/__init__.py @@ -12,6 +12,7 @@ from models.assignment_tag_membership import assignment_tag_membership from models.assignment_group import AssignmentGroup from models.assignment_group_membership import AssignmentGroupMembership +from models.assignment_group_variation import AssignmentGroupVariation from models.authentication import Authentication from models.log_tables import AccessLog, ErrorLog, CourseLog, RoleLog, AssignmentLog, SubmissionLog from models.review import Review @@ -51,7 +52,7 @@ def init_database(app: Flask) -> Flask: #: A listing of all the tables ALL_TABLES = ( - Assignment, AssignmentGroup, AssignmentGroupMembership, + Assignment, AssignmentGroup, AssignmentGroupMembership, AssignmentGroupVariation, AssignmentTag, assignment_tag_membership, Course, Invite, Submission, Review, SampleSubmission, diff --git a/models/assignment.py b/models/assignment.py index fc1cd68f2..97a7823ba 100644 --- a/models/assignment.py +++ b/models/assignment.py @@ -88,6 +88,7 @@ class Assignment(EnhancedBase): assignment_logs: Mapped[list["AssignmentLog"]] = db.relationship(back_populates="assignment") submissions: Mapped[list["Submission"]] = db.relationship(back_populates="assignment") memberships: Mapped[list["AssignmentGroupMembership"]] = db.relationship(back_populates="assignment") + group_variations: Mapped[list["AssignmentGroupVariation"]] = db.relationship(back_populates="assignment") reports: Mapped[list["Report"]] = db.relationship(back_populates="assignment") __table_args__ = (Index("assignment_url_index", "url"), diff --git a/models/assignment_group.py b/models/assignment_group.py index 7b1467507..48df26246 100644 --- a/models/assignment_group.py +++ b/models/assignment_group.py @@ -1,3 +1,4 @@ +import random from typing import List, Optional, TYPE_CHECKING from flask import url_for @@ -32,12 +33,16 @@ class AssignmentGroup(EnhancedBase): course_id: Mapped[int] = mapped_column(Integer(), ForeignKey('course.id')) position: Mapped[int] = mapped_column(Integer(), default=0) version: Mapped[int] = mapped_column(Integer(), default=0) + #: When > 0, students are assigned this many variation assignments from the pool + #: instead of completing all assignments. 0 means every student does all assignments. + variation_count: Mapped[int] = mapped_column(Integer(), default=0) forked: Mapped["AssignmentGroup"] = db.relationship("AssignmentGroup", remote_side="AssignmentGroup.id") owner: Mapped["User"] = db.relationship(back_populates="assignment_groups") course: Mapped["Course"] = db.relationship(back_populates="assignment_groups") memberships: Mapped[list["AssignmentGroupMembership"]] = db.relationship(back_populates="assignment_group") submissions: Mapped[list["Submission"]] = db.relationship(back_populates="assignment_group") + variations: Mapped[list["AssignmentGroupVariation"]] = db.relationship(back_populates="assignment_group") __table_args__ = (Index("assignment_group_url_index", "url"), Index('assignment_group_course_index', "course_id")) @@ -57,6 +62,7 @@ def encode_json(self): 'owner_id__email': user.email if user else '', 'course_id': self.course_id, 'position': self.position, + 'variation_count': self.variation_count, 'id': self.id, 'date_modified': datetime_to_string(self.date_modified), 'date_created': datetime_to_string(self.date_created)} @@ -160,7 +166,43 @@ def get_ungrouped_assignments(course_id): # TODO # SELECT assignment, grp FROM assignment JOIN assignment_group_membership as members ON members.assignment_id = assignment.id JOIN assignment_group as grp ON grp.id = members.assignment_group_id WHERE assignment.id IN (SELECT DISTINCT submission.assignment_id FROM submission WHERE submission.course_id=11) AND (grp.course_id=11 OR assignment.course_id = grp.course_id) - def get_assignments(self) -> 'List[models.Assignment]': + def get_assignments(self, user_id: int = None) -> 'List[models.Assignment]': + """ + Return the assignments in this group. + + If *user_id* is supplied **and** this group has ``variation_count > 0``, + the result is the union of: + + * Required assignments (those whose membership has ``variation_group`` IS NULL) + * The variation assignments that have been specifically assigned to *user_id* + (recorded in ``AssignmentGroupVariation``). + + When ``variation_count == 0`` (the default) or no *user_id* is given, all + assignments in the group are returned, preserving the existing behaviour. + """ + if user_id is not None and self.variation_count > 0: + # Required assignments: membership has no variation_group + required = (models.Assignment.query + .join(models.AssignmentGroupMembership, + models.AssignmentGroupMembership.assignment_id == models.Assignment.id) + .filter(models.AssignmentGroupMembership.assignment_group_id == self.id) + .filter(models.AssignmentGroupMembership.variation_group.is_(None)) + .order_by(models.Assignment.name, + models.AssignmentGroupMembership.position) + .all()) + # Variation assignments specifically assigned to this user + from models.assignment_group_variation import AssignmentGroupVariation + assigned_ids = AssignmentGroupVariation.get_assignment_ids_for_user( + user_id, self.id) + if assigned_ids: + variation_assignments = (models.Assignment.query + .filter(models.Assignment.id.in_(assigned_ids)) + .all()) + else: + variation_assignments = [] + combined = list({a.id: a for a in required + variation_assignments}.values()) + return natsorted(combined, key=lambda a: a.title()) + # Default: return all assignments assignments = (models.Assignment.query .join(models.AssignmentGroupMembership, models.AssignmentGroupMembership.assignment_id == models.Assignment.id) @@ -188,6 +230,44 @@ def get_filename(self, extension=".json"): else: return secure_filename(self.name) + extension + def get_variation_groups(self) -> 'List[int]': + """Return the distinct variation_group values used by memberships in this group.""" + rows = (db.session.query(models.AssignmentGroupMembership.variation_group) + .filter(models.AssignmentGroupMembership.assignment_group_id == self.id) + .filter(models.AssignmentGroupMembership.variation_group.isnot(None)) + .distinct() + .all()) + return sorted(row[0] for row in rows) + + def assign_variation_to_user(self, user_id: int, + assignment_ids: 'List[int]'): + """ + Explicitly set the variation assignments for *user_id* in this group. + + Replaces any previously stored variation assignments for that user. + """ + from models.assignment_group_variation import AssignmentGroupVariation + AssignmentGroupVariation.assign_to_user(user_id, self.id, assignment_ids) + + def assign_random_variation(self, user_id: int) -> 'List[int]': + """ + Randomly choose *variation_count* variation groups and assign the + corresponding assignments to *user_id*. + + Returns the list of assigned assignment IDs. + """ + available_groups = self.get_variation_groups() + count = min(self.variation_count, len(available_groups)) + selected_groups = random.sample(available_groups, count) + # Collect assignments that belong to the selected groups + memberships = (models.AssignmentGroupMembership.query + .filter(models.AssignmentGroupMembership.assignment_group_id == self.id) + .filter(models.AssignmentGroupMembership.variation_group.in_(selected_groups)) + .all()) + assignment_ids = [m.assignment_id for m in memberships] + self.assign_variation_to_user(user_id, assignment_ids) + return assignment_ids + def find_all_linked_resources(self) -> dict[str, list[Base]]: # Get any assignments that are forked from this one forked = AssignmentGroup.query.filter_by(forked_id=self.id).all() diff --git a/models/assignment_group_membership.py b/models/assignment_group_membership.py index 3c413035a..61b10794a 100644 --- a/models/assignment_group_membership.py +++ b/models/assignment_group_membership.py @@ -21,6 +21,10 @@ class AssignmentGroupMembership(EnhancedBase): #: Logic for grading and visibility policy: Mapped[Optional[str]] = mapped_column(String(255), default="{}") + #: When None, this assignment is required for all students. + #: When set to a positive integer, this assignment belongs to variation pool N + #: and is only shown to students who have been assigned that variation. + variation_group: Mapped[Optional[int]] = mapped_column(Integer(), nullable=True, default=None) assignment_group: Mapped[list["AssignmentGroup"]] = db.relationship(back_populates="memberships") assignment: Mapped[list["Assignment"]] = db.relationship(back_populates="memberships") diff --git a/models/assignment_group_variation.py b/models/assignment_group_variation.py new file mode 100644 index 000000000..c0da271b6 --- /dev/null +++ b/models/assignment_group_variation.py @@ -0,0 +1,96 @@ +import random +from typing import Optional, List, TYPE_CHECKING + +from sqlalchemy.orm import Mapped, mapped_column +from sqlalchemy import Integer, ForeignKey, Index, UniqueConstraint + +from models.generics.models import db +from models.generics.base import EnhancedBase +import models + +if TYPE_CHECKING: + from models import * + + +class AssignmentGroupVariation(EnhancedBase): + """ + Tracks which assignments from a variation pool have been assigned to a specific student + within an assignment group. + + When an AssignmentGroup has variation_count > 0, some AssignmentGroupMemberships will + have a non-null variation_group integer indicating they belong to a pool. This model + records the exact (user, assignment) pairs that result from assigning a student their + specific variation, so they only see the subset of assignments chosen for them. + """ + __tablename__ = 'assignment_group_variation' + + user_id: Mapped[int] = mapped_column(Integer(), ForeignKey('user.id')) + assignment_group_id: Mapped[int] = mapped_column(Integer(), ForeignKey('assignment_group.id')) + assignment_id: Mapped[int] = mapped_column(Integer(), ForeignKey('assignment.id')) + + user: Mapped["User"] = db.relationship(back_populates="assignment_group_variations") + assignment_group: Mapped["AssignmentGroup"] = db.relationship(back_populates="variations") + assignment: Mapped["Assignment"] = db.relationship(back_populates="group_variations") + + __table_args__ = ( + UniqueConstraint('user_id', 'assignment_group_id', 'assignment_id', + name='uq_agv_user_group_assignment'), + Index('agv_user_group_index', 'user_id', 'assignment_group_id'), + Index('agv_group_index', 'assignment_group_id'), + ) + + def __str__(self): + return ''.format( + self.user_id, self.assignment_group_id, self.assignment_id) + + def encode_json(self): + return { + '_schema_version': 1, + 'user_id': self.user_id, + 'assignment_group_id': self.assignment_group_id, + 'assignment_id': self.assignment_id, + 'id': self.id, + } + + @staticmethod + def get_for_user(user_id: int, assignment_group_id: int) -> 'List[AssignmentGroupVariation]': + """Get all variation assignments for a user in a specific group.""" + return (AssignmentGroupVariation.query + .filter_by(user_id=user_id, assignment_group_id=assignment_group_id) + .all()) + + @staticmethod + def get_assignment_ids_for_user(user_id: int, assignment_group_id: int) -> 'List[int]': + """Get the assignment IDs that a user has been assigned in a group.""" + rows = (AssignmentGroupVariation.query + .filter_by(user_id=user_id, assignment_group_id=assignment_group_id) + .all()) + return [row.assignment_id for row in rows] + + @staticmethod + def clear_for_user(user_id: int, assignment_group_id: int): + """Remove all variation assignments for a user in a group.""" + (AssignmentGroupVariation.query + .filter_by(user_id=user_id, assignment_group_id=assignment_group_id) + .delete()) + db.session.commit() + + @staticmethod + def assign_to_user(user_id: int, assignment_group_id: int, + assignment_ids: 'List[int]'): + """ + Assign specific variation assignments to a user. + Replaces any existing variation assignments for this user/group. + """ + AssignmentGroupVariation.clear_for_user(user_id, assignment_group_id) + for assignment_id in assignment_ids: + variation = AssignmentGroupVariation( + user_id=user_id, + assignment_group_id=assignment_group_id, + assignment_id=assignment_id, + ) + db.session.add(variation) + db.session.commit() + + def find_all_linked_resources(self) -> dict: + return {} diff --git a/models/user.py b/models/user.py index 4925663e4..6d7c82f3b 100644 --- a/models/user.py +++ b/models/user.py @@ -50,6 +50,7 @@ class User(Base, UserMixin): authentications: Mapped[list["Authentication"]] = db.relationship(back_populates="user") assignments: Mapped[list["Assignment"]] = db.relationship(back_populates="owner") assignment_groups: Mapped[list["AssignmentGroup"]] = db.relationship(back_populates="owner") + assignment_group_variations: Mapped[list["AssignmentGroupVariation"]] = db.relationship(back_populates="user") courses: Mapped[list["Course"]] = db.relationship(back_populates="owner") tags: Mapped[list["AssignmentTag"]] = db.relationship(back_populates="owner") access_logs: Mapped[list["AccessLog"]] = db.relationship(back_populates="subject") diff --git a/tests/controllers/test_variations.py b/tests/controllers/test_variations.py new file mode 100644 index 000000000..e8ffc92c6 --- /dev/null +++ b/tests/controllers/test_variations.py @@ -0,0 +1,414 @@ +""" +Tests for question variations support in assignment groups. + +Covers: + - AssignmentGroupMembership.variation_group field + - AssignmentGroup.variation_count field + - AssignmentGroup.get_assignments(user_id) filtering + - AssignmentGroup.assign_variation_to_user / assign_random_variation helpers + - GET /assignment_group/get_variations + - POST /assignment_group/assign_variation + - POST /assignment_group/assign_variations_random + - GET /assignment_group/edit_variation_settings + - POST /assignment_group/edit_variation_settings +""" +import json +import pytest + +from models import db +from models.assignment_group import AssignmentGroup +from models.assignment_group_membership import AssignmentGroupMembership +from models.assignment_group_variation import AssignmentGroupVariation +from tests.factory.factories import ( + AssignmentGroupFactory, + AssignmentFactory, + CourseFactory, + UserFactory, +) +from models.role import Role +from models.enums.roles import UserRoles + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _add_membership(group, assignment, position=1, variation_group=None): + membership = AssignmentGroupMembership( + assignment_group_id=group.id, + assignment_id=assignment.id, + position=position, + variation_group=variation_group, + ) + db.session.add(membership) + db.session.commit() + return membership + + +# --------------------------------------------------------------------------- +# Model-level tests +# --------------------------------------------------------------------------- + +class TestVariationGroupMembership: + """AssignmentGroupMembership.variation_group field.""" + + def test_default_variation_group_is_none(self, app): + group = AssignmentGroupFactory.create_assignment_group() + assignment = AssignmentFactory.create_assignment(course=group.course) + membership = _add_membership(group, assignment) + assert membership.variation_group is None + + def test_variation_group_can_be_set(self, app): + group = AssignmentGroupFactory.create_assignment_group() + assignment = AssignmentFactory.create_assignment(course=group.course) + membership = _add_membership(group, assignment, variation_group=1) + assert membership.variation_group == 1 + + +class TestAssignmentGroupVariationCount: + """AssignmentGroup.variation_count field.""" + + def test_default_variation_count_is_zero(self, app): + group = AssignmentGroupFactory.create_assignment_group() + assert group.variation_count == 0 + + def test_variation_count_can_be_set(self, app): + group = AssignmentGroupFactory.create_assignment_group(variation_count=2) + assert group.variation_count == 2 + + +class TestGetAssignmentsVariations: + """AssignmentGroup.get_assignments() with user_id filtering.""" + + def _setup_group_with_variations(self): + """Create a group with 1 required + 2 variation pools (each with 1 assignment).""" + course = CourseFactory.create_course() + group = AssignmentGroupFactory.create_assignment_group(course=course, + variation_count=1) + required_a = AssignmentFactory.create_assignment(course=course, name="Required A") + pool1_a = AssignmentFactory.create_assignment(course=course, name="Pool1 A") + pool2_a = AssignmentFactory.create_assignment(course=course, name="Pool2 A") + _add_membership(group, required_a, position=1, variation_group=None) + _add_membership(group, pool1_a, position=2, variation_group=1) + _add_membership(group, pool2_a, position=3, variation_group=2) + return group, required_a, pool1_a, pool2_a + + def test_get_assignments_no_user_returns_all(self, app): + group, required_a, pool1_a, pool2_a = self._setup_group_with_variations() + all_assignments = group.get_assignments() + ids = {a.id for a in all_assignments} + assert {required_a.id, pool1_a.id, pool2_a.id} == ids + + def test_get_assignments_with_user_no_variation_assigned(self, app): + """User with no variation assignment sees only required assignments.""" + group, required_a, pool1_a, pool2_a = self._setup_group_with_variations() + user = UserFactory.create_student(course=group.course) + assignments = group.get_assignments(user_id=user.id) + ids = {a.id for a in assignments} + assert ids == {required_a.id} + + def test_get_assignments_with_user_variation_assigned(self, app): + """User with a variation assignment sees required + their variation.""" + group, required_a, pool1_a, pool2_a = self._setup_group_with_variations() + user = UserFactory.create_student(course=group.course) + group.assign_variation_to_user(user.id, [pool1_a.id]) + assignments = group.get_assignments(user_id=user.id) + ids = {a.id for a in assignments} + assert ids == {required_a.id, pool1_a.id} + + def test_get_assignments_variation_count_zero_ignores_user(self, app): + """When variation_count == 0, user_id is ignored and all are returned.""" + course = CourseFactory.create_course() + group = AssignmentGroupFactory.create_assignment_group(course=course, + variation_count=0) + a1 = AssignmentFactory.create_assignment(course=course) + a2 = AssignmentFactory.create_assignment(course=course) + _add_membership(group, a1, variation_group=1) + _add_membership(group, a2, variation_group=2) + user = UserFactory.create_student(course=course) + assignments = group.get_assignments(user_id=user.id) + ids = {a.id for a in assignments} + assert ids == {a1.id, a2.id} + + def test_two_users_get_different_variations(self, app): + """Two users assigned to different variation groups see different assignments.""" + course = CourseFactory.create_course() + group = AssignmentGroupFactory.create_assignment_group(course=course, + variation_count=1) + required = AssignmentFactory.create_assignment(course=course, name="Required") + pool1 = AssignmentFactory.create_assignment(course=course, name="Pool1") + pool2 = AssignmentFactory.create_assignment(course=course, name="Pool2") + _add_membership(group, required, position=1, variation_group=None) + _add_membership(group, pool1, position=2, variation_group=1) + _add_membership(group, pool2, position=3, variation_group=2) + + user1 = UserFactory.create_student(course=course) + user2 = UserFactory.create_student(course=course) + group.assign_variation_to_user(user1.id, [pool1.id]) + group.assign_variation_to_user(user2.id, [pool2.id]) + + ids1 = {a.id for a in group.get_assignments(user_id=user1.id)} + ids2 = {a.id for a in group.get_assignments(user_id=user2.id)} + assert ids1 == {required.id, pool1.id} + assert ids2 == {required.id, pool2.id} + + +class TestAssignRandomVariation: + """AssignmentGroup.assign_random_variation()""" + + def test_assigns_correct_number_of_groups(self, app): + course = CourseFactory.create_course() + group = AssignmentGroupFactory.create_assignment_group(course=course, + variation_count=1) + a1 = AssignmentFactory.create_assignment(course=course) + a2 = AssignmentFactory.create_assignment(course=course) + _add_membership(group, a1, variation_group=1) + _add_membership(group, a2, variation_group=2) + user = UserFactory.create_student(course=course) + assigned_ids = group.assign_random_variation(user.id) + # variation_count=1 means 1 variation group selected → 1 assignment from pool + assert len(assigned_ids) == 1 + assert assigned_ids[0] in {a1.id, a2.id} + + def test_get_variation_groups_returns_distinct_values(self, app): + course = CourseFactory.create_course() + group = AssignmentGroupFactory.create_assignment_group(course=course) + a1 = AssignmentFactory.create_assignment(course=course) + a2 = AssignmentFactory.create_assignment(course=course) + a3 = AssignmentFactory.create_assignment(course=course) + _add_membership(group, a1, variation_group=None) + _add_membership(group, a2, variation_group=1) + _add_membership(group, a3, variation_group=2) + assert group.get_variation_groups() == [1, 2] + + def test_clear_and_reassign(self, app): + course = CourseFactory.create_course() + group = AssignmentGroupFactory.create_assignment_group(course=course, + variation_count=1) + a1 = AssignmentFactory.create_assignment(course=course) + a2 = AssignmentFactory.create_assignment(course=course) + _add_membership(group, a1, variation_group=1) + _add_membership(group, a2, variation_group=2) + user = UserFactory.create_student(course=course) + group.assign_variation_to_user(user.id, [a1.id]) + assert AssignmentGroupVariation.get_assignment_ids_for_user(user.id, group.id) == [a1.id] + # Reassign to different set + group.assign_variation_to_user(user.id, [a2.id]) + assert AssignmentGroupVariation.get_assignment_ids_for_user(user.id, group.id) == [a2.id] + + +# --------------------------------------------------------------------------- +# Endpoint tests +# --------------------------------------------------------------------------- + +class TestGetVariationsEndpoint: + """GET /assignment_group/get_variations""" + + def test_get_variations_anonymous_blocked(self, client, test_data): + response = client.get('/assignment_group/get_variations', + query_string={'assignment_group_id': 1}) + assert response.json['success'] is False + + def test_get_variations_student_own_view(self, client, test_data, act_as): + """Students can view their own variation assignments.""" + act_as(test_data.user("lulu@blockpy.com")) + response = client.get('/assignment_group/get_variations', + query_string={'assignment_group_id': 1}) + assert response.status_code == 200 + data = response.get_json() + assert data['success'] is True + assert 'variation_count' in data + assert 'assigned_assignment_ids' in data + + def test_get_variations_student_blocked_other_user(self, client, test_data, act_as): + """Students cannot view another student's variation assignments.""" + act_as(test_data.user("lulu@blockpy.com")) + # user_id 10 is Ada – requesting another user's data as a student should be blocked + response = client.get('/assignment_group/get_variations', + query_string={'assignment_group_id': 1, 'user_id': 10}) + assert response.json['success'] is False + + def test_get_variations_instructor_can_view_student(self, client, test_data, act_as): + """Instructors can view any student's variation assignments.""" + act_as(test_data.user("ada@blockpy.com")) + response = client.get('/assignment_group/get_variations', + query_string={'assignment_group_id': 1, 'user_id': 100}) + assert response.status_code == 200 + data = response.get_json() + assert data['success'] is True + + def test_get_variations_includes_variation_groups(self, client, test_data, act_as): + act_as(test_data.user("ada@blockpy.com")) + response = client.get('/assignment_group/get_variations', + query_string={'assignment_group_id': 1}) + data = response.get_json() + assert 'variation_groups' in data + assert isinstance(data['variation_groups'], list) + + +class TestAssignVariationEndpoint: + """POST /assignment_group/assign_variation""" + + def test_assign_variation_anonymous_blocked(self, client, test_data): + response = client.post('/assignment_group/assign_variation', data={ + 'assignment_group_id': 1, + 'user_id': 100, + 'assignment_ids': json.dumps([100]), + }) + assert response.json['success'] is False + + def test_assign_variation_student_blocked(self, client, test_data, act_as): + act_as(test_data.user("lulu@blockpy.com")) + response = client.post('/assignment_group/assign_variation', data={ + 'assignment_group_id': 1, + 'user_id': 100, + 'assignment_ids': json.dumps([100]), + }) + assert response.json['success'] is False + + def test_assign_variation_instructor_success(self, client, test_data, act_as): + act_as(test_data.user("ada@blockpy.com")) + response = client.post('/assignment_group/assign_variation', data={ + 'assignment_group_id': 1, + 'user_id': 100, + 'assignment_ids': json.dumps([100, 101]), + }) + assert response.status_code == 200 + data = response.get_json() + assert data['success'] is True + assert sorted(data['assigned_assignment_ids']) == [100, 101] + + def test_assign_variation_invalid_ids(self, client, test_data, act_as): + """Assignment IDs that don't belong to this group should be rejected.""" + act_as(test_data.user("ada@blockpy.com")) + response = client.post('/assignment_group/assign_variation', data={ + 'assignment_group_id': 1, + 'user_id': 100, + 'assignment_ids': json.dumps([9999]), # Not in group 1 + }) + assert response.json['success'] is False + + def test_assign_variation_invalid_json(self, client, test_data, act_as): + act_as(test_data.user("ada@blockpy.com")) + response = client.post('/assignment_group/assign_variation', data={ + 'assignment_group_id': 1, + 'user_id': 100, + 'assignment_ids': 'not-valid-json', + }) + assert response.json['success'] is False + + +class TestAssignVariationsRandomEndpoint: + """POST /assignment_group/assign_variations_random""" + + def test_anonymous_blocked(self, client, test_data): + response = client.post('/assignment_group/assign_variations_random', + data={'assignment_group_id': 1}) + assert response.json['success'] is False + + def test_student_blocked(self, client, test_data, act_as): + act_as(test_data.user("lulu@blockpy.com")) + response = client.post('/assignment_group/assign_variations_random', + data={'assignment_group_id': 1}) + assert response.json['success'] is False + + def test_variation_count_zero_fails(self, client, test_data, act_as): + """Groups with variation_count=0 cannot use random assignment.""" + act_as(test_data.user("ada@blockpy.com")) + # Group 1 has variation_count=0 by default + response = client.post('/assignment_group/assign_variations_random', + data={'assignment_group_id': 1}) + data = response.get_json() + assert data['success'] is False + assert 'variation_count' in data['message'] + + def test_instructor_can_assign_random(self, client, test_data, act_as): + """When variation groups are set up, instructor can randomly assign.""" + act_as(test_data.user("ada@blockpy.com")) + # First configure group 1 to have variation_count=1 and add variation groups + response = client.post('/assignment_group/edit_variation_settings', data={ + 'assignment_group_id': 1, + 'variation_count': 1, + # membership IDs 1-3 belong to group 1 (assignments 100, 101, 102) + 'variation_group[1]': '1', + 'variation_group[2]': '2', + 'variation_group[3]': '', + }) + assert response.get_json()['success'] is True + + # Now assign randomly; should work since variation groups exist + response = client.post('/assignment_group/assign_variations_random', data={ + 'assignment_group_id': 1, + 'course_id': 6, + }) + data = response.get_json() + assert data['success'] is True + assert 'assigned_count' in data + + +class TestEditVariationSettingsEndpoint: + """GET/POST /assignment_group/edit_variation_settings""" + + def test_get_variation_settings_anonymous_blocked(self, client, test_data): + response = client.get('/assignment_group/edit_variation_settings', + query_string={'assignment_group_id': 1}) + assert response.json['success'] is False + + def test_get_variation_settings_instructor(self, client, test_data, act_as): + act_as(test_data.user("ada@blockpy.com")) + response = client.get('/assignment_group/edit_variation_settings', + query_string={'assignment_group_id': 1}) + assert response.status_code == 200 + data = response.get_json() + assert data['success'] is True + assert 'variation_count' in data + assert 'memberships' in data + + def test_post_variation_settings_updates_count(self, client, test_data, act_as): + act_as(test_data.user("ada@blockpy.com")) + response = client.post('/assignment_group/edit_variation_settings', data={ + 'assignment_group_id': 2, + 'variation_count': 2, + }) + assert response.status_code == 200 + data = response.get_json() + assert data['success'] is True + assert data['variation_count'] == 2 + + def test_post_variation_settings_invalid_count(self, client, test_data, act_as): + act_as(test_data.user("ada@blockpy.com")) + response = client.post('/assignment_group/edit_variation_settings', data={ + 'assignment_group_id': 1, + 'variation_count': -1, + }) + assert response.json['success'] is False + + def test_post_variation_settings_student_blocked(self, client, test_data, act_as): + act_as(test_data.user("lulu@blockpy.com")) + response = client.post('/assignment_group/edit_variation_settings', data={ + 'assignment_group_id': 1, + 'variation_count': 1, + }) + assert response.json['success'] is False + + def test_post_updates_membership_variation_group(self, client, test_data, act_as): + act_as(test_data.user("ada@blockpy.com")) + # group 1 memberships: 1 (assignment 100), 2 (assignment 101), 3 (assignment 102) + response = client.post('/assignment_group/edit_variation_settings', data={ + 'assignment_group_id': 1, + 'variation_count': 1, + 'variation_group[1]': '1', + 'variation_group[2]': '2', + 'variation_group[3]': '', # empty → NULL + }) + data = response.get_json() + assert data['success'] is True + # Re-GET to verify persistence + response2 = client.get('/assignment_group/edit_variation_settings', + query_string={'assignment_group_id': 1}) + data2 = response2.get_json() + assert data2['variation_count'] == 1 + mem_map = {m['id']: m['variation_group'] for m in data2['memberships']} + assert mem_map[1] == 1 + assert mem_map[2] == 2 + assert mem_map[3] is None