Improve test coverage for course groups

common/djangoapps/course_groups/cohorts.py

@@ -4,6 +4,7 @@ forums, and to the cohort admin views.
 """
 
 from django.http import Http404
+
 import logging
 import random
 
@@ -86,14 +87,14 @@ def is_commentable_cohorted(course_key, commentable_id):
 
 def get_cohorted_commentables(course_key):
     """
-    Given a course_key return a list of strings representing cohorted commentables
+    Given a course_key return a set of strings representing cohorted commentables.
     """
 
     course = courses.get_course_by_id(course_key)
 
     if not course.is_cohorted:
         # this is the easy case :)
-        ans = []
+        ans = set()
     else:
         ans = course.cohorted_discussions
 
@@ -213,8 +214,13 @@ def add_cohort(course_key, name):
                                       name=name).exists():
         raise ValueError("Can't create two cohorts with the same name")
 
+    try:
+        course = courses.get_course_by_id(course_key)
+    except Http404:
+        raise ValueError("Invalid course_key")
+
     return CourseUserGroup.objects.create(
-        course_id=course_key,
+        course_id=course.id,
         group_type=CourseUserGroup.COHORT,
         name=name
     )
@@ -240,7 +246,6 @@ def add_user_to_cohort(cohort, username_or_email):
 
     Raises:
         User.DoesNotExist if can't find user.
-
         ValueError if user already present in this cohort.
     """
     user = get_user_by_username_or_email(username_or_email)

common/djangoapps/course_groups/tests/test_cohorts.py

 import django.test
 from django.contrib.auth.models import User
 from django.conf import settings
+from django.http import Http404
 
 from django.test.utils import override_settings
 
+from student.models import CourseEnrollment
 from course_groups.models import CourseUserGroup
-from course_groups.cohorts import (get_cohort, get_course_cohorts,
-                                   is_commentable_cohorted, get_cohort_by_name)
+from course_groups import cohorts
 
 from xmodule.modulestore.django import modulestore, clear_existing_modulestores
 from opaque_keys.edx.locations import SlashSeparatedCourseKey
@@ -87,9 +88,49 @@ class TestCohorts(django.test.TestCase):
         clear_existing_modulestores()
         self.toy_course_key = SlashSeparatedCourseKey("edX", "toy", "2012_Fall")
 
+    def test_is_course_cohorted(self):
+        """
+        Make sure cohorts.is_course_cohorted() correctly reports if a course is cohorted or not.
+        """
+        course = modulestore().get_course(self.toy_course_key)
+        self.assertFalse(course.is_cohorted)
+        self.assertFalse(cohorts.is_course_cohorted(course.id))
+
+        self.config_course_cohorts(course, [], cohorted=True)
+
+        self.assertTrue(course.is_cohorted)
+        self.assertTrue(cohorts.is_course_cohorted(course.id))
+
+        # Make sure we get a Http404 if there's no course
+        fake_key = SlashSeparatedCourseKey('a', 'b', 'c')
+        self.assertRaises(Http404, lambda: cohorts.is_course_cohorted(fake_key))
+
+    def test_get_cohort_id(self):
+        """
+        Make sure that cohorts.get_cohort_id() correctly returns the cohort id, or raises a ValueError when given an
+        invalid course key.
+        """
+        course = modulestore().get_course(self.toy_course_key)
+        self.assertFalse(course.is_cohorted)
+
+        user = User.objects.create(username="test", email="a@b.com")
+        self.assertIsNone(cohorts.get_cohort_id(user, course.id))
+
+        self.config_course_cohorts(course, [], cohorted=True)
+        cohort = CourseUserGroup.objects.create(name="TestCohort",
+                                                course_id=course.id,
+                                                group_type=CourseUserGroup.COHORT)
+        cohort.users.add(user)
+        self.assertEqual(cohorts.get_cohort_id(user, course.id), cohort.id)
+
+        self.assertRaises(
+            ValueError,
+            lambda: cohorts.get_cohort_id(user, SlashSeparatedCourseKey("course", "does_not", "exist"))
+        )
+
     def test_get_cohort(self):
         """
-        Make sure get_cohort() does the right thing when the course is cohorted
+        Make sure cohorts.get_cohort() does the right thing when the course is cohorted
         """
         course = modulestore().get_course(self.toy_course_key)
         self.assertEqual(course.id, self.toy_course_key)
@@ -98,7 +139,7 @@ class TestCohorts(django.test.TestCase):
         user = User.objects.create(username="test", email="a@b.com")
         other_user = User.objects.create(username="test2", email="a2@b.com")
 
-        self.assertIsNone(get_cohort(user, course.id), "No cohort created yet")
+        self.assertIsNone(cohorts.get_cohort(user, course.id), "No cohort created yet")
 
         cohort = CourseUserGroup.objects.create(name="TestCohort",
                                                 course_id=course.id,
@@ -106,21 +147,21 @@ class TestCohorts(django.test.TestCase):
 
         cohort.users.add(user)
 
-        self.assertIsNone(get_cohort(user, course.id),
+        self.assertIsNone(cohorts.get_cohort(user, course.id),
                           "Course isn't cohorted, so shouldn't have a cohort")
 
         # Make the course cohorted...
         self.config_course_cohorts(course, [], cohorted=True)
 
-        self.assertEquals(get_cohort(user, course.id).id, cohort.id,
+        self.assertEquals(cohorts.get_cohort(user, course.id).id, cohort.id,
                           "Should find the right cohort")
 
-        self.assertEquals(get_cohort(other_user, course.id), None,
+        self.assertEquals(cohorts.get_cohort(other_user, course.id), None,
                           "other_user shouldn't have a cohort")
 
     def test_auto_cohorting(self):
         """
-        Make sure get_cohort() does the right thing when the course is auto_cohorted
+        Make sure cohorts.get_cohort() does the right thing when the course is auto_cohorted
         """
         course = modulestore().get_course(self.toy_course_key)
         self.assertFalse(course.is_cohorted)
@@ -141,10 +182,10 @@ class TestCohorts(django.test.TestCase):
                                    auto_cohort=True,
                                    auto_cohort_groups=["AutoGroup"])
 
-        self.assertEquals(get_cohort(user1, course.id).id, cohort.id,
+        self.assertEquals(cohorts.get_cohort(user1, course.id).id, cohort.id,
                           "user1 should stay put")
 
-        self.assertEquals(get_cohort(user2, course.id).name, "AutoGroup",
+        self.assertEquals(cohorts.get_cohort(user2, course.id).name, "AutoGroup",
                           "user2 should be auto-cohorted")
 
         # Now make the group list empty
@@ -152,7 +193,7 @@ class TestCohorts(django.test.TestCase):
                                    auto_cohort=True,
                                    auto_cohort_groups=[])
 
-        self.assertEquals(get_cohort(user3, course.id), None,
+        self.assertEquals(cohorts.get_cohort(user3, course.id), None,
                           "No groups->no auto-cohorting")
 
         # Now make it different
@@ -160,14 +201,14 @@ class TestCohorts(django.test.TestCase):
                                    auto_cohort=True,
                                    auto_cohort_groups=["OtherGroup"])
 
-        self.assertEquals(get_cohort(user3, course.id).name, "OtherGroup",
+        self.assertEquals(cohorts.get_cohort(user3, course.id).name, "OtherGroup",
                           "New list->new group")
-        self.assertEquals(get_cohort(user2, course.id).name, "AutoGroup",
+        self.assertEquals(cohorts.get_cohort(user2, course.id).name, "AutoGroup",
                           "user2 should still be in originally placed cohort")
 
     def test_auto_cohorting_randomization(self):
         """
-        Make sure get_cohort() randomizes properly.
+        Make sure cohorts.get_cohort() randomizes properly.
         """
         course = modulestore().get_course(self.toy_course_key)
         self.assertFalse(course.is_cohorted)
@@ -181,14 +222,14 @@ class TestCohorts(django.test.TestCase):
         for i in range(100):
             user = User.objects.create(username="test_{0}".format(i),
                                        email="a@b{0}.com".format(i))
-            get_cohort(user, course.id)
+            cohorts.get_cohort(user, course.id)
 
         # Now make sure that the assignment was at least vaguely random:
         # each cohort should have at least 1, and fewer than 50 students.
         # (with 5 groups, probability of 0 users in any group is about
         # .8**100= 2.0e-10)
         for cohort_name in groups:
-            cohort = get_cohort_by_name(course.id, cohort_name)
+            cohort = cohorts.get_cohort_by_name(course.id, cohort_name)
             num_users = cohort.users.count()
             self.assertGreater(num_users, 1)
             self.assertLess(num_users, 50)
@@ -207,10 +248,10 @@ class TestCohorts(django.test.TestCase):
                                                 group_type=CourseUserGroup.COHORT)
 
         # second course should have no cohorts
-        self.assertEqual(get_course_cohorts(course2_key), [])
+        self.assertEqual(cohorts.get_course_cohorts(course2_key), [])
 
-        cohorts = sorted([c.name for c in get_course_cohorts(course1_key)])
-        self.assertEqual(cohorts, ['TestCohort', 'TestCohort2'])
+        course1_cohorts = sorted([c.name for c in cohorts.get_course_cohorts(course1_key)])
+        self.assertEqual(course1_cohorts, ['TestCohort', 'TestCohort2'])
 
     def test_is_commentable_cohorted(self):
         course = modulestore().get_course(self.toy_course_key)
@@ -220,14 +261,14 @@ class TestCohorts(django.test.TestCase):
             return self.topic_name_to_id(course, name)
 
         # no topics
-        self.assertFalse(is_commentable_cohorted(course.id, to_id("General")),
+        self.assertFalse(cohorts.is_commentable_cohorted(course.id, to_id("General")),
                          "Course doesn't even have a 'General' topic")
 
         # not cohorted
         self.config_course_cohorts(course, ["General", "Feedback"],
                                    cohorted=False)
 
-        self.assertFalse(is_commentable_cohorted(course.id, to_id("General")),
+        self.assertFalse(cohorts.is_commentable_cohorted(course.id, to_id("General")),
                          "Course isn't cohorted")
 
         # cohorted, but top level topics aren't
@@ -235,11 +276,11 @@ class TestCohorts(django.test.TestCase):
                                    cohorted=True)
 
         self.assertTrue(course.is_cohorted)
-        self.assertFalse(is_commentable_cohorted(course.id, to_id("General")),
+        self.assertFalse(cohorts.is_commentable_cohorted(course.id, to_id("General")),
                          "Course is cohorted, but 'General' isn't.")
 
         self.assertTrue(
-            is_commentable_cohorted(course.id, to_id("random")),
+            cohorts.is_commentable_cohorted(course.id, to_id("random")),
             "Non-top-level discussion is always cohorted in cohorted courses.")
 
         # cohorted, including "Feedback" top-level topics aren't
@@ -248,9 +289,236 @@ class TestCohorts(django.test.TestCase):
                                    cohorted_discussions=["Feedback"])
 
         self.assertTrue(course.is_cohorted)
-        self.assertFalse(is_commentable_cohorted(course.id, to_id("General")),
+        self.assertFalse(cohorts.is_commentable_cohorted(course.id, to_id("General")),
                          "Course is cohorted, but 'General' isn't.")
 
         self.assertTrue(
-            is_commentable_cohorted(course.id, to_id("Feedback")),
+            cohorts.is_commentable_cohorted(course.id, to_id("Feedback")),
             "Feedback was listed as cohorted.  Should be.")
+
+    def test_get_cohorted_commentables(self):
+        """
+        Make sure cohorts.get_cohorted_commentables() correctly returns a list of strings representing cohorted
+        commentables.  Also verify that we can't get the cohorted commentables from a course which does not exist.
+        """
+        course = modulestore().get_course(self.toy_course_key)
+
+        self.assertEqual(cohorts.get_cohorted_commentables(course.id), set())
+
+        self.config_course_cohorts(course, [], cohorted=True)
+        self.assertEqual(cohorts.get_cohorted_commentables(course.id), set())
+
+        self.config_course_cohorts(
+            course, ["General", "Feedback"],
+            cohorted=True,
+            cohorted_discussions=["Feedback"]
+        )
+        self.assertItemsEqual(
+            cohorts.get_cohorted_commentables(course.id),
+            set([self.topic_name_to_id(course, "Feedback")])
+        )
+
+        self.config_course_cohorts(
+            course, ["General", "Feedback"],
+            cohorted=True,
+            cohorted_discussions=["General", "Feedback"]
+        )
+        self.assertItemsEqual(
+            cohorts.get_cohorted_commentables(course.id),
+            set([self.topic_name_to_id(course, "General"),
+                 self.topic_name_to_id(course, "Feedback")])
+        )
+        self.assertRaises(
+            Http404,
+            lambda: cohorts.get_cohorted_commentables(SlashSeparatedCourseKey("course", "does_not", "exist"))
+        )
+
+    def test_get_cohort_by_name(self):
+        """
+        Make sure cohorts.get_cohort_by_name() properly finds a cohort by name for a given course.  Also verify that it
+        raises an error when the cohort is not found.
+        """
+        course = modulestore().get_course(self.toy_course_key)
+
+        self.assertRaises(
+            CourseUserGroup.DoesNotExist,
+            lambda: cohorts.get_cohort_by_name(course.id, "CohortDoesNotExist")
+        )
+
+        cohort = CourseUserGroup.objects.create(
+            name="MyCohort",
+            course_id=course.id,
+            group_type=CourseUserGroup.COHORT
+        )
+
+        self.assertEqual(cohorts.get_cohort_by_name(course.id, "MyCohort"), cohort)
+
+        self.assertRaises(
+            CourseUserGroup.DoesNotExist,
+            lambda: cohorts.get_cohort_by_name(SlashSeparatedCourseKey("course", "does_not", "exist"), cohort)
+        )
+
+    def test_get_cohort_by_id(self):
+        """
+        Make sure cohorts.get_cohort_by_id() properly finds a cohort by id for a given
+        course.
+        """
+        course = modulestore().get_course(self.toy_course_key)
+        cohort = CourseUserGroup.objects.create(
+            name="MyCohort",
+            course_id=course.id,
+            group_type=CourseUserGroup.COHORT
+        )
+
+        self.assertEqual(cohorts.get_cohort_by_id(course.id, cohort.id), cohort)
+
+        cohort.delete()
+
+        self.assertRaises(
+            CourseUserGroup.DoesNotExist,
+            lambda: cohorts.get_cohort_by_id(course.id, cohort.id)
+        )
+
+    def test_add_cohort(self):
+        """
+        Make sure cohorts.add_cohort() properly adds a cohort to a course and handles
+        errors.
+        """
+        course = modulestore().get_course(self.toy_course_key)
+        added_cohort = cohorts.add_cohort(course.id, "My Cohort")
+
+        self.assertEqual(added_cohort.name, "My Cohort")
+        self.assertRaises(
+            ValueError,
+            lambda: cohorts.add_cohort(course.id, "My Cohort")
+        )
+        self.assertRaises(
+            ValueError,
+            lambda: cohorts.add_cohort(SlashSeparatedCourseKey("course", "does_not", "exist"), "My Cohort")
+        )
+
+    def test_add_user_to_cohort(self):
+        """
+        Make sure cohorts.add_user_to_cohort() properly adds a user to a cohort and
+        handles errors.
+        """
+        course_user = User.objects.create(username="Username", email="a@b.com")
+        User.objects.create(username="RandomUsername", email="b@b.com")
+        course = modulestore().get_course(self.toy_course_key)
+        CourseEnrollment.enroll(course_user, self.toy_course_key)
+        first_cohort = CourseUserGroup.objects.create(
+            name="FirstCohort",
+            course_id=course.id,
+            group_type=CourseUserGroup.COHORT
+        )
+        second_cohort = CourseUserGroup.objects.create(
+            name="SecondCohort",
+            course_id=course.id,
+            group_type=CourseUserGroup.COHORT
+        )
+
+        # Success cases
+        # We shouldn't get back a previous cohort, since the user wasn't in one
+        self.assertEqual(
+            cohorts.add_user_to_cohort(first_cohort, "Username"),
+            (course_user, None)
+        )
+        # Should get (user, previous_cohort_name) when moved from one cohort to
+        # another
+        self.assertEqual(
+            cohorts.add_user_to_cohort(second_cohort, "Username"),
+            (course_user, "FirstCohort")
+        )
+
+        # Error cases
+        # Should get ValueError if user already in cohort
+        self.assertRaises(
+            ValueError,
+            lambda: cohorts.add_user_to_cohort(second_cohort, "Username")
+        )
+        # UserDoesNotExist if user truly does not exist
+        self.assertRaises(
+            User.DoesNotExist,
+            lambda: cohorts.add_user_to_cohort(first_cohort, "non_existent_username")
+        )
+
+    def test_get_course_cohort_names(self):
+        """
+        Make sure cohorts.get_course_cohort_names() properly returns a list of cohort
+        names for a given course.
+        """
+        course = modulestore().get_course(self.toy_course_key)
+
+        self.assertItemsEqual(
+            cohorts.get_course_cohort_names(course.id),
+            []
+        )
+        self.assertItemsEqual(
+            cohorts.get_course_cohort_names(SlashSeparatedCourseKey('course', 'does_not', 'exist')),
+            []
+        )
+
+        CourseUserGroup.objects.create(
+            name="FirstCohort",
+            course_id=course.id,
+            group_type=CourseUserGroup.COHORT
+        )
+
+        self.assertItemsEqual(
+            cohorts.get_course_cohort_names(course.id),
+            ["FirstCohort"]
+        )
+
+        CourseUserGroup.objects.create(
+            name="SecondCohort",
+            course_id=course.id,
+            group_type=CourseUserGroup.COHORT
+        )
+
+        self.assertItemsEqual(
+            cohorts.get_course_cohort_names(course.id),
+            ["FirstCohort", "SecondCohort"]
+        )
+
+    def test_delete_empty_cohort(self):
+        """
+        Make sure that cohorts.delete_empty_cohort() properly removes an empty cohort
+        for a given course.
+        """
+        course = modulestore().get_course(self.toy_course_key)
+        user = User.objects.create(username="Username", email="a@b.com")
+        empty_cohort = CourseUserGroup.objects.create(
+            name="EmptyCohort",
+            course_id=course.id,
+            group_type=CourseUserGroup.COHORT
+        )
+        nonempty_cohort = CourseUserGroup.objects.create(
+            name="NonemptyCohort",
+            course_id=course.id,
+            group_type=CourseUserGroup.COHORT
+        )
+        nonempty_cohort.users.add(user)
+
+        cohorts.delete_empty_cohort(course.id, "EmptyCohort")
+
+        # Make sure we cannot access the deleted cohort
+        self.assertRaises(
+            CourseUserGroup.DoesNotExist,
+            lambda: CourseUserGroup.objects.get(
+                course_id=course.id,
+                group_type=CourseUserGroup.COHORT,
+                id=empty_cohort.id
+            )
+        )
+        self.assertRaises(
+            ValueError,
+            lambda: cohorts.delete_empty_cohort(course.id, "NonemptyCohort")
+        )
+        self.assertRaises(
+            CourseUserGroup.DoesNotExist,
+            lambda: cohorts.delete_empty_cohort(SlashSeparatedCourseKey('course', 'does_not', 'exist'), "EmptyCohort")
+        )
+        self.assertRaises(
+            CourseUserGroup.DoesNotExist,
+            lambda: cohorts.delete_empty_cohort(course.id, "NonExistentCohort")
+        )

common/djangoapps/course_groups/tests/test_views.py

@@ -5,13 +5,17 @@ from django.test.utils import override_settings
 from factory import post_generation, Sequence
 from factory.django import DjangoModelFactory
 
+from django.http import Http404
+from django.contrib.auth.models import User
 from course_groups.models import CourseUserGroup
-from course_groups.views import add_users_to_cohort
 from courseware.tests.tests import TEST_DATA_MIXED_MODULESTORE
+from student.models import CourseEnrollment
 from student.tests.factories import UserFactory
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
+from opaque_keys.edx.locations import SlashSeparatedCourseKey
 
+from course_groups.views import list_cohorts, add_cohort, users_in_cohort, add_users_to_cohort, remove_user_from_cohort
 
 class CohortFactory(DjangoModelFactory):
     FACTORY_FOR = CourseUserGroup
@@ -26,60 +30,381 @@ class CohortFactory(DjangoModelFactory):
 
 
 @override_settings(MODULESTORE=TEST_DATA_MIXED_MODULESTORE)
-class AddUsersToCohortTestCase(ModuleStoreTestCase):
+class CohortViewsTestCase(ModuleStoreTestCase):
+    """
+    Base class which sets up a course and staff/non-staff users.
+    """
     def setUp(self):
         self.course = CourseFactory.create()
-        self.staff_user = UserFactory.create(is_staff=True)
+        self.staff_user = UserFactory.create(is_staff=True, username="staff")
+        self.non_staff_user = UserFactory.create(username="nonstaff")
+
+    def _enroll_users(self, users, course_key):
+        """Enroll each user in the specified course"""
+        for user in users:
+            CourseEnrollment.enroll(user, course_key)
+
+    def _create_cohorts(self):
+        """Creates cohorts for testing"""
         self.cohort1_users = [UserFactory.create() for _ in range(3)]
         self.cohort2_users = [UserFactory.create() for _ in range(2)]
         self.cohort3_users = [UserFactory.create() for _ in range(2)]
         self.cohortless_users = [UserFactory.create() for _ in range(3)]
+        self.unenrolled_users = [UserFactory.create() for _ in range(3)]
+        self._enroll_users(
+            self.cohort1_users + self.cohort2_users + self.cohort3_users + self.cohortless_users,
+            self.course.id
+        )
         self.cohort1 = CohortFactory.create(course_id=self.course.id, users=self.cohort1_users)
         self.cohort2 = CohortFactory.create(course_id=self.course.id, users=self.cohort2_users)
         self.cohort3 = CohortFactory.create(course_id=self.course.id, users=self.cohort3_users)
 
-    def check_request(
-            self,
-            users_string,
-            expected_added=None,
-            expected_changed=None,
-            expected_present=None,
-            expected_unknown=None
-    ):
+    def _cohort_in_course(self, cohort_name, course):
         """
-        Check that add_users_to_cohort returns the expected result and has the
-        expected side effects. The given users will be added to cohort1.
+        Returns true iff `course` contains a cohort with the name
+        `cohort_name`.
+        """
+        try:
+            CourseUserGroup.objects.get(
+                course_id=course.id,
+                group_type=CourseUserGroup.COHORT,
+                name=cohort_name
+            )
+        except CourseUserGroup.DoesNotExist:
+            return False
+        else:
+            return True
 
-        users_string is the string input entered by the client
+    def _user_in_cohort(self, username, cohort):
+        """
+        Return true iff a user with `username` exists in `cohort`.
+        """
+        return username in [user.username for user in cohort.users.all()]
 
-        expected_added is a list of users
+    def _verify_non_staff_cannot_access(self, view, request_method, view_args):
+        """
+        Verify that a non-staff user cannot access a given view.
 
-        expected_changed is a list of (user, previous_cohort) tuples
+        `view` is the view to test.
+        `view_args` is a list of arguments (not including the request) to pass
+            to the view.
+        """
+        if request_method == "GET":
+            request = RequestFactory().get("dummy_url")
+        elif request_method == "POST":
+            request = RequestFactory().post("dummy_url")
+        else:
+            request = RequestFactory().request()
+        request.user = self.non_staff_user
+        view_args.insert(0, request)
+        self.assertRaises(Http404, view, *view_args)
 
-        expected_present is a list of (user, email/username) tuples where
-        email/username corresponds to the input
 
-        expected_unknown is a list of strings corresponding to the input
+class ListCohortsTestCase(CohortViewsTestCase):
+    """
+    Tests the `list_cohorts` view.
+    """
+    def request_list_cohorts(self, course):
         """
-        expected_added = expected_added or []
-        expected_changed = expected_changed or []
-        expected_present = expected_present or []
-        expected_unknown = expected_unknown or []
-        request = RequestFactory().post("dummy_url", {"users": users_string})
+        Call `list_cohorts` for a given `course` and return its response as a
+        dict.
+        """
+        request = RequestFactory().get("dummy_url")
         request.user = self.staff_user
-        response = add_users_to_cohort(request, self.course.id.to_deprecated_string(), self.cohort1.id)
+        response = list_cohorts(request, course.id.to_deprecated_string())
         self.assertEqual(response.status_code, 200)
-        result = json.loads(response.content)
-        self.assertEqual(result.get("success"), True)
+        return json.loads(response.content)
+
+    def verify_lists_expected_cohorts(self, response_dict, expected_cohorts):
+        """
+        Verify that the server response contains the expected_cohorts.
+        """
+        self.assertTrue(response_dict.get("success"))
         self.assertItemsEqual(
-            result.get("added"),
+            response_dict.get("cohorts"),
+            [
+                {"name": cohort.name, "id": cohort.id}
+                for cohort in expected_cohorts
+            ]
+        )
+
+    def test_non_staff(self):
+        """
+        Verify that we cannot access list_cohorts if we're a non-staff user.
+        """
+        self._verify_non_staff_cannot_access(list_cohorts, "GET", [self.course.id.to_deprecated_string()])
+
+    def test_no_cohorts(self):
+        """
+        Verify that no cohorts are in response for a course with no cohorts.
+        """
+        self.verify_lists_expected_cohorts(self.request_list_cohorts(self.course), [])
+
+    def test_some_cohorts(self):
+        """
+        Verify that cohorts are in response for a course with some cohorts.
+        """
+        self._create_cohorts()
+        expected_cohorts = CourseUserGroup.objects.filter(
+            course_id=self.course.id,
+            group_type=CourseUserGroup.COHORT
+        )
+        self.verify_lists_expected_cohorts(self.request_list_cohorts(self.course), expected_cohorts)
+
+
+class AddCohortTestCase(CohortViewsTestCase):
+    """
+    Tests the `add_cohort` view.
+    """
+    def request_add_cohort(self, cohort_name, course):
+        """
+        Call `add_cohort` and return its response as a dict.
+        """
+        request = RequestFactory().post("dummy_url", {"name": cohort_name})
+        request.user = self.staff_user
+        response = add_cohort(request, course.id.to_deprecated_string())
+        self.assertEqual(response.status_code, 200)
+        return json.loads(response.content)
+
+    def verify_contains_added_cohort(self, response_dict, cohort_name, course, expected_error_msg=None):
+        """
+        Check that `add_cohort`'s response correctly returns the newly added
+        cohort (or error) in the response.  Also verify that the cohort was
+        actually created/exists.
+        """
+        if expected_error_msg is not None:
+            self.assertFalse(response_dict.get("success"))
+            self.assertEqual(
+                response_dict.get("msg"),
+                expected_error_msg
+            )
+        else:
+            self.assertTrue(response_dict.get("success"))
+            self.assertEqual(
+                response_dict.get("cohort").get("name"),
+                cohort_name
+            )
+        self.assertTrue(self._cohort_in_course(cohort_name, course))
+
+    def test_non_staff(self):
+        """
+        Verify that non-staff users cannot access add_cohort.
+        """
+        self._verify_non_staff_cannot_access(add_cohort, "POST", [self.course.id.to_deprecated_string()])
+
+    def test_new_cohort(self):
+        """
+        Verify that we can add a new cohort.
+        """
+        cohort_name = "New Cohort"
+        self.verify_contains_added_cohort(
+            self.request_add_cohort(cohort_name, self.course),
+            cohort_name,
+            self.course
+        )
+
+    def test_no_cohort(self):
+        """
+        Verify that we cannot explicitly add no cohort.
+        """
+        response_dict = self.request_add_cohort("", self.course)
+        self.assertFalse(response_dict.get("success"))
+        self.assertEqual(response_dict.get("msg"), "No name specified")
+
+    def test_existing_cohort(self):
+        """
+        Verify that we cannot add a cohort with the same name as an existing
+        cohort.
+        """
+        self._create_cohorts()
+        cohort_name = self.cohort1.name
+        self.verify_contains_added_cohort(
+            self.request_add_cohort(cohort_name, self.course),
+            cohort_name,
+            self.course,
+            expected_error_msg="Can't create two cohorts with the same name"
+        )
+
+
+class UsersInCohortTestCase(CohortViewsTestCase):
+    """
+    Tests the `users_in_cohort` view.
+    """
+    def request_users_in_cohort(self, cohort, course, requested_page, should_return_bad_request=False):
+        """
+        Call `users_in_cohort` for a given cohort/requested page, and return
+        its response as a dict.  When `should_return_bad_request` is True,
+        verify that the response indicates a bad request.
+        """
+        request = RequestFactory().get("dummy_url", {"page": requested_page})
+        request.user = self.staff_user
+        response = users_in_cohort(request, course.id.to_deprecated_string(), cohort.id)
+
+        if should_return_bad_request:
+            self.assertEqual(response.status_code, 400)
+            return
+
+        self.assertEqual(response.status_code, 200)
+        return json.loads(response.content)
+
+    def verify_users_in_cohort_and_response(self, cohort, response_dict, expected_users, expected_page,
+                                            expected_num_pages):
+        """
+        Check that the `users_in_cohort` response contains the expected list of
+        users, page number, and total number of pages for a given cohort.  Also
+        verify that those users are actually in the given cohort.
+        """
+        self.assertTrue(response_dict.get("success"))
+        self.assertEqual(response_dict.get("page"), expected_page)
+        self.assertEqual(response_dict.get("num_pages"), expected_num_pages)
+
+        returned_users = User.objects.filter(username__in=[user.get("username") for user in response_dict.get("users")])
+        self.assertItemsEqual(returned_users, expected_users)
+        self.assertTrue(set(returned_users).issubset(cohort.users.all()))
+
+    def test_non_staff(self):
+        """
+        Verify that non-staff users cannot access `check_users_in_cohort`.
+        """
+        cohort = CohortFactory.create(course_id=self.course.id, users=[])
+        self._verify_non_staff_cannot_access(users_in_cohort, "GET", [self.course.id.to_deprecated_string(), cohort.id])
+
+    def test_no_users(self):
+        """
+        Verify that we don't get back any users for a cohort with no users.
+        """
+        cohort = CohortFactory.create(course_id=self.course.id, users=[])
+        response_dict = self.request_users_in_cohort(cohort, self.course, 1)
+        self.verify_users_in_cohort_and_response(
+            cohort,
+            response_dict,
+            expected_users=[],
+            expected_page=1,
+            expected_num_pages=1
+        )
+
+    def test_few_users(self):
+        """
+        Verify that we get back all users for a cohort when the cohort has
+        <=100 users.
+        """
+        users = [UserFactory.create() for _ in range(5)]
+        cohort = CohortFactory.create(course_id=self.course.id, users=users)
+        response_dict = self.request_users_in_cohort(cohort, self.course, 1)
+        self.verify_users_in_cohort_and_response(
+            cohort,
+            response_dict,
+            expected_users=users,
+            expected_page=1,
+            expected_num_pages=1
+        )
+
+    def test_many_users(self):
+        """
+        Verify that pagination works correctly for cohorts with >100 users.
+        """
+        users = [UserFactory.create() for _ in range(101)]
+        cohort = CohortFactory.create(course_id=self.course.id, users=users)
+        response_dict_1 = self.request_users_in_cohort(cohort, self.course, 1)
+        response_dict_2 = self.request_users_in_cohort(cohort, self.course, 2)
+        self.verify_users_in_cohort_and_response(
+            cohort,
+            response_dict_1,
+            expected_users=users[:100],
+            expected_page=1,
+            expected_num_pages=2
+        )
+        self.verify_users_in_cohort_and_response(
+            cohort,
+            response_dict_2,
+            expected_users=users[100:],
+            expected_page=2,
+            expected_num_pages=2
+        )
+
+    def test_out_of_range(self):
+        """
+        Verify that we get a blank page of users when requesting page 0 or a
+        page greater than the actual number of pages.
+        """
+        users = [UserFactory.create() for _ in range(5)]
+        cohort = CohortFactory.create(course_id=self.course.id, users=users)
+        response = self.request_users_in_cohort(cohort, self.course, 0)
+        self.verify_users_in_cohort_and_response(
+            cohort,
+            response,
+            expected_users=[],
+            expected_page=0,
+            expected_num_pages=1
+        )
+        response = self.request_users_in_cohort(cohort, self.course, 2)
+        self.verify_users_in_cohort_and_response(
+            cohort,
+            response,
+            expected_users=[],
+            expected_page=2,
+            expected_num_pages=1
+        )
+
+    def test_non_positive_page(self):
+        """
+        Verify that we get a `HttpResponseBadRequest` (bad request) when the
+        page we request isn't a positive integer.
+        """
+        users = [UserFactory.create() for _ in range(5)]
+        cohort = CohortFactory.create(course_id=self.course.id, users=users)
+        self.request_users_in_cohort(cohort, self.course, "invalid", should_return_bad_request=True)
+        self.request_users_in_cohort(cohort, self.course, -1, should_return_bad_request=True)
+
+
+class AddUsersToCohortTestCase(CohortViewsTestCase):
+    """
+    Tests the `add_users_to_cohort` view.
+    """
+    def setUp(self):
+        super(AddUsersToCohortTestCase, self).setUp()
+        self._create_cohorts()
+
+    def request_add_users_to_cohort(self, users_string, cohort, course, should_raise_404=False):
+        """
+        Call `add_users_to_cohort` for a given cohort, course, and list of
+        users, returning its response as a dict.  When `should_raise_404` is
+        True, verify that the request raised a Http404.
+        """
+        request = RequestFactory().post("dummy_url", {"users": users_string})
+        request.user = self.staff_user
+        if should_raise_404:
+            self.assertRaises(
+                Http404,
+                lambda: add_users_to_cohort(request, course.id.to_deprecated_string(), cohort.id)
+            )
+        else:
+            response = add_users_to_cohort(request, course.id.to_deprecated_string(), cohort.id)
+            self.assertEqual(response.status_code, 200)
+            return json.loads(response.content)
+
+    def verify_added_users_to_cohort(self, response_dict, cohort, course, expected_added, expected_changed,
+                                     expected_present, expected_unknown):
+        """
+        Check that add_users_to_cohort returned the expected response and has
+        the expected side effects.
+
+        `expected_added` is a list of users
+        `expected_changed` is a list of (user, previous_cohort) tuples
+        `expected_present` is a list of (user, email/username) tuples where
+            email/username corresponds to the input
+        `expected_unknown` is a list of strings corresponding to the input
+        """
+        self.assertTrue(response_dict.get("success"))
+        self.assertItemsEqual(
+            response_dict.get("added"),
             [
                 {"username": user.username, "name": user.profile.name, "email": user.email}
                 for user in expected_added
             ]
         )
         self.assertItemsEqual(
-            result.get("changed"),
+            response_dict.get("changed"),
             [
                 {
                     "username": user.username,
@@ -91,55 +416,133 @@ class AddUsersToCohortTestCase(ModuleStoreTestCase):
             ]
         )
         self.assertItemsEqual(
-            result.get("present"),
+            response_dict.get("present"),
             [username_or_email for (_, username_or_email) in expected_present]
         )
-        self.assertItemsEqual(result.get("unknown"), expected_unknown)
+        self.assertItemsEqual(response_dict.get("unknown"), expected_unknown)
         for user in expected_added + [user for (user, _) in expected_changed + expected_present]:
             self.assertEqual(
                 CourseUserGroup.objects.get(
-                    course_id=self.course.id,
+                    course_id=course.id,
                     group_type=CourseUserGroup.COHORT,
                     users__id=user.id
                 ),
-                self.cohort1
+                cohort
             )
 
+    def test_non_staff(self):
+        """
+        Verify that non-staff users cannot access `check_users_in_cohort`.
+        """
+        cohort = CohortFactory.create(course_id=self.course.id, users=[])
+        self._verify_non_staff_cannot_access(
+            add_users_to_cohort,
+            "POST",
+            [self.course.id.to_deprecated_string(), cohort.id]
+        )
+
     def test_empty(self):
-        self.check_request("")
+        """
+        Verify that adding an empty list of users to a cohort has no result.
+        """
+        response_dict = self.request_add_users_to_cohort("", self.cohort1, self.course)
+        self.verify_added_users_to_cohort(
+            response_dict,
+            self.cohort1,
+            self.course,
+            expected_added=[],
+            expected_changed=[],
+            expected_present=[],
+            expected_unknown=[]
+        )
 
     def test_only_added(self):
-        self.check_request(
+        """
+        Verify that we can add users to their first cohort.
+        """
+        response_dict = self.request_add_users_to_cohort(
             ",".join([user.username for user in self.cohortless_users]),
-            expected_added=self.cohortless_users
+            self.cohort1,
+            self.course
+        )
+        self.verify_added_users_to_cohort(
+            response_dict,
+            self.cohort1,
+            self.course,
+            expected_added=self.cohortless_users,
+            expected_changed=[],
+            expected_present=[],
+            expected_unknown=[]
         )
 
     def test_only_changed(self):
-        self.check_request(
+        """
+        Verify that we can move users to a different cohort.
+        """
+        response_dict = self.request_add_users_to_cohort(
             ",".join([user.username for user in self.cohort2_users + self.cohort3_users]),
+            self.cohort1,
+            self.course
+        )
+        self.verify_added_users_to_cohort(
+            response_dict,
+            self.cohort1,
+            self.course,
+            expected_added=[],
             expected_changed=(
                 [(user, self.cohort2.name) for user in self.cohort2_users] +
                 [(user, self.cohort3.name) for user in self.cohort3_users]
-            )
+            ),
+            expected_present=[],
+            expected_unknown=[]
         )
 
     def test_only_present(self):
+        """
+        Verify that we can 'add' users to their current cohort.
+        """
         usernames = [user.username for user in self.cohort1_users]
-        self.check_request(
+        response_dict = self.request_add_users_to_cohort(
             ",".join(usernames),
-            expected_present=[(user, user.username) for user in self.cohort1_users]
+            self.cohort1,
+            self.course
+        )
+        self.verify_added_users_to_cohort(
+            response_dict,
+            self.cohort1,
+            self.course,
+            expected_added=[],
+            expected_changed=[],
+            expected_present=[(user, user.username) for user in self.cohort1_users],
+            expected_unknown=[]
         )
 
     def test_only_unknown(self):
+        """
+        Verify that non-existent users are not added.
+        """
         usernames = ["unknown_user{}".format(i) for i in range(3)]
-        self.check_request(
+        response_dict = self.request_add_users_to_cohort(
             ",".join(usernames),
+            self.cohort1,
+            self.course
+        )
+        self.verify_added_users_to_cohort(
+            response_dict,
+            self.cohort1,
+            self.course,
+            expected_added=[],
+            expected_changed=[],
+            expected_present=[],
             expected_unknown=usernames
         )
 
     def test_all(self):
+        """
+        Test all adding conditions together.
+        """
         unknowns = ["unknown_user{}".format(i) for i in range(3)]
-        self.check_request(
+        response_dict = self.request_add_users_to_cohort(
             ",".join(
                 unknowns +
                 [
@@ -147,41 +550,200 @@ class AddUsersToCohortTestCase(ModuleStoreTestCase):
                     for user in self.cohortless_users + self.cohort1_users + self.cohort2_users + self.cohort3_users
                 ]
             ),
-            self.cohortless_users,
-            (
+            self.cohort1,
+            self.course
+        )
+        self.verify_added_users_to_cohort(
+            response_dict,
+            self.cohort1,
+            self.course,
+            expected_added=self.cohortless_users,
+            expected_changed=(
                 [(user, self.cohort2.name) for user in self.cohort2_users] +
                 [(user, self.cohort3.name) for user in self.cohort3_users]
             ),
-            [(user, user.username) for user in self.cohort1_users],
-            unknowns
+            expected_present=[(user, user.username) for user in self.cohort1_users],
+            expected_unknown=unknowns
         )
 
     def test_emails(self):
+        """
+        Verify that we can use emails to identify users.
+        """
         unknown = "unknown_user@example.com"
-        self.check_request(
+        response_dict = self.request_add_users_to_cohort(
             ",".join([
                 self.cohort1_users[0].email,
                 self.cohort2_users[0].email,
                 self.cohortless_users[0].email,
                 unknown
             ]),
-            [self.cohortless_users[0]],
-            [(self.cohort2_users[0], self.cohort2.name)],
-            [(self.cohort1_users[0], self.cohort1_users[0].email)],
-            [unknown]
+            self.cohort1,
+            self.course
+        )
+        self.verify_added_users_to_cohort(
+            response_dict,
+            self.cohort1,
+            self.course,
+            expected_added=[self.cohortless_users[0]],
+            expected_changed=[(self.cohort2_users[0], self.cohort2.name)],
+            expected_present=[(self.cohort1_users[0], self.cohort1_users[0].email)],
+            expected_unknown=[unknown]
         )
 
     def test_delimiters(self):
+        """
+        Verify that we can use different types of whitespace to delimit
+        usernames in the user string.
+        """
         unknown = "unknown_user"
-        self.check_request(
+        response_dict = self.request_add_users_to_cohort(
             " {} {}\t{}, \r\n{}".format(
                 unknown,
                 self.cohort1_users[0].username,
                 self.cohort2_users[0].username,
                 self.cohortless_users[0].username
             ),
-            [self.cohortless_users[0]],
-            [(self.cohort2_users[0], self.cohort2.name)],
-            [(self.cohort1_users[0], self.cohort1_users[0].username)],
-            [unknown]
+            self.cohort1,
+            self.course
+        )
+        self.verify_added_users_to_cohort(
+            response_dict,
+            self.cohort1,
+            self.course,
+            expected_added=[self.cohortless_users[0]],
+            expected_changed=[(self.cohort2_users[0], self.cohort2.name)],
+            expected_present=[(self.cohort1_users[0], self.cohort1_users[0].username)],
+            expected_unknown=[unknown]
+        )
+
+    def test_can_cohort_unenrolled_users(self):
+        """
+        Verify that users can be added to a cohort of a course they're not
+        enrolled in.  This feature is currently used to pre-cohort users that
+        are expected to enroll in a course.
+        """
+        unenrolled_usernames = [user.username for user in self.unenrolled_users]
+        response_dict = self.request_add_users_to_cohort(
+            ",".join(unenrolled_usernames),
+            self.cohort1,
+            self.course
+        )
+        self.verify_added_users_to_cohort(
+            response_dict,
+            self.cohort1,
+            self.course,
+            expected_added=self.unenrolled_users,
+            expected_changed=[],
+            expected_present=[],
+            expected_unknown=[]
+        )
+
+    def test_non_existent_cohort(self):
+        """
+        Verify that an error is raised when trying to add users to a cohort
+        which does not belong to the given course.
+        """
+        users = [UserFactory.create(username="user{0}".format(i)) for i in range(3)]
+        usernames = [user.username for user in users]
+        wrong_course_key = SlashSeparatedCourseKey("some", "arbitrary", "course")
+        wrong_course_cohort = CohortFactory.create(name="wrong_cohort", course_id=wrong_course_key, users=[])
+        self.request_add_users_to_cohort(
+            ",".join(usernames),
+            wrong_course_cohort,
+            self.course,
+            should_raise_404=True
         )
+
+
+class RemoveUserFromCohortTestCase(CohortViewsTestCase):
+    """
+    Tests the `remove_user_from_cohort` view.
+    """
+    def request_remove_user_from_cohort(self, username, cohort):
+        """
+        Call `remove_user_from_cohort` with the given username and cohort.
+        """
+        if username is not None:
+            request = RequestFactory().post("dummy_url", {"username": username})
+        else:
+            request = RequestFactory().post("dummy_url")
+        request.user = self.staff_user
+        response = remove_user_from_cohort(request, self.course.id.to_deprecated_string(), cohort.id)
+        self.assertEqual(response.status_code, 200)
+        return json.loads(response.content)
+
+    def verify_removed_user_from_cohort(self, username, response_dict, cohort, expected_error_msg=None):
+        """
+        Check that `remove_user_from_cohort` properly removes a user from a
+        cohort and returns appropriate success.  If the removal should fail,
+        verify that the returned error message matches the expected one.
+        """
+        if expected_error_msg is None:
+            self.assertTrue(response_dict.get("success"))
+            self.assertIsNone(response_dict.get("msg"))
+            self.assertFalse(self._user_in_cohort(username, cohort))
+        else:
+            self.assertFalse(response_dict.get("success"))
+            self.assertEqual(response_dict.get("msg"), expected_error_msg)
+
+    def test_non_staff(self):
+        """
+        Verify that non-staff users cannot access `check_users_in_cohort`.
+        """
+        cohort = CohortFactory.create(course_id=self.course.id, users=[])
+        self._verify_non_staff_cannot_access(
+            remove_user_from_cohort,
+            "POST",
+            [self.course.id.to_deprecated_string(), cohort.id]
+        )
+
+    def test_no_username_given(self):
+        """
+        Verify that we get an error message when omitting a username.
+        """
+        cohort = CohortFactory.create(course_id=self.course.id, users=[])
+        response_dict = self.request_remove_user_from_cohort(None, cohort)
+        self.verify_removed_user_from_cohort(
+            None,
+            response_dict,
+            cohort,
+            expected_error_msg='No username specified'
+        )
+
+    def test_user_does_not_exist(self):
+        """
+        Verify that we get an error message when the requested user to remove
+        does not exist.
+        """
+        username = "bogus"
+        cohort = CohortFactory.create(course_id=self.course.id, users=[])
+        response_dict = self.request_remove_user_from_cohort(
+            username,
+            cohort
+        )
+        self.verify_removed_user_from_cohort(
+            username,
+            response_dict,
+            cohort,
+            expected_error_msg='No user \'{0}\''.format(username)
+        )
+
+    def test_can_remove_user_not_in_cohort(self):
+        """
+        Verify that we can "remove" a user from a cohort even if they are not a
+        member of that cohort.
+        """
+        user = UserFactory.create()
+        cohort = CohortFactory.create(course_id=self.course.id, users=[])
+        response_dict = self.request_remove_user_from_cohort(user.username, cohort)
+        self.verify_removed_user_from_cohort(user.username, response_dict, cohort)
+
+    def test_can_remove_user_from_cohort(self):
+        """
+        Verify that we can remove a user from a cohort.
+        """
+        user = UserFactory.create()
+        cohort = CohortFactory.create(course_id=self.course.id, users=[user])
+        response_dict = self.request_remove_user_from_cohort(user.username, cohort)
+        self.verify_removed_user_from_cohort(user.username, response_dict, cohort)

common/djangoapps/course_groups/views.py

@@ -3,7 +3,8 @@ from django.views.decorators.http import require_POST
 from django.contrib.auth.models import User
 from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger
 from django.core.urlresolvers import reverse
-from django.http import HttpResponse
+from django.http import Http404, HttpResponse, HttpResponseBadRequest
+from django.utils.translation import ugettext as _
 import json
 import logging
 import re
@@ -13,7 +14,7 @@ from courseware.courses import get_course_with_access
 from edxmako.shortcuts import render_to_response
 
 from . import cohorts
-
+from .models import CourseUserGroup
 
 log = logging.getLogger(__name__)
 
@@ -115,17 +116,18 @@ def users_in_cohort(request, course_key_string, cohort_id):
     cohort = cohorts.get_cohort_by_id(course_key, int(cohort_id))
 
     paginator = Paginator(cohort.users.all(), 100)
-    page = request.GET.get('page')
     try:
-        users = paginator.page(page)
-    except PageNotAnInteger:
-        # return the first page
-        page = 1
+        page = int(request.GET.get('page'))
+    except (TypeError, ValueError):
+        return HttpResponseBadRequest(_('Requested page must be numeric'))
+    else:
+        if page < 0:
+            return HttpResponseBadRequest(_('Requested page must be greater than zero'))
+
+    try:
         users = paginator.page(page)
     except EmptyPage:
-        # Page is out of range.  Return last page
-        page = paginator.num_pages
-        contacts = paginator.page(page)
+        users = []  # When page > number of pages, return a blank page
 
     user_info = [{'username': u.username,
                   'email': u.email,
@@ -154,12 +156,20 @@ def add_users_to_cohort(request, course_key_string, cohort_id):
                   'previous_cohort': ...}, ...],
      'present': [str1, str2, ...],    # already there
      'unknown': [str1, str2, ...]}
+
+     Raises Http404 if the cohort cannot be found for the given course.
     """
     # this is a string when we get it here
     course_key = SlashSeparatedCourseKey.from_deprecated_string(course_key_string)
     get_course_with_access(request.user, 'staff', course_key)
 
-    cohort = cohorts.get_cohort_by_id(course_key, cohort_id)
+    try:
+        cohort = cohorts.get_cohort_by_id(course_key, cohort_id)
+    except CourseUserGroup.DoesNotExist:
+        raise Http404("Cohort (ID {cohort_id}) not found for {course_key_string}".format(
+            cohort_id=cohort_id,
+            course_key_string=course_key_string
+        ))
 
     users = request.POST.get('users', '')
     added = []

common/static/js/course_groups/cohorts.js

@@ -220,7 +220,7 @@ var CohortManager = (function ($) {
                 });
             } else if (state == state_detail) {
                 detail_header.text("Members of " + cohort_title);
-                $.ajax(detail_url).done(show_users).fail(function() {
+                $.ajax(detail_url, {data: {page: 1}}).done(show_users).fail(function() {
                     log_error("Error trying to load users in cohort");
                 });
             }