diff --git a/lms/djangoapps/program_enrollments/api/tests/test_reading.py b/lms/djangoapps/program_enrollments/api/tests/test_reading.py
index 6f4b3ab0e96c6d43f5d82f3078eaa87bda43ec95..44a8604e5cdd192411e012a12ee2ffb848ba49ff 100644
--- a/lms/djangoapps/program_enrollments/api/tests/test_reading.py
+++ b/lms/djangoapps/program_enrollments/api/tests/test_reading.py
@@ -90,6 +90,7 @@ class ProgramEnrollmentReadingTests(TestCase):
             (cls.user_3, cls.ext_3, cls.program_uuid_y, cls.curriculum_uuid_c, PEStatuses.CANCELED),  # 7
             (None, cls.ext_4, cls.program_uuid_y, cls.curriculum_uuid_c, PEStatuses.ENROLLED),        # 8
             (cls.user_1, None, cls.program_uuid_x, cls.curriculum_uuid_b, PEStatuses.SUSPENDED),      # 9
+            (cls.user_2, None, cls.program_uuid_y, cls.curriculum_uuid_c, PEStatuses.ENDED),          # 10
         ]
         for user, external_user_key, program_uuid, curriculum_uuid, status in enrollment_test_data:
             ProgramEnrollmentFactory(
@@ -148,6 +149,7 @@ class ProgramEnrollmentReadingTests(TestCase):
         # Specifying no curriculum (because ext_6 only has Program Y
         # enrollments in one curriculum, so it's not ambiguous).
         (program_uuid_y, None, None, ext_6, 6),
+        (program_uuid_y, None, username_2, None, 10),
     )
     @ddt.unpack
     def test_get_program_enrollment(
diff --git a/lms/djangoapps/program_enrollments/api/tests/test_writing.py b/lms/djangoapps/program_enrollments/api/tests/test_writing.py
index f3f2ba2083c4a509b0cc781bf18b5b2532ec18df..5db2decbca9fb918308d76c96f99b03708e4b2ae 100644
--- a/lms/djangoapps/program_enrollments/api/tests/test_writing.py
+++ b/lms/djangoapps/program_enrollments/api/tests/test_writing.py
@@ -1,12 +1,75 @@
 """
-(Future home of) Tests for program enrollment writing Python API.
+Tests for program enrollment writing Python API.
 
-Currently, we do not directly unit test the functions in api/writing.py.
+Currently, we do not directly unit test the functions in api/writing.py extensively.
 This is okay for now because they are all used in
 `rest_api.v1.views` and is thus tested through `rest_api.v1.tests.test_views`.
 Eventually it would be good to directly test the Python API function and just use
 mocks in the view tests.
-This file serves as a placeholder and reminder to do that the next time there
-is development on the program_enrollments writing API.
 """
 from __future__ import absolute_import, unicode_literals
+
+from uuid import UUID
+
+from organizations.tests.factories import OrganizationFactory
+from django.core.cache import cache
+
+from lms.djangoapps.program_enrollments.constants import ProgramEnrollmentStatuses as PEStatuses
+from lms.djangoapps.program_enrollments.models import ProgramEnrollment
+from openedx.core.djangoapps.catalog.cache import PROGRAM_CACHE_KEY_TPL
+from openedx.core.djangoapps.catalog.tests.factories import OrganizationFactory as CatalogOrganizationFactory
+from openedx.core.djangoapps.catalog.tests.factories import ProgramFactory
+from openedx.core.djangolib.testing.utils import CacheIsolationTestCase
+from third_party_auth.tests.factories import SAMLProviderConfigFactory
+
+from ..writing import (
+    write_program_enrollments
+)
+
+
+class WritingProgramEnrollmentTest(CacheIsolationTestCase):
+    """
+    Test cases for program enrollment writing functions.
+    """
+    ENABLED_CACHES = ['default']
+
+    organization_key = 'test'
+
+    program_uuid_x = UUID('dddddddd-5f48-493d-9910-84e1d36c657f')
+
+    curriculum_uuid_a = UUID('aaaaaaaa-bd26-4370-94b8-b4063858210b')
+
+    user_0 = 'user-0'
+
+    def setUp(self):
+        """
+        Set up test data
+        """
+        super(WritingProgramEnrollmentTest, self).setUp()
+        catalog_org = CatalogOrganizationFactory.create(key=self.organization_key)
+        program = ProgramFactory.create(
+            uuid=self.program_uuid_x,
+            authoring_organizations=[catalog_org]
+        )
+        organization = OrganizationFactory.create(short_name=self.organization_key)
+        SAMLProviderConfigFactory.create(organization=organization)
+        cache.set(PROGRAM_CACHE_KEY_TPL.format(uuid=self.program_uuid_x), program, None)
+
+    def test_write_program_enrollments_status_ended(self):
+        """
+        Successfully updates program enrollment to status ended if requested
+        """
+        assert ProgramEnrollment.objects.count() == 0
+        write_program_enrollments(self.program_uuid_x, [{
+            'external_user_key': self.user_0,
+            'status': PEStatuses.PENDING,
+            'curriculum_uuid': self.curriculum_uuid_a,
+        }], True, False)
+        assert ProgramEnrollment.objects.count() == 1
+        write_program_enrollments(self.program_uuid_x, [{
+            'external_user_key': self.user_0,
+            'status': PEStatuses.ENDED,
+            'curriculum_uuid': self.curriculum_uuid_a,
+        }], False, True)
+        assert ProgramEnrollment.objects.count() == 1
+        assert ProgramEnrollment.objects.filter(status=PEStatuses.ENDED).exists()
diff --git a/lms/djangoapps/program_enrollments/api/writing.py b/lms/djangoapps/program_enrollments/api/writing.py
index f318418e0c82f73e36ecfd20ba5fb36cbf9bb37e..6c9fdab91c430c20cfd1f3288caf506608d9b676 100644
--- a/lms/djangoapps/program_enrollments/api/writing.py
+++ b/lms/djangoapps/program_enrollments/api/writing.py
@@ -220,13 +220,13 @@ def write_program_course_enrollments(
     to_save = []
     for external_key, request in requests_by_key.items():
         status = request['status']
+        if status not in ProgramCourseEnrollmentStatuses.__ALL__:
+            results[external_key] = ProgramCourseOpStatuses.INVALID_STATUS
+            continue
         program_enrollment = program_enrollments_by_key.get(external_key)
         if not program_enrollment:
             results[external_key] = ProgramCourseOpStatuses.NOT_IN_PROGRAM
             continue
-        if status not in ProgramCourseEnrollmentStatuses.__ALL__:
-            results[external_key] = ProgramCourseOpStatuses.INVALID_STATUS
-            continue
         existing_course_enrollment = existing_course_enrollments_by_key[external_key]
         if existing_course_enrollment:
             if not update:
diff --git a/lms/djangoapps/program_enrollments/constants.py b/lms/djangoapps/program_enrollments/constants.py
index 6b1e0efaeb6adb8cab8d6ff1d5244427d80fca60..b02ba1834d1f99be6c1775a5f5272b4ca18721ae 100644
--- a/lms/djangoapps/program_enrollments/constants.py
+++ b/lms/djangoapps/program_enrollments/constants.py
@@ -15,8 +15,9 @@ class ProgramEnrollmentStatuses(object):
     PENDING = 'pending'
     SUSPENDED = 'suspended'
     CANCELED = 'canceled'
+    ENDED = 'ended'
     __ACTIVE__ = (ENROLLED, PENDING)
-    __ALL__ = (ENROLLED, PENDING, SUSPENDED, CANCELED)
+    __ALL__ = (ENROLLED, PENDING, SUSPENDED, CANCELED, ENDED)
 
     # Note: Any changes to this value will trigger a migration on
     # ProgramEnrollment!
diff --git a/lms/djangoapps/program_enrollments/migrations/0008_add_ended_programenrollment_status.py b/lms/djangoapps/program_enrollments/migrations/0008_add_ended_programenrollment_status.py
new file mode 100644
index 0000000000000000000000000000000000000000..1e60ae929ad69e6310d8a61be1fb49eb011f004d
--- /dev/null
+++ b/lms/djangoapps/program_enrollments/migrations/0008_add_ended_programenrollment_status.py
@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+# Generated by Django 1.11.24 on 2019-10-09 16:49
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('program_enrollments', '0007_waiting_programcourseenrollment_constraint'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='historicalprogramenrollment',
+            name='status',
+            field=models.CharField(choices=[('enrolled', 'enrolled'), ('pending', 'pending'), ('suspended', 'suspended'), ('canceled', 'canceled'), ('ended', 'ended')], max_length=9),
+        ),
+        migrations.AlterField(
+            model_name='programenrollment',
+            name='status',
+            field=models.CharField(choices=[('enrolled', 'enrolled'), ('pending', 'pending'), ('suspended', 'suspended'), ('canceled', 'canceled'), ('ended', 'ended')], max_length=9),
+        ),
+    ]
diff --git a/lms/djangoapps/program_enrollments/rest_api/v1/tests/test_views.py b/lms/djangoapps/program_enrollments/rest_api/v1/tests/test_views.py
index 0b91f00ce543c0e39beb1e5770fb8caa1be5d6ea..fdf56edebb07c2fdbc40a1f67521acc19b65f431 100644
--- a/lms/djangoapps/program_enrollments/rest_api/v1/tests/test_views.py
+++ b/lms/djangoapps/program_enrollments/rest_api/v1/tests/test_views.py
@@ -448,8 +448,6 @@ class ProgramEnrollmentsPostTests(ProgramEnrollmentsWriteMixin, APITestCase):
     """
     add_uuid = True
 
-    view_name = 'programs_api:v1:program_enrollments'
-
     def setUp(self):
         super(ProgramEnrollmentsPostTests, self).setUp()
         self.request = self.client.post
@@ -460,9 +458,9 @@ class ProgramEnrollmentsPostTests(ProgramEnrollmentsWriteMixin, APITestCase):
         ProgramEnrollment.objects.all().delete()
 
     def test_successful_program_enrollments_no_existing_user(self):
-        statuses = ['pending', 'enrolled', 'pending']
-        external_user_keys = ['abc1', 'efg2', 'hij3']
-        curriculum_uuids = [self.curriculum_uuid, self.curriculum_uuid, uuid4()]
+        statuses = ['pending', 'enrolled', 'pending', 'ended']
+        external_user_keys = ['abc1', 'efg2', 'hij3', 'klm4']
+        curriculum_uuids = [self.curriculum_uuid, self.curriculum_uuid, uuid4(), uuid4()]
         post_data = [
             {
                 REQUEST_STUDENT_KEY: e,
@@ -478,7 +476,7 @@ class ProgramEnrollmentsPostTests(ProgramEnrollmentsWriteMixin, APITestCase):
 
         self.assertEqual(response.status_code, 200)
 
-        for i in range(3):
+        for i in range(4):
             enrollment = ProgramEnrollment.objects.get(external_user_key=external_user_keys[i])
 
             self.assertEqual(enrollment.external_user_key, external_user_keys[i])
diff --git a/lms/djangoapps/program_enrollments/rest_api/v1/views.py b/lms/djangoapps/program_enrollments/rest_api/v1/views.py
index b9d18e9c870972a46863c2afa99903e4a1754506..24e0a096c00fd83a4f58f2d2e8b76dbe40c58b0c 100644
--- a/lms/djangoapps/program_enrollments/rest_api/v1/views.py
+++ b/lms/djangoapps/program_enrollments/rest_api/v1/views.py
@@ -192,7 +192,7 @@ class ProgramEnrollmentsView(
     Request body:
         * The request body will be a list of one or more students to enroll with the following schema:
             {
-                'status': A choice of the following statuses: ['enrolled', 'pending', 'canceled', 'suspended'],
+                'status': A choice of the following statuses: ['enrolled', 'pending', 'canceled', 'suspended', 'ended'],
                 student_key: string representation of a learner in partner systems,
                 'curriculum_uuid': string representation of a curriculum
             }
@@ -226,10 +226,12 @@ class ProgramEnrollmentsView(
                 * 'pending'
                 * 'canceled'
                 * 'suspended'
+                * 'ended'
             * failure statuses:
                 * 'duplicated' - the request body listed the same learner twice
                 * 'conflict' - there is an existing enrollment for that learner, curriculum and program combo
-                * 'invalid-status' - a status other than 'enrolled', 'pending', 'canceled', 'suspended' was entered
+                * 'invalid-status' - a status other than 'enrolled', 'pending', 'canceled', 'suspended',
+                  or 'ended' was entered
       * 200: OK - All students were successfully enrolled.
         * Example json response:
             {
@@ -260,7 +262,13 @@ class ProgramEnrollmentsView(
     Request body:
         * The request body will be a list of one or more students with their updated enrollment status:
             {
-                'status': A choice of the following statuses: ['enrolled', 'pending', 'canceled', 'suspended'],
+                'status': A choice of the following statuses: [
+                    'enrolled',
+                    'pending',
+                    'canceled',
+                    'suspended',
+                    'ended',
+                ],
                 student_key: string representation of a learner in partner systems
             }
         Example:
@@ -289,10 +297,12 @@ class ProgramEnrollmentsView(
                 * 'pending'
                 * 'canceled'
                 * 'suspended'
+                * 'ended'
             * failure statuses:
                 * 'duplicated' - the request body listed the same learner twice
                 * 'conflict' - there is an existing enrollment for that learner, curriculum and program combo
-                * 'invalid-status' - a status other than 'enrolled', 'pending', 'canceled', 'suspended' was entered
+                * 'invalid-status' - a status other than 'enrolled', 'pending', 'canceled', 'suspended', 'ended'
+                                     was entered
                 * 'not-in-program' - the user is not in the program and cannot be updated
       * 200: OK - All students were successfully enrolled.
         * Example json response: