coldcall_lti/tests/test_calls.py

import datetime
import random
from collections import Counter

from coldcall_lti import calls, models

DAY = datetime.date(2026, 10, 1)


def make_course(db, n=4, mode=models.SELECTION_WEIGHTED):
    course = models.Course(lti_context_id="ctx-1", selection_mode=mode)
    db.add(course)
    students = []
    for i in range(n):
        s = models.Student(canvas_user_id=f"u{i}", name=f"Student {i}")
        db.add(s)
        students.append(s)
    db.flush()
    for s in students:
        db.add(models.Enrollment(course_id=course.id, student_id=s.id))
    db.flush()
    return course, students


def test_pick_next_honors_optouts(db_session):
    course, students = make_course(db_session, n=2)
    db_session.add(
        models.OptOut(
            course_id=course.id, student_id=students[0].id, date=DAY
        )
    )
    db_session.flush()
    rng = random.Random(1)
    for _ in range(20):
        assert calls.pick_next_student(db_session, course, DAY, rng).id == students[1].id


def test_pick_next_weighted_downweights(db_session):
    course, students = make_course(db_session)
    heavy = students[0]
    for _ in range(4):
        db_session.add(
            models.Call(
                course_id=course.id,
                student_id=heavy.id,
                session_date=DAY - datetime.timedelta(days=7),
                status=models.STATUS_ANSWERED,
            )
        )
    db_session.flush()

    rng = random.Random(42)
    picks = Counter(
        calls.pick_next_student(db_session, course, DAY, rng).id
        for _ in range(400)
    )
    assert picks[heavy.id] < min(picks[s.id] for s in students[1:]) / 2


def test_cycle_mode_calls_everyone_before_repeating(db_session):
    course, students = make_course(db_session, mode=models.SELECTION_CYCLE)
    rng = random.Random(7)
    called = []
    for _ in range(len(students)):
        call = calls.create_live_call(db_session, course, DAY, rng)
        call.status = models.STATUS_ANSWERED
        db_session.flush()
        called.append(call.student_id)
    assert sorted(called) == sorted(s.id for s in students)

    # The next draw starts a fresh pass rather than failing.
    assert calls.create_live_call(db_session, course, DAY, rng) is not None


def test_cycle_mode_recalls_skipped_students(db_session):
    course, students = make_course(db_session, n=2, mode=models.SELECTION_CYCLE)
    rng = random.Random(3)
    call = calls.create_live_call(db_session, course, DAY, rng)
    call.status = models.STATUS_SKIPPED
    db_session.flush()
    # A skipped call doesn't count as having been called this pass.
    pool_ids = {
        calls.pick_next_student(db_session, course, DAY, rng).id
        for _ in range(20)
    }
    assert call.student_id in pool_ids


def test_generate_day_list_cycle_covers_roster(db_session):
    course, students = make_course(db_session, mode=models.SELECTION_CYCLE)
    generated = calls.generate_day_list(db_session, course, DAY, rng=random.Random(1))
    assert sorted(c.student_id for c in generated) == sorted(s.id for s in students)
    assert all(c.status == models.STATUS_PENDING for c in generated)


def test_generate_day_list_cycle_batches_continue_across_days(db_session):
    course, students = make_course(db_session, mode=models.SELECTION_CYCLE)
    rng = random.Random(1)
    day2 = DAY + datetime.timedelta(days=2)

    first = calls.generate_day_list(db_session, course, DAY, n=3, rng=rng)
    assert len(first) == 3
    # Next day's batch continues the pass: the remaining student only.
    second = calls.generate_day_list(db_session, course, day2, n=3, rng=rng)
    assert len(second) == 1
    assert sorted(c.student_id for c in first + second) == sorted(
        s.id for s in students
    )
    # Pass complete: the next batch starts a fresh pass.
    third = calls.generate_day_list(db_session, course, day2, n=3, rng=rng)
    assert len(third) == 3


def test_generate_day_list_weighted_length_and_optouts(db_session):
    course, students = make_course(db_session)
    db_session.add(
        models.OptOut(
            course_id=course.id, student_id=students[0].id, date=DAY
        )
    )
    db_session.flush()
    generated = calls.generate_day_list(
        db_session, course, DAY, n=10, rng=random.Random(1)
    )
    assert len(generated) == 10
    assert students[0].id not in {c.student_id for c in generated}


def test_clear_pending_keeps_resolved(db_session):
    course, students = make_course(db_session)
    generated = calls.generate_day_list(
        db_session, course, DAY, n=5, rng=random.Random(1)
    )
    generated[0].status = models.STATUS_ANSWERED
    db_session.flush()

    removed = calls.clear_pending(db_session, course.id, DAY)
    assert removed == 4
    remaining = calls.calls_for_day(db_session, course.id, DAY)
    assert len(remaining) == 1
    assert remaining[0].status == models.STATUS_ANSWERED


def test_unresolved_call_is_oldest_pending(db_session):
    course, students = make_course(db_session)
    generated = calls.generate_day_list(
        db_session, course, DAY, n=3, rng=random.Random(1)
    )
    assert calls.unresolved_call(db_session, course.id, DAY).id == generated[0].id