feat: bulk add courses to base catalog

partner_catalog/admin.py

@@ -2,13 +2,17 @@
 
 from urllib.parse import urlencode
 
+from django import forms
 from django.contrib import admin
+from django.contrib.admin.widgets import FilteredSelectMultiple
+from django.core.exceptions import FieldDoesNotExist
 from django.db.models import Count
 from django.http import HttpResponseRedirect
 from django.urls import reverse
 from django.utils.html import format_html
 
 from flex_catalog.admin import CourseKeysMixin
+from partner_catalog.edxapp_wrapper.course_module import course_overview
 from partner_catalog.models import (
     BaseCatalog,
     BaseCatalogCourse,
@@ -23,81 +27,103 @@
 )
 
 
+class BaseCatalogAdminForm(forms.ModelForm):
+    """ModelForm for BaseCatalog with an inline dual-list course manager."""
+
+    courses = forms.ModelMultipleChoiceField(
+        queryset=None,
+        widget=FilteredSelectMultiple("Courses", is_stacked=False),
+        required=False,
+        label="",
+        help_text=(
+            "Manage courses in this catalog. "
+            "Use the search box to filter, Ctrl+click to select multiple, "
+            "then use the arrow buttons to add or remove them. "
+            "Saving will apply all additions and removals at once."
+        ),
+    )
+
+    def __init__(self, *args, **kwargs):
+        """Pre-populate the courses field with the catalog's current courses."""
+        super().__init__(*args, **kwargs)
+        CourseOverview = course_overview()
+        try:
+            CourseOverview._meta.get_field('display_name')
+            qs = CourseOverview.objects.order_by('display_name')
+        except FieldDoesNotExist:
+            qs = CourseOverview.objects.all()
+        self.fields['courses'].queryset = qs
+        if self.instance.pk:
+            self.fields['courses'].initial = self.instance.courses.all()
+
+    class Meta:
+        """Meta options for BaseCatalogAdminForm."""
+
+        model = BaseCatalog
+        fields = '__all__'
+
+
 @admin.register(BaseCatalog)
 class BaseCatalogAdmin(admin.ModelAdmin):
     """Admin interface for BaseCatalog model."""
 
-    list_display = ('name', 'slug', 'course_count', 'course_ids', 'add_course')
-    readonly_fields = ('created', 'modified', 'course_count', 'course_ids_display', 'add_course_button')
-    fields = ('name', 'slug', 'created', 'modified', 'course_count', 'course_ids_display', 'add_course_button')
+    form = BaseCatalogAdminForm
+
+    list_display = ('name', 'slug', 'course_count', 'course_ids', 'manage_courses')
+    readonly_fields = ('created', 'modified', 'course_count')
     search_fields = ('name', 'slug')
 
+    def get_fields(self, request, obj=None):
+        """Return fields for the change form, including the custom courses widget."""
+        return ('name', 'slug', 'created', 'modified', 'course_count', 'courses')
+
     def get_queryset(self, request):
         """Optimize queryset with prefetch."""
         qs = super().get_queryset(request)
         return qs.prefetch_related('courses', 'base_catalog_courses')
 
+    def save_related(self, request, form, formsets, change):
+        """Sync the courses M2M: add newly selected courses and remove deselected ones."""
+        super().save_related(request, form, formsets, change)
+
+        selected_courses = set(form.cleaned_data.get('courses', []))
+        selected_ids = {course.pk for course in selected_courses}
+
+        current_entries = form.instance.base_catalog_courses.select_related('course_overview')
+        current_ids = {entry.course_overview_id for entry in current_entries}
+
+        for course in selected_courses:
+            if course.pk not in current_ids:
+                BaseCatalogCourse.objects.create(
+                    base_catalog=form.instance,
+                    course_overview=course,
+                    added_by=request.user,
+                )
+
+        form.instance.base_catalog_courses.filter(
+            course_overview_id__in=current_ids - selected_ids
+        ).delete()
+
     def course_count(self, obj):
         """Display the total number of courses in the catalog."""
         return obj.courses.count()
     course_count.short_description = 'Total Courses'
 
     def course_ids(self, obj):
-        """Display preview of course IDs in the list view."""
+        """Display course IDs in the list view."""
         course_runs = obj.get_course_runs()
-
         if course_runs:
-            course_ids = [str(course.id) for course in course_runs]
-            return format_html('<br>'.join(course_ids))
-
+            return format_html('<br>'.join(str(c.id) for c in course_runs))
         return format_html('<em style="color: #999;">No courses</em>')
-
     course_ids.short_description = 'Course IDs'
 
-    def course_ids_display(self, obj):
-        """Display all course IDs in the detail view."""
-        course_runs = obj.get_course_runs()
-
-        if course_runs:
-            course_ids = [str(course.id) for course in course_runs]
-            return format_html('<br>'.join(course_ids))
-
-        return format_html('<em style="color: #999;">No courses</em>')
-
-    course_ids_display.short_description = 'Course IDs'
-
-    def add_course_button(self, obj):
-        """Genera un botón para agregar un nuevo curso a este BaseCatalog."""
+    def manage_courses(self, obj):
+        """Link to the catalog change page to manage its courses."""
         if not obj.pk:
-            return format_html('<em style="color: #999;">Guarda el catálogo primero</em>')
-
-        course_model = BaseCatalogCourse
-        add_course_url = reverse(
-            f"admin:{course_model._meta.app_label}_{course_model._meta.model_name}_add"
-        )
-        full_url = f"{add_course_url}?base_catalog={obj.pk}"
-
-        return format_html(
-            '<a href="{}" class="button" style="background-color: #417690;">+ Add Course</a>',
-            full_url,
-        )
-
-    add_course_button.short_description = "Add Courses"
-
-    def add_course(self, obj):
-        """Genera un link para agregar un nuevo curso a este BaseCatalog."""
-        course_model = BaseCatalogCourse
-        add_course_url = reverse(
-            f"admin:{course_model._meta.app_label}_{course_model._meta.model_name}_add"
-        )
-        full_url = f"{add_course_url}?base_catalog={obj.pk}"
-
-        return format_html(
-            '<a href="{}" style="font-weight: bold;">+ Add Course</a>',
-            full_url,
-        )
-
-    add_course.short_description = "+ Add Course"
+            return format_html('')
+        url = reverse('admin:partner_catalog_basecatalog_change', args=[obj.pk])
+        return format_html('<a href="{}" style="font-weight:bold;">Manage Courses</a>', url)
+    manage_courses.short_description = 'Manage Courses'
 
 
 @admin.register(BaseCatalogCourse)

partner_catalog/migrations/0010_basecatalogcourse_unique_base_catalog_course.py

@@ -0,0 +1,21 @@
+"""Migration to add a uniqueness constraint on (base_catalog, course_overview) for BaseCatalogCourse."""
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    """Add UniqueConstraint to prevent duplicate BaseCatalogCourse entries."""
+
+    dependencies = [
+        ("partner_catalog", "0009_alter_catalogcourseenrollment_course_overview"),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name="basecatalogcourse",
+            constraint=models.UniqueConstraint(
+                fields=["base_catalog", "course_overview"],
+                name="unique_base_catalog_course",
+            ),
+        ),
+    ]

partner_catalog/models.py

@@ -95,6 +95,12 @@ class Meta:
         verbose_name = "Base Catalog Course"
         verbose_name_plural = "Base Catalog Courses"
         ordering = ["-added_at"]
+        constraints = [
+            models.UniqueConstraint(
+                fields=["base_catalog", "course_overview"],
+                name="unique_base_catalog_course",
+            )
+        ]
 
     def __str__(self):
         """Return string representation."""

tests/test_base_catalog_admin.py

@@ -0,0 +1,178 @@
+"""
+Tests for BaseCatalog admin — form pre-population and save_related sync logic (Suite 10).
+
+Covers:
+- BaseCatalogAdminForm.__init__: courses queryset is set on all forms
+- BaseCatalogAdminForm.__init__: courses initial is empty for a new catalog
+- BaseCatalogAdminForm.__init__: courses initial is pre-populated for an existing catalog
+- BaseCatalogAdmin.save_related: adds newly selected courses
+- BaseCatalogAdmin.save_related: removes deselected courses
+- BaseCatalogAdmin.save_related: no-op when selection matches current state
+- BaseCatalogAdmin.save_related: records added_by from request.user
+- BaseCatalogAdmin.save_related: clears all courses when selection is empty
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+from django.contrib.admin.sites import AdminSite
+
+from partner_catalog.admin import BaseCatalogAdmin, BaseCatalogAdminForm
+from partner_catalog.models import BaseCatalog, BaseCatalogCourse
+from partner_catalog.services.catalog_courses import CourseOverview
+from tests.factories import make_user
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def make_base_catalog(slug_suffix="1"):
+    """Create and return a BaseCatalog for testing."""
+    return BaseCatalog.objects.create(name=f"Test Catalog {slug_suffix}", slug=f"test-catalog-{slug_suffix}")
+
+
+def make_course():
+    """Create and return a CourseOverview (test backend) instance."""
+    return CourseOverview.objects.create()
+
+
+def _admin():
+    """Return a BaseCatalogAdmin instance bound to a fresh AdminSite."""
+    return BaseCatalogAdmin(BaseCatalog, AdminSite())
+
+
+def _request(user=None):
+    """Return a mock request with the given user (or a new staff user)."""
+    req = MagicMock()
+    req.user = user or make_user(is_staff=True)
+    return req
+
+
+def _form(instance, selected_courses):
+    """Return a mock form with cleaned_data and instance set."""
+    frm = MagicMock()
+    frm.instance = instance
+    frm.cleaned_data = {'courses': selected_courses}
+    return frm
+
+
+# ---------------------------------------------------------------------------
+# BaseCatalogAdminForm — __init__ pre-population
+# ---------------------------------------------------------------------------
+
+class TestBaseCatalogAdminFormInit:
+    """Tests for BaseCatalogAdminForm.__init__ initialization behaviour."""
+
+    @pytest.mark.django_db
+    def test_courses_queryset_includes_all_courses(self):
+        """The courses queryset covers all CourseOverview objects."""
+        make_course()
+        make_course()
+
+        form = BaseCatalogAdminForm()
+
+        assert form.fields['courses'].queryset.count() == 2
+
+    @pytest.mark.django_db
+    def test_courses_initial_is_empty_for_new_catalog(self):
+        """Without an existing instance the courses initial is not set."""
+        form = BaseCatalogAdminForm()
+
+        assert not form.fields['courses'].initial
+
+    @pytest.mark.django_db
+    def test_courses_initial_pre_populates_existing_courses(self):
+        """With an existing catalog the initial value matches its current courses."""
+        catalog = make_base_catalog()
+        course1 = make_course()
+        course2 = make_course()
+        BaseCatalogCourse.objects.create(base_catalog=catalog, course_overview=course1)
+        BaseCatalogCourse.objects.create(base_catalog=catalog, course_overview=course2)
+
+        form = BaseCatalogAdminForm(instance=catalog)
+
+        initial_ids = {c.pk for c in form.fields['courses'].initial}
+        assert initial_ids == {course1.pk, course2.pk}
+
+    @pytest.mark.django_db
+    def test_courses_initial_is_empty_for_catalog_with_no_courses(self):
+        """A catalog with no courses yields an empty initial queryset."""
+        catalog = make_base_catalog()
+
+        form = BaseCatalogAdminForm(instance=catalog)
+
+        assert not list(form.fields['courses'].initial)
+
+
+# ---------------------------------------------------------------------------
+# BaseCatalogAdmin.save_related — course sync logic
+# ---------------------------------------------------------------------------
+
+class TestBaseCatalogAdminSaveRelated:
+    """Tests for BaseCatalogAdmin.save_related diff/sync logic."""
+
+    @pytest.mark.django_db
+    def test_adds_newly_selected_courses(self):
+        """save_related creates BaseCatalogCourse entries for newly selected courses."""
+        catalog = make_base_catalog(slug_suffix="a")
+        course1 = make_course()
+        course2 = make_course()
+
+        _admin().save_related(_request(), _form(catalog, [course1, course2]), [], change=True)
+
+        assert catalog.courses.count() == 2
+        assert BaseCatalogCourse.objects.filter(base_catalog=catalog, course_overview=course1).exists()
+        assert BaseCatalogCourse.objects.filter(base_catalog=catalog, course_overview=course2).exists()
+
+    @pytest.mark.django_db
+    def test_removes_deselected_courses(self):
+        """save_related deletes BaseCatalogCourse entries for deselected courses."""
+        catalog = make_base_catalog(slug_suffix="b")
+        course1 = make_course()
+        course2 = make_course()
+        BaseCatalogCourse.objects.create(base_catalog=catalog, course_overview=course1)
+        BaseCatalogCourse.objects.create(base_catalog=catalog, course_overview=course2)
+
+        _admin().save_related(_request(), _form(catalog, [course1]), [], change=True)
+
+        assert catalog.courses.count() == 1
+        assert BaseCatalogCourse.objects.filter(base_catalog=catalog, course_overview=course1).exists()
+        assert not BaseCatalogCourse.objects.filter(base_catalog=catalog, course_overview=course2).exists()
+
+    @pytest.mark.django_db
+    def test_no_op_when_selection_matches_current_state(self):
+        """save_related does not create duplicates when the selection is unchanged."""
+        catalog = make_base_catalog(slug_suffix="c")
+        course = make_course()
+        BaseCatalogCourse.objects.create(base_catalog=catalog, course_overview=course)
+
+        _admin().save_related(_request(), _form(catalog, [course]), [], change=True)
+
+        assert catalog.courses.count() == 1
+        assert BaseCatalogCourse.objects.filter(base_catalog=catalog).count() == 1
+
+    @pytest.mark.django_db
+    def test_records_added_by_from_request_user(self):
+        """save_related sets the added_by field to the current request user."""
+        catalog = make_base_catalog(slug_suffix="d")
+        course = make_course()
+        user = make_user()
+
+        _admin().save_related(_request(user=user), _form(catalog, [course]), [], change=True)
+
+        entry = BaseCatalogCourse.objects.get(base_catalog=catalog, course_overview=course)
+        assert entry.added_by == user
+
+    @pytest.mark.django_db
+    def test_clears_all_courses_when_selection_is_empty(self):
+        """save_related removes all entries when the submitted selection is empty."""
+        catalog = make_base_catalog(slug_suffix="e")
+        course1 = make_course()
+        course2 = make_course()
+        BaseCatalogCourse.objects.create(base_catalog=catalog, course_overview=course1)
+        BaseCatalogCourse.objects.create(base_catalog=catalog, course_overview=course2)
+
+        _admin().save_related(_request(), _form(catalog, []), [], change=True)
+
+        assert not catalog.courses.count()