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 over every PublishableEntity in the system.
+ if obj: # editing an existing object
+ return ("content_entity", "org", "pathway_code")
+ return ("content_entity",)
+
+ @admin.display(description="Organization", ordering="org__short_name")
+ def org_display(self, obj: CatalogPathway) -> str:
+ """Display the organization, only showing the short_name if different from full name"""
+ if obj.org.name == obj.org.short_name:
+ return obj.org.short_name
+ return str(obj.org)
+
+ @admin.display(description=_("Created"), ordering="created")
+ def created_date(self, obj: CatalogPathway) -> datetime.date:
+ """Display the created date without the timestamp"""
+ return obj.created.date()
+
+
+admin.site.register(CatalogPathway, CatalogPathwayAdmin)
+
+
+class PathwayEnrollmentAdmin(admin.ModelAdmin):
+ """
+ The PathwayEnrollment model admin.
+ """
+
+ list_display = ["user", "catalog_pathway", "is_active", "created_date", "modified"]
+ list_filter = ["is_active", "catalog_pathway__category"]
+ # There may be very many users and a fair number of pathways, so don't use
+ 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}"