From fa1d75b6aa37d2614072041c7ed6a0849d807cf3 Mon Sep 17 00:00:00 2001
From: Agrendalath <piotr@surowiec.it>
Date: Mon, 14 Sep 2026 22:41:42 +0200
Subject: [PATCH] feat(pathways): implement CatalogPathway

---
 .importlinter                                 |  15 +-
 src/openedx_catalog/admin.py                  |  91 +++++-
 src/openedx_catalog/api_impl.py               | 214 ++++++++++++-
 .../migrations/0002_pathways.py               | 286 ++++++++++++++++++
 src/openedx_catalog/models/__init__.py        |   3 +
 src/openedx_catalog/models/catalog_pathway.py | 182 +++++++++++
 .../models/pathway_category.py                |  99 ++++++
 .../models/pathway_enrollment.py              |  88 ++++++
 src/openedx_catalog/models_api.py             |   2 +-
 tests/openedx_catalog/test_pathway_api.py     | 200 ++++++++++++
 tests/openedx_catalog/test_pathway_models.py  | 206 +++++++++++++
 11 files changed, 1379 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..02e191db2 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,14 +19,20 @@ 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 content, catalog 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.
+    # Content: authoring-side models and APIs. May reference catalog entities, e.g. to record
+    # which content implements a given CourseRun.
     openedx_content
 
+    # Catalog: the enrollable things (CatalogCourse, CourseRun, CatalogPathway) that content and
+    # everything above it may point at. It stays unaware of content, so that changes in how
+    # content is represented never reach it. See the openedx_catalog ADR 0001, decision 2.
+    openedx_catalog
+
     # Tagging is very simple & fundamental. Should probably not depend on any other Django apps.
     openedx_tagging
 
diff --git a/src/openedx_catalog/admin.py b/src/openedx_catalog/admin.py
index 2aa4f3ca6..64a8bb596 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,89 @@ 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('<a href="{}">{}</a>', 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", "created_date", "modified"]
+    search_fields = ["title", "pathway_code"]
+
+    def get_readonly_fields(self, request, obj: CatalogPathway | None = None) -> tuple[str, ...]:
+        if obj:  # editing an existing object
+            return ("org", "pathway_code")
+        return tuple()
+
+    @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 <select>
+    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..354c95d1f 100644
--- a/src/openedx_catalog/api_impl.py
+++ b/src/openedx_catalog/api_impl.py
@@ -5,11 +5,15 @@
 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 .models import CatalogCourse, CatalogPathway, CourseRun, PathwayCategory, PathwayEnrollment
+from .models.pathway_category import get_default_pathway_category
 
 log = logging.getLogger(__name__)
 
@@ -22,6 +26,16 @@
     "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",
+    "delete_catalog_pathway",
+    "enroll_in_pathway",
+    "unenroll_from_pathway",
+    "is_enrolled_in_pathway",
+    "get_pathway_enrollments",
 ]
 
 
@@ -249,3 +263,201 @@ 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`.
+
+
+# `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 any content implementing it yet. Because `openedx_catalog` never knows about the
+    content side, this function cannot tell you whether it does; ask `openedx_learning.api` instead.
+    """
+    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 content that
+    implements it.
+
+    ⚠️ 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 delete_catalog_pathway(catalog_pathway: CatalogPathway | CatalogPathway.ID) -> None:
+    """
+    Delete a `CatalogPathway`, along with its enrollments.
+
+    This will fail with a `ProtectedError` if any Pathway content still implements it, because that link is a `PROTECT`
+    foreign key on the content side.
+
+    ⚠️ 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..c25ca530d
--- /dev/null
+++ b/src/openedx_catalog/migrations/0002_pathways.py
@@ -0,0 +1,286 @@
+"""
+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"),
+        ("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",
+                    ),
+                ),
+            ],
+            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..f2310ef15
--- /dev/null
+++ b/src/openedx_catalog/models/catalog_pathway.py
@@ -0,0 +1,182 @@
+"""
+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_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 link between the two halves lives on the content side (`PathwayVersion.catalog_pathway`), never here: this app
+    stays unaware of content, so that changes in how content is represented never reach it (see the openedx_catalog ADR
+    0001, decision 2). A `CatalogPathway` therefore cannot, on its own, tell you which content implements it; queries in
+    that direction start from the content side.
+
+    A `CatalogPathway` may exist before any content implements it, 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."),
+    )
+    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..745d4f6d6
--- /dev/null
+++ b/tests/openedx_catalog/test_pathway_api.py
@@ -0,0 +1,200 @@
+"""
+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
+
+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")
+
+
+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_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..20714c02f
--- /dev/null
+++ b/tests/openedx_catalog/test_pathway_models.py
@@ -0,0 +1,206 @@
+"""
+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
+
+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")
+
+
+# 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)"
+
+
+# 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}"