Phase 2: LTI plumbing, roster sync, and fake-launch dev mode

OIDC login, launch, and JWKS endpoints built on pylti1p3next's Flask
adapter, with launch-claim processing split into a testable module.
Instructor launches refresh the roster through NRPS; the sync code is
source-agnostic and also drives the dev-mode fake roster. Dev mode
(COLDCALL_DEV_MODE=1) provides fake instructor and student personas
that set up the same session state as a real launch, so the rest of
the app can be developed before a Canvas Developer Key exists.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

README.md

@@ -61,5 +61,43 @@ Create the database and run the tests:
 ```
 
 Configuration is by environment variable: `COLDCALL_DATABASE_URL` (any
-SQLAlchemy URL; defaults to an SQLite file under `instance/`) and
+SQLAlchemy URL; defaults to an SQLite file under `instance/`),
-`COLDCALL_SECRET_KEY` for Flask sessions.
+`COLDCALL_SECRET_KEY` for Flask sessions, `COLDCALL_LTI_CONFIG` (path
+to the LTI platform configuration), and `COLDCALL_DEV_MODE=1` to enable
+the fake-launch pages.
+
+## Connecting to Canvas
+
+The tool speaks LTI 1.3, which requires a Developer Key created by a
+Canvas account admin. The key points Canvas at three endpoints here:
+`/lti/login` (OIDC initiation), `/lti/launch` (the launch target), and
+`/lti/jwks` (this tool's public keys). The platform side is described
+in a JSON file — copy `lti_config.example.json` to
+`instance/lti_config.json` and fill in the client id and deployment id
+from the Developer Key. Generate the tool's keypair alongside it:
+
+```
+openssl genrsa -out instance/private.key 4096
+openssl rsa -in instance/private.key -pubout -out instance/public.key
+```
+
+On each instructor launch the tool refreshes the course roster from
+Canvas through the Names and Role Provisioning Service, so enrollment
+changes appear without any manual step.
+
+## Developing without Canvas
+
+Because a Developer Key takes institutional approval to get, the app
+has a fake-launch mode for local development:
+
+```
+COLDCALL_DEV_MODE=1 .venv/bin/flask --app coldcall_lti run --debug
+```
+
+Then open http://localhost:5000/dev and launch as the fake instructor
+or any of the fake students. This sets up exactly the session state a
+real launch would, and the fake roster flows through the same sync code
+as real NRPS data, so everything past the launch behaves identically.
+Dev mode also relaxes the cookie settings that Canvas's iframe
+embedding requires in production (SameSite=None; Secure), which would
+otherwise break plain-http localhost use.

coldcall_lti/__init__.py

@@ -3,7 +3,7 @@ import os
 from flask import Flask
 
 from .config import Config
-from .db import make_engine, make_session_factory
+from .db import close_db, make_engine, make_session_factory
 
 
 def create_app(config=Config):
@@ -14,6 +14,24 @@ def create_app(config=Config):
     engine = make_engine(app.config["DATABASE_URL"])
     app.extensions["db_engine"] = engine
     app.extensions["db_session_factory"] = make_session_factory(engine)
+    app.teardown_appcontext(close_db)
+
+    # Canvas launches the tool in an iframe, so the session cookie must
+    # be usable in a third-party context. Dev mode runs over plain
+    # http://localhost, where Secure cookies would be dropped.
+    if not app.config["DEV_MODE"]:
+        app.config["SESSION_COOKIE_SAMESITE"] = "None"
+        app.config["SESSION_COOKIE_SECURE"] = True
+
+    from . import lti, views
+
+    app.register_blueprint(lti.bp)
+    app.register_blueprint(views.bp)
+
+    if app.config["DEV_MODE"]:
+        from . import dev
+
+        app.register_blueprint(dev.bp)
 
     @app.get("/healthz")
     def healthz():

coldcall_lti/config.py

@@ -6,3 +6,7 @@ class Config:
     DATABASE_URL = os.environ.get(
         "COLDCALL_DATABASE_URL", "sqlite:///instance/coldcall.sqlite3"
     )
+    LTI_CONFIG_PATH = os.environ.get(
+        "COLDCALL_LTI_CONFIG", "instance/lti_config.json"
+    )
+    DEV_MODE = os.environ.get("COLDCALL_DEV_MODE") == "1"

coldcall_lti/db.py

@@ -1,3 +1,4 @@
+from flask import current_app, g
 from sqlalchemy import create_engine
 from sqlalchemy.orm import DeclarativeBase, sessionmaker
 
@@ -12,3 +13,15 @@ def make_engine(url, echo=False):
 
 def make_session_factory(engine):
     return sessionmaker(bind=engine, expire_on_commit=False)
+
+
+def get_db():
+    if "db_session" not in g:
+        g.db_session = current_app.extensions["db_session_factory"]()
+    return g.db_session
+
+
+def close_db(exc=None):
+    session = g.pop("db_session", None)
+    if session is not None:
+        session.close()

coldcall_lti/dev.py

@@ -0,0 +1,97 @@
+"""Fake-launch mode for development without Canvas.
+
+When COLDCALL_DEV_MODE=1, /dev offers a page of personas (one
+instructor, a small roster of students) in a fake course. Launching as
+one sets up the same Flask session state a real LTI launch would, so
+every other part of the app behaves identically. The fake roster goes
+through the same sync_roster path as real NRPS data.
+"""
+
+from flask import Blueprint, abort, redirect, render_template, session, url_for
+from sqlalchemy import select
+
+from .db import get_db
+from .launch import ROLE_INSTRUCTOR, ROLE_LEARNER
+from .models import Course, Student
+from .roster import sync_roster
+
+bp = Blueprint("dev", __name__, url_prefix="/dev")
+
+DEV_CONTEXT_ID = "dev-course-1"
+
+DEV_MEMBERS = [
+    {
+        "user_id": "dev-instructor",
+        "name": "Ida Instructor",
+        "given_name": "Ida",
+        "family_name": "Instructor",
+        "email": "instructor@example.edu",
+        "roles": [ROLE_INSTRUCTOR],
+        "status": "Active",
+    }
+] + [
+    {
+        "user_id": f"dev-student-{i}",
+        "name": f"{given} {family}",
+        "given_name": given,
+        "family_name": family,
+        "email": f"{given.lower()}@example.edu",
+        "roles": [ROLE_LEARNER],
+        "status": "Active",
+    }
+    for i, (given, family) in enumerate(
+        [
+            ("Ada", "Lovelace"),
+            ("Grace", "Hopper"),
+            ("Alan", "Turing"),
+            ("Annie", "Easley"),
+            ("Edsger", "Dijkstra"),
+            ("Katherine", "Johnson"),
+            ("Donald", "Knuth"),
+            ("Radia", "Perlman"),
+        ],
+        start=1,
+    )
+]
+
+
+def _ensure_dev_course(db):
+    course = db.execute(
+        select(Course).where(Course.lti_context_id == DEV_CONTEXT_ID)
+    ).scalar_one_or_none()
+    if course is None:
+        course = Course(lti_context_id=DEV_CONTEXT_ID, title="Development Course")
+        db.add(course)
+        db.flush()
+    sync_roster(db, course, DEV_MEMBERS)
+    db.commit()
+    return course
+
+
+@bp.get("/")
+def persona_list():
+    db = get_db()
+    _ensure_dev_course(db)
+    return render_template("dev_launch.html", members=DEV_MEMBERS)
+
+
+@bp.post("/launch/<user_id>")
+def fake_launch(user_id):
+    member = next((m for m in DEV_MEMBERS if m["user_id"] == user_id), None)
+    if member is None:
+        abort(404)
+
+    db = get_db()
+    course = _ensure_dev_course(db)
+    student = db.execute(
+        select(Student).where(Student.canvas_user_id == user_id)
+    ).scalar_one()
+
+    is_instructor = ROLE_INSTRUCTOR in member["roles"]
+    session["course_id"] = course.id
+    session["user_db_id"] = student.id
+    session["is_instructor"] = is_instructor
+
+    if is_instructor:
+        return redirect(url_for("views.instructor_home"))
+    return redirect(url_for("views.student_home"))

coldcall_lti/launch.py

@@ -0,0 +1,95 @@
+"""Turn validated LTI launch claims into database rows.
+
+Kept separate from the Flask/pylti1p3 plumbing so it can be tested with
+plain dicts of claims and reused by the dev-mode fake launcher.
+"""
+
+from dataclasses import dataclass
+
+from sqlalchemy import select
+
+from .models import Course, Enrollment, Student
+
+CLAIM_CONTEXT = "https://purl.imsglobal.org/spec/lti/claim/context"
+CLAIM_ROLES = "https://purl.imsglobal.org/spec/lti/claim/roles"
+CLAIM_DEPLOYMENT = "https://purl.imsglobal.org/spec/lti/claim/deployment_id"
+
+ROLE_INSTRUCTOR = "http://purl.imsglobal.org/vocab/lis/v2/membership#Instructor"
+ROLE_LEARNER = "http://purl.imsglobal.org/vocab/lis/v2/membership#Learner"
+
+
+@dataclass
+class LaunchInfo:
+    context_id: str
+    deployment_id: str | None
+    course_title: str | None
+    user_id: str
+    name: str | None
+    email: str | None
+    picture: str | None
+    is_instructor: bool
+
+
+def roles_include_instructor(roles):
+    return any("membership#Instructor" in r for r in roles)
+
+
+def extract_launch_info(launch_data):
+    context = launch_data.get(CLAIM_CONTEXT, {})
+    return LaunchInfo(
+        context_id=context["id"],
+        deployment_id=launch_data.get(CLAIM_DEPLOYMENT),
+        course_title=context.get("title"),
+        user_id=launch_data["sub"],
+        name=launch_data.get("name"),
+        email=launch_data.get("email"),
+        picture=launch_data.get("picture"),
+        is_instructor=roles_include_instructor(launch_data.get(CLAIM_ROLES, [])),
+    )
+
+
+def upsert_launch(db, info):
+    """Create or refresh the course, user, and enrollment for a launch."""
+    course = db.execute(
+        select(Course).where(Course.lti_context_id == info.context_id)
+    ).scalar_one_or_none()
+    if course is None:
+        course = Course(lti_context_id=info.context_id)
+        db.add(course)
+    if info.course_title:
+        course.title = info.course_title
+    if info.deployment_id:
+        course.lti_deployment_id = info.deployment_id
+
+    student = db.execute(
+        select(Student).where(Student.canvas_user_id == info.user_id)
+    ).scalar_one_or_none()
+    if student is None:
+        student = Student(canvas_user_id=info.user_id)
+        db.add(student)
+    if info.name:
+        student.name = info.name
+    if info.email:
+        student.email = info.email
+    if info.picture:
+        student.avatar_url = info.picture
+    db.flush()
+
+    enrollment = db.execute(
+        select(Enrollment).where(
+            Enrollment.course_id == course.id,
+            Enrollment.student_id == student.id,
+        )
+    ).scalar_one_or_none()
+    role = "instructor" if info.is_instructor else "student"
+    if enrollment is None:
+        enrollment = Enrollment(
+            course_id=course.id, student_id=student.id, role=role
+        )
+        db.add(enrollment)
+    else:
+        enrollment.role = role
+        enrollment.active = True
+    db.flush()
+
+    return course, student

coldcall_lti/lti.py

@@ -0,0 +1,71 @@
+from flask import Blueprint, abort, current_app, jsonify, redirect, session, url_for
+from pylti1p3.contrib.flask import (
+    FlaskCookieService,
+    FlaskMessageLaunch,
+    FlaskOIDCLogin,
+    FlaskRequest,
+    FlaskSessionService,
+)
+from pylti1p3.tool_config import ToolConfJsonFile
+
+from .db import get_db
+from .launch import extract_launch_info, upsert_launch
+from .roster import sync_roster
+
+bp = Blueprint("lti", __name__, url_prefix="/lti")
+
+
+def get_tool_conf():
+    return ToolConfJsonFile(current_app.config["LTI_CONFIG_PATH"])
+
+
+def _services(flask_request):
+    return {
+        "session_service": FlaskSessionService(flask_request),
+        "cookie_service": FlaskCookieService(flask_request),
+    }
+
+
+@bp.route("/login", methods=["GET", "POST"])
+def login():
+    flask_request = FlaskRequest()
+    target_link_uri = flask_request.get_param("target_link_uri")
+    if not target_link_uri:
+        abort(400, "Missing target_link_uri")
+    oidc_login = FlaskOIDCLogin(
+        flask_request, get_tool_conf(), **_services(flask_request)
+    )
+    return oidc_login.enable_check_cookies().redirect(target_link_uri)
+
+
+@bp.route("/launch", methods=["POST"])
+def launch():
+    flask_request = FlaskRequest()
+    message_launch = FlaskMessageLaunch(
+        flask_request, get_tool_conf(), **_services(flask_request)
+    )
+    launch_data = message_launch.get_launch_data()
+    info = extract_launch_info(launch_data)
+
+    db = get_db()
+    course, student = upsert_launch(db, info)
+
+    # Instructor launches refresh the roster, so adds and drops are
+    # picked up before every class without a separate sync step.
+    if info.is_instructor and message_launch.has_nrps():
+        members = message_launch.get_nrps().get_members()
+        sync_roster(db, course, members)
+    db.commit()
+
+    session["course_id"] = course.id
+    session["user_db_id"] = student.id
+    session["is_instructor"] = info.is_instructor
+
+    if info.is_instructor:
+        return redirect(url_for("views.instructor_home"))
+    return redirect(url_for("views.student_home"))
+
+
+@bp.get("/jwks")
+def jwks():
+    return jsonify(get_tool_conf().get_jwks())

coldcall_lti/roster.py

@@ -0,0 +1,77 @@
+"""Sync course rosters from NRPS membership data.
+
+`sync_roster` takes plain member dicts in the NRPS shape, so the same
+code serves real Names and Role Provisioning Service responses and the
+dev-mode fake roster.
+"""
+
+from sqlalchemy import select
+
+from .models import Enrollment, Student
+
+
+def _member_role(roles):
+    return "instructor" if any("Instructor" in r for r in roles) else "student"
+
+
+def _sortable_name(member):
+    family = member.get("family_name")
+    given = member.get("given_name")
+    if family and given:
+        return f"{family}, {given}"
+    return member.get("name")
+
+
+def sync_roster(db, course, members):
+    """Upsert students and enrollments for every active member;
+    deactivate enrollments for anyone no longer on the roster (drops).
+    Returns (active, deactivated) counts."""
+    seen_student_ids = set()
+
+    for member in members:
+        if member.get("status", "Active") != "Active":
+            continue
+
+        student = db.execute(
+            select(Student).where(
+                Student.canvas_user_id == member["user_id"]
+            )
+        ).scalar_one_or_none()
+        if student is None:
+            student = Student(canvas_user_id=member["user_id"])
+            db.add(student)
+        if member.get("name"):
+            student.name = member["name"]
+        sortable = _sortable_name(member)
+        if sortable:
+            student.sortable_name = sortable
+        if member.get("email"):
+            student.email = member["email"]
+        if member.get("picture"):
+            student.avatar_url = member["picture"]
+        db.flush()
+        seen_student_ids.add(student.id)
+
+        enrollment = db.execute(
+            select(Enrollment).where(
+                Enrollment.course_id == course.id,
+                Enrollment.student_id == student.id,
+            )
+        ).scalar_one_or_none()
+        if enrollment is None:
+            enrollment = Enrollment(course_id=course.id, student_id=student.id)
+            db.add(enrollment)
+        enrollment.role = _member_role(member.get("roles", []))
+        enrollment.active = True
+        db.flush()
+
+    deactivated = 0
+    for enrollment in db.execute(
+        select(Enrollment).where(Enrollment.course_id == course.id)
+    ).scalars():
+        if enrollment.student_id not in seen_student_ids and enrollment.active:
+            enrollment.active = False
+            deactivated += 1
+    db.flush()
+
+    return len(seen_student_ids), deactivated

coldcall_lti/templates/base.html

@@ -0,0 +1,18 @@
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>{% block title %}Cold Call{% endblock %}</title>
+  <style>
+    body { font-family: system-ui, sans-serif; margin: 2rem auto; max-width: 50rem; padding: 0 1rem; line-height: 1.5; }
+    h1 { font-size: 1.5rem; }
+    table { border-collapse: collapse; }
+    th, td { text-align: left; padding: 0.25rem 0.75rem 0.25rem 0; }
+    .muted { color: #666; }
+  </style>
+</head>
+<body>
+  {% block body %}{% endblock %}
+</body>
+</html>

coldcall_lti/templates/dev_launch.html

@@ -0,0 +1,19 @@
+{% extends "base.html" %}
+{% block title %}Dev launch{% endblock %}
+{% block body %}
+<h1>Development fake launch</h1>
+<p>Pick a persona; this sets up the same session a real LTI launch would.</p>
+<table>
+  {% for m in members %}
+  <tr>
+    <td>{{ m.name }}</td>
+    <td class="muted">{{ "instructor" if "Instructor" in m.roles|join(" ") else "student" }}</td>
+    <td>
+      <form method="post" action="{{ url_for('dev.fake_launch', user_id=m.user_id) }}">
+        <button type="submit">Launch</button>
+      </form>
+    </td>
+  </tr>
+  {% endfor %}
+</table>
+{% endblock %}

coldcall_lti/templates/index.html

@@ -0,0 +1,9 @@
+{% extends "base.html" %}
+{% block body %}
+<h1>Cold Call</h1>
+{% if course %}
+  <p>{{ user.name }} in {{ course.title or course.lti_context_id }}.</p>
+{% else %}
+  <p>This tool is meant to be launched from a Canvas course.</p>
+{% endif %}
+{% endblock %}

coldcall_lti/templates/instructor_home.html

@@ -0,0 +1,15 @@
+{% extends "base.html" %}
+{% block title %}{{ course.title or "Cold Call" }}{% endblock %}
+{% block body %}
+<h1>{{ course.title or course.lti_context_id }}</h1>
+<p>{{ roster|length }} active students on the roster.</p>
+<table>
+  <tr><th>Name</th><th>Email</th></tr>
+  {% for student in roster %}
+  <tr>
+    <td>{{ student.name }}</td>
+    <td class="muted">{{ student.email or "" }}</td>
+  </tr>
+  {% endfor %}
+</table>
+{% endblock %}

coldcall_lti/templates/student_home.html

@@ -0,0 +1,7 @@
+{% extends "base.html" %}
+{% block title %}{{ course.title or "Cold Call" }}{% endblock %}
+{% block body %}
+<h1>{{ course.title or course.lti_context_id }}</h1>
+<p>Hi {{ user.name }}.</p>
+<p class="muted">Your call history and the absence form will appear here.</p>
+{% endblock %}

coldcall_lti/views.py

@@ -0,0 +1,75 @@
+import functools
+
+from flask import Blueprint, abort, render_template, session
+from sqlalchemy import select
+
+from .db import get_db
+from .models import Course, Enrollment, Student
+
+bp = Blueprint("views", __name__)
+
+
+def current_course():
+    return get_db().get(Course, session["course_id"])
+
+
+def current_user():
+    return get_db().get(Student, session["user_db_id"])
+
+
+def require_launch(view):
+    @functools.wraps(view)
+    def wrapped(**kwargs):
+        if "course_id" not in session:
+            abort(403, "Launch this tool from your Canvas course.")
+        return view(**kwargs)
+
+    return wrapped
+
+
+def require_instructor(view):
+    @functools.wraps(view)
+    @require_launch
+    def wrapped(**kwargs):
+        if not session.get("is_instructor"):
+            abort(403)
+        return view(**kwargs)
+
+    return wrapped
+
+
+@bp.get("/")
+def index():
+    if "course_id" in session:
+        return render_template(
+            "index.html", course=current_course(), user=current_user()
+        )
+    return render_template("index.html", course=None, user=None)
+
+
+@bp.get("/instructor")
+@require_instructor
+def instructor_home():
+    db = get_db()
+    course = current_course()
+    roster = db.execute(
+        select(Student)
+        .join(Enrollment, Enrollment.student_id == Student.id)
+        .where(
+            Enrollment.course_id == course.id,
+            Enrollment.role == "student",
+            Enrollment.active.is_(True),
+        )
+        .order_by(Student.sortable_name, Student.name)
+    ).scalars().all()
+    return render_template(
+        "instructor_home.html", course=course, roster=roster
+    )
+
+
+@bp.get("/me")
+@require_launch
+def student_home():
+    return render_template(
+        "student_home.html", course=current_course(), user=current_user()
+    )

lti_config.example.json

@@ -0,0 +1,14 @@
+{
+  "https://canvas.instructure.com": [
+    {
+      "default": true,
+      "client_id": "FIXME-developer-key-client-id",
+      "auth_login_url": "https://sso.canvaslms.com/api/lti/authorize_redirect",
+      "auth_token_url": "https://sso.canvaslms.com/login/oauth2/token",
+      "key_set_url": "https://sso.canvaslms.com/api/lti/security/jwks",
+      "private_key_file": "private.key",
+      "public_key_file": "public.key",
+      "deployment_ids": ["FIXME-deployment-id"]
+    }
+  ]
+}

tests/test_app.py

@@ -0,0 +1,58 @@
+import pytest
+
+from coldcall_lti import create_app
+from coldcall_lti.config import Config
+from coldcall_lti.db import Base
+
+
+def make_app(tmp_path, dev_mode):
+    class TestConfig(Config):
+        TESTING = True
+        DATABASE_URL = f"sqlite:///{tmp_path}/test.sqlite3"
+        DEV_MODE = dev_mode
+        SECRET_KEY = "test"
+
+    app = create_app(TestConfig)
+    Base.metadata.create_all(app.extensions["db_engine"])
+    return app
+
+
+@pytest.fixture
+def dev_client(tmp_path):
+    return make_app(tmp_path, dev_mode=True).test_client()
+
+
+def test_healthz(tmp_path):
+    client = make_app(tmp_path, dev_mode=False).test_client()
+    assert client.get("/healthz").json == {"status": "ok"}
+
+
+def test_dev_routes_absent_outside_dev_mode(tmp_path):
+    client = make_app(tmp_path, dev_mode=False).test_client()
+    assert client.get("/dev/").status_code == 404
+
+
+def test_views_require_launch(dev_client):
+    assert dev_client.get("/instructor").status_code == 403
+    assert dev_client.get("/me").status_code == 403
+
+
+def test_dev_instructor_launch_shows_roster(dev_client):
+    assert dev_client.get("/dev/").status_code == 200
+    resp = dev_client.post(
+        "/dev/launch/dev-instructor", follow_redirects=True
+    )
+    assert resp.status_code == 200
+    page = resp.get_data(as_text=True)
+    assert "8 active students" in page
+    assert "Ada Lovelace" in page
+
+
+def test_dev_student_launch(dev_client):
+    resp = dev_client.post(
+        "/dev/launch/dev-student-1", follow_redirects=True
+    )
+    page = resp.get_data(as_text=True)
+    assert "Ada Lovelace" in page
+    # A student cannot reach the instructor view.
+    assert dev_client.get("/instructor").status_code == 403

tests/test_launch.py

@@ -0,0 +1,51 @@
+from coldcall_lti import launch, models
+
+
+def sample_launch_data(is_instructor=False):
+    roles = [launch.ROLE_INSTRUCTOR] if is_instructor else [launch.ROLE_LEARNER]
+    return {
+        "sub": "canvas-user-42",
+        "name": "Pat Example",
+        "email": "pat@example.edu",
+        "picture": "https://example.edu/pat.png",
+        launch.CLAIM_DEPLOYMENT: "dep-1",
+        launch.CLAIM_CONTEXT: {"id": "ctx-abc", "title": "COM 999"},
+        launch.CLAIM_ROLES: roles,
+    }
+
+
+def test_extract_launch_info_student():
+    info = launch.extract_launch_info(sample_launch_data())
+    assert info.context_id == "ctx-abc"
+    assert info.course_title == "COM 999"
+    assert info.user_id == "canvas-user-42"
+    assert info.deployment_id == "dep-1"
+    assert not info.is_instructor
+
+
+def test_extract_launch_info_instructor():
+    info = launch.extract_launch_info(sample_launch_data(is_instructor=True))
+    assert info.is_instructor
+
+
+def test_upsert_launch_creates_and_is_idempotent(db_session):
+    info = launch.extract_launch_info(sample_launch_data())
+    course, student = launch.upsert_launch(db_session, info)
+    assert course.title == "COM 999"
+    assert student.canvas_user_id == "canvas-user-42"
+
+    course2, student2 = launch.upsert_launch(db_session, info)
+    assert course2.id == course.id
+    assert student2.id == student.id
+    assert db_session.query(models.Enrollment).count() == 1
+
+
+def test_upsert_launch_reactivates_dropped_enrollment(db_session):
+    info = launch.extract_launch_info(sample_launch_data())
+    course, student = launch.upsert_launch(db_session, info)
+    enrollment = db_session.query(models.Enrollment).one()
+    enrollment.active = False
+    db_session.flush()
+
+    launch.upsert_launch(db_session, info)
+    assert db_session.query(models.Enrollment).one().active

tests/test_roster.py

@@ -0,0 +1,82 @@
+from coldcall_lti import models
+from coldcall_lti.launch import ROLE_INSTRUCTOR, ROLE_LEARNER
+from coldcall_lti.roster import sync_roster
+
+
+def member(uid, name="A Student", roles=(ROLE_LEARNER,), status="Active", **kw):
+    return {
+        "user_id": uid,
+        "name": name,
+        "roles": list(roles),
+        "status": status,
+        **kw,
+    }
+
+
+def make_course(db):
+    course = models.Course(lti_context_id="ctx-1")
+    db.add(course)
+    db.flush()
+    return course
+
+
+def test_sync_creates_students_and_enrollments(db_session):
+    course = make_course(db_session)
+    active, deactivated = sync_roster(
+        db_session,
+        course,
+        [
+            member("u1", "Ada Lovelace", given_name="Ada", family_name="Lovelace"),
+            member("u2", "Grace Hopper"),
+            member("t1", "Teacher One", roles=[ROLE_INSTRUCTOR]),
+        ],
+    )
+    assert (active, deactivated) == (3, 0)
+
+    students = {
+        s.canvas_user_id: s for s in db_session.query(models.Student).all()
+    }
+    assert students["u1"].sortable_name == "Lovelace, Ada"
+
+    roles = {
+        e.student.canvas_user_id: e.role
+        for e in db_session.query(models.Enrollment).all()
+    }
+    assert roles == {"u1": "student", "u2": "student", "t1": "instructor"}
+
+
+def test_sync_deactivates_dropped_and_reactivates_returning(db_session):
+    course = make_course(db_session)
+    sync_roster(db_session, course, [member("u1"), member("u2")])
+
+    active, deactivated = sync_roster(db_session, course, [member("u1")])
+    assert (active, deactivated) == (1, 1)
+    dropped = (
+        db_session.query(models.Enrollment)
+        .join(models.Student)
+        .filter(models.Student.canvas_user_id == "u2")
+        .one()
+    )
+    assert not dropped.active
+
+    sync_roster(db_session, course, [member("u1"), member("u2")])
+    assert dropped.active
+
+
+def test_sync_skips_inactive_members(db_session):
+    course = make_course(db_session)
+    active, _ = sync_roster(
+        db_session,
+        course,
+        [member("u1"), member("u2", status="Inactive")],
+    )
+    assert active == 1
+    assert db_session.query(models.Student).count() == 1
+
+
+def test_sync_updates_changed_names(db_session):
+    course = make_course(db_session)
+    sync_roster(db_session, course, [member("u1", "Old Name")])
+    sync_roster(db_session, course, [member("u1", "New Name")])
+    student = db_session.query(models.Student).one()
+    assert student.name == "New Name"