From 643644c1725f5550ee8053f0e53986b8846829b5 Mon Sep 17 00:00:00 2001 From: Agrendalath Date: Mon, 14 Sep 2026 22:41:42 +0200 Subject: [PATCH] feat(pathways): implement CatalogPathway --- .importlinter | 16 +- src/openedx_catalog/admin.py | 103 +++++- src/openedx_catalog/api_impl.py | 246 ++++++++++++++- .../migrations/0002_pathways.py | 298 ++++++++++++++++++ src/openedx_catalog/models/__init__.py | 3 + src/openedx_catalog/models/catalog_pathway.py | 199 ++++++++++++ .../models/pathway_category.py | 99 ++++++ .../models/pathway_enrollment.py | 88 ++++++ src/openedx_catalog/models_api.py | 2 +- tests/openedx_catalog/test_pathway_api.py | 230 ++++++++++++++ tests/openedx_catalog/test_pathway_models.py | 245 ++++++++++++++ 11 files changed, 1522 insertions(+), 7 deletions(-) create mode 100644 src/openedx_catalog/migrations/0002_pathways.py create mode 100644 src/openedx_catalog/models/catalog_pathway.py create mode 100644 src/openedx_catalog/models/pathway_category.py create mode 100644 src/openedx_catalog/models/pathway_enrollment.py create mode 100644 tests/openedx_catalog/test_pathway_api.py create mode 100644 tests/openedx_catalog/test_pathway_models.py diff --git a/.importlinter b/.importlinter index 17dd176f6..85bba4a0d 100644 --- a/.importlinter +++ b/.importlinter @@ -8,6 +8,7 @@ root_packages = openedx_learning openedx_content openedx_tagging + openedx_catalog openedx_django_lib openedx_core @@ -18,12 +19,19 @@ root_packages = name = "top-level source folders are layered correctly" type = layers layers = - # Learning-domain features (currently CBE; Learning Pathways to follow). - # May build on content and tagging. Nothing below may import it: in - # particular, openedx_tagging must never know that CBE exists. + # Learning-domain features (CBE and Pathways). May build on catalog, content and + # tagging. Nothing below may import it: in particular, openedx_tagging must never + # know that CBE exists. openedx_learning - # Content: authoring-side models and APIs. + # Catalog holds the models learners browse and enroll against (CatalogCourse, CourseRun, + # CatalogPathway). A catalog entry points at its content, never the reverse, so catalog + # sits above content: it may hold (nullable) foreign keys to content, while content stays + # ignorant of what a learning package represents. See the openedx_catalog ADR 0001. + openedx_catalog + + # Content: authoring-side models and APIs. Generic infrastructure for courses, libraries, + # pathways and future context types alike. openedx_content # Tagging is very simple & fundamental. Should probably not depend on any other Django apps. diff --git a/src/openedx_catalog/admin.py b/src/openedx_catalog/admin.py index 2aa4f3ca6..6aeaedaa3 100644 --- a/src/openedx_catalog/admin.py +++ b/src/openedx_catalog/admin.py @@ -13,13 +13,16 @@ from django.utils.html import format_html from django.utils.translation import gettext_lazy as _ -from .models import CatalogCourse, CourseRun +from .models import CatalogCourse, CatalogPathway, CourseRun, PathwayCategory, PathwayEnrollment if TYPE_CHECKING: class CatalogCourseWithRunCount(CatalogCourse): run_count: int + class PathwayCategoryWithPathwayCount(PathwayCategory): + pathway_count: int + class CatalogCourseAdmin(admin.ModelAdmin): """ @@ -110,3 +113,101 @@ def warnings(self, obj: CourseRun) -> str | None: admin.site.register(CourseRun, CourseRunAdmin) + + +class PathwayCategoryAdmin(admin.ModelAdmin): + """ + The PathwayCategory model admin. + + Renaming a category changes what learners see. It does not change the authoring-side terminology, which is always + "Pathway". + """ + + list_display = ["name", "category_code", "pathways_summary"] + search_fields = ["name", "category_code"] + + def get_readonly_fields(self, request, obj: PathwayCategory | None = None) -> tuple[str, ...]: + if obj: # editing an existing object; the code is what other systems key off + return ("category_code",) + return tuple() + + def get_queryset(self, request) -> QuerySet[PathwayCategoryWithPathwayCount]: + """Add the 'pathway_count' to the list_display queryset""" + qs = super().get_queryset(request) + qs = qs.annotate(pathway_count=Count("pathways")) + return qs + + @admin.display(description=_("Pathways"), ordering="pathway_count") + def pathways_summary(self, obj: PathwayCategoryWithPathwayCount) -> str: + """Link to the catalog pathways using this category""" + if obj.pathway_count == 0: + return "-" + url = reverse("admin:openedx_catalog_catalogpathway_changelist") + f"?category={obj.pk}" + return format_html('{}', url, obj.pathway_count) + + +admin.site.register(PathwayCategory, PathwayCategoryAdmin) + + +class CatalogPathwayAdmin(admin.ModelAdmin): + """ + The CatalogPathway model admin. + + This edits only the catalog half of a Pathway. The Items a learner must complete live on the content side, in the + openedx_learning app, and are versioned there. + """ + + list_filter = ["org__short_name", "category"] + list_display = [ + "title", + "category", + "org_display", + "pathway_code", + "key_str", + "content_entity", + "created_date", + "modified", + ] + list_select_related = ["org", "category", "content_entity"] + search_fields = ["title", "pathway_code"] + + def get_readonly_fields(self, request, obj: CatalogPathway | None = None) -> tuple[str, ...]: + # The definition is linked through openedx_learning.api, which is the only place that can check that the entity + # really is a Pathway. Show it, but don't offer a + raw_id_fields = ["user", "catalog_pathway"] + + @admin.display(description=_("Enrolled"), ordering="created") + def created_date(self, obj: PathwayEnrollment) -> datetime.date: + """Display the enrollment date without the timestamp""" + return obj.created.date() + + +admin.site.register(PathwayEnrollment, PathwayEnrollmentAdmin) diff --git a/src/openedx_catalog/api_impl.py b/src/openedx_catalog/api_impl.py index 03c9a3ee7..3fd3f8c5a 100644 --- a/src/openedx_catalog/api_impl.py +++ b/src/openedx_catalog/api_impl.py @@ -5,11 +5,17 @@ import logging from typing import overload +from django.db import transaction +from django.db.models import QuerySet +from django.utils import timezone from opaque_keys.edx.keys import CourseKey from organizations.api import ensure_organization # type: ignore[import] from organizations.api import exceptions as org_exceptions -from .models import CatalogCourse, CourseRun +from openedx_content.models_api import PublishableEntity + +from .models import CatalogCourse, CatalogPathway, CourseRun, PathwayCategory, PathwayEnrollment +from .models.pathway_category import get_default_pathway_category log = logging.getLogger(__name__) @@ -22,6 +28,17 @@ "sync_course_run_details", "create_course_run_for_modulestore_course_with", "delete_course_run", + "get_default_pathway_category", + "get_pathway_category", + "get_catalog_pathway", + "create_catalog_pathway", + "update_catalog_pathway", + "set_catalog_pathway_content", + "delete_catalog_pathway", + "enroll_in_pathway", + "unenroll_from_pathway", + "is_enrolled_in_pathway", + "get_pathway_enrollments", ] @@ -249,3 +266,230 @@ def delete_course_run(course_key: CourseKey) -> None: ⚠️ Does not emit any course lifecycle events. """ CourseRun.objects.get(course_key=course_key).delete() + + +# Pathways (catalog side). +# +# A Pathway is split into a catalog half (these models) and a versioned content half in +# `openedx_learning.applets.pathways`. See the openedx_learning ADR 0007. The functions below only touch the catalog +# half; creating and versioning the *definition* of a Pathway is done through `openedx_learning.api`, which also links +# the definition to its `CatalogPathway` via `set_catalog_pathway_content()`. + + +# `get_default_pathway_category` is part of this API too, and is re-exported via `__all__`. It's defined next to the +# model because the `CatalogPathway.category` field default needs it as well. + + +def get_pathway_category(category_code: str) -> PathwayCategory: + """ + Get a `PathwayCategory` by its stable code. + + ⚠️ Does not check permissions. + """ + return PathwayCategory.objects.get(category_code=category_code) + + +@overload +def get_catalog_pathway(*, org_code: str, pathway_code: str) -> CatalogPathway: ... +@overload +def get_catalog_pathway(*, key_str: str) -> CatalogPathway: ... +@overload +def get_catalog_pathway(*, pk: CatalogPathway.ID) -> CatalogPathway: ... + + +def get_catalog_pathway( + pk: CatalogPathway.ID | None = None, + key_str: str = "", + org_code: str = "", + pathway_code: str = "", +) -> CatalogPathway: + """ + Get a catalog pathway. + + ⚠️ Does not check permissions or visibility rules. + + The `CatalogPathway` may not have a definition yet: `content_entity` is `None` until `openedx_learning.api` links + one. To resolve it to the actual Pathway, use `openedx_learning.api.get_pathway_for_catalog_pathway()`. + """ + assert pk or key_str or (org_code and pathway_code) + if pk: + assert not org_code + assert not key_str + return CatalogPathway.objects.get(pk=pk) + if key_str: + assert key_str.startswith("catalog-pathway:") + assert not org_code + assert not pathway_code + _, org_code, pathway_code = key_str.split(":", 2) + # We might as well select_related org because we're joining to check the org__short_name field anyways. + return CatalogPathway.objects.select_related("org").get(org__short_name=org_code, pathway_code=pathway_code) + + +def create_catalog_pathway( + *, + org_code: str, + pathway_code: str, + title: str = "", + category: PathwayCategory | None = None, + description: str = "", +) -> CatalogPathway: + """ + Create a `CatalogPathway`. + + The `Organization` identified by `org_code` must already exist. Pass `category=None` to use the default category. + + This creates only the catalog half of a Pathway. Use `openedx_learning.api` to create the versioned definition and + link it to this entry. + + ⚠️ Does not check permissions. + """ + pathway = CatalogPathway( + pathway_code=pathway_code, + title=title, + description=description, + # Only pass the category if given, so that the field default (which queries for the shipped category) runs + # only when it's actually needed. + **({"category": category} if category is not None else {}), + ) + pathway.org_code = org_code # Resolves the Organization by short_name; raises Organization.DoesNotExist. + pathway.save() + return pathway + + +def update_catalog_pathway( + catalog_pathway: CatalogPathway | CatalogPathway.ID, + *, + title: str | None = None, + category: PathwayCategory | None = None, + description: str | None = None, +) -> None: + """ + Update a `CatalogPathway`. Pass `None` for a field to leave it unchanged. + + None of these edits create a new content version: catalog copy and the Pathway definition change at different rates + and are edited by different people, which is the whole point of the split. + + ⚠️ Does not check permissions. + """ + if isinstance(catalog_pathway, CatalogPathway): + cp = catalog_pathway + else: + cp = CatalogPathway.objects.get(pk=catalog_pathway) + + update_fields = [] + for field_name, value in ( + ("title", title), + ("category", category), + ("description", description), + ): + if value is not None: + setattr(cp, field_name, value) + update_fields.append(field_name) + if update_fields: + cp.save(update_fields=update_fields + ["modified"]) + + +def set_catalog_pathway_content( + catalog_pathway: CatalogPathway | CatalogPathway.ID, + content_entity: PublishableEntity | PublishableEntity.ID | None, +) -> None: + """ + Point a `CatalogPathway` at the `PublishableEntity` holding its versioned definition, or pass `None` to unlink it. + + A definition can serve only one catalog entry, so this raises an `IntegrityError` if another `CatalogPathway` + already points at the same entity. + + `openedx_catalog` cannot tell a Pathway entity apart from any other `PublishableEntity`, so no such check happens + here. Prefer `openedx_learning.api` (`create_pathway()`, `link_catalog_pathway()`), which only ever passes entities + it created as Pathways. + + ⚠️ Does not check permissions. + """ + if isinstance(catalog_pathway, CatalogPathway): + cp = catalog_pathway + else: + cp = CatalogPathway.objects.get(pk=catalog_pathway) + + if content_entity is None or isinstance(content_entity, PublishableEntity): + cp.content_entity = content_entity + else: + cp.content_entity_id = content_entity + cp.save(update_fields=["content_entity", "modified"]) + + +def delete_catalog_pathway(catalog_pathway: CatalogPathway | CatalogPathway.ID) -> None: + """ + Delete a `CatalogPathway`, along with its enrollments. + + The versioned definition it pointed at, if any, is left in place in its learning package; it is simply no longer + linked from the catalog. + + ⚠️ Does not check permissions. + """ + if isinstance(catalog_pathway, CatalogPathway): + cp = catalog_pathway + else: + cp = CatalogPathway.objects.get(pk=catalog_pathway) + cp.delete() + + +def enroll_in_pathway(user_id: int, catalog_pathway: CatalogPathway | CatalogPathway.ID) -> PathwayEnrollment: + """ + Enroll a learner in a `CatalogPathway`, or return their existing active enrollment. + + If the learner had previously unenrolled, their existing row is reactivated rather than replaced, so the original + enrollment date is kept. + + Enrollment does not pin a content version: progress is always evaluated against whatever is published at the time, + so that authoring changes reach learners who are already enrolled. + + ⚠️ Does not check permissions. + """ + pathway_id = catalog_pathway.id if isinstance(catalog_pathway, CatalogPathway) else catalog_pathway + with transaction.atomic(): + # Lock the row so a concurrent unenroll can't slip in between reading `is_active` and writing it back. + enrollment, created = PathwayEnrollment.objects.select_for_update().get_or_create( + user_id=user_id, catalog_pathway_id=pathway_id + ) + if not created and not enrollment.is_active: + enrollment.is_active = True + enrollment.save(update_fields=["is_active", "modified"]) + return enrollment + + +def unenroll_from_pathway(user_id: int, catalog_pathway: CatalogPathway | CatalogPathway.ID) -> None: + """ + Unenroll a learner from a `CatalogPathway`. A no-op if they aren't enrolled. + + The enrollment row is deactivated, not deleted. + + ⚠️ Does not check permissions. + """ + pathway_id = catalog_pathway.id if isinstance(catalog_pathway, CatalogPathway) else catalog_pathway + PathwayEnrollment.objects.filter(user_id=user_id, catalog_pathway_id=pathway_id, is_active=True).update( + is_active=False, modified=timezone.now() + ) + + +def is_enrolled_in_pathway(user_id: int, catalog_pathway: CatalogPathway | CatalogPathway.ID) -> bool: + """ + Check whether this learner is actively enrolled in this `CatalogPathway`. + + ⚠️ Does not check permissions. + """ + pathway_id = catalog_pathway.id if isinstance(catalog_pathway, CatalogPathway) else catalog_pathway + return PathwayEnrollment.objects.filter(user_id=user_id, catalog_pathway_id=pathway_id, is_active=True).exists() + + +def get_pathway_enrollments(user_id: int, *, include_inactive: bool = False) -> QuerySet[PathwayEnrollment]: + """ + Get a learner's pathway enrollments, most recent first. + + Only active enrollments are returned unless ``include_inactive`` is set. + + ⚠️ Does not check permissions or visibility rules. + """ + enrollments = PathwayEnrollment.objects.filter(user_id=user_id).select_related("catalog_pathway") + if not include_inactive: + enrollments = enrollments.filter(is_active=True) + return enrollments diff --git a/src/openedx_catalog/migrations/0002_pathways.py b/src/openedx_catalog/migrations/0002_pathways.py new file mode 100644 index 000000000..05c7f1fc4 --- /dev/null +++ b/src/openedx_catalog/migrations/0002_pathways.py @@ -0,0 +1,298 @@ +""" +Create the catalog half of a Pathway: PathwayCategory, CatalogPathway, and PathwayEnrollment. + +Every CatalogPathway must have a category. Rather than falling back to the word "Pathway" in code, we ship a database +row with that name, so that the behavior is uniform and operators can rename it or add categories of their own without a +code change (ADR 0007, decision 2). The default row is created right after its table and before CatalogPathway exists. +Because Django unapplies operations in reverse order, the reverse step deletes that row only after the CatalogPathway +table is already gone, so nothing can still reference it. +""" + +import re + +import django.core.validators +import django.db.models.deletion +import django.db.models.functions.text +import django.db.models.lookups +from django.conf import settings +from django.db import migrations, models + +import openedx_catalog.models.pathway_category +import openedx_django_lib.fields +import openedx_django_lib.validators + +# These values are duplicated from openedx_catalog.models.pathway_category rather than imported, because a migration +# should represent a point-in-time transformation and must not change if those constants later do. +DEFAULT_CATEGORY_CODE = "pathway" +DEFAULT_CATEGORY_NAME = "Pathway" + + +def create_default_pathway_category(apps, schema_editor): + """Create the default category.""" + PathwayCategory = apps.get_model("openedx_catalog", "PathwayCategory") + PathwayCategory.objects.get_or_create( + category_code=DEFAULT_CATEGORY_CODE, + defaults={"name": DEFAULT_CATEGORY_NAME}, + ) + + +def delete_default_pathway_category(apps, schema_editor): + """Remove the default category on reverse.""" + PathwayCategory = apps.get_model("openedx_catalog", "PathwayCategory") + PathwayCategory.objects.filter(category_code=DEFAULT_CATEGORY_CODE).delete() + + +class Migration(migrations.Migration): + + dependencies = [ + ("openedx_catalog", "0001_initial"), + ("openedx_content", "0014_typed_media_id"), + ("organizations", "0004_auto_20230727_2054"), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name="PathwayCategory", + fields=[ + ( + "id", + models.BigAutoField( + editable=False, + help_text="The internal database ID for this pathway category. Should not be exposed to users nor in APIs.", + primary_key=True, + serialize=False, + verbose_name="Primary Key", + ), + ), + ( + "category_code", + openedx_django_lib.fields.MultiCollationCharField( + db_collations={"mysql": "utf8mb4_bin", "sqlite": "BINARY"}, + help_text='A stable slug identifying this category, e.g. "masters-degree". Not shown to learners.', + max_length=255, + validators=[ + django.core.validators.RegexValidator( + re.compile("^[a-zA-Z0-9_.-]+\\Z"), + 'Enter a valid "code name" consisting of latin letters (A-Z, a-z), numbers, underscores, hyphens, or periods.', + "invalid", + ) + ], + ), + ), + ( + "name", + openedx_django_lib.fields.MultiCollationCharField( + db_collations={"mysql": "utf8mb4_unicode_ci", "sqlite": "NOCASE"}, + help_text='The learner-facing name of this category, e.g. "Master\'s Degree". Operators may change this.', + max_length=255, + ), + ), + ], + options={ + "verbose_name": "Pathway Category", + "verbose_name_plural": "Pathway Categories", + "ordering": ("name",), + "constraints": [ + models.UniqueConstraint( + django.db.models.functions.text.Lower("category_code"), + name="oex_catalog_pathwaycategory_code_uniq_ci", + ), + models.CheckConstraint( + condition=django.db.models.lookups.Regex(models.F("category_code"), "^[a-zA-Z0-9_.-]+\\Z"), + name="oex_catalog_pathwaycategory_code_regex", + violation_error_message='Enter a valid "code name" consisting of latin letters (A-Z, a-z), numbers, underscores, hyphens, or periods.', + ), + models.CheckConstraint( + condition=models.Q(("name__length__gt", 0)), name="oex_catalog_pathwaycategory_name_not_blank" + ), + ], + }, + ), + migrations.RunPython( + create_default_pathway_category, + reverse_code=delete_default_pathway_category, + ), + migrations.CreateModel( + name="CatalogPathway", + fields=[ + ( + "id", + models.BigAutoField( + editable=False, + help_text="The internal database ID for this catalog pathway. Should not be exposed to users nor in APIs.", + primary_key=True, + serialize=False, + verbose_name="Primary Key", + ), + ), + ( + "pathway_code", + openedx_django_lib.fields.MultiCollationCharField( + db_collations={"mysql": "utf8mb4_bin", "sqlite": "BINARY"}, + help_text='The pathway code/number, e.g. "DataScience2026".', + max_length=255, + validators=[ + django.core.validators.RegexValidator( + re.compile("^[a-zA-Z0-9_.-]+\\Z"), + 'Enter a valid "code name" consisting of latin letters (A-Z, a-z), numbers, underscores, hyphens, or periods.', + "invalid", + ) + ], + ), + ), + ( + "created", + models.DateTimeField( + auto_now_add=True, validators=[openedx_django_lib.validators.validate_utc_datetime] + ), + ), + ( + "modified", + models.DateTimeField( + auto_now=True, + help_text="When the catalog fields of this pathway were last edited. Unrelated to its content.", + validators=[openedx_django_lib.validators.validate_utc_datetime], + ), + ), + ( + "title", + openedx_django_lib.fields.MultiCollationCharField( + blank=True, + db_collations={"mysql": "utf8mb4_unicode_ci", "sqlite": "NOCASE"}, + help_text='The full title (display name) of this pathway, e.g. "Data Science Professional Certificate". Leave blank to use the pathway code as the title.', + max_length=255, + ), + ), + ( + "description", + openedx_django_lib.fields.MultiCollationTextField( + blank=True, + db_collations={"mysql": "utf8mb4_unicode_ci", "sqlite": "NOCASE"}, + default="", + help_text="The description shown to learners browsing the catalog.", + max_length=10000, + ), + ), + ( + "org", + models.ForeignKey( + on_delete=django.db.models.deletion.PROTECT, + related_name="catalog_pathways", + to="organizations.organization", + ), + ), + ( + "category", + models.ForeignKey( + default=openedx_catalog.models.pathway_category.get_default_pathway_category_id, + help_text="The learner-facing kind of pathway this is. Always required; defaults to a category we ship.", + on_delete=django.db.models.deletion.PROTECT, + related_name="pathways", + to="openedx_catalog.pathwaycategory", + ), + ), + ( + "content_entity", + models.OneToOneField( + blank=True, + help_text="The publishable entity holding this pathway's versioned definition (a Pathway in openedx_learning). Blank until a definition has been created and linked through openedx_learning.api.", + null=True, + on_delete=django.db.models.deletion.PROTECT, + related_name="catalog_pathway", + to="openedx_content.publishableentity", + ), + ), + ], + options={ + "verbose_name": "Catalog Pathway", + "verbose_name_plural": "Catalog Pathways", + "ordering": ("-created",), + }, + ), + migrations.CreateModel( + name="PathwayEnrollment", + fields=[ + ( + "id", + models.BigAutoField( + editable=False, + help_text="The internal database ID for this enrollment. Should not be exposed to users nor in APIs.", + primary_key=True, + serialize=False, + verbose_name="Primary Key", + ), + ), + ( + "created", + models.DateTimeField( + auto_now_add=True, validators=[openedx_django_lib.validators.validate_utc_datetime] + ), + ), + ( + "modified", + models.DateTimeField(auto_now=True, validators=[openedx_django_lib.validators.validate_utc_datetime]), + ), + ( + "is_active", + models.BooleanField( + default=True, + help_text="False once the learner has unenrolled. The row is kept so re-enrolling reuses it.", + ), + ), + ( + "catalog_pathway", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="enrollments", + to="openedx_catalog.catalogpathway", + ), + ), + ( + "user", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="pathway_enrollments", + to=settings.AUTH_USER_MODEL, + ), + ), + ], + options={ + "verbose_name": "Pathway Enrollment", + "verbose_name_plural": "Pathway Enrollments", + "ordering": ("-created",), + }, + ), + migrations.AddIndex( + model_name="catalogpathway", + index=models.Index(fields=["org", "pathway_code"], name="openedx_cat_org_id_037ff8_idx"), + ), + migrations.AddConstraint( + model_name="catalogpathway", + constraint=models.UniqueConstraint( + models.F("org"), + django.db.models.functions.text.Lower("pathway_code"), + name="oex_catalog_catalogpathway_org_code_uniq_ci", + ), + ), + migrations.AddConstraint( + model_name="catalogpathway", + constraint=models.CheckConstraint( + condition=django.db.models.lookups.Regex(models.F("pathway_code"), "^[a-zA-Z0-9_.-]+\\Z"), + name="oex_catalog_catalogpathway_pathway_code_regex", + violation_error_message='Enter a valid "code name" consisting of latin letters (A-Z, a-z), numbers, underscores, hyphens, or periods.', + ), + ), + migrations.AddConstraint( + model_name="catalogpathway", + constraint=models.CheckConstraint( + condition=models.Q(("title__length__gt", 0)), name="oex_catalog_catalogpathway_title_not_blank" + ), + ), + migrations.AddConstraint( + model_name="pathwayenrollment", + constraint=models.UniqueConstraint( + fields=("user", "catalog_pathway"), name="oex_catalog_pathwayenrollment_uniq_user_pathway" + ), + ), + ] diff --git a/src/openedx_catalog/models/__init__.py b/src/openedx_catalog/models/__init__.py index 31da59962..513141541 100644 --- a/src/openedx_catalog/models/__init__.py +++ b/src/openedx_catalog/models/__init__.py @@ -3,4 +3,7 @@ """ from .catalog_course import CatalogCourse +from .catalog_pathway import CatalogPathway from .course_run import CourseRun +from .pathway_category import PathwayCategory +from .pathway_enrollment import PathwayEnrollment diff --git a/src/openedx_catalog/models/catalog_pathway.py b/src/openedx_catalog/models/catalog_pathway.py new file mode 100644 index 000000000..3761e9a30 --- /dev/null +++ b/src/openedx_catalog/models/catalog_pathway.py @@ -0,0 +1,199 @@ +""" +CatalogPathway model +""" + +import logging +from typing import NewType + +from django.contrib import admin +from django.db import models +from django.db.models.functions import Length, Lower +from django.utils.translation import gettext_lazy as _ +from organizations.models import Organization # type: ignore[import] + +from openedx_content.models_api import PublishableEntity +from openedx_django_lib.fields import ( + MultiCollationTextField, + TypedBigAutoField, + case_insensitive_char_field, + code_field, + code_field_check, +) +from openedx_django_lib.validators import validate_utc_datetime + +from .pathway_category import PathwayCategory, get_default_pathway_category_id + +log = logging.getLogger(__name__) + +# Make 'length' available for CHECK constraints. OK if this is called multiple times. +models.CharField.register_lookup(Length) + + +class CatalogPathway(models.Model): + """ + The learner-browsable, enrollable half of a Pathway. + + A Pathway is split in two (see the openedx_learning ADR 0007). This model is the catalog half: the display name, the + description shown in the catalog, and the `PathwayCategory`. It is **not versioned**, because marketing copy is + revised frequently and casually and versioning it would be pure overhead. + + The other half - the *definition* of the Pathway, meaning its Items and completion criteria - lives in + `openedx_learning.applets.pathways` and *is* versioned, so that progress and credentials can be judged against the + definition that was in effect at the time. + + The one link between the two halves is `content_entity`, which points at the `PublishableEntity` that carries the + versioned definition. It lives here because a context always points at its content, never the reverse (see the + openedx_catalog ADR 0001): `openedx_catalog` sits above `openedx_content` and may reference it, while the definition + models themselves live in `openedx_learning`, above this app. That is why the field is typed as a bare + `PublishableEntity` rather than as a Pathway; `openedx_learning.api` is what sets it and resolves it. + + A `CatalogPathway` may exist before its definition does, in the same way that a `CatalogCourse` may exist as a + marketing placeholder for a course that has no content yet. + + Like `CatalogCourse`, this model is intentionally minimal. Additional catalog-side fields should generally go in a + related model in your own app, with a `ForeignKey` or `OneToOneField` to this one. + + .. no_pii: + """ + + CatalogPathwayID = NewType("CatalogPathwayID", int) + type ID = CatalogPathwayID + + class IDField(TypedBigAutoField[ID]): # Boilerplate for fully-typed ID field. + pass + + id = IDField( + primary_key=True, + verbose_name=_("Primary Key"), + help_text=_("The internal database ID for this catalog pathway. Should not be exposed to users nor in APIs."), + editable=False, + ) + org = models.ForeignKey( + Organization, + on_delete=models.PROTECT, + null=False, + related_name="catalog_pathways", + ) + pathway_code = code_field( + unicode=False, + help_text=_('The pathway code/number, e.g. "DataScience2026".'), + ) + created = models.DateTimeField( + auto_now_add=True, + validators=[validate_utc_datetime], + editable=False, + ) + # This reflects edits to the *catalog* fields on this row only (title, category, description). It says nothing about + # when the pathway's *definition* - its Items and completion criteria - last changed. That happens on the content + # side and is versioned there; ask `openedx_learning.api` for it. + modified = models.DateTimeField( + auto_now=True, + validators=[validate_utc_datetime], + editable=False, + help_text=_("When the catalog fields of this pathway were last edited. Unrelated to its content."), + ) + title = case_insensitive_char_field( + max_length=255, + blank=True, # Only allowed to be blank temporarily when creating a new instance in the Django admin form. + help_text=_( + 'The full title (display name) of this pathway, e.g. "Data Science Professional Certificate". ' + "Leave blank to use the pathway code as the title." + ), + ) + category = models.ForeignKey( + PathwayCategory, + on_delete=models.PROTECT, + null=False, + default=get_default_pathway_category_id, + related_name="pathways", + help_text=_("The learner-facing kind of pathway this is. Always required; defaults to a category we ship."), + ) + content_entity = models.OneToOneField( # One definition serves one catalog entry. + # The link is deliberately unversioned: a catalog entry follows whichever version of its Pathway is currently + # published, which is what lets authoring changes reach learners who are already enrolled. + PublishableEntity, + # Deleting a definition out from under enrolled learners would leave them enrolled in something with + # no requirements. We must unlink it first. + on_delete=models.PROTECT, + null=True, # A `CatalogPathway` may exist as a placeholder before any definition does. + blank=True, + related_name="catalog_pathway", + help_text=_( + "The publishable entity holding this pathway's versioned definition (a Pathway in openedx_learning). " + "Blank until a definition has been created and linked through openedx_learning.api." + ), + ) + description = MultiCollationTextField( + blank=True, + null=False, + default="", + max_length=10_000, + # We don't expect to sort by this column, but we may want case-insensitive searches over it. + db_collations={ + "sqlite": "NOCASE", + "mysql": "utf8mb4_unicode_ci", + }, + help_text=_("The description shown to learners browsing the catalog."), + ) + + # 🛑 Avoid adding additional fields here. Anything that describes what a learner must *do* belongs on the content + # side, where it is versioned. Anything else catalog-related should go in a related model in your own app. + + @property + @admin.display(ordering="org__short_name") + def org_code(self) -> str: + """ + Get the org code (Organization short_name) of this pathway, e.g. "MITx". + """ + return self.org.short_name + + @org_code.setter + def org_code(self, org_code: str) -> None: + """ + Convenience method to set the related organization using its short_name. + """ + # As with CatalogCourse, we don't use `get_organization_by_short_name` because it filters for active orgs only, + # and we need to allow inactive orgs to support historical data and backfills. + self.org = Organization.objects.get(short_name__iexact=org_code) + + @property + def key_str(self) -> str: + """ + A string key that can be used to identify this catalog pathway in URLs or APIs. + + As with `CatalogCourse.key_str`, this may become based on an editable `SlugField` or an opaque key in the + future, so don't assume it never changes. + """ + return f"catalog-pathway:{self.org_code}:{self.pathway_code}" + + def clean(self) -> None: + """Validate/normalize fields when edited via Django admin.""" + # Set a default value for title: + if not self.title: + self.title = self.pathway_code + + def save(self, *args, **kwargs): + """Save the model, with some defaults and validation.""" + self.clean() + super().save(*args, **kwargs) + + def __str__(self) -> str: + return f"{self.title} ({self.org_code} {self.pathway_code})" + + class Meta: + verbose_name = _("Catalog Pathway") + verbose_name_plural = _("Catalog Pathways") + ordering = ("-created",) + indexes = [ + # We need fast lookups by (org, pathway_code) pairs. We generally want this lookup to be case sensitive. + models.Index(fields=["org", "pathway_code"]), + ] + constraints = [ + # The pathway_code must be case-insensitively unique per org: + models.UniqueConstraint("org", Lower("pathway_code"), name="oex_catalog_catalogpathway_org_code_uniq_ci"), + code_field_check("pathway_code", name="oex_catalog_catalogpathway_pathway_code_regex", unicode=False), + # Enforce at the DB level that these required fields are not blank: + models.CheckConstraint( + condition=models.Q(title__length__gt=0), name="oex_catalog_catalogpathway_title_not_blank" + ), + ] diff --git a/src/openedx_catalog/models/pathway_category.py b/src/openedx_catalog/models/pathway_category.py new file mode 100644 index 000000000..de41c6e3d --- /dev/null +++ b/src/openedx_catalog/models/pathway_category.py @@ -0,0 +1,99 @@ +""" +PathwayCategory model +""" + +import logging +from typing import NewType + +from django.db import models +from django.db.models.functions import Length, Lower +from django.utils.translation import gettext_lazy as _ + +from openedx_django_lib.fields import TypedBigAutoField, case_insensitive_char_field, code_field, code_field_check + +log = logging.getLogger(__name__) + +# Make 'length' available for CHECK constraints. OK if this is called multiple times. +models.CharField.register_lookup(Length) + +DEFAULT_PATHWAY_CATEGORY_CODE = "pathway" +DEFAULT_PATHWAY_CATEGORY_NAME = "Pathway" + + +class PathwayCategory(models.Model): + """ + A student-facing label for a kind of Pathway. + + Learners are shown the category ("Master's Degree", "Annual Training") rather than the word "Pathway". In authoring + contexts - Studio, Django admin, code, docs - the terminology stays "Pathway", with the category shown explicitly; + relabelling is a learner-facing concern of the catalog side only. + + The ``category_code`` is the stable identifier that code and imports may key off. The ``name`` is what learners see, + and operators are free to change it - including on the default category shipped by the initial migration. + + .. no_pii: + """ + + PathwayCategoryID = NewType("PathwayCategoryID", int) + type ID = PathwayCategoryID + + class IDField(TypedBigAutoField[ID]): # Boilerplate for fully-typed ID field. + pass + + id = IDField( + primary_key=True, + verbose_name=_("Primary Key"), + help_text=_("The internal database ID for this pathway category. Should not be exposed to users nor in APIs."), + editable=False, + ) + category_code = code_field( + unicode=False, + help_text=_('A stable slug identifying this category, e.g. "masters-degree". Not shown to learners.'), + ) + name = case_insensitive_char_field( + max_length=255, + blank=False, + help_text=_('The learner-facing name of this category, e.g. "Master\'s Degree". Operators may change this.'), + ) + + def __str__(self) -> str: + return str(self.name) + + class Meta: + verbose_name = _("Pathway Category") + verbose_name_plural = _("Pathway Categories") + ordering = ("name",) + constraints = [ + # The category_code must be case-insensitively unique: + models.UniqueConstraint(Lower("category_code"), name="oex_catalog_pathwaycategory_code_uniq_ci"), + code_field_check("category_code", name="oex_catalog_pathwaycategory_code_regex", unicode=False), + # Enforce at the DB level that this required field is not blank: + models.CheckConstraint( + condition=models.Q(name__length__gt=0), name="oex_catalog_pathwaycategory_name_not_blank" + ), + ] + + +def get_default_pathway_category() -> PathwayCategory: + """ + Get the default `PathwayCategory`, creating it if it doesn't exist. + + Every `CatalogPathway` must have a category. Rather than falling back to the word "Pathway" in code, we ship a + database row with that name, so that operators can rename it or add categories of their own without a code change. + See the openedx_learning ADR 0007. + """ + category, _created = PathwayCategory.objects.get_or_create( + category_code=DEFAULT_PATHWAY_CATEGORY_CODE, + defaults={"name": DEFAULT_PATHWAY_CATEGORY_NAME}, + ) + return category + + +def get_default_pathway_category_id() -> PathwayCategory.ID: + """ + Get the ID of the default `PathwayCategory`, creating it if it doesn't exist. + + Note: this function is used as a field default and is therefore referenced from migrations, so update those + migrations if moving it or changing its signature. + """ + return get_default_pathway_category().id diff --git a/src/openedx_catalog/models/pathway_enrollment.py b/src/openedx_catalog/models/pathway_enrollment.py new file mode 100644 index 000000000..59371dea1 --- /dev/null +++ b/src/openedx_catalog/models/pathway_enrollment.py @@ -0,0 +1,88 @@ +""" +PathwayEnrollment model +""" + +import logging +from typing import NewType + +from django.conf import settings +from django.db import models +from django.utils.translation import gettext_lazy as _ + +from openedx_django_lib.fields import TypedBigAutoField +from openedx_django_lib.validators import validate_utc_datetime + +from .catalog_pathway import CatalogPathway + +log = logging.getLogger(__name__) + + +class PathwayEnrollment(models.Model): + """ + Ties a learner to a `CatalogPathway`. + + Enrollment is against the *catalog* half of a Pathway, never against a version of its content. Progress is evaluated + against whichever content version is published at the time of evaluation, not against a version frozen at enrollment + time, so that authoring changes reach learners who are already enrolled. That is why this model pins no version. See + the openedx_learning ADR 0007, decision 5. + + Unenrolling sets ``is_active`` to False rather than deleting the row, so that "unenrolled" can be told apart from + "never enrolled" and the original enrollment date survives. Re-enrolling reactivates the same row. + + .. no_pii: + """ + + PathwayEnrollmentID = NewType("PathwayEnrollmentID", int) + type ID = PathwayEnrollmentID + + class IDField(TypedBigAutoField[ID]): # Boilerplate for fully-typed ID field. + pass + + id = IDField( + primary_key=True, + verbose_name=_("Primary Key"), + help_text=_("The internal database ID for this enrollment. Should not be exposed to users nor in APIs."), + editable=False, + ) + user = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.CASCADE, + null=False, + related_name="pathway_enrollments", + ) + catalog_pathway = models.ForeignKey( + CatalogPathway, + on_delete=models.CASCADE, + null=False, + related_name="enrollments", + ) + created = models.DateTimeField( + auto_now_add=True, + validators=[validate_utc_datetime], + editable=False, + ) + modified = models.DateTimeField( + auto_now=True, + validators=[validate_utc_datetime], + editable=False, + ) + is_active = models.BooleanField( + default=True, + help_text=_("False once the learner has unenrolled. The row is kept so re-enrolling reuses it."), + ) + + def __str__(self) -> str: + return f"{self.user} in {self.catalog_pathway}" + + class Meta: + verbose_name = _("Pathway Enrollment") + verbose_name_plural = _("Pathway Enrollments") + ordering = ("-created",) + constraints = [ + # There is only ever one row per (learner, pathway) pair; unenrolling flips `is_active` rather than + # deleting or adding a row. + models.UniqueConstraint( + fields=["user", "catalog_pathway"], + name="oex_catalog_pathwayenrollment_uniq_user_pathway", + ), + ] diff --git a/src/openedx_catalog/models_api.py b/src/openedx_catalog/models_api.py index 2836a2aa1..90a67b1a3 100644 --- a/src/openedx_catalog/models_api.py +++ b/src/openedx_catalog/models_api.py @@ -7,4 +7,4 @@ """ # pylint: disable=unused-import -from .models import CatalogCourse, CourseRun +from .models import CatalogCourse, CatalogPathway, CourseRun, PathwayCategory, PathwayEnrollment diff --git a/tests/openedx_catalog/test_pathway_api.py b/tests/openedx_catalog/test_pathway_api.py new file mode 100644 index 000000000..c6831929b --- /dev/null +++ b/tests/openedx_catalog/test_pathway_api.py @@ -0,0 +1,230 @@ +""" +Tests of the catalog-side Pathway API. +""" +# pylint: disable=unused-argument + +from datetime import datetime, timezone + +import pytest +from django.contrib.auth import get_user_model +from freezegun import freeze_time +from organizations.api import ensure_organization # type: ignore[import] + +from openedx_catalog import api as catalog_api +from openedx_catalog.models import CatalogPathway, PathwayCategory +from openedx_catalog.models.pathway_category import DEFAULT_PATHWAY_CATEGORY_CODE +from openedx_content import api as content_api +from openedx_content.models_api import PublishableEntity + +User = get_user_model() + +pytestmark = pytest.mark.django_db + + +@pytest.fixture(name="org1") +def _org1() -> None: + """Create an "Org1" organization for use in these tests""" + ensure_organization("Org1") + + +@pytest.fixture(name="data_science") +def _data_science(org1) -> CatalogPathway: + """Create a CatalogPathway for use in these tests""" + return catalog_api.create_catalog_pathway( + org_code="Org1", + pathway_code="DataScience", + title="Data Science Professional Certificate", + description="Learn data science.", + ) + + +@pytest.fixture(name="learner") +def _learner(): + """Create a learner for use in these tests""" + return User.objects.create(username="learner", email="learner@example.com") + + +@pytest.fixture(name="definition") +def _definition() -> PublishableEntity: + """Create a bare PublishableEntity to stand in for a Pathway definition.""" + package = content_api.create_learning_package(package_ref="pathway-tests", title="Pathway tests") + return content_api.create_publishable_entity( + package.id, "pathway:DataScience", datetime(2026, 1, 1, tzinfo=timezone.utc), None + ) + + +def test_get_default_pathway_category() -> None: + assert catalog_api.get_default_pathway_category().category_code == DEFAULT_PATHWAY_CATEGORY_CODE + + +def test_get_pathway_category() -> None: + """Categories are looked up by their stable code, not by the learner-facing name.""" + masters = PathwayCategory.objects.create(category_code="masters-degree", name="Master's Degree") + assert catalog_api.get_pathway_category("masters-degree") == masters + with pytest.raises(PathwayCategory.DoesNotExist): + catalog_api.get_pathway_category("Master's Degree") + + +def test_create_with_default_category(org1) -> None: + """Omitting the category picks the shipped default, and a blank title falls back to the code.""" + pathway = catalog_api.create_catalog_pathway(org_code="Org1", pathway_code="CompSci") + assert pathway.category.category_code == DEFAULT_PATHWAY_CATEGORY_CODE + assert pathway.title == "CompSci" + + +def test_create_with_explicit_category(org1) -> None: + """Operators can add categories of their own; the default is only a default.""" + masters = PathwayCategory.objects.create(category_code="masters-degree", name="Master's Degree") + pathway = catalog_api.create_catalog_pathway( + org_code="Org1", + pathway_code="CompSci", + title="Computer Science", + category=masters, + ) + assert pathway.category == masters + + +def test_get_catalog_pathway(data_science) -> None: + """A catalog pathway can be looked up by pk, key string, or org + code.""" + assert catalog_api.get_catalog_pathway(pk=data_science.id) == data_science + assert catalog_api.get_catalog_pathway(key_str=data_science.key_str) == data_science + assert catalog_api.get_catalog_pathway(org_code="Org1", pathway_code="DataScience") == data_science + with pytest.raises(CatalogPathway.DoesNotExist): + catalog_api.get_catalog_pathway(org_code="Org1", pathway_code="Nope") + + +def test_update_catalog_pathway_by_id_and_category(data_science) -> None: + """The pathway may be given by ID, and the category can be changed like any other catalog field.""" + masters = PathwayCategory.objects.create(category_code="masters-degree", name="Master's Degree") + catalog_api.update_catalog_pathway(data_science.id, category=masters) + assert catalog_api.get_catalog_pathway(pk=data_science.id).category == masters + + +def test_update_catalog_pathway(data_science) -> None: + """Only the fields passed are changed; the rest are left alone. `modified` records the edit.""" + edited_at = datetime(2026, 3, 1, tzinfo=timezone.utc) + with freeze_time(edited_at): + catalog_api.update_catalog_pathway(data_science, description="Learn even more data science.") + + reloaded = catalog_api.get_catalog_pathway(pk=data_science.id) + assert reloaded.description == "Learn even more data science." + assert reloaded.title == "Data Science Professional Certificate" + assert reloaded.modified == edited_at + + +def test_update_catalog_pathway_with_nothing_to_change(data_science) -> None: + """Passing no fields is a no-op, so `modified` is not bumped.""" + before = catalog_api.get_catalog_pathway(pk=data_science.id).modified + catalog_api.update_catalog_pathway(data_science) + assert catalog_api.get_catalog_pathway(pk=data_science.id).modified == before + + +def test_delete_catalog_pathway(data_science) -> None: + catalog_api.delete_catalog_pathway(data_science.id) + with pytest.raises(CatalogPathway.DoesNotExist): + catalog_api.get_catalog_pathway(pk=data_science.id) + + +def test_set_catalog_pathway_content(data_science, definition) -> None: + """The catalog entry points at its definition; the link can be set by instance or ID, and cleared with None.""" + catalog_api.set_catalog_pathway_content(data_science, definition) + assert catalog_api.get_catalog_pathway(pk=data_science.id).content_entity == definition + + catalog_api.set_catalog_pathway_content(data_science.id, None) + assert catalog_api.get_catalog_pathway(pk=data_science.id).content_entity is None + + catalog_api.set_catalog_pathway_content(data_science.id, definition.id) + assert catalog_api.get_catalog_pathway(pk=data_science.id).content_entity == definition + + +def test_deleting_catalog_pathway_leaves_definition_in_place(data_science, definition) -> None: + """Deleting the catalog half unlinks the definition; it does not delete it.""" + catalog_api.set_catalog_pathway_content(data_science, definition) + catalog_api.delete_catalog_pathway(data_science) + assert PublishableEntity.objects.filter(id=definition.id).exists() + + +def test_enrollment_round_trip(data_science, learner) -> None: + assert not catalog_api.is_enrolled_in_pathway(learner.id, data_science) + + enrollment = catalog_api.enroll_in_pathway(learner.id, data_science) + assert catalog_api.is_enrolled_in_pathway(learner.id, data_science) + assert list(catalog_api.get_pathway_enrollments(learner.id)) == [enrollment] + + catalog_api.unenroll_from_pathway(learner.id, data_science) + assert not catalog_api.is_enrolled_in_pathway(learner.id, data_science) + + +def test_enrolling_twice_is_idempotent(data_science, learner) -> None: + """Enrolling again returns the existing enrollment rather than failing, and doesn't touch `modified`.""" + with freeze_time(datetime(2026, 1, 1, tzinfo=timezone.utc)): + first = catalog_api.enroll_in_pathway(learner.id, data_science) + with freeze_time(datetime(2026, 2, 1, tzinfo=timezone.utc)): + second = catalog_api.enroll_in_pathway(learner.id, data_science.id) + assert first == second + assert second.modified == datetime(2026, 1, 1, tzinfo=timezone.utc) + assert catalog_api.get_pathway_enrollments(learner.id).count() == 1 + + +def test_unenrolling_keeps_the_row(data_science, learner) -> None: + """Unenrolling deactivates rather than deletes, so history survives.""" + enrollment = catalog_api.enroll_in_pathway(learner.id, data_science) + catalog_api.unenroll_from_pathway(learner.id, data_science) + + assert not catalog_api.is_enrolled_in_pathway(learner.id, data_science) + assert catalog_api.get_pathway_enrollments(learner.id).count() == 0 + inactive = catalog_api.get_pathway_enrollments(learner.id, include_inactive=True) + assert list(inactive) == [enrollment] + assert not inactive[0].is_active + + +def test_re_enrolling_reactivates_the_same_row(data_science, learner) -> None: + """ + Re-enrolling reuses the original row, keeping the original enrollment date. `modified` tracks each flip of + `is_active`, so it ends up as "when this learner last enrolled or unenrolled". + """ + enrolled_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + unenrolled_at = datetime(2026, 2, 1, tzinfo=timezone.utc) + re_enrolled_at = datetime(2026, 3, 1, tzinfo=timezone.utc) + + with freeze_time(enrolled_at): + original = catalog_api.enroll_in_pathway(learner.id, data_science) + assert original.modified == enrolled_at + + with freeze_time(unenrolled_at): + catalog_api.unenroll_from_pathway(learner.id, data_science) + inactive = catalog_api.get_pathway_enrollments(learner.id, include_inactive=True).get() + assert not inactive.is_active + assert inactive.modified == unenrolled_at + + with freeze_time(re_enrolled_at): + reactivated = catalog_api.enroll_in_pathway(learner.id, data_science) + assert reactivated.id == original.id + assert reactivated.is_active + assert reactivated.created == enrolled_at + assert reactivated.modified == re_enrolled_at + assert catalog_api.is_enrolled_in_pathway(learner.id, data_science) + + +def test_unenrolling_when_not_enrolled_is_a_no_op(data_science, learner) -> None: + catalog_api.unenroll_from_pathway(learner.id, data_science) # Should not raise. + assert not catalog_api.is_enrolled_in_pathway(learner.id, data_science) + assert catalog_api.get_pathway_enrollments(learner.id, include_inactive=True).count() == 0 + + +def test_unenrolling_twice_does_not_bump_modified(data_science, learner) -> None: + """Unenrolling only touches rows that are actually active, so a repeat call leaves `modified` alone.""" + catalog_api.enroll_in_pathway(learner.id, data_science) + with freeze_time(datetime(2026, 2, 1, tzinfo=timezone.utc)): + catalog_api.unenroll_from_pathway(learner.id, data_science) + with freeze_time(datetime(2026, 3, 1, tzinfo=timezone.utc)): + catalog_api.unenroll_from_pathway(learner.id, data_science) + + row = catalog_api.get_pathway_enrollments(learner.id, include_inactive=True).get() + assert row.modified == datetime(2026, 2, 1, tzinfo=timezone.utc) + + +def test_deleting_pathway_removes_enrollments(data_science, learner) -> None: + catalog_api.enroll_in_pathway(learner.id, data_science) + catalog_api.delete_catalog_pathway(data_science) + assert catalog_api.get_pathway_enrollments(learner.id).count() == 0 diff --git a/tests/openedx_catalog/test_pathway_models.py b/tests/openedx_catalog/test_pathway_models.py new file mode 100644 index 000000000..cb718274a --- /dev/null +++ b/tests/openedx_catalog/test_pathway_models.py @@ -0,0 +1,245 @@ +""" +Tests related to the catalog half of Pathways. +""" +# pylint: disable=unused-argument +# mypy: disable-error-code="misc" +# (Ignore 'Unexpected attribute "org_code" for model "CatalogPathway"' until +# https://github.com/typeddjango/django-stubs/issues/1034 is fixed.) + +from datetime import datetime, timezone + +import pytest +from django.contrib.auth import get_user_model +from django.db import transaction +from django.db.models import ProtectedError +from django.db.utils import IntegrityError +from freezegun import freeze_time +from organizations.api import ensure_organization # type: ignore[import] +from organizations.models import Organization # type: ignore[import] + +from openedx_catalog.models import CatalogPathway, PathwayCategory, PathwayEnrollment +from openedx_catalog.models.pathway_category import DEFAULT_PATHWAY_CATEGORY_CODE, DEFAULT_PATHWAY_CATEGORY_NAME +from openedx_content import api as content_api +from openedx_content.models_api import PublishableEntity + +User = get_user_model() + +pytestmark = pytest.mark.django_db + + +@pytest.fixture(name="org1") +def _org1() -> None: + """Create an "Org1" organization for use in these tests""" + ensure_organization("Org1") + + +@pytest.fixture(name="org2") +def _org2() -> None: + """Create an "Org2" organization for use in these tests""" + ensure_organization("Org2") + + +@pytest.fixture(name="data_science") +def _data_science(org1) -> CatalogPathway: + """Create a CatalogPathway for use in these tests""" + return CatalogPathway.objects.create(org_code="Org1", pathway_code="DataScience") + + +@pytest.fixture(name="learner") +def _learner(): + """Create a learner for use in these tests""" + return User.objects.create(username="learner", email="learner@example.com") + + +@pytest.fixture(name="definition") +def _definition() -> PublishableEntity: + """ + Create a bare PublishableEntity to stand in for a Pathway definition. + + The catalog can't tell a Pathway entity from any other, so a bare entity exercises the same constraints. + """ + package = content_api.create_learning_package(package_ref="pathway-tests", title="Pathway tests") + return content_api.create_publishable_entity( + package.id, "pathway:DataScience", datetime(2026, 1, 1, tzinfo=timezone.utc), None + ) + + +# PathwayCategory + + +def test_default_category_is_shipped() -> None: + """ + The default category is a database row, not a fallback in code, so that operators can rename it without a code + change (ADR 0007, decision 2). + """ + category = PathwayCategory.objects.get(category_code=DEFAULT_PATHWAY_CATEGORY_CODE) + assert category.name == DEFAULT_PATHWAY_CATEGORY_NAME + + +def test_category_is_always_provided(org1) -> None: + """A CatalogPathway created without a category gets the default one.""" + pathway = CatalogPathway.objects.create(org_code="Org1", pathway_code="NoCategory") + assert pathway.category.category_code == DEFAULT_PATHWAY_CATEGORY_CODE + + +def test_default_category_can_be_renamed(org1) -> None: + """ + Renaming the default changes what learners see, and nothing else. The code stays put, so existing pathways keep + pointing at the same row. + """ + category = PathwayCategory.objects.get(category_code=DEFAULT_PATHWAY_CATEGORY_CODE) + category.name = "Program" + category.save() + + pathway = CatalogPathway.objects.create(org_code="Org1", pathway_code="Renamed") + assert pathway.category.name == "Program" + assert pathway.category.category_code == DEFAULT_PATHWAY_CATEGORY_CODE + + +def test_category_code_unique_ci() -> None: + """Category codes are case-insensitively unique.""" + PathwayCategory.objects.create(category_code="masters-degree", name="Master's Degree") + with pytest.raises(IntegrityError), transaction.atomic(): + PathwayCategory.objects.create(category_code="Masters-Degree", name="Duplicate") + + +def test_category_name_cannot_be_blank() -> None: + """The learner-facing name is required at the database level.""" + with pytest.raises(IntegrityError), transaction.atomic(): + PathwayCategory.objects.create(category_code="blank-name", name="") + + +def test_category_in_use_cannot_be_deleted(data_science) -> None: + """Deleting a category out from under a pathway would leave it without one.""" + with pytest.raises(ProtectedError): + data_science.category.delete() + + +def test_category_string_representation() -> None: + """The string representation of a category is its name.""" + category = PathwayCategory.objects.get(category_code=DEFAULT_PATHWAY_CATEGORY_CODE) + assert str(category) == DEFAULT_PATHWAY_CATEGORY_NAME + +# CatalogPathway + + +def test_invalid_org() -> None: + """The Organization must exist in the DB before a CatalogPathway can be created""" + with pytest.raises(Organization.DoesNotExist): + CatalogPathway.objects.create(org_code="NewOrg", pathway_code="Whatever") + + +def test_pathway_code_unique_per_org_ci(org1, org2) -> None: + """The pathway_code is case-insensitively unique per org, but not across orgs.""" + CatalogPathway.objects.create(org_code="Org1", pathway_code="DataScience") + with pytest.raises(IntegrityError), transaction.atomic(): + CatalogPathway.objects.create(org_code="Org1", pathway_code="datascience") + # A different org may use the same code: + CatalogPathway.objects.create(org_code="Org2", pathway_code="DataScience") + + +def test_title_defaults_to_pathway_code(data_science) -> None: + """A blank title falls back to the code, rather than failing the not-blank constraint.""" + assert data_science.title == "DataScience" + + +def test_key_str(data_science) -> None: + """The key is derived from the org and pathway codes.""" + assert data_science.key_str == "catalog-pathway:Org1:DataScience" + + +def test_catalog_edits_are_free(data_science) -> None: + """ + Catalog copy is not versioned. Editing it is an ordinary save, with no version to create and no trace left behind. + """ + data_science.title = "Data Science Professional Program" + data_science.description = "Learn data science." + data_science.save() + + reloaded = CatalogPathway.objects.get(pk=data_science.pk) + assert reloaded.title == "Data Science Professional Program" + assert reloaded.description == "Learn data science." + + +def test_modified_tracks_catalog_edits(org1) -> None: + """ + `modified` moves when the catalog fields change and `created` does not. + """ + created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + edited_at = datetime(2026, 2, 1, tzinfo=timezone.utc) + with freeze_time(created_at): + pathway = CatalogPathway.objects.create(org_code="Org1", pathway_code="Timestamps") + assert pathway.created == created_at + assert pathway.modified == created_at + + with freeze_time(edited_at): + pathway.title = "Renamed" + pathway.save() + + reloaded = CatalogPathway.objects.get(pk=pathway.pk) + assert reloaded.created == created_at + assert reloaded.modified == edited_at + + +def test_pathway_string_representation(data_science) -> None: + """Test the string representation of a pathway.""" + data_science.title = "Data Science Professional Program" + data_science.save() + data_science.refresh_from_db() + assert str(data_science) == "Data Science Professional Program (Org1 DataScience)" + + +def test_content_entity_is_optional(data_science) -> None: + """A catalog pathway may exist as a placeholder before its definition does.""" + assert data_science.content_entity is None + + +def test_one_catalog_pathway_per_definition(org1, definition) -> None: + """A definition serves exactly one catalog entry.""" + CatalogPathway.objects.create(org_code="Org1", pathway_code="First", content_entity=definition) + with pytest.raises(IntegrityError), transaction.atomic(): + CatalogPathway.objects.create(org_code="Org1", pathway_code="Second", content_entity=definition) + + +def test_linked_definition_cannot_be_deleted(data_science, definition) -> None: + """PROTECT: deleting a definition out from under a catalog entry would leave its learners with no requirements.""" + data_science.content_entity = definition + data_science.save() + with pytest.raises(ProtectedError), transaction.atomic(): + definition.delete() + + data_science.content_entity = None + data_science.save() + definition.delete() # Unlinked, so this is fine. + + +# PathwayEnrollment + + +def test_enrollment_is_unique_per_learner(data_science, learner) -> None: + """A learner is either enrolled in a pathway or not; there is never a second row.""" + PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science) + with pytest.raises(IntegrityError), transaction.atomic(): + PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science) + + +def test_enrollment_pins_no_version(data_science, learner) -> None: + """ + Enrollment ties a learner to the catalog half only. There is deliberately no field pinning a content version, + because progress is evaluated against whatever is published at the time (ADR 0007, decision 5). + """ + enrollment = PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science) + field_names = {field.name for field in enrollment._meta.get_fields()} + assert not any("version" in name for name in field_names) + + +def test_enrollment_is_active_by_default(data_science, learner) -> None: + """A fresh enrollment is active; deactivating it is how unenrolling is recorded.""" + enrollment = PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science) + assert enrollment.is_active + + +def test_enrollment_string_representation(data_science, learner) -> None: + """Test the string representation of a pathway enrollment.""" + enrollment = PathwayEnrollment.objects.create(user=learner, catalog_pathway=data_science) + assert str(enrollment) == f"{learner} in {data_science}"