From 88ad29d1666da8d905bddc10287a7256ff791ee8 Mon Sep 17 00:00:00 2001
From: Nimisha Asthagiri <nasthagiri@edx.org>
Date: Sun, 27 Jul 2014 21:33:23 -0400
Subject: [PATCH] LMS-11016 Studio server-side for course_listing and
 course_rerun.

Conflicts:
	cms/djangoapps/contentstore/tests/test_course_listing.py
---
 .../contentstore/tests/test_contentstore.py   | 104 +++++++-
 .../contentstore/tests/test_course_listing.py |  78 ++++--
 cms/djangoapps/contentstore/tests/utils.py    |   1 +
 cms/djangoapps/contentstore/utils.py          |  67 ++++--
 cms/djangoapps/contentstore/views/course.py   | 227 ++++++++++--------
 cms/djangoapps/contentstore/views/tasks.py    |  32 +++
 6 files changed, 373 insertions(+), 136 deletions(-)
 create mode 100644 cms/djangoapps/contentstore/views/tasks.py

diff --git a/cms/djangoapps/contentstore/tests/test_contentstore.py b/cms/djangoapps/contentstore/tests/test_contentstore.py
index 41702b64df..101533a467 100644
--- a/cms/djangoapps/contentstore/tests/test_contentstore.py
+++ b/cms/djangoapps/contentstore/tests/test_contentstore.py
@@ -29,7 +29,7 @@ from xmodule.modulestore import ModuleStoreEnum
 from xmodule.modulestore.exceptions import ItemNotFoundError
 from xmodule.modulestore.inheritance import own_metadata
 from opaque_keys.edx.keys import UsageKey
-from opaque_keys.edx.locations import SlashSeparatedCourseKey, AssetLocation
+from opaque_keys.edx.locations import SlashSeparatedCourseKey, AssetLocation, CourseLocator
 from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
 from xmodule.modulestore.xml_exporter import export_to_xml
 from xmodule.modulestore.xml_importer import import_from_xml, perform_xlint
@@ -46,6 +46,7 @@ from student.models import CourseEnrollment
 from student.roles import CourseCreatorRole, CourseInstructorRole
 from opaque_keys import InvalidKeyError
 from contentstore.tests.utils import get_url
+from course_action_state.models import CourseRerunState, CourseRerunUIStateManager
 
 
 TEST_DATA_CONTENTSTORE = copy.deepcopy(settings.CONTENTSTORE)
@@ -1561,6 +1562,107 @@ class MetadataSaveTestCase(ContentStoreTestCase):
         pass
 
 
+class RerunCourseTest(ContentStoreTestCase):
+    """
+    Tests for Rerunning a course via the view handler
+    """
+    def setUp(self):
+        super(RerunCourseTest, self).setUp()
+        self.destination_course_data = {
+            'org': 'MITx',
+            'number': '111',
+            'display_name': 'Robot Super Course',
+            'run': '2013_Spring'
+        }
+        self.destination_course_key = _get_course_id(self.destination_course_data)
+
+    def post_rerun_request(self, source_course_key, response_code=200):
+        """Create and send an ajax post for the rerun request"""
+
+        # create data to post
+        rerun_course_data = {'source_course_key': unicode(source_course_key)}
+        rerun_course_data.update(self.destination_course_data)
+
+        # post the request
+        course_url = get_url('course_handler', self.destination_course_key, 'course_key_string')
+        response = self.client.ajax_post(course_url, rerun_course_data)
+
+        # verify response
+        self.assertEqual(response.status_code, response_code)
+        if response_code == 200:
+            self.assertNotIn('ErrMsg', parse_json(response))
+
+    def create_course_listing_html(self, course_key):
+        """Creates html fragment that is created for the given course_key in the course listing section"""
+        return '<a class="course-link" href="/course/{}"'.format(course_key)
+
+    def create_unsucceeded_course_action_html(self, course_key):
+        """Creates html fragment that is created for the given course_key in the unsucceeded course action section"""
+        # TODO LMS-11011 Update this once the Rerun UI is implemented.
+        return '<div class="unsucceeded-course-action" href="/course/{}"'.format(course_key)
+
+    def assertInCourseListing(self, course_key):
+        """
+        Asserts that the given course key is in the accessible course listing section of the html
+        and NOT in the unsucceeded course action section of the html.
+        """
+        course_listing_html = self.client.get_html('/course/')
+        self.assertIn(self.create_course_listing_html(course_key), course_listing_html.content)
+        self.assertNotIn(self.create_unsucceeded_course_action_html(course_key), course_listing_html.content)
+
+    def assertInUnsucceededCourseActions(self, course_key):
+        """
+        Asserts that the given course key is in the unsucceeded course action section of the html
+        and NOT in the accessible course listing section of the html.
+        """
+        course_listing_html = self.client.get_html('/course/')
+        self.assertNotIn(self.create_course_listing_html(course_key), course_listing_html.content)
+        # TODO Uncomment this once LMS-11011 is implemented.
+        # self.assertIn(self.create_unsucceeded_course_action_html(course_key), course_listing_html.content)
+
+    def test_rerun_course_success(self):
+        source_course = CourseFactory.create()
+        self.post_rerun_request(source_course.id)
+
+        # Verify that the course rerun action is marked succeeded
+        rerun_state = CourseRerunState.objects.find_first(course_key=self.destination_course_key)
+        self.assertEquals(rerun_state.state, CourseRerunUIStateManager.State.SUCCEEDED)
+
+        # Verify that the creator is now enrolled in the course.
+        self.assertTrue(CourseEnrollment.is_enrolled(self.user, self.destination_course_key))
+
+        # Verify both courses are in the course listing section
+        self.assertInCourseListing(source_course.id)
+        self.assertInCourseListing(self.destination_course_key)
+
+    def test_rerun_course_fail(self):
+        existent_course_key = CourseFactory.create().id
+        non_existent_course_key = CourseLocator("org", "non_existent_course", "run")
+        self.post_rerun_request(non_existent_course_key)
+
+        # Verify that the course rerun action is marked failed
+        rerun_state = CourseRerunState.objects.find_first(course_key=self.destination_course_key)
+        self.assertEquals(rerun_state.state, CourseRerunUIStateManager.State.FAILED)
+        self.assertIn("Cannot find a course at", rerun_state.message)
+
+        # Verify that the creator is not enrolled in the course.
+        self.assertFalse(CourseEnrollment.is_enrolled(self.user, non_existent_course_key))
+
+        # Verify that the existing course continues to be in the course listings
+        self.assertInCourseListing(existent_course_key)
+
+        # Verify that the failed course is NOT in the course listings
+        self.assertInUnsucceededCourseActions(non_existent_course_key)
+
+    def test_rerun_with_permission_denied(self):
+        with mock.patch.dict('django.conf.settings.FEATURES', {"ENABLE_CREATOR_GROUP": True}):
+            source_course = CourseFactory.create()
+            auth.add_users(self.user, CourseCreatorRole(), self.user)
+            self.user.is_staff = False
+            self.user.save()
+            self.post_rerun_request(source_course.id, 403)
+
+
 class EntryPageTestCase(TestCase):
     """
     Tests entry pages that aren't specific to a course.
diff --git a/cms/djangoapps/contentstore/tests/test_course_listing.py b/cms/djangoapps/contentstore/tests/test_course_listing.py
index ce6fd24791..cf2a49f7f7 100644
--- a/cms/djangoapps/contentstore/tests/test_course_listing.py
+++ b/cms/djangoapps/contentstore/tests/test_course_listing.py
@@ -18,9 +18,10 @@ from student.roles import CourseInstructorRole, CourseStaffRole, GlobalStaff, Or
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory, check_mongo_calls
 from xmodule.modulestore import ModuleStoreEnum
-from opaque_keys.edx.locations import SlashSeparatedCourseKey
+from opaque_keys.edx.locations import SlashSeparatedCourseKey, CourseLocator
 from xmodule.modulestore.django import modulestore
 from xmodule.error_module import ErrorDescriptor
+from course_action_state.models import CourseRerunState
 
 TOTAL_COURSES_COUNT = 500
 USER_COURSES_COUNT = 50
@@ -76,11 +77,11 @@ class TestCourseListing(ModuleStoreTestCase):
         self._create_course_with_access_groups(course_location, self.user)
 
         # get courses through iterating all courses
-        courses_list = _accessible_courses_list(self.request)
+        courses_list, __ = _accessible_courses_list(self.request)
         self.assertEqual(len(courses_list), 1)
 
         # get courses by reversing group name formats
-        courses_list_by_groups = _accessible_courses_list_from_groups(self.request)
+        courses_list_by_groups, __ = _accessible_courses_list_from_groups(self.request)
         self.assertEqual(len(courses_list_by_groups), 1)
         # check both course lists have same courses
         self.assertEqual(courses_list, courses_list_by_groups)
@@ -98,11 +99,11 @@ class TestCourseListing(ModuleStoreTestCase):
             self.assertIsInstance(modulestore().get_course(course_key), ErrorDescriptor)
 
             # get courses through iterating all courses
-            courses_list = _accessible_courses_list(self.request)
+            courses_list, __ = _accessible_courses_list(self.request)
             self.assertEqual(courses_list, [])
 
             # get courses by reversing group name formats
-            courses_list_by_groups = _accessible_courses_list_from_groups(self.request)
+            courses_list_by_groups, __ = _accessible_courses_list_from_groups(self.request)
             self.assertEqual(courses_list_by_groups, [])
 
     def test_errored_course_regular_access(self):
@@ -119,11 +120,11 @@ class TestCourseListing(ModuleStoreTestCase):
             self.assertIsInstance(modulestore().get_course(course_key), ErrorDescriptor)
 
             # get courses through iterating all courses
-            courses_list = _accessible_courses_list(self.request)
+            courses_list, __ = _accessible_courses_list(self.request)
             self.assertEqual(courses_list, [])
 
             # get courses by reversing group name formats
-            courses_list_by_groups = _accessible_courses_list_from_groups(self.request)
+            courses_list_by_groups, __ = _accessible_courses_list_from_groups(self.request)
             self.assertEqual(courses_list_by_groups, [])
             self.assertEqual(courses_list, courses_list_by_groups)
 
@@ -135,11 +136,11 @@ class TestCourseListing(ModuleStoreTestCase):
         self._create_course_with_access_groups(course_key, self.user)
 
         # get courses through iterating all courses
-        courses_list = _accessible_courses_list(self.request)
+        courses_list, __ = _accessible_courses_list(self.request)
         self.assertEqual(len(courses_list), 1)
 
         # get courses by reversing group name formats
-        courses_list_by_groups = _accessible_courses_list_from_groups(self.request)
+        courses_list_by_groups, __ = _accessible_courses_list_from_groups(self.request)
         self.assertEqual(len(courses_list_by_groups), 1)
         # check both course lists have same courses
         self.assertEqual(courses_list, courses_list_by_groups)
@@ -150,7 +151,7 @@ class TestCourseListing(ModuleStoreTestCase):
         CourseInstructorRole(course_key).add_users(self.user)
 
         # test that get courses through iterating all courses now returns no course
-        courses_list = _accessible_courses_list(self.request)
+        courses_list, __ = _accessible_courses_list(self.request)
         self.assertEqual(len(courses_list), 0)
 
     def test_course_listing_performance(self):
@@ -175,22 +176,22 @@ class TestCourseListing(ModuleStoreTestCase):
 
         # time the get courses by iterating through all courses
         with Timer() as iteration_over_courses_time_1:
-            courses_list = _accessible_courses_list(self.request)
+            courses_list, __ = _accessible_courses_list(self.request)
         self.assertEqual(len(courses_list), USER_COURSES_COUNT)
 
         # time again the get courses by iterating through all courses
         with Timer() as iteration_over_courses_time_2:
-            courses_list = _accessible_courses_list(self.request)
+            courses_list, __ = _accessible_courses_list(self.request)
         self.assertEqual(len(courses_list), USER_COURSES_COUNT)
 
         # time the get courses by reversing django groups
         with Timer() as iteration_over_groups_time_1:
-            courses_list = _accessible_courses_list_from_groups(self.request)
+            courses_list, __ = _accessible_courses_list_from_groups(self.request)
         self.assertEqual(len(courses_list), USER_COURSES_COUNT)
 
         # time again the get courses by reversing django groups
         with Timer() as iteration_over_groups_time_2:
-            courses_list = _accessible_courses_list_from_groups(self.request)
+            courses_list, __ = _accessible_courses_list_from_groups(self.request)
         self.assertEqual(len(courses_list), USER_COURSES_COUNT)
 
         # test that the time taken by getting courses through reversing django groups is lower then the time
@@ -201,10 +202,10 @@ class TestCourseListing(ModuleStoreTestCase):
         # Now count the db queries
         store = modulestore()._get_modulestore_by_type(ModuleStoreEnum.Type.mongo)
         with check_mongo_calls(store, USER_COURSES_COUNT):
-            courses_list = _accessible_courses_list_from_groups(self.request)
+            _accessible_courses_list_from_groups(self.request)
 
         with check_mongo_calls(store, 1):
-            courses_list = _accessible_courses_list(self.request)
+            _accessible_courses_list(self.request)
 
     def test_get_course_list_with_same_course_id(self):
         """
@@ -215,11 +216,11 @@ class TestCourseListing(ModuleStoreTestCase):
         self._create_course_with_access_groups(course_location_caps, self.user)
 
         # get courses through iterating all courses
-        courses_list = _accessible_courses_list(self.request)
+        courses_list, __ = _accessible_courses_list(self.request)
         self.assertEqual(len(courses_list), 1)
 
         # get courses by reversing group name formats
-        courses_list_by_groups = _accessible_courses_list_from_groups(self.request)
+        courses_list_by_groups, __ = _accessible_courses_list_from_groups(self.request)
         self.assertEqual(len(courses_list_by_groups), 1)
         # check both course lists have same courses
         self.assertEqual(courses_list, courses_list_by_groups)
@@ -229,22 +230,22 @@ class TestCourseListing(ModuleStoreTestCase):
         self._create_course_with_access_groups(course_location_camel, self.user)
 
         # test that get courses through iterating all courses returns both courses
-        courses_list = _accessible_courses_list(self.request)
+        courses_list, __ = _accessible_courses_list(self.request)
         self.assertEqual(len(courses_list), 2)
 
         # test that get courses by reversing group name formats returns both courses
-        courses_list_by_groups = _accessible_courses_list_from_groups(self.request)
+        courses_list_by_groups, __ = _accessible_courses_list_from_groups(self.request)
         self.assertEqual(len(courses_list_by_groups), 2)
 
         # now delete first course (course_location_caps) and check that it is no longer accessible
         delete_course_and_groups(course_location_caps, self.user.id)
 
         # test that get courses through iterating all courses now returns one course
-        courses_list = _accessible_courses_list(self.request)
+        courses_list, __ = _accessible_courses_list(self.request)
         self.assertEqual(len(courses_list), 1)
 
         # test that get courses by reversing group name formats also returns one course
-        courses_list_by_groups = _accessible_courses_list_from_groups(self.request)
+        courses_list_by_groups, __ = _accessible_courses_list_from_groups(self.request)
         self.assertEqual(len(courses_list_by_groups), 1)
 
         # now check that deleted course is not accessible
@@ -282,7 +283,7 @@ class TestCourseListing(ModuleStoreTestCase):
             }},
         )
 
-        courses_list = _accessible_courses_list_from_groups(self.request)
+        courses_list, __ = _accessible_courses_list_from_groups(self.request)
         self.assertEqual(len(courses_list), 1, courses_list)
 
     @ddt.data(OrgStaffRole('AwesomeOrg'), OrgInstructorRole('AwesomeOrg'))
@@ -310,5 +311,34 @@ class TestCourseListing(ModuleStoreTestCase):
 
         with self.assertRaises(AccessListFallback):
             _accessible_courses_list_from_groups(self.request)
-        courses_list = _accessible_courses_list(self.request)
+        courses_list, __ = _accessible_courses_list(self.request)
         self.assertEqual(len(courses_list), 2)
+
+    def test_course_listing_with_actions_in_progress(self):
+        sourse_course_key = CourseLocator('source-Org', 'source-Course', 'source-Run')
+
+        num_courses_to_create = 3
+        courses = [
+            self._create_course_with_access_groups(CourseLocator('Org', 'CreatedCourse' + str(num), 'Run'), self.user)
+            for num in range(0, num_courses_to_create)
+        ]
+        courses_in_progress = [
+            self._create_course_with_access_groups(CourseLocator('Org', 'InProgressCourse' + str(num), 'Run'), self.user)
+            for num in range(0, num_courses_to_create)
+        ]
+
+        # simulate initiation of course actions
+        for course in courses_in_progress:
+            CourseRerunState.objects.initiated(sourse_course_key, destination_course_key=course.id, user=self.user)
+
+        # verify return values
+        for method in (_accessible_courses_list_from_groups, _accessible_courses_list):
+            def set_of_course_keys(course_list, key_attribute_name='id'):
+                """Returns a python set of course keys by accessing the key with the given attribute name."""
+                return set(getattr(c, key_attribute_name) for c in course_list)
+
+            found_courses, unsucceeded_course_actions = method(self.request)
+            self.assertSetEqual(set_of_course_keys(courses + courses_in_progress), set_of_course_keys(found_courses))
+            self.assertSetEqual(
+                set_of_course_keys(courses_in_progress), set_of_course_keys(unsucceeded_course_actions, 'course_key')
+            )
diff --git a/cms/djangoapps/contentstore/tests/utils.py b/cms/djangoapps/contentstore/tests/utils.py
index ec425f0e44..e4b19c1771 100644
--- a/cms/djangoapps/contentstore/tests/utils.py
+++ b/cms/djangoapps/contentstore/tests/utils.py
@@ -95,6 +95,7 @@ class CourseTestCase(ModuleStoreTestCase):
         client = Client()
         if authenticate:
             client.login(username=nonstaff.username, password=password)
+            nonstaff.is_authenticated = True
         return client, nonstaff
 
     def populate_course(self):
diff --git a/cms/djangoapps/contentstore/utils.py b/cms/djangoapps/contentstore/utils.py
index f1ccdbc717..0f54190c15 100644
--- a/cms/djangoapps/contentstore/utils.py
+++ b/cms/djangoapps/contentstore/utils.py
@@ -9,6 +9,8 @@ from pytz import UTC
 from django.conf import settings
 from django.utils.translation import ugettext as _
 from django.core.urlresolvers import reverse
+from django_comment_common.models import assign_default_role
+from django_comment_common.utils import seed_permissions_roles
 
 from xmodule.contentstore.content import StaticContent
 from xmodule.modulestore import ModuleStoreEnum
@@ -16,6 +18,8 @@ from xmodule.modulestore.django import modulestore
 from xmodule.modulestore.exceptions import ItemNotFoundError
 from opaque_keys.edx.keys import UsageKey, CourseKey
 from student.roles import CourseInstructorRole, CourseStaffRole
+from student.models import CourseEnrollment
+from student import auth
 
 
 log = logging.getLogger(__name__)
@@ -26,25 +30,58 @@ NOTES_PANEL = {"name": _("My Notes"), "type": "notes"}
 EXTRA_TAB_PANELS = dict([(p['type'], p) for p in [OPEN_ENDED_PANEL, NOTES_PANEL]])
 
 
-def delete_course_and_groups(course_id, user_id):
+def add_instructor(course_key, requesting_user, new_instructor):
     """
-    This deletes the courseware associated with a course_id as well as cleaning update_item
+    Adds given user as instructor and staff to the given course,
+    after verifying that the requesting_user has permission to do so.
+    """
+    # can't use auth.add_users here b/c it requires user to already have Instructor perms in this course
+    CourseInstructorRole(course_key).add_users(new_instructor)
+    auth.add_users(requesting_user, CourseStaffRole(course_key), new_instructor)
+
+
+def initialize_permissions(course_key, user_who_created_course):
+    """
+    Initializes a new course by enrolling the course creator as a student,
+    and initializing Forum by seeding its permissions and assigning default roles.
+    """
+    # seed the forums
+    seed_permissions_roles(course_key)
+
+    # auto-enroll the course creator in the course so that "View Live" will work.
+    CourseEnrollment.enroll(user_who_created_course, course_key)
+
+    # set default forum roles (assign 'Student' role)
+    assign_default_role(course_key, user_who_created_course)
+
+
+def remove_all_instructors(course_key):
+    """
+    Removes given user as instructor and staff to the given course,
+    after verifying that the requesting_user has permission to do so.
+    """
+    staff_role = CourseStaffRole(course_key)
+    staff_role.remove_users(*staff_role.users_with_role())
+    instructor_role = CourseInstructorRole(course_key)
+    instructor_role.remove_users(*instructor_role.users_with_role())
+
+
+def delete_course_and_groups(course_key, user_id):
+    """
+    This deletes the courseware associated with a course_key as well as cleaning update_item
     the various user table stuff (groups, permissions, etc.)
     """
     module_store = modulestore()
 
-    with module_store.bulk_write_operations(course_id):
-        module_store.delete_course(course_id, user_id)
+    with module_store.bulk_write_operations(course_key):
+        module_store.delete_course(course_key, user_id)
 
         print 'removing User permissions from course....'
         # in the django layer, we need to remove all the user permissions groups associated with this course
         try:
-            staff_role = CourseStaffRole(course_id)
-            staff_role.remove_users(*staff_role.users_with_role())
-            instructor_role = CourseInstructorRole(course_id)
-            instructor_role.remove_users(*instructor_role.users_with_role())
+            remove_all_instructors(course_key)
         except Exception as err:
-            log.error("Error in deleting course groups for {0}: {1}".format(course_id, err))
+            log.error("Error in deleting course groups for {0}: {1}".format(course_key, err))
 
 
 def get_lms_link_for_item(location, preview=False):
@@ -64,19 +101,19 @@ def get_lms_link_for_item(location, preview=False):
     else:
         lms_base = settings.LMS_BASE
 
-    return u"//{lms_base}/courses/{course_id}/jump_to/{location}".format(
+    return u"//{lms_base}/courses/{course_key}/jump_to/{location}".format(
         lms_base=lms_base,
-        course_id=location.course_key.to_deprecated_string(),
+        course_key=location.course_key.to_deprecated_string(),
         location=location.to_deprecated_string(),
     )
 
 
-def get_lms_link_for_about_page(course_id):
+def get_lms_link_for_about_page(course_key):
     """
     Returns the url to the course about page from the location tuple.
     """
 
-    assert(isinstance(course_id, CourseKey))
+    assert(isinstance(course_key, CourseKey))
 
     if settings.FEATURES.get('ENABLE_MKTG_SITE', False):
         if not hasattr(settings, 'MKTG_URLS'):
@@ -101,9 +138,9 @@ def get_lms_link_for_about_page(course_id):
     else:
         return None
 
-    return u"//{about_base_url}/courses/{course_id}/about".format(
+    return u"//{about_base_url}/courses/{course_key}/about".format(
         about_base_url=about_base,
-        course_id=course_id.to_deprecated_string()
+        course_key=course_key.to_deprecated_string()
     )
 
 
diff --git a/cms/djangoapps/contentstore/views/course.py b/cms/djangoapps/contentstore/views/course.py
index 4f76a6b8bd..736965777e 100644
--- a/cms/djangoapps/contentstore/views/course.py
+++ b/cms/djangoapps/contentstore/views/course.py
@@ -30,14 +30,16 @@ from opaque_keys.edx.locations import Location, SlashSeparatedCourseKey
 
 from contentstore.course_info_model import get_course_updates, update_course_updates, delete_course_update
 from contentstore.utils import (
+    add_instructor,
+    initialize_permissions,
     get_lms_link_for_item,
     add_extra_panel_tab,
     remove_extra_panel_tab,
     reverse_course_url,
     reverse_usage_url,
+    reverse_url,
 )
 from models.settings.course_details import CourseDetails, CourseSettingsEncoder
-
 from models.settings.course_grading import CourseGradingModel
 from models.settings.course_metadata import CourseMetadata
 from util.json_request import expect_json
@@ -50,21 +52,20 @@ from .component import (
     ADVANCED_COMPONENT_POLICY_KEY,
     SPLIT_TEST_COMPONENT_TYPE,
 )
-
-from django_comment_common.models import assign_default_role
-from django_comment_common.utils import seed_permissions_roles
-
-from student.models import CourseEnrollment
-from student.roles import CourseRole, UserBasedRole
+from .tasks import rerun_course
 
 from opaque_keys.edx.keys import CourseKey
 from course_creators.views import get_course_creator_status, add_user_with_status_unrequested
 from contentstore import utils
-from student.roles import CourseInstructorRole, CourseStaffRole, CourseCreatorRole, GlobalStaff
+from student.roles import (
+    CourseInstructorRole, CourseStaffRole, CourseCreatorRole, GlobalStaff, UserBasedRole
+)
 from student import auth
+from course_action_state.models import CourseRerunState, CourseRerunUIStateManager
 
 from microsite_configuration import microsite
 
+
 __all__ = ['course_info_handler', 'course_handler', 'course_info_update_handler',
            'settings_handler',
            'grading_handler',
@@ -123,7 +124,7 @@ def course_handler(request, course_key_string=None):
         if request.method == 'GET':
             return JsonResponse(_course_json(request, CourseKey.from_string(course_key_string)))
         elif request.method == 'POST':  # not sure if this is only post. If one will have ids, it goes after access
-            return create_new_course(request)
+            return _create_or_rerun_course(request)
         elif not has_course_access(request.user, CourseKey.from_string(course_key_string)):
             raise PermissionDenied()
         elif request.method == 'PUT':
@@ -171,26 +172,36 @@ def _accessible_courses_list(request):
     """
     List all courses available to the logged in user by iterating through all the courses
     """
-    courses = modulestore().get_courses()
+    def course_permission_filter(course_key):
+        """Filter out courses that user doesn't have access to"""
+        if GlobalStaff().has_user(request.user):
+            return True
+        else:
+            return has_course_access(request.user, course_key)
 
-    # filter out courses that we don't have access to
     def course_filter(course):
         """
-        Get courses to which this user has access
+        Filter out unusable and inaccessible courses
         """
         if isinstance(course, ErrorDescriptor):
             return False
 
-        if GlobalStaff().has_user(request.user):
-            return course.location.course != 'templates'
+        # pylint: disable=fixme
+        # TODO remove this condition when templates purged from db
+        if course.location.course == 'templates':
+            return False
 
-        return (has_course_access(request.user, course.id)
-                # pylint: disable=fixme
-                # TODO remove this condition when templates purged from db
-                and course.location.course != 'templates'
-                )
-    courses = filter(course_filter, courses)
-    return courses
+        return course_permission_filter(course.id)
+
+    courses = filter(course_filter, modulestore().get_courses())
+    unsucceeded_course_actions = [
+        crs for crs in
+        CourseRerunState.objects.find_all(
+            exclude_args={'state': CourseRerunUIStateManager.State.SUCCEEDED}, should_display=True
+        )
+        if course_permission_filter(crs.course_key)
+    ]
+    return courses, unsucceeded_course_actions
 
 
 def _accessible_courses_list_from_groups(request):
@@ -198,6 +209,7 @@ def _accessible_courses_list_from_groups(request):
     List all courses available to the logged in user by reversing access group names
     """
     courses_list = {}
+    unsucceeded_course_actions = []
 
     instructor_courses = UserBasedRole(request.user, CourseInstructorRole.ROLE).courses_with_role()
     staff_courses = UserBasedRole(request.user, CourseStaffRole.ROLE).courses_with_role()
@@ -209,6 +221,15 @@ def _accessible_courses_list_from_groups(request):
             # If the course_access does not have a course_id, it's an org-based role, so we fall back
             raise AccessListFallback
         if course_key not in courses_list:
+            # check for any course action state for this course
+            unsucceeded_course_actions.extend(
+                CourseRerunState.objects.find_all(
+                    exclude_args={'state': CourseRerunUIStateManager.State.SUCCEEDED},
+                    should_display=True,
+                    course_key=course_key,
+                )
+            )
+            # check for the course itself
             try:
                 course = modulestore().get_course(course_key)
             except ItemNotFoundError:
@@ -218,7 +239,7 @@ def _accessible_courses_list_from_groups(request):
                 # ignore deleted or errored courses
                 courses_list[course_key] = course
 
-    return courses_list.values()
+    return courses_list.values(), unsucceeded_course_actions
 
 
 @login_required
@@ -231,14 +252,14 @@ def course_listing(request):
     """
     if GlobalStaff().has_user(request.user):
         # user has global access so no need to get courses from django groups
-        courses = _accessible_courses_list(request)
+        courses, unsucceeded_course_actions = _accessible_courses_list(request)
     else:
         try:
-            courses = _accessible_courses_list_from_groups(request)
+            courses, unsucceeded_course_actions = _accessible_courses_list_from_groups(request)
         except AccessListFallback:
             # user have some old groups or there was some error getting courses from django groups
             # so fallback to iterating through all courses
-            courses = _accessible_courses_list(request)
+            courses, unsucceeded_course_actions = _accessible_courses_list(request)
 
     def format_course_for_view(course):
         """
@@ -253,8 +274,17 @@ def course_listing(request):
             course.location.name
         )
 
+    # remove any courses in courses that are also in the unsucceeded_course_actions list
+    unsucceeded_action_course_keys = [uca.course_key for uca in unsucceeded_course_actions]
+    courses = [
+        format_course_for_view(c)
+        for c in courses
+        if not isinstance(c, ErrorDescriptor) and (c.id not in unsucceeded_action_course_keys)
+    ]
+
     return render_to_response('index.html', {
-        'courses': [format_course_for_view(c) for c in courses if not isinstance(c, ErrorDescriptor)],
+        'courses': courses,
+        'unsucceeded_course_actions': unsucceeded_course_actions,
         'user': request.user,
         'request_course_creator_url': reverse('contentstore.views.request_course_creator'),
         'course_creator_status': _get_course_creator_status(request.user),
@@ -289,80 +319,36 @@ def course_index(request, course_key):
 
 
 @expect_json
-def create_new_course(request):
+def _create_or_rerun_course(request):
     """
-    Create a new course.
-
-    Returns the URL for the course overview page.
+    To be called by requests that create a new destination course (i.e., create_new_course and rerun_course)
+    Returns the destination course_key and overriding fields for the new course.
+    Raises InvalidLocationError and InvalidKeyError
     """
     if not auth.has_access(request.user, CourseCreatorRole()):
         raise PermissionDenied()
 
-    org = request.json.get('org')
-    number = request.json.get('number')
-    display_name = request.json.get('display_name')
-    run = request.json.get('run')
-
-    # allow/disable unicode characters in course_id according to settings
-    if not settings.FEATURES.get('ALLOW_UNICODE_COURSE_ID'):
-        if _has_non_ascii_characters(org) or _has_non_ascii_characters(number) or _has_non_ascii_characters(run):
-            return JsonResponse(
-                {'error': _('Special characters not allowed in organization, course number, and course run.')},
-                status=400
-            )
-
     try:
+        org = request.json.get('org')
+        number = request.json.get('number')
+        display_name = request.json.get('display_name')
+        run = request.json.get('run')
+
+        # allow/disable unicode characters in course_id according to settings
+        if not settings.FEATURES.get('ALLOW_UNICODE_COURSE_ID'):
+            if _has_non_ascii_characters(org) or _has_non_ascii_characters(number) or _has_non_ascii_characters(run):
+                return JsonResponse(
+                    {'error': _('Special characters not allowed in organization, course number, and course run.')},
+                    status=400
+                )
+
         course_key = SlashSeparatedCourseKey(org, number, run)
+        fields = {'display_name': display_name} if display_name is not None else {}
 
-        # instantiate the CourseDescriptor and then persist it
-        # note: no system to pass
-        if display_name is None:
-            metadata = {}
+        if 'source_course_key' in request.json:
+            return _rerun_course(request, course_key, fields)
         else:
-            metadata = {'display_name': display_name}
-
-        # Set a unique wiki_slug for newly created courses. To maintain active wiki_slugs for
-        # existing xml courses this cannot be changed in CourseDescriptor.
-        # # TODO get rid of defining wiki slug in this org/course/run specific way and reconcile
-        # w/ xmodule.course_module.CourseDescriptor.__init__
-        wiki_slug = u"{0}.{1}.{2}".format(course_key.org, course_key.course, course_key.run)
-        definition_data = {'wiki_slug': wiki_slug}
-
-        # Create the course then fetch it from the modulestore
-        # Check if role permissions group for a course named like this already exists
-        # Important because role groups are case insensitive
-        if CourseRole.course_group_already_exists(course_key):
-            raise InvalidLocationError()
-
-        fields = {}
-        fields.update(definition_data)
-        fields.update(metadata)
-
-        # Creating the course raises InvalidLocationError if an existing course with this org/name is found
-        new_course = modulestore().create_course(
-            course_key.org,
-            course_key.course,
-            course_key.run,
-            request.user.id,
-            fields=fields,
-        )
-
-        # can't use auth.add_users here b/c it requires request.user to already have Instructor perms in this course
-        # however, we can assume that b/c this user had authority to create the course, the user can add themselves
-        CourseInstructorRole(new_course.id).add_users(request.user)
-        auth.add_users(request.user, CourseStaffRole(new_course.id), request.user)
-
-        # seed the forums
-        seed_permissions_roles(new_course.id)
-
-        # auto-enroll the course creator in the course so that "View Live" will
-        # work.
-        CourseEnrollment.enroll(request.user, new_course.id)
-        _users_assign_default_role(new_course.id)
-
-        return JsonResponse({
-            'url': reverse_course_url('course_handler', new_course.id)
-        })
+            return _create_new_course(request, course_key, fields)
 
     except InvalidLocationError:
         return JsonResponse({
@@ -384,13 +370,62 @@ def create_new_course(request):
         )
 
 
-def _users_assign_default_role(course_id):
+def _create_new_course(request, course_key, fields):
     """
-    Assign 'Student' role to all previous users (if any) for this course
+    Create a new course.
+    Returns the URL for the course overview page.
     """
-    enrollments = CourseEnrollment.objects.filter(course_id=course_id)
-    for enrollment in enrollments:
-        assign_default_role(course_id, enrollment.user)
+    # Set a unique wiki_slug for newly created courses. To maintain active wiki_slugs for
+    # existing xml courses this cannot be changed in CourseDescriptor.
+    # # TODO get rid of defining wiki slug in this org/course/run specific way and reconcile
+    # w/ xmodule.course_module.CourseDescriptor.__init__
+    wiki_slug = u"{0}.{1}.{2}".format(course_key.org, course_key.course, course_key.run)
+    definition_data = {'wiki_slug': wiki_slug}
+    fields.update(definition_data)
+
+    # Creating the course raises InvalidLocationError if an existing course with this org/name is found
+    new_course = modulestore().create_course(
+        course_key.org,
+        course_key.course,
+        course_key.run,
+        request.user.id,
+        fields=fields,
+    )
+
+    # Make sure user has instructor and staff access to the new course
+    add_instructor(new_course.id, request.user, request.user)
+
+    # Initialize permissions for user in the new course
+    initialize_permissions(new_course.id, request.user)
+
+    return JsonResponse({
+        'url': reverse_course_url('course_handler', new_course.id)
+    })
+
+
+def _rerun_course(request, destination_course_key, fields):
+    """
+    Reruns an existing course.
+    Returns the URL for the course listing page.
+    """
+    source_course_key = CourseKey.from_string(request.json.get('source_course_key'))
+
+    # verify user has access to the original course
+    if not has_course_access(request.user, source_course_key):
+        raise PermissionDenied()
+
+    # Make sure user has instructor and staff access to the destination course
+    # so the user can see the updated status for that course
+    add_instructor(destination_course_key, request.user, request.user)
+
+    # Mark the action as initiated
+    CourseRerunState.objects.initiated(source_course_key, destination_course_key, request.user)
+
+    # Rerun the course as a new celery task
+    rerun_course.delay(source_course_key, destination_course_key, request.user.id, fields)
+
+    # Return course listing page
+    return JsonResponse({'url': reverse_url('course_handler')})
 
 
 # pylint: disable=unused-argument
diff --git a/cms/djangoapps/contentstore/views/tasks.py b/cms/djangoapps/contentstore/views/tasks.py
new file mode 100644
index 0000000000..473e0c46dc
--- /dev/null
+++ b/cms/djangoapps/contentstore/views/tasks.py
@@ -0,0 +1,32 @@
+"""
+This file contains celery tasks for contentstore views
+"""
+
+from celery.task import task
+from django.contrib.auth.models import User
+from xmodule.modulestore.django import modulestore
+from course_action_state.models import CourseRerunState
+from contentstore.utils import initialize_permissions
+
+
+@task()
+def rerun_course(source_course_key, destination_course_key, user_id, fields=None):
+    """
+    Reruns a course in a new celery task.
+    """
+    try:
+        modulestore().clone_course(source_course_key, destination_course_key, user_id, fields=fields)
+
+        # set initial permissions for the user to access the course.
+        initialize_permissions(destination_course_key, User.objects.get(id=user_id))
+
+        # update state: Succeeded
+        CourseRerunState.objects.succeeded(course_key=destination_course_key)
+
+    # catch all exceptions so we can update the state and properly cleanup the course.
+    except Exception as exc:  # pylint: disable=broad-except
+        # update state: Failed
+        CourseRerunState.objects.failed(course_key=destination_course_key, exception=exc)
+
+        # cleanup any remnants of the course
+        modulestore().delete_course(destination_course_key, user_id)