Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

revert: remove learner_recommendations app

Browse Source
This commit is contained in:
mubbsharanwar
2023-11-21 14:36:58 +05:00
committed by Mubbshar Anwar
parent 1734fdc0dc
commit 487b870ae4
14 changed files with 1 additions and 2709 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/pylint-checks.yml vendored
View File

@@ -16,7 +16,7 @@ jobs:
- module-name: lms-1
path: "--django-settings-module=lms.envs.test lms/djangoapps/badges/ lms/djangoapps/branding/ lms/djangoapps/bulk_email/ lms/djangoapps/bulk_enroll/ lms/djangoapps/bulk_user_retirement/ lms/djangoapps/ccx/ lms/djangoapps/certificates/ lms/djangoapps/commerce/ lms/djangoapps/course_api/ lms/djangoapps/course_blocks/ lms/djangoapps/course_home_api/ lms/djangoapps/course_wiki/ lms/djangoapps/coursewarehistoryextended/ lms/djangoapps/debug/ lms/djangoapps/courseware/ lms/djangoapps/course_goals/ lms/djangoapps/rss_proxy/"
- module-name: lms-2
path: "--django-settings-module=lms.envs.test lms/djangoapps/gating/ lms/djangoapps/grades/ lms/djangoapps/instructor/ lms/djangoapps/instructor_analytics/ lms/djangoapps/discussion/ lms/djangoapps/edxnotes/ lms/djangoapps/email_marketing/ lms/djangoapps/experiments/ lms/djangoapps/instructor_task/ lms/djangoapps/learner_dashboard/ lms/djangoapps/learner_recommendations/ lms/djangoapps/learner_home/ lms/djangoapps/lms_initialization/ lms/djangoapps/lms_xblock/ lms/djangoapps/lti_provider/ lms/djangoapps/mailing/ lms/djangoapps/mobile_api/ lms/djangoapps/monitoring/ lms/djangoapps/ora_staff_grader/ lms/djangoapps/program_enrollments/ lms/djangoapps/rss_proxy lms/djangoapps/static_template_view/ lms/djangoapps/staticbook/ lms/djangoapps/support/ lms/djangoapps/survey/ lms/djangoapps/teams/ lms/djangoapps/tests/ lms/djangoapps/user_tours/ lms/djangoapps/verify_student/ lms/djangoapps/mfe_config_api/ lms/envs/ lms/lib/ lms/tests.py"
path: "--django-settings-module=lms.envs.test lms/djangoapps/gating/ lms/djangoapps/grades/ lms/djangoapps/instructor/ lms/djangoapps/instructor_analytics/ lms/djangoapps/discussion/ lms/djangoapps/edxnotes/ lms/djangoapps/email_marketing/ lms/djangoapps/experiments/ lms/djangoapps/instructor_task/ lms/djangoapps/learner_dashboard/ lms/djangoapps/learner_home/ lms/djangoapps/lms_initialization/ lms/djangoapps/lms_xblock/ lms/djangoapps/lti_provider/ lms/djangoapps/mailing/ lms/djangoapps/mobile_api/ lms/djangoapps/monitoring/ lms/djangoapps/ora_staff_grader/ lms/djangoapps/program_enrollments/ lms/djangoapps/rss_proxy lms/djangoapps/static_template_view/ lms/djangoapps/staticbook/ lms/djangoapps/support/ lms/djangoapps/survey/ lms/djangoapps/teams/ lms/djangoapps/tests/ lms/djangoapps/user_tours/ lms/djangoapps/verify_student/ lms/djangoapps/mfe_config_api/ lms/envs/ lms/lib/ lms/tests.py"
- module-name: openedx-1
path: "--django-settings-module=lms.envs.test openedx/core/types/ openedx/core/djangoapps/ace_common/ openedx/core/djangoapps/agreements/ openedx/core/djangoapps/api_admin/ openedx/core/djangoapps/auth_exchange/ openedx/core/djangoapps/bookmarks/ openedx/core/djangoapps/cache_toolbox/ openedx/core/djangoapps/catalog/ openedx/core/djangoapps/ccxcon/ openedx/core/djangoapps/commerce/ openedx/core/djangoapps/common_initialization/ openedx/core/djangoapps/common_views/ openedx/core/djangoapps/config_model_utils/ openedx/core/djangoapps/content/ openedx/core/djangoapps/content_libraries/ openedx/core/djangoapps/content_staging/ openedx/core/djangoapps/contentserver/ openedx/core/djangoapps/cookie_metadata/ openedx/core/djangoapps/cors_csrf/ openedx/core/djangoapps/course_apps/ openedx/core/djangoapps/course_date_signals/ openedx/core/djangoapps/course_groups/ openedx/core/djangoapps/courseware_api/ openedx/core/djangoapps/crawlers/ openedx/core/djangoapps/credentials/ openedx/core/djangoapps/credit/ openedx/core/djangoapps/dark_lang/ openedx/core/djangoapps/debug/ openedx/core/djangoapps/demographics/ openedx/core/djangoapps/discussions/ openedx/core/djangoapps/django_comment_common/ openedx/core/djangoapps/embargo/ openedx/core/djangoapps/enrollments/ openedx/core/djangoapps/external_user_ids/ openedx/core/djangoapps/zendesk_proxy/ openedx/core/djangolib/ openedx/core/lib/ openedx/core/tests/ openedx/core/djangoapps/course_live/"
- module-name: openedx-2

1
.github/workflows/unit-test-shards.json vendored
View File

@@ -53,7 +53,6 @@
"lms/djangoapps/instructor_task/",
"lms/djangoapps/learner_dashboard/",
"lms/djangoapps/learner_home/",
"lms/djangoapps/learner_recommendations/",
"lms/djangoapps/lms_initialization/",
"lms/djangoapps/lms_xblock/",
"lms/djangoapps/lti_provider/",

0
lms/djangoapps/learner_recommendations/__init__.py
View File

142
lms/djangoapps/learner_recommendations/serializers.py
View File

@@ -1,142 +0,0 @@
"""
Serializers for learner recommendations APIs.
"""
from rest_framework import serializers
class ActiveCourseRunSerializer(serializers.Serializer):
"""Serializer for active course run for course about page recommendations API"""
key = serializers.CharField()
marketingUrl = serializers.URLField(source="marketing_url")
class CourseOwnersSerializer(serializers.Serializer):
"""Serializer for course owners for course about page recommendations API"""
key = serializers.CharField()
name = serializers.CharField()
logoImageUrl = serializers.URLField(source="logo_image_url")
class CourseImageSerializer(serializers.Serializer):
"""Serializer for course image for course about page recommendations API"""
src = serializers.URLField()
class RecommendedCourseSerializer(serializers.Serializer):
"""Serializer for a recommended course from the recommendation engine"""
key = serializers.CharField()
uuid = serializers.UUIDField()
title = serializers.CharField()
image = CourseImageSerializer()
prospectusPath = serializers.SerializerMethodField()
owners = serializers.ListField(
child=CourseOwnersSerializer(), allow_empty=True
)
activeCourseRun = ActiveCourseRunSerializer(source="active_course_run")
def get_prospectusPath(self, instance):
url_slug = instance.get("url_slug")
return f"course/{url_slug}"
class AboutPageProductRecommendationsSerializer(serializers.Serializer):
"""Serializer for a cross product recommended course for the course about page"""
key = serializers.CharField()
uuid = serializers.UUIDField()
title = serializers.CharField()
image = CourseImageSerializer()
prospectusPath = serializers.SerializerMethodField()
owners = serializers.ListField(
child=CourseOwnersSerializer(), allow_empty=True
)
activeCourseRun = ActiveCourseRunSerializer(source="active_course_run")
courseType = serializers.CharField(source="course_type")
def get_prospectusPath(self, instance):
url_slug = instance.get("url_slug")
return f"course/{url_slug}"
class LearnerDashboardProductRecommendationsSerializer(serializers.Serializer):
"""Serializer for product recommendations for the Learner Dashboard"""
title = serializers.CharField()
courseRunKey = serializers.SerializerMethodField()
marketingUrl = serializers.URLField(source="marketing_url")
courseType = serializers.CharField(source="course_type")
image = CourseImageSerializer()
owners = serializers.ListField(
child=CourseOwnersSerializer(), allow_empty=True
)
def get_courseRunKey(self, instance):
active_course_run_key = instance.get('active_course_run_key')
return active_course_run_key if active_course_run_key else instance.get('course_runs')[0]['key']
class AboutPageRecommendationsSerializer(serializers.Serializer):
"""Recommended courses for course about page"""
courses = serializers.ListField(
child=RecommendedCourseSerializer(), allow_empty=True
)
isControl = serializers.BooleanField(
source="is_control",
default=None
)
class RecommendationsContextSerializer(serializers.Serializer):
"""Serializer for recommendations context"""
countryCode = serializers.CharField(allow_blank=True)
class CrossProductRecommendationsSerializer(serializers.Serializer):
"""
Cross product recommendation courses for course about page
"""
courses = serializers.ListField(
child=AboutPageProductRecommendationsSerializer(), allow_empty=True
)
class AmplitudeRecommendationsSerializer(serializers.Serializer):
"""Serializer for Amplitude recommendations for Learner Dashboard"""
amplitudeCourses = serializers.ListField(
child=LearnerDashboardProductRecommendationsSerializer(), allow_empty=True
)
class CrossProductAndAmplitudeRecommendationsSerializer(serializers.Serializer):
"""
Cross product recommendation courses and
Amplitude recommendations for Learner Dashboard
"""
crossProductCourses = serializers.ListField(
child=LearnerDashboardProductRecommendationsSerializer(), allow_empty=True
)
amplitudeCourses = serializers.ListField(
child=LearnerDashboardProductRecommendationsSerializer(), allow_empty=True
)
class CourseSerializer(serializers.Serializer):
"""Serializer for a recommended course from the recommendation engine"""
courseKey = serializers.CharField(source="course_key")
logoImageUrl = serializers.URLField(source="logo_image_url")
marketingUrl = serializers.URLField(source="marketing_url")
title = serializers.CharField()
class DashboardRecommendationsSerializer(serializers.Serializer):
"""Recommended courses for learner dashboard"""
courses = serializers.ListField(
child=CourseSerializer(), allow_empty=True
)
isControl = serializers.BooleanField(
source="is_control",
default=None
)

0
lms/djangoapps/learner_recommendations/tests/__init__.py
View File

171
lms/djangoapps/learner_recommendations/tests/test_data.py
View File

@@ -1,171 +0,0 @@
""" Mocked data for testing """
mock_course_data = [
{
"key": "edx+HL0",
"uuid": "0f8cb2c9-589b-4d1e-88c1-b01a02db3a9c",
"title": "Title 0",
"image": {
"src": "https://www.logo_image_url0.com"
},
"prospectusPath": "course/https://www.marketing_url0.com",
"owners": [
{
"key": "org-0",
"name": "org 0",
"logoImageUrl": "https://discovery.com/organization/logos/org-0.png"
}
],
"activeCourseRun": {
"key": "course-v1:Test+2023_T0",
"marketingUrl": "https://www.marketing_url0.com"
},
"courseType": "executive-education"
},
{
"key": "edx+HL1",
"uuid": "1f8cb2c9-589b-4d1e-88c1-b01a02db3a9c",
"title": "Title 1",
"image": {
"src": "https://www.logo_image_url1.com"
},
"prospectusPath": "course/https://www.marketing_url1.com",
"owners": [
{
"key": "org-1",
"name": "org 1",
"logoImageUrl": "https://discovery.com/organization/logos/org-1.png"
}
],
"activeCourseRun": {
"key": "course-v1:Test+2023_T1",
"marketingUrl": "https://www.marketing_url1.com"
},
"courseType": "executive-education"
}
]
mock_cross_product_data = [
{
"title": "Title 0",
"courseRunKey": "course-v1:Test+2023_T0",
"marketingUrl": "https://www.marketing_url0.com",
"courseType": "executive-education",
"image": {
"src": "https://www.logo_image_url0.com"
},
"owners": [
{
"key": "org-0",
"name": "org 0",
"logoImageUrl": "https://discovery.com/organization/logos/org-0.png"
}
],
},
{
"title": "Title 1",
"courseRunKey": "course-v1:Test+2023_T1",
"marketingUrl": "https://www.marketing_url1.com",
"courseType": "executive-education",
"image": {
"src": "https://www.logo_image_url1.com"
},
"owners": [
{
"key": "org-1",
"name": "org 1",
"logoImageUrl": "https://discovery.com/organization/logos/org-1.png"
}
],
},
]
mock_amplitude_data = [
*mock_cross_product_data,
{
"title": "Title 2",
"courseRunKey": "course-v1:Test+2023_T2",
"marketingUrl": "https://www.marketing_url2.com",
"courseType": "executive-education",
"image": {
"src": "https://www.logo_image_url2.com"
},
"owners": [
{
"key": "org-2",
"name": "org 2",
"logoImageUrl": "https://discovery.com/organization/logos/org-2.png"
}
],
},
{
"title": "Title 3",
"courseRunKey": "course-v1:Test+2023_T3",
"marketingUrl": "https://www.marketing_url3.com",
"courseType": "executive-education",
"image": {
"src": "https://www.logo_image_url3.com"
},
"owners": [
{
"key": "org-3",
"name": "org 3",
"logoImageUrl": "https://discovery.com/organization/logos/org-3.png"
}
],
}
]
def get_general_recommendations():
"""Returns 5 general recommendations with the necessary fields"""
courses = []
base_course = {
"course_key": "MITx+1.00",
"title": "Introduction to Computer Science and Programming Using Python",
"url_slug": "introduction-to-computer-science-and-programming-7",
"course_type": "credit-verified-audit",
"logo_image_url": "https://discovery.com/organization/logos/org-1.png",
"marketing_url": "https://www.marketing_url.com",
"course_runs": [
{
"key": "course-v1:MITx+6.00.1x+2T2023",
}
],
"owners": [
{
"key": "MITx",
"name": "Massachusetts Institute of Technology",
"logo_image_url": "https://discovery.com/organization/logos/org-1.png",
}
],
"image": {
"src": "https://link.to.an.image.png"
},
}
for _ in range(5):
courses.append(base_course)
return courses
mock_amplitude_and_cross_product_course_data = {
"crossProductCourses": mock_cross_product_data,
"amplitudeCourses": mock_amplitude_data
}
mock_cross_product_course_data = {
"courses": mock_course_data
}
mock_amplitude_course_data = {
"amplitudeCourses": mock_amplitude_data
}
mock_cross_product_recommendation_keys = {
"edx+HL0": ["edx+HL1", "edx+HL2"],
"edx+BZ0": ["edx+BZ1", "edx+BZ2"],
}

264
lms/djangoapps/learner_recommendations/tests/test_serializers.py
View File

@@ -1,264 +0,0 @@
"""Tests for serializers for the Learner Recommendations"""
from uuid import uuid4
from django.test import TestCase
from lms.djangoapps.learner_recommendations.serializers import (
DashboardRecommendationsSerializer,
RecommendationsContextSerializer,
CrossProductRecommendationsSerializer,
CrossProductAndAmplitudeRecommendationsSerializer,
AmplitudeRecommendationsSerializer
)
from lms.djangoapps.learner_recommendations.tests.test_data import (
mock_amplitude_and_cross_product_course_data,
mock_cross_product_course_data,
mock_amplitude_course_data
)
class TestDashboardRecommendationsSerializer(TestCase):
"""High-level tests for DashboardRecommendationsSerializer"""
@classmethod
def mock_recommended_courses(cls, courses_count=2):
"""Sample course data"""
recommended_courses = []
for _ in range(courses_count):
recommended_courses.append(
{
"course_key": str(uuid4()),
"logo_image_url": "http://edx.org/images/test.png",
"marketing_url": "http://edx.org/courses/AI",
"title": str(uuid4()),
},
)
return recommended_courses
def test_no_recommended_courses(self):
"""That that data serializes correctly for empty courses list"""
recommended_courses = self.mock_recommended_courses(courses_count=0)
output_data = DashboardRecommendationsSerializer(
{
"courses": recommended_courses,
}
).data
self.assertDictEqual(
output_data,
{
"courses": [],
"isControl": None,
},
)
def test_happy_path(self):
"""Test that data serializes correctly"""
recommended_courses = self.mock_recommended_courses()
output_data = DashboardRecommendationsSerializer(
{
"courses": recommended_courses,
"is_control": False,
}
).data
self.assertDictEqual(
output_data,
{
"courses": [
{
"courseKey": recommended_courses[0]["course_key"],
"logoImageUrl": recommended_courses[0]["logo_image_url"],
"marketingUrl": recommended_courses[0]["marketing_url"],
"title": recommended_courses[0]["title"],
},
{
"courseKey": recommended_courses[1]["course_key"],
"logoImageUrl": recommended_courses[1]["logo_image_url"],
"marketingUrl": recommended_courses[1]["marketing_url"],
"title": recommended_courses[1]["title"],
},
],
"isControl": False,
},
)
class TestRecommendationsContextSerializer(TestCase):
"""Tests for RecommendationsContextSerializer"""
def test_successful_serialization(self):
"""Test that context data serializes correctly"""
serialized_data = RecommendationsContextSerializer(
{
"countryCode": "US",
}
).data
self.assertDictEqual(
serialized_data,
{
"countryCode": "US",
},
)
def test_empty_response_serialization(self):
"""Test that an empty response serializes correctly"""
serialized_data = RecommendationsContextSerializer(
{
"countryCode": "",
}
).data
self.assertDictEqual(
serialized_data,
{
"countryCode": "",
},
)
class TestCrossProductRecommendationsSerializers(TestCase):
"""
Tests for the CrossProductRecommendationsSerializer,
AmplitudeRecommendationsSerializer, and CrossProductAndAmplitudeRecommendations Serializer
"""
def mock_recommended_courses(self, num_of_courses=2):
"""Course data mock"""
recommended_courses = []
for index in range(num_of_courses):
recommended_courses.append(
{
"key": f"edx+HL{index}",
"uuid": f"{index}f8cb2c9-589b-4d1e-88c1-b01a02db3a9c",
"title": f"Title {index}",
"image": {
"src": f"https://www.logo_image_url{index}.com",
},
"url_slug": f"https://www.marketing_url{index}.com",
"course_type": "executive-education",
"owners": [
{
"key": f"org-{index}",
"name": f"org {index}",
"logo_image_url": f"https://discovery.com/organization/logos/org-{index}.png",
},
],
"course_runs": [
{
"key": f"course-v1:Test+2023_T{index}",
"marketing_url": f"https://www.marketing_url{index}.com",
"availability": "Current",
}
],
"active_course_run": {
"key": f"course-v1:Test+2023_T{index}",
"marketing_url": f"https://www.marketing_url{index}.com",
"availability": "Current",
},
"active_course_run_key": f"course-v1:Test+2023_T{index}",
"marketing_url": f"https://www.marketing_url{index}.com",
"location_restriction": None
},
)
return recommended_courses
def test_successful_cross_product_recommendation_serialization(self):
"""Test that course data serializes correctly for CrossProductRecommendationSerializer"""
courses = self.mock_recommended_courses(num_of_courses=2)
serialized_data = CrossProductRecommendationsSerializer({
"courses": courses,
}).data
self.assertDictEqual(
serialized_data,
mock_cross_product_course_data
)
def test_successful_amplitude_recommendations_serialization(self):
"""Test the course data serializes correctly for AmplitudeRecommendationsSerializer"""
courses = self.mock_recommended_courses(num_of_courses=4)
serialized_data = AmplitudeRecommendationsSerializer({
"amplitudeCourses": courses
}).data
self.assertDictEqual(
serialized_data,
mock_amplitude_course_data
)
def test_successful_cross_product_and_amplitude_recommendations_serializer(self):
"""Test that course data serializes correctly for CrossProductAndAmplitudeRecommendationSerializer"""
cross_product_courses = self.mock_recommended_courses(num_of_courses=2)
amplitude_courses = self.mock_recommended_courses(num_of_courses=4)
serialized_data = CrossProductAndAmplitudeRecommendationsSerializer({
"crossProductCourses": cross_product_courses,
"amplitudeCourses": amplitude_courses,
}).data
self.assertDictEqual(
serialized_data,
mock_amplitude_and_cross_product_course_data
)
def test_no_cross_product_course_serialization(self):
"""Tests that empty course data for CrossProductRecommendationsSerializer serializes properly"""
serialized_data = CrossProductRecommendationsSerializer({
"courses": [],
}).data
self.assertDictEqual(
serialized_data,
{
"courses": [],
},
)
def test_no_amplitude_courses_serialization(self):
"""Tests that empty course data for AmplitudeRecommendationsSerializer serializes properly"""
serialized_data = AmplitudeRecommendationsSerializer({
"amplitudeCourses": [],
}).data
self.assertDictEqual(
serialized_data,
{
"amplitudeCourses": [],
},
)
def test_no_amplitude_and_cross_product_and_course_serialization(self):
"""Tests that empty course data for CrossProductRecommendationsSerializer serializes properly"""
serialized_data = CrossProductAndAmplitudeRecommendationsSerializer({
"crossProductCourses": [],
"amplitudeCourses": []
}).data
self.assertDictEqual(
serialized_data,
{
"crossProductCourses": [],
"amplitudeCourses": []
},
)

283
lms/djangoapps/learner_recommendations/tests/test_utils.py
View File

@@ -1,283 +0,0 @@
""" Test Recommendations helpers methods """
import ddt
from django.test import TestCase
from unittest.mock import Mock, patch
from common.djangoapps.student.tests.factories import (
CourseEnrollmentFactory,
UserFactory,
)
from lms.djangoapps.learner_recommendations.utils import (
_has_country_restrictions,
filter_recommended_courses,
get_amplitude_course_recommendations,
get_cross_product_recommendations,
get_active_course_run
)
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from lms.djangoapps.learner_recommendations.tests.test_data import mock_cross_product_recommendation_keys
@ddt.ddt
class TestRecommendationsHelper(TestCase):
"""Test course recommendations helper methods."""
def setUp(self):
super().setUp()
self.user = UserFactory()
@ddt.data(
({}, 0),
({"userData": {}}, 0),
({"userData": {"recommendations": []}}, 0),
(
{
"userData": {
"recommendations": [
{
"items": ["MITx+6.00.1x", "IBM+PY0101EN", "HarvardX+CS50P"],
"is_control": True,
"has_is_control": False,
}
],
}
},
3,
),
)
@patch("lms.djangoapps.learner_recommendations.utils.requests.get")
@ddt.unpack
def test_get_amplitude_course_recommendations_method(
self, mocked_response, expected_recommendations_count, mock_get
):
"""
Tests the get_amplitude_recommendations method returns course key list.
"""
mock_get.return_value = Mock(status_code=200, json=lambda: mocked_response)
_, _, course_keys = get_amplitude_course_recommendations(
self.user.id, "amplitude-rec-id"
)
self.assertEqual(len(course_keys), expected_recommendations_count)
@ddt.data(
({}, False),
({"restriction_type": "blocklist", "countries": []}, False),
({"restriction_type": "blocklist", "countries": ["SA"]}, False),
({"restriction_type": "blocklist", "countries": ["US"]}, True),
({"restriction_type": "allowlist", "countries": []}, False),
({"restriction_type": "allowlist", "countries": ["SA"]}, True),
({"restriction_type": "allowlist", "countries": ["US"]}, False),
)
@ddt.unpack
def test_has_country_restrictions_method(
self,
location_restriction,
expected_response,
):
"""
Helper method to test the _has_country_restrictions method.
"""
product = {"location_restriction": location_restriction}
assert _has_country_restrictions(product, "US") == expected_response
class TestFilterRecommendedCourses(ModuleStoreTestCase):
"""Test for filter_recommended_courses helper method."""
def setUp(self):
super().setUp()
self.user = UserFactory()
self.recommended_course_keys = [
"MITx+6.00.1x",
"IBM+PY0101EN",
"HarvardX+CS50P",
"UQx+IELTSx",
"HarvardX+CS50x",
"Harvard+CS50z",
"BabsonX+EPS03x",
"TUMx+QPLS2x",
"NYUx+FCS.NET.1",
"MichinX+101x",
]
self.unrestricted_course_keys = self.recommended_course_keys[0:2]
self.course_run_keys = [f"course-v1:{course_key}+Run_0" for course_key in self.recommended_course_keys]
self.course_keys_with_active_course_runs = self.recommended_course_keys[0:8]
self.enrolled_course_run_keys = self.course_run_keys[4:10]
def _mock_get_course_data(self, course_id, fields=None, querystring=None): # pylint: disable=unused-argument
"""
Mocked response for the get_course_data call
"""
course_data = {
"course_key": course_id,
"title": "Mocked course title",
"owners": [{"logo_image_url": "https://www.logo_image_url.com"}],
"marketing_url": "https://www.marketing_url.com",
}
if course_id not in self.unrestricted_course_keys:
course_data.update(
{
"location_restriction": {
"restriction_type": "blocklist",
"countries": ["US"],
}
}
)
if course_id in self.course_keys_with_active_course_runs:
course_data.update(
{
"course_runs": [
{
"key": f"course-v1:{course_id}+Run_0",
}
]
}
)
return course_data
@patch("lms.djangoapps.learner_recommendations.utils.get_course_data")
def test_enrolled_courses_are_removed_from_recommendations(
self, mocked_get_course_data
):
"""
Tests that given a recommended course list, the filter_recommended_courses
method removes the enrolled courses from it.
"""
total_enrolled_courses = len(self.enrolled_course_run_keys)
total_recommendations = len(self.recommended_course_keys)
mocked_get_course_data.side_effect = self._mock_get_course_data
for course_run_key in self.enrolled_course_run_keys:
CourseEnrollmentFactory(course_id=course_run_key, user=self.user)
filtered_courses = filter_recommended_courses(
self.user, self.recommended_course_keys, total_recommendations
)
assert len(filtered_courses) == (total_recommendations - total_enrolled_courses)
@patch("lms.djangoapps.learner_recommendations.utils.get_course_data")
def test_request_course_is_removed_from_the_recommendations(
self,
mocked_get_course_data,
):
"""
Test that if the "request course" is one of the recommended courses,
we filter that from the final recommendation list.
"""
request_course = self.course_run_keys[0]
mocked_get_course_data.side_effect = self._mock_get_course_data
filtered_courses = filter_recommended_courses(
self.user,
self.recommended_course_keys,
request_course_key=request_course,
)
assert all(course["course_key"] != request_course for course in filtered_courses) is True
@patch("lms.djangoapps.learner_recommendations.utils.get_course_data")
def test_country_restrictions_for_the_recommended_course(
self,
mocked_get_course_data,
):
"""
Test that if a recommended course is restricted in the country the user
is logged from, the course is filtered out.
"""
mocked_get_course_data.side_effect = self._mock_get_course_data
filtered_courses = filter_recommended_courses(
self.user, self.recommended_course_keys, user_country_code="US"
)
expected_recommendations = []
for course_key in self.unrestricted_course_keys:
expected_recommendations.append(self._mock_get_course_data(course_key))
assert filtered_courses == expected_recommendations
@patch("lms.djangoapps.learner_recommendations.utils.get_course_data")
def test_recommend_only_active_courses(
self,
mocked_get_course_data,
):
"""
Test that courses having no active course runs are filtered out from recommended courses.
"""
mocked_get_course_data.side_effect = self._mock_get_course_data
filtered_courses = filter_recommended_courses(
self.user, self.recommended_course_keys
)
expected_recommendations = []
for course_key in self.course_keys_with_active_course_runs:
expected_recommendations.append(self._mock_get_course_data(course_key))
assert filtered_courses == expected_recommendations
@ddt.ddt
class TestGetCrossProductRecommendationsMethod(TestCase):
"""Test for get_cross_product_recommendations method"""
@ddt.data(
("edx+HL0", ["edx+HL1", "edx+HL2"]),
("edx+BZ0", ["edx+BZ1", "edx+BZ2"]),
('NoKeyAssociated', None)
)
@patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@ddt.unpack
def test_get_cross_product_recommendations_method(self, course_key, expected_response):
assert get_cross_product_recommendations(course_key) == expected_response
class TestGetActiveCourseRunMethod(TestCase):
"""Tests for get_active_course_run method"""
advertised_course_run_uuid = "jh76b2c9-589b-4d1e-88c1-b01a02db3a9c"
def _mock_get_course_data(self, active_course_run=False):
"""
Returns a course with details based on the status passed
"""
return {
"key": "edx+BLN",
"uuid": "6f8cb2c9-589b-4d1e-88c1-b01a02db3a9c",
"course_runs": [
{
"key": "course-v1:Test+2023_T1",
"uuid": "hb86b3cf-589b-4d1e-88c1-b01a02db3a9c",
"status": "published",
},
{
"key": "course-v1:Test+2023_T2",
"uuid": self.advertised_course_run_uuid if active_course_run else "other-uuid",
"status": "published",
}
],
"course_run_statuses": ["published"],
"advertised_course_run_uuid": self.advertised_course_run_uuid if active_course_run else None,
}
def test_advertised_course_run_returned(self):
"""
Test that the course run with the uuid that matches the advertised_uuid_course_run_uuid is returned
"""
course = self._mock_get_course_data(active_course_run=True)
active_course_run = get_active_course_run(course)
self.assertDictEqual(
active_course_run,
{
"key": "course-v1:Test+2023_T2",
"uuid": self.advertised_course_run_uuid,
"status": "published"
}
)
def test_no_course_run_returned(self):
"""
Test that if there is no advertised_course_run_uuid value, no course run is returned
"""
course = self._mock_get_course_data(active_course_run=False)
active_course_run = get_active_course_run(course)
self.assertIsNone(active_course_run)

1048
lms/djangoapps/learner_recommendations/tests/test_views.py
View File

@@ -1,1048 +0,0 @@
"""
Tests for Learner Recommendations views and related functions.
"""
import json
from django.urls import reverse_lazy
from edx_toggles.toggles.testutils import override_waffle_flag
from rest_framework.test import APITestCase
from unittest import mock
import ddt
from common.djangoapps.student.tests.factories import UserFactory
from common.djangoapps.student.toggles import ENABLE_FALLBACK_RECOMMENDATIONS
from lms.djangoapps.learner_recommendations.toggles import (
ENABLE_COURSE_ABOUT_PAGE_RECOMMENDATIONS,
ENABLE_DASHBOARD_RECOMMENDATIONS,
)
from lms.djangoapps.learner_recommendations.tests.test_data import (
mock_cross_product_recommendation_keys,
get_general_recommendations
)
class TestRecommendationsBase(APITestCase):
"""Recommendations test base class"""
def setUp(self):
super().setUp()
self.TEST_PASSWORD = 'Password1234'
self.user = UserFactory(password=self.TEST_PASSWORD)
self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
self.recommended_courses = [
"MITx+6.00.1x",
"IBM+PY0101EN",
"HarvardX+CS50P",
"UQx+IELTSx",
"HarvardX+CS50x",
"Harvard+CS50z",
"BabsonX+EPS03x",
"TUMx+QPLS2x",
"NYUx+FCS.NET.1",
"MichinX+101x",
]
@override_waffle_flag(ENABLE_COURSE_ABOUT_PAGE_RECOMMENDATIONS, active=True)
class TestAboutPageRecommendationsView(TestRecommendationsBase):
"""Unit tests for the Amplitude recommendations API"""
url = reverse_lazy(
"learner_recommendations:amplitude_recommendations",
kwargs={'course_id': 'course-v1:test+TestX+Test_Course'}
)
def _get_filtered_courses(self):
"""
Returns the filtered course data
"""
filtered_course = []
for course_key in self.recommended_courses[0:4]:
filtered_course.append({
"key": course_key,
"uuid": "4f8cb2c9-589b-4d1e-88c1-b01a02db3a9c",
"title": f"Title for {course_key}",
"image": {
"src": "https://www.logo_image_url.com",
},
"url_slug": "https://www.marketing_url.com",
"owners": [
{
"key": "org-1",
"name": "org 1",
"logo_image_url": "https://discovery.com/organization/logos/org-1.png",
},
{
"key": "org-2",
"name": "org 2",
"logo_image_url": "https://discovery.com/organization/logos/org-2.png",
}
],
"course_runs": [
{
"key": "course-v1:Test+2023_T1",
"marketing_url": "https://www.marketing_url.com",
"availability": "Current",
},
{
"key": "course-v1:Test+2023_T2",
"marketing_url": "https://www.marketing_url.com",
"availability": "Upcoming",
}
]
})
return filtered_course
@override_waffle_flag(ENABLE_COURSE_ABOUT_PAGE_RECOMMENDATIONS, active=False)
def test_waffle_flag_off(self):
"""
Verify API returns 404 (Not Found) if waffle flag is off.
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 404)
self.assertEqual(response.data, None)
@mock.patch('lms.djangoapps.learner_recommendations.views.is_enterprise_learner', mock.Mock(return_value=True))
def test_enterprise_user_access(self):
"""
Verify API returns 403 (Forbidden) for an enterprise user.
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 403)
@mock.patch(
"lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations",
mock.Mock(side_effect=Exception),
)
def test_amplitude_api_unexpected_error(self):
"""
Test that if the Amplitude API gives an unexpected error,
API returns 404 (Not Found).
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 404)
self.assertEqual(response.data, None)
@mock.patch("lms.djangoapps.learner_recommendations.views.segment.track")
@mock.patch(
"lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations"
)
@mock.patch("lms.djangoapps.learner_recommendations.views.filter_recommended_courses")
def test_successful_response(
self, filter_recommended_courses_mock, get_amplitude_course_recommendations_mock, segment_mock,
):
"""
Verify API returns course recommendations.
"""
expected_recommendations_length = 4
filter_recommended_courses_mock.return_value = self._get_filtered_courses()
get_amplitude_course_recommendations_mock.return_value = [
False,
True,
self.recommended_courses,
]
segment_mock.return_value = None
response = self.client.get(self.url)
response_content = json.loads(response.content)
self.assertEqual(response.status_code, 200)
self.assertEqual(response_content.get("isControl"), False)
self.assertEqual(
len(response_content.get("courses")), expected_recommendations_length
)
# Verify that the segment event was fired
assert segment_mock.call_count == 1
assert segment_mock.call_args[0][1] == "edx.bi.user.recommendations.viewed"
class TestRecommendationsContextView(APITestCase):
"""Unit tests for the Recommendations Context View"""
def setUp(self):
super().setUp()
self.user = UserFactory()
self.password = 'Password1234'
self.url = reverse_lazy("learner_recommendations:recommendations_context")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
def test_successful_response(self, country_code_from_ip_mock):
"""Test that country code gets sent back when authenticated"""
country_code_from_ip_mock.return_value = "za"
self.client.login(username=self.user.username, password=self.password)
response = self.client.get(self.url)
response_data = json.loads(response.content)
self.assertEqual(response_data["countryCode"], "za")
def test_unauthenticated_response(self):
"""
Test that a 401 is sent back if an anauthenticated user calls endpoint
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 401)
class TestCrossProductRecommendationsView(APITestCase):
"""Unit tests for the Cross Product Recommendations View"""
def setUp(self):
super().setUp()
self.associated_course_keys = ["edx+HL1", "edx+HL2"]
def _get_url(self, course_key):
"""
Returns the url with a sepcific course id
"""
return reverse_lazy(
"learner_recommendations:cross_product_recommendations",
kwargs={'course_id': f'course-v1:{course_key}+Test_Course'}
)
def _get_recommended_courses(self, num_of_courses_with_restriction=0, active_course_run=True):
"""
Returns an array of 2 discovery courses with or without country restrictions
"""
courses = []
restriction_obj = {
"restriction_type": "blocklist",
"countries": ["CN"],
"states": []
}
for course_key in enumerate(self.associated_course_keys):
location_restriction = restriction_obj if num_of_courses_with_restriction > 0 else None
advertised_course_run_uuid = "jh76b2c9-589b-4d1e-88c1-b01a02db3a9c" if active_course_run else None
courses.append({
"key": course_key[1],
"uuid": "6f8cb2c9-589b-4d1e-88c1-b01a02db3a9c",
"title": f"Title {course_key[0]}",
"image": {
"src": "https://www.logo_image_url.com",
},
"url_slug": "https://www.marketing_url.com",
"course_type": "executive-education",
"owners": [
{
"key": "org-1",
"name": "org 1",
"logo_image_url": "https://discovery.com/organization/logos/org-1.png",
},
],
"course_runs": [
{
"key": "course-v1:Test+2023_T2",
"marketing_url": "https://www.marketing_url.com",
"availability": "Current",
"uuid": "jh76b2c9-589b-4d1e-88c1-b01a02db3a9c",
"status": "published"
}
],
"advertised_course_run_uuid": advertised_course_run_uuid,
"location_restriction": location_restriction,
})
if num_of_courses_with_restriction > 0:
num_of_courses_with_restriction -= 1
return courses
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
def test_successful_response(
self, country_code_from_ip_mock, get_course_data_mock,
):
"""
Verify 2 cross product course recommendations are returned.
"""
country_code_from_ip_mock.return_value = "za"
mock_course_data = self._get_recommended_courses()
get_course_data_mock.side_effect = [mock_course_data[0], mock_course_data[1]]
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
course_data = response_content["courses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(course_data), 2)
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
def test_one_course_country_restriction_response(
self, country_code_from_ip_mock, get_course_data_mock,
):
"""
Verify 1 cross product course recommendation is returned
if there is a location restriction for one course for the users country
"""
country_code_from_ip_mock.return_value = "cn"
mock_course_data = self._get_recommended_courses(1)
get_course_data_mock.side_effect = [mock_course_data[0], mock_course_data[1]]
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
course_data = response_content["courses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(course_data), 1)
self.assertEqual(course_data[0]["title"], "Title 1")
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
def test_both_course_country_restriction_response(
self, country_code_from_ip_mock, get_course_data_mock,
):
"""
Verify no courses are returned if both courses have a location restriction
for the users country.
"""
country_code_from_ip_mock.return_value = "cn"
mock_course_data = self._get_recommended_courses(2)
get_course_data_mock.side_effect = [mock_course_data[0], mock_course_data[1]]
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
course_data = response_content["courses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(course_data), 0)
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
def test_no_associated_course_response(self):
"""
Verify an empty array of courses is returned if there are no associated course keys.
"""
response = self.client.get(self._get_url('No+Associations'))
response_content = json.loads(response.content)
course_data = response_content["courses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(course_data), 0)
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
def test_no_response_from_discovery(self, country_code_from_ip_mock, get_course_data_mock):
"""
Verify an empty array of courses is returned if discovery returns two empty dictionaries.
"""
country_code_from_ip_mock.return_value = "za"
get_course_data_mock.side_effect = [{}, {}]
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
course_data = response_content["courses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(course_data), 0)
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
def test_no_active_course_runs_response(self, country_code_from_ip_mock, get_course_data_mock):
"""
Verify that an empty array of courses is returned if courses do not have an active course run.
"""
country_code_from_ip_mock.return_value = "za"
mock_course_data = self._get_recommended_courses(0, active_course_run=False)
get_course_data_mock.side_effect = [mock_course_data[0], mock_course_data[1]]
response = self.client.get(self._get_url('edx+HL0'))
reponse_content = json.loads(response.content)
course_data = reponse_content["courses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(course_data), 0)
class TestProductRecommendationsView(APITestCase):
"""Unit tests for ProductRecommendations View"""
def setUp(self):
super().setUp()
self.TEST_PASSWORD = 'Password1234'
self.user = UserFactory(password=self.TEST_PASSWORD)
self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
self.associated_course_keys = ["edx+HL1", "edx+HL2"]
self.amplitude_keys = [
"edx+CS0",
"edx+CS10",
"edx+CS20",
"edx+CS30",
"edx+CS40",
"edx+CS50",
"edx+CS60",
"edx+CS70",
"edx+CS80",
"edx+CS90",
]
self.amplitude_course_run_keys = [f"course-v1:{course_key}+2023_T2" for course_key in self.amplitude_keys]
self.enrolled_course_run_keys = self.amplitude_course_run_keys[3:8]
self.enrolled_course_keys = self.amplitude_keys[3:8]
self.amplitude_location_restriction_keys = self.amplitude_keys[0:3]
self.cross_product_location_restriction_keys = self.associated_course_keys[0]
def _get_url(self, course_key=None):
"""
Returns the product recommendations url with or without the course key
"""
if course_key:
return reverse_lazy(
"learner_recommendations:product_recommendations",
kwargs={'course_id': f'course-v1:{course_key}+Test_Course'}
)
return reverse_lazy(
"learner_recommendations:product_recommendations_amplitude_only"
)
def _get_product_recommendations(self, course_keys, keys_with_restriction=None):
"""
Returns course data based on the number of course keys passed in
with a location restriction object if a list of keys for location restriction courses is passed in
"""
courses = []
for key in course_keys:
course = {
"title": f"Title for {key}",
"image": {
"src": "https://www.logo_image_url.com",
},
"course_type": "executive-education",
"owners": [
{
"key": "org-1",
"name": "org 1",
"logo_image_url": "https://discovery.com/organization/logos/org-1.png",
},
],
"course_runs": [
{
"key": f"course-v1:{key}+2023_T2",
"marketing_url": "https://www.marketing_url.com",
"availability": "Current",
"uuid": "jh76b2c9-589b-4d1e-88c1-b01a02db3a9c",
"status": "published"
}
],
"marketing_url": "https://www.marketing_url.com/course/some-course",
"advertised_course_run_uuid": f"course-v1:{key}+2023_T2",
}
if keys_with_restriction and key in keys_with_restriction:
course.update({
"location_restriction": {
"restriction_type": "blocklist",
"countries": ["CN"],
"states": []
}
})
courses.append(course)
return courses
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("lms.djangoapps.learner_recommendations.utils._get_user_enrolled_course_keys")
@mock.patch("lms.djangoapps.learner_recommendations.utils.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
@mock.patch("lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program")
def test_successful_response(
self,
is_user_enrolled_in_ut_austin_masters_program_mock,
country_code_from_ip_mock,
get_amplitude_course_recommendations_mock,
get_course_data_view_mock,
get_course_data_util_mock,
get_user_enrolled_course_keys_mock,
):
"""
Verify 2 cross product course recommendations are returned
and 4 amplitude courses are returned
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = False
country_code_from_ip_mock.return_value = "za"
get_user_enrolled_course_keys_mock.return_value = []
get_amplitude_course_recommendations_mock.return_value = [False, True, self.amplitude_keys]
mock_cross_product_course_data = self._get_product_recommendations(self.associated_course_keys)
mock_amplitude_course_data = self._get_product_recommendations(self.amplitude_keys)
get_course_data_view_mock.side_effect = mock_cross_product_course_data
get_course_data_util_mock.side_effect = mock_amplitude_course_data
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
cross_product_course_data = response_content["crossProductCourses"]
amplitude_course_data = response_content["amplitudeCourses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(cross_product_course_data), 2)
self.assertEqual(len(amplitude_course_data), 4)
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("lms.djangoapps.learner_recommendations.utils._get_user_enrolled_course_keys")
@mock.patch("lms.djangoapps.learner_recommendations.utils.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
@mock.patch("lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program")
def test_successful_course_filtering(
self,
is_user_enrolled_in_ut_austin_masters_program_mock,
country_code_from_ip_mock,
get_amplitude_course_recommendations_mock,
get_course_data_view_mock,
get_course_data_util_mock,
get_user_enrolled_course_keys_mock,
):
"""
Verify 1 cross product course recommendation is returned
and 2 amplitude courses are returned with filtering done for
enrolled courses and courses with country restrictions
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = False
country_code_from_ip_mock.return_value = "cn"
get_user_enrolled_course_keys_mock.return_value = self.enrolled_course_run_keys
get_amplitude_course_recommendations_mock.return_value = [False, True, self.amplitude_keys]
mock_cross_product_course_data = self._get_product_recommendations(
self.associated_course_keys, self.cross_product_location_restriction_keys
)
mock_amplitude_course_data = self._get_product_recommendations(
self.amplitude_keys, self.amplitude_location_restriction_keys
)
get_course_data_view_mock.side_effect = mock_cross_product_course_data
get_course_data_util_mock.side_effect = mock_amplitude_course_data
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
cross_product_course_data = response_content["crossProductCourses"]
amplitude_course_data = response_content["amplitudeCourses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(cross_product_course_data), 1)
self.assertEqual(len(amplitude_course_data), 2)
for course in amplitude_course_data:
course_key = course["title"][2]
assert course_key not in [*self.amplitude_location_restriction_keys, *self.enrolled_course_keys]
for course in cross_product_course_data:
course_key = course["title"][2]
assert course_key not in self.cross_product_location_restriction_keys
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("django.conf.settings.GENERAL_RECOMMENDATIONS", get_general_recommendations())
@mock.patch("lms.djangoapps.learner_recommendations.utils._get_user_enrolled_course_keys")
@mock.patch("lms.djangoapps.learner_recommendations.utils.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
@mock.patch("lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program")
def test_fallback_recommendations_when_enrolled_courses_removed(
self,
is_user_enrolled_in_ut_austin_masters_program_mock,
country_code_from_ip_mock,
get_amplitude_course_recommendations_mock,
get_course_data_view_mock,
get_course_data_util_mock,
get_user_enrolled_course_keys_mock
):
"""
Verify 2 cross product course recommendations are returned
and 4 fallback amplitude recommendations are returned if no courses are left
after filtering due to courses being already enrolled in
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = False
country_code_from_ip_mock.return_value = "za"
get_user_enrolled_course_keys_mock.return_value = self.amplitude_course_run_keys
get_amplitude_course_recommendations_mock.return_value = [False, True, self.amplitude_keys]
mock_cross_product_course_data = self._get_product_recommendations(self.associated_course_keys)
mock_amplitude_course_data = self._get_product_recommendations(self.amplitude_keys)
get_course_data_view_mock.side_effect = mock_cross_product_course_data
get_course_data_util_mock.side_effect = mock_amplitude_course_data
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
cross_product_course_data = response_content["crossProductCourses"]
amplitude_course_data = response_content["amplitudeCourses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(cross_product_course_data), 2)
self.assertEqual(len(amplitude_course_data), 4)
for course in amplitude_course_data:
self.assertEqual(course["title"], "Introduction to Computer Science and Programming Using Python")
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("django.conf.settings.GENERAL_RECOMMENDATIONS", get_general_recommendations())
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
@mock.patch("lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program")
def test_fallback_recommendations_when_error_querying_amplitude(
self,
is_user_enrolled_in_ut_austin_masters_program_mock,
country_code_from_ip_mock,
get_amplitude_course_recommendations_mock,
get_course_data_mock,
):
"""
Verify 2 cross product course recommendations are returned
and 4 fallback amplitude recommendations are returned
if there was an error querying amplitude for recommendations
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = False
country_code_from_ip_mock.return_value = "za"
get_amplitude_course_recommendations_mock.side_effect = Exception()
mock_cross_product_course_data = self._get_product_recommendations(self.associated_course_keys)
get_course_data_mock.side_effect = mock_cross_product_course_data
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
cross_product_course_data = response_content["crossProductCourses"]
amplitude_course_data = response_content["amplitudeCourses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(cross_product_course_data), 2)
self.assertEqual(len(amplitude_course_data), 4)
for course in amplitude_course_data:
self.assertEqual(course["title"], "Introduction to Computer Science and Programming Using Python")
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("django.conf.settings.GENERAL_RECOMMENDATIONS", get_general_recommendations())
@mock.patch("lms.djangoapps.learner_recommendations.views.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
@mock.patch("lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program")
def test_fallback_recommendations_when_no_amplitude_recommended_keys(
self,
is_user_enrolled_in_ut_austin_masters_program_mock,
country_code_from_ip_mock,
get_amplitude_course_recommendations_mock,
get_course_data_mock,
):
"""
Verify 2 cross product course recommendations are returned
and 4 fallback amplitude recommendations are returned
if amplitude gave back no course keys
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = False
country_code_from_ip_mock.return_value = "za"
get_amplitude_course_recommendations_mock.side_effect = [False, True, []]
mock_cross_product_course_data = self._get_product_recommendations(self.associated_course_keys)
get_course_data_mock.side_effect = mock_cross_product_course_data
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
cross_product_course_data = response_content["crossProductCourses"]
amplitude_course_data = response_content["amplitudeCourses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(cross_product_course_data), 2)
self.assertEqual(len(amplitude_course_data), 4)
for course in amplitude_course_data:
self.assertEqual(course["title"], "Introduction to Computer Science and Programming Using Python")
@mock.patch("django.conf.settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS", mock_cross_product_recommendation_keys)
@mock.patch("lms.djangoapps.learner_recommendations.utils._get_user_enrolled_course_keys")
@mock.patch("lms.djangoapps.learner_recommendations.utils.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
@mock.patch("lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program")
def test_response_with_amplitude_and_no_cross_product_courses(
self,
is_user_enrolled_in_ut_austin_masters_program_mock,
country_code_from_ip_mock,
get_amplitude_course_recommendations_mock,
get_course_data_mock,
get_user_enrolled_course_keys_mock
):
"""
Verify that if no cross product courses are returned,
then 4 fallback amplitude recommendations will still be returned
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = False
country_code_from_ip_mock.return_value = "za"
get_user_enrolled_course_keys_mock.return_value = self.enrolled_course_run_keys
get_amplitude_course_recommendations_mock.return_value = [False, True, self.amplitude_keys]
mock_amplitude_course_data = self._get_product_recommendations(self.amplitude_keys)
get_course_data_mock.side_effect = mock_amplitude_course_data
response = self.client.get(self._get_url('No+Association'))
response_content = json.loads(response.content)
cross_product_course_data = response_content["crossProductCourses"]
amplitude_course_data = response_content["amplitudeCourses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(cross_product_course_data), 0)
self.assertEqual(len(amplitude_course_data), 4)
@mock.patch("lms.djangoapps.learner_recommendations.utils._get_user_enrolled_course_keys")
@mock.patch("lms.djangoapps.learner_recommendations.utils.get_course_data")
@mock.patch("lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations")
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
@mock.patch("lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program")
def test_amplitude_only_url_response(
self,
is_user_enrolled_in_ut_austin_masters_program_mock,
country_code_from_ip_mock,
get_amplitude_course_recommendations_mock,
get_course_data_mock,
get_user_enrolled_course_keys_mock
):
"""
Verify that if no course key was provided in the url,
only 1 field for amplitude courses are sent back
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = False
country_code_from_ip_mock.return_value = "za"
get_user_enrolled_course_keys_mock.return_value = self.enrolled_course_run_keys
get_amplitude_course_recommendations_mock.return_value = [False, True, self.amplitude_keys]
mock_amplitude_course_data = self._get_product_recommendations(self.amplitude_keys)
get_course_data_mock.side_effect = mock_amplitude_course_data
response = self.client.get(self._get_url())
response_content = json.loads(response.content)
amplitude_course_data = response_content["amplitudeCourses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response_content), 1)
self.assertEqual(len(amplitude_course_data), 4)
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
@mock.patch("lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program")
def test_zero_cross_product_and_amplitude_recommendations(
self,
is_user_enrolled_in_ut_austin_masters_program_mock,
country_code_from_ip_mock,
):
"""
Verify 0 cross product course recommendations are returned
and 0 amplitude courses are returned if the user is enrolled in ut austin masters program
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = True
country_code_from_ip_mock.return_value = "za"
response = self.client.get(self._get_url('edx+HL0'))
response_content = json.loads(response.content)
cross_product_course_data = response_content["crossProductCourses"]
amplitude_course_data = response_content["amplitudeCourses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(cross_product_course_data), 0)
self.assertEqual(len(amplitude_course_data), 0)
@mock.patch("lms.djangoapps.learner_recommendations.views.country_code_from_ip")
@mock.patch("lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program")
def test_zero_amplitude_recommendations(
self,
is_user_enrolled_in_ut_austin_masters_program_mock,
country_code_from_ip_mock,
):
"""
Verify that 0 amplitude courses are returned
if the user is enrolled in ut austin masters program
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = True
country_code_from_ip_mock.return_value = "za"
response = self.client.get(self._get_url())
response_content = json.loads(response.content)
amplitude_course_data = response_content["amplitudeCourses"]
self.assertEqual(response.status_code, 200)
self.assertEqual(len(amplitude_course_data), 0)
@ddt.ddt
class TestDashboardRecommendationsApiView(TestRecommendationsBase):
"""Unit tests for the course recommendations on learner home page."""
url = reverse_lazy("learner_recommendations:courses")
GENERAL_RECOMMENDATIONS = [
{
"course_key": "HogwartsX+6.00.1x",
"logo_image_url": "http://edx.org/images/test.png",
"marketing_url": "http://edx.org/courses/AI",
"title": "Defense Against the Dark Arts",
},
{
"course_key": "MonstersX+SC101EN",
"logo_image_url": "http://edx.org/images/test.png",
"marketing_url": "http://edx.org/courses/AI",
"title": "Scaring 101",
},
]
SERIALIZED_GENERAL_RECOMMENDATIONS = [
{
"courseKey": GENERAL_RECOMMENDATIONS[0]["course_key"],
"logoImageUrl": GENERAL_RECOMMENDATIONS[0]["logo_image_url"],
"marketingUrl": GENERAL_RECOMMENDATIONS[0]["marketing_url"],
"title": GENERAL_RECOMMENDATIONS[0]["title"],
},
{
"courseKey": GENERAL_RECOMMENDATIONS[1]["course_key"],
"logoImageUrl": GENERAL_RECOMMENDATIONS[1]["logo_image_url"],
"marketingUrl": GENERAL_RECOMMENDATIONS[1]["marketing_url"],
"title": GENERAL_RECOMMENDATIONS[1]["title"],
},
]
def setUp(self):
super().setUp()
self.course_run_keys = [f"course-v1:{course}+Run_0" for course in self.recommended_courses]
def _get_filtered_courses(self):
"""
Returns the filtered course data
"""
filtered_course = []
for course_key in self.recommended_courses[:5]:
filtered_course.append({
"key": course_key,
"title": f"Title for {course_key}",
"logo_image_url": "https://www.logo_image_url.com",
"marketing_url": "https://www.marketing_url.com",
})
return filtered_course
@override_waffle_flag(ENABLE_DASHBOARD_RECOMMENDATIONS, active=False)
def test_waffle_flag_off(self):
"""
Verify API returns 404 if waffle flag is off.
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 404)
self.assertEqual(response.data, None)
@override_waffle_flag(ENABLE_FALLBACK_RECOMMENDATIONS, active=True)
@override_waffle_flag(ENABLE_DASHBOARD_RECOMMENDATIONS, active=True)
@mock.patch("django.conf.settings.GENERAL_RECOMMENDATIONS", GENERAL_RECOMMENDATIONS)
@mock.patch(
"lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations"
)
def test_no_recommendations_from_amplitude(
self, get_amplitude_course_recommendations_mock
):
"""
Verify API returns general recommendations if no course recommendations from amplitude.
"""
get_amplitude_course_recommendations_mock.return_value = [False, True, []]
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
response_content = json.loads(response.content)
self.assertEqual(response_content.get("isControl"), False)
self.assertEqual(
response_content.get("courses"),
self.SERIALIZED_GENERAL_RECOMMENDATIONS,
)
@override_waffle_flag(ENABLE_FALLBACK_RECOMMENDATIONS, active=True)
@override_waffle_flag(ENABLE_DASHBOARD_RECOMMENDATIONS, active=True)
@mock.patch("django.conf.settings.GENERAL_RECOMMENDATIONS", GENERAL_RECOMMENDATIONS)
@mock.patch(
"lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations",
mock.Mock(side_effect=Exception),
)
def test_amplitude_api_unexpected_error(self):
"""
Test that if the Amplitude API gives an unexpected error, general recommendations are returned.
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
response_content = json.loads(response.content)
self.assertEqual(response_content.get("isControl"), None)
self.assertEqual(
response_content.get("courses"),
self.SERIALIZED_GENERAL_RECOMMENDATIONS,
)
@override_waffle_flag(ENABLE_DASHBOARD_RECOMMENDATIONS, active=True)
@mock.patch(
"lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations"
)
@mock.patch("lms.djangoapps.learner_recommendations.views.filter_recommended_courses")
def test_get_course_recommendations(
self, filter_recommended_courses_mock, get_amplitude_course_recommendations_mock
):
"""
Verify API returns course recommendations.
"""
get_amplitude_course_recommendations_mock.return_value = [
False,
True,
self.recommended_courses,
]
filter_recommended_courses_mock.return_value = self._get_filtered_courses()
expected_recommendations_length = 5
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
response_content = json.loads(response.content)
self.assertEqual(response_content.get("isControl"), False)
self.assertEqual(
len(response_content.get("courses")), expected_recommendations_length
)
@override_waffle_flag(ENABLE_FALLBACK_RECOMMENDATIONS, active=True)
@override_waffle_flag(ENABLE_DASHBOARD_RECOMMENDATIONS, active=True)
@mock.patch("django.conf.settings.GENERAL_RECOMMENDATIONS", GENERAL_RECOMMENDATIONS)
@mock.patch(
"lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations"
)
def test_general_recommendations(
self, get_amplitude_course_recommendations_mock
):
"""
Test that a user gets general recommendations for the control group.
"""
get_amplitude_course_recommendations_mock.return_value = [
True,
True,
self.recommended_courses,
]
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
response_content = json.loads(response.content)
self.assertEqual(response_content.get("isControl"), True)
self.assertEqual(
response_content.get("courses"),
self.SERIALIZED_GENERAL_RECOMMENDATIONS,
)
@override_waffle_flag(ENABLE_FALLBACK_RECOMMENDATIONS, active=False)
@override_waffle_flag(ENABLE_DASHBOARD_RECOMMENDATIONS, active=True)
@mock.patch("django.conf.settings.GENERAL_RECOMMENDATIONS", GENERAL_RECOMMENDATIONS)
@mock.patch(
"lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations"
)
def test_fallback_recommendations_disabled(
self, get_amplitude_course_recommendations_mock
):
"""
Test that a user gets no recommendations for the control group.
"""
get_amplitude_course_recommendations_mock.return_value = [
True,
True,
[],
]
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
response_content = json.loads(response.content)
self.assertEqual(response_content.get("isControl"), True)
self.assertEqual(response_content.get("courses"), [])
@override_waffle_flag(ENABLE_FALLBACK_RECOMMENDATIONS, active=True)
@override_waffle_flag(ENABLE_DASHBOARD_RECOMMENDATIONS, active=True)
@mock.patch("django.conf.settings.GENERAL_RECOMMENDATIONS", GENERAL_RECOMMENDATIONS)
@mock.patch(
"lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations"
)
@mock.patch("lms.djangoapps.learner_recommendations.views.filter_recommended_courses")
def test_no_recommended_courses_after_filtration(
self, filter_recommended_courses_mock, get_amplitude_course_recommendations_mock
):
"""
Test that if after filtering already enrolled courses from Amplitude recommendations
we are left with zero personalized recommendations, we return general recommendations.
"""
filter_recommended_courses_mock.return_value = []
get_amplitude_course_recommendations_mock.return_value = [
False,
True,
self.recommended_courses,
]
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
response_content = json.loads(response.content)
self.assertEqual(response_content.get("isControl"), False)
self.assertEqual(
response_content.get("courses"),
self.SERIALIZED_GENERAL_RECOMMENDATIONS,
)
@ddt.data(
(True, False, None),
(False, True, False),
(False, False, None),
(True, True, True),
)
@mock.patch("lms.djangoapps.learner_recommendations.views.segment.track")
@mock.patch("lms.djangoapps.learner_recommendations.views.filter_recommended_courses")
@mock.patch(
"lms.djangoapps.learner_recommendations.views.get_amplitude_course_recommendations"
)
@override_waffle_flag(ENABLE_DASHBOARD_RECOMMENDATIONS, active=True)
@ddt.unpack
def test_recommendations_viewed_segment_event(
self,
is_control,
has_is_control,
expected_is_control,
get_amplitude_course_recommendations_mock,
filter_recommended_courses_mock,
segment_track_mock
):
"""
Test that Segment event is emitted with desired properties.
"""
get_amplitude_course_recommendations_mock.return_value = [
is_control,
has_is_control,
self.recommended_courses,
]
filter_recommended_courses_mock.return_value = self._get_filtered_courses()
self.client.get(self.url)
assert segment_track_mock.call_count == 1
assert segment_track_mock.call_args[0][1] == "edx.bi.user.recommendations.viewed"
self.assertEqual(segment_track_mock.call_args[0][2]["is_control"], expected_is_control)
@override_waffle_flag(ENABLE_DASHBOARD_RECOMMENDATIONS, active=True)
@mock.patch(
"lms.djangoapps.learner_recommendations.views.is_user_enrolled_in_ut_austin_masters_program"
)
def test_no_recommendations_for_masters_program_learners(
self, is_user_enrolled_in_ut_austin_masters_program_mock
):
"""
Verify API returns no recommendations if a user is enrolled in UT Austin masters program.
"""
is_user_enrolled_in_ut_austin_masters_program_mock.return_value = True
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
response_content = json.loads(response.content)
self.assertEqual(response_content.get("isControl"), None)
self.assertEqual(response_content.get("courses"), [])

44
lms/djangoapps/learner_recommendations/toggles.py
View File

@@ -1,44 +0,0 @@
"""
Toggles for learner recommendations.
"""
from edx_toggles.toggles import WaffleFlag
# Namespace for learner_recommendations waffle flags.
WAFFLE_FLAG_NAMESPACE = 'learner_recommendations'
# Waffle flag to enable course about page recommendations.
# .. toggle_name: learner_recommendations.enable_course_about_page_recommendations
# .. toggle_implementation: WaffleFlag
# .. toggle_default: False
# .. toggle_description: Enable recommendations on course about page
# .. toggle_use_cases: temporary
# .. toggle_creation_date: 2023-01-30
# .. toggle_target_removal_date: None
# .. toggle_warning: None
# .. toggle_tickets: VAN-1259
ENABLE_COURSE_ABOUT_PAGE_RECOMMENDATIONS = WaffleFlag(
f'{WAFFLE_FLAG_NAMESPACE}.enable_course_about_page_recommendations', __name__
)
# Waffle flag to enable to recommendation panel on learner dashboard
# .. toggle_name: learner_recommendations.enable_dashboard_recommendations
# .. toggle_implementation: WaffleFlag
# .. toggle_default: False
# .. toggle_description: Waffle flag to enable to recommendation panel on learner dashboard
# .. toggle_use_cases: temporary
# .. toggle_creation_date: 2023-03-24
# .. toggle_target_removal_date: None
# .. toggle_warning: None
# .. toggle_tickets: VAN-1310
ENABLE_DASHBOARD_RECOMMENDATIONS = WaffleFlag(
f"{WAFFLE_FLAG_NAMESPACE}.enable_dashboard_recommendations", __name__
)
def enable_dashboard_recommendations():
return ENABLE_DASHBOARD_RECOMMENDATIONS.is_enabled()
def enable_course_about_page_recommendations():
return ENABLE_COURSE_ABOUT_PAGE_RECOMMENDATIONS.is_enabled()

32
lms/djangoapps/learner_recommendations/urls.py
View File

@@ -1,32 +0,0 @@
"""
Learner Recommendations URL routing configuration.
"""
from django.conf import settings
from django.urls import path
from django.urls import re_path
from lms.djangoapps.learner_recommendations import views
app_name = "learner_recommendations"
urlpatterns = [
re_path(fr'^amplitude/{settings.COURSE_ID_PATTERN}/$',
views.AboutPageRecommendationsView.as_view(),
name='amplitude_recommendations'),
re_path(fr'^cross_product/{settings.COURSE_ID_PATTERN}/$',
views.CrossProductRecommendationsView.as_view(),
name='cross_product_recommendations'),
path('product_recommendations/',
views.ProductRecommendationsView.as_view(),
name='product_recommendations_amplitude_only'),
re_path(fr'^product_recommendations/{settings.COURSE_ID_PATTERN}/$',
views.ProductRecommendationsView.as_view(),
name='product_recommendations'),
path("courses/",
views.DashboardRecommendationsApiView.as_view(),
name="courses"),
path('recommendations_context/',
views.RecommendationsContextView.as_view(),
name='recommendations_context'),
]

234
lms/djangoapps/learner_recommendations/utils.py
View File

@@ -1,234 +0,0 @@
"""
Additional utilities for Learner Recommendations.
"""
import logging
import requests
try:
from algoliasearch.search_client import SearchClient
except ImportError:
SearchClient = None
from django.conf import settings
from common.djangoapps.student.models import CourseEnrollment
from lms.djangoapps.program_enrollments.constants import ProgramEnrollmentStatuses
from openedx.core.djangoapps.catalog.utils import get_course_data, get_programs
from lms.djangoapps.program_enrollments.api import fetch_program_enrollments_by_student
log = logging.getLogger(__name__)
COURSE_LEVELS = [
'Introductory',
'Intermediate',
'Advanced'
]
class AlgoliaClient:
""" Class for instantiating an Algolia search client instance. """
algolia_client = None
algolia_app_id = settings.ALGOLIA_APP_ID
algolia_search_api_key = settings.ALGOLIA_SEARCH_API_KEY
@classmethod
def get_algolia_client(cls):
""" Get Algolia client instance. """
if not SearchClient:
return None
if not cls.algolia_client:
if not (cls.algolia_app_id and cls.algolia_search_api_key):
return None
cls.algolia_client = SearchClient.create(cls.algolia_app_id, cls.algolia_search_api_key)
return cls.algolia_client
def _get_user_enrolled_course_keys(user):
"""
Returns course ids in which the user is enrolled in.
"""
course_enrollments = CourseEnrollment.enrollments_for_user(user)
return [str(course_enrollment.course_id) for course_enrollment in course_enrollments]
def _is_enrolled_in_course(course_runs, enrolled_course_keys):
"""
Returns True if a user is enrolled in any course run of the course else false.
"""
return any(course_run.get("key", None) in enrolled_course_keys for course_run in course_runs)
def _has_country_restrictions(product, user_country):
"""
Helper method that tell whether the product (course or program) has any country restrictions.
A product is restricted for the user if the country in which user is logged in from:
- is in the "block list" or
- is not in the "allow list" if the "allow list" is not empty. If it is empty, then all locations can access it.
Args:
product: course/program
user_country (string): country the user is logged in from
Returns:
True if the product is restricted in the country and False otherwise
"""
if not user_country:
return False
allow_list, block_list = [], []
location_restriction = product.get("location_restriction", None)
if location_restriction:
restriction_type = location_restriction.get("restriction_type")
countries = location_restriction.get("countries")
if restriction_type == "allowlist":
allow_list = countries
elif restriction_type == "blocklist":
block_list = countries
return user_country in block_list or (bool(allow_list) and user_country not in allow_list)
def get_amplitude_course_recommendations(user_id, recommendation_id):
"""
Get personalized recommendations from Amplitude.
Args:
user_id: The user for which the recommendations need to be pulled
recommendation_id: Amplitude model id
Returns:
is_control (bool): Control group value for the user
has_is_control (bool): Boolean value indicating if the control group for
the user has been decided.
recommended_course_keys (list): Course keys returned by Amplitude.
"""
headers = {
"Authorization": f"Api-Key {settings.AMPLITUDE_API_KEY}",
"Content-Type": "application/json",
}
params = {
"user_id": user_id,
"get_recs": True,
"rec_id": recommendation_id,
}
response = requests.get(settings.AMPLITUDE_URL, params=params, headers=headers)
if response.status_code == 200:
response = response.json()
recommendations = response.get("userData", {}).get("recommendations", [])
if recommendations:
is_control = recommendations[0].get("is_control")
has_is_control = recommendations[0].get("has_is_control")
recommended_course_keys = recommendations[0].get("items")
return is_control, has_is_control, recommended_course_keys
return True, False, []
def is_user_enrolled_in_ut_austin_masters_program(user):
"""
Checks if a user is enrolled in any masters program
Args:
user: The user object
Returns:
True if the user is enrolled in UT Austin masters program otherwise False
"""
program_enrollments = fetch_program_enrollments_by_student(
user=user,
program_enrollment_statuses=ProgramEnrollmentStatuses.__ACTIVE__,
)
uuids = [enrollment.program_uuid for enrollment in program_enrollments]
enrolled_programs = get_programs(uuids=uuids) or []
for enrolled_program in enrolled_programs:
if enrolled_program.get("type", None) == "Masters":
authoring_organizations = enrolled_program.get("authoring_organizations", [])
if any(org.get("key", None) == "UTAustinX" for org in authoring_organizations):
return True
return False
def filter_recommended_courses(
user,
unfiltered_course_keys,
recommendation_count=10,
user_country_code=None,
request_course_key=None,
course_fields=None,
):
"""
Returns the filtered course recommendations. The unfiltered course keys
pass through the following filters:
1. Remove courses that a user is already enrolled in.
2. If user is seeing the recommendations on a course about pages, filter that course out of recommendations.
3. Remove the courses which is restricted in user region.
Args:
user: The user for which the recommendations need to be pulled
unfiltered_course_keys: recommended course keys that needs to be filtered
recommendation_count: the maximum count of recommendations to be returned
user_country_code: if provided, will apply location restrictions to recommendations
request_course_key: if provided, will filter out that course from recommendations (used for course about page)
fields: if provided, collects those fields on each course being queried, otherwise collects default fields
Returns:
filtered_recommended_courses (list): A list of filtered course objects.
"""
filtered_recommended_courses = []
fields = [
"key",
"uuid",
"title",
"owners",
"image",
"url_slug",
"course_runs",
"location_restriction",
"marketing_url",
"programs",
] if not course_fields else course_fields
# Filter out enrolled courses .
course_keys_to_filter_out = _get_user_enrolled_course_keys(user)
# If user is seeing the recommendations on a course about page, filter that course out of recommendations
if request_course_key:
course_keys_to_filter_out.append(request_course_key)
for course_id in unfiltered_course_keys:
if len(filtered_recommended_courses) >= recommendation_count:
break
course_data = get_course_data(course_id, fields, querystring={'marketable_course_runs_only': 1})
if (
course_data
and course_data.get("course_runs", [])
and not _is_enrolled_in_course(course_data.get("course_runs", []), course_keys_to_filter_out)
and not _has_country_restrictions(course_data, user_country_code)
):
filtered_recommended_courses.append(course_data)
return filtered_recommended_courses
def get_cross_product_recommendations(course_key):
"""
Helper method to get associated course keys based on the key passed
"""
return settings.CROSS_PRODUCT_RECOMMENDATIONS_KEYS.get(course_key)
def get_active_course_run(course):
"""
Returns an active course run based on prospectus frontend logic
for what defines an active course run
"""
course_runs = course.get("course_runs")
advertised_course_run_uuid = course.get("advertised_course_run_uuid")
if advertised_course_run_uuid:
for course_run in course_runs:
if course_run.get("uuid") == advertised_course_run_uuid:
return course_run
return None

483
lms/djangoapps/learner_recommendations/views.py
View File

@@ -1,483 +0,0 @@
"""
Views for Learner Recommendations.
"""
import logging
from django.conf import settings
from ipware.ip import get_client_ip
from edx_rest_framework_extensions.auth.jwt.authentication import JwtAuthentication
from edx_rest_framework_extensions.auth.session.authentication import (
SessionAuthenticationAllowInactiveUser,
)
from edx_rest_framework_extensions.permissions import NotJwtRestrictedApplication
from opaque_keys.edx.keys import CourseKey
from django.core.exceptions import PermissionDenied
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from common.djangoapps.track import segment
from common.djangoapps.student.toggles import show_fallback_recommendations
from openedx.core.djangoapps.geoinfo.api import country_code_from_ip
from openedx.core.djangoapps.catalog.utils import get_course_data
from openedx.features.enterprise_support.utils import is_enterprise_learner
from lms.djangoapps.learner_recommendations.toggles import (
enable_course_about_page_recommendations,
enable_dashboard_recommendations,
)
from lms.djangoapps.learner_recommendations.utils import (
_has_country_restrictions,
get_amplitude_course_recommendations,
filter_recommended_courses,
is_user_enrolled_in_ut_austin_masters_program,
get_cross_product_recommendations,
get_active_course_run,
)
from lms.djangoapps.learner_recommendations.serializers import (
AboutPageRecommendationsSerializer,
DashboardRecommendationsSerializer,
RecommendationsContextSerializer,
CrossProductAndAmplitudeRecommendationsSerializer,
CrossProductRecommendationsSerializer,
AmplitudeRecommendationsSerializer,
)
log = logging.getLogger(__name__)
class AboutPageRecommendationsView(APIView):
"""
IMPORTANT: Please do not update or use this API. This code has been moved to edx-recommendations plugin.
Please use that plugin for further code changes. This API will be removed as part of VAN-1427.
**Example Request**
GET api/learner_recommendations/amplitude/{course_id}/
"""
authentication_classes = (JwtAuthentication, SessionAuthenticationAllowInactiveUser,)
permission_classes = (IsAuthenticated,)
recommendations_count = 4
def _emit_recommendations_viewed_event(
self,
user_id,
is_control,
recommended_courses,
amplitude_recommendations=True,
):
"""Emits an event to track recommendation experiment views."""
segment.track(
user_id,
"edx.bi.user.recommendations.viewed",
{
"is_control": is_control,
"amplitude_recommendations": amplitude_recommendations,
"course_key_array": [
course["key"] for course in recommended_courses
],
"page": "course_about_page",
},
)
def get(self, request, course_id):
"""
Returns
- Amplitude course recommendations for course about page
"""
if not enable_course_about_page_recommendations():
return Response(status=404)
if is_enterprise_learner(request.user):
raise PermissionDenied()
user = request.user
try:
is_control, has_is_control, course_keys = get_amplitude_course_recommendations(
user.id, settings.COURSE_ABOUT_PAGE_AMPLITUDE_RECOMMENDATION_ID
)
except Exception as err: # pylint: disable=broad-except
log.warning(f"Amplitude API failed for {user.id} due to: {err}")
return Response(status=404)
is_control = is_control if has_is_control else None
recommended_courses = []
if not (is_control or is_control is None):
ip_address = get_client_ip(request)[0]
user_country_code = country_code_from_ip(ip_address).upper()
recommended_courses = filter_recommended_courses(
user,
course_keys,
user_country_code=user_country_code,
request_course_key=course_id,
recommendation_count=self.recommendations_count
)
for course in recommended_courses:
course.update({
"active_course_run": course.get("course_runs")[0]
})
self._emit_recommendations_viewed_event(
user.id, is_control, recommended_courses
)
return Response(
AboutPageRecommendationsSerializer(
{
"courses": recommended_courses,
"is_control": is_control,
}
).data,
status=200,
)
class CrossProductRecommendationsView(APIView):
"""
IMPORTANT: Please do not update or use this API. This code has been moved to edx_recommendations plugin.
Please use that plugin for further code changes. This API will be removed as part of VAN-1427.
**Example Request**
GET api/learner_recommendations/cross_product/{course_id}/
"""
def _empty_response(self):
return Response({"courses": []}, status=200)
def get(self, request, course_id):
"""
Returns cross product recommendation courses
"""
course_locator = CourseKey.from_string(course_id)
course_key = f'{course_locator.org}+{course_locator.course}'
associated_course_keys = get_cross_product_recommendations(course_key)
if not associated_course_keys:
return self._empty_response()
fields = [
"key",
"uuid",
"title",
"owners",
"image",
"url_slug",
"course_type",
"course_runs",
"location_restriction",
"advertised_course_run_uuid",
]
course_data = [get_course_data(key, fields) for key in associated_course_keys]
filtered_courses = [course for course in course_data if course and course.get("course_runs")]
ip_address = get_client_ip(request)[0]
user_country_code = country_code_from_ip(ip_address).upper()
unrestricted_courses = []
for course in filtered_courses:
if _has_country_restrictions(course, user_country_code):
continue
active_course_run = get_active_course_run(course)
if active_course_run:
course.update({"active_course_run": active_course_run})
unrestricted_courses.append(course)
if not unrestricted_courses:
return self._empty_response()
return Response(
CrossProductRecommendationsSerializer(
{
"courses": unrestricted_courses
}).data,
status=200
)
class RecommendationsContextView(APIView):
"""
IMPORTANT: Please do not update or use this API. This code has been moved to edx-recommendations plugin.
Please use that plugin for further code changes. This API will be removed as part of VAN-1427.
*Example Request*
GET /api/learner_recommendations/recommendations_context/
"""
authentication_classes = (
JwtAuthentication,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (IsAuthenticated, NotJwtRestrictedApplication)
def get(self, request):
"""
Returns the context needed for the recommendations experiment:
- Country Code
"""
ip_address = get_client_ip(request)[0]
country_code = country_code_from_ip(ip_address)
return Response(
RecommendationsContextSerializer(
{
"countryCode": country_code,
}
).data,
status=200,
)
class ProductRecommendationsView(APIView):
"""
IMPORTANT: Please do not update or use this API. This code has been moved to edx-recommendations plugin.
Please use that plugin for further code changes. This API will be removed as part of VAN-1427.
**Example Request**
GET api/learner_recommendations/product_recommendations/
GET api/learner_recommendations/product_recommendations/{course_id}/
"""
authentication_classes = (
JwtAuthentication,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (IsAuthenticated, NotJwtRestrictedApplication)
fields = [
"title",
"owners",
"image",
"course_type",
"course_runs",
"location_restriction",
"marketing_url",
"advertised_course_run_uuid",
]
def _get_amplitude_recommendations(self, user, user_country_code):
"""
Helper for getting amplitude recommendations
"""
fallback_recommendations = settings.GENERAL_RECOMMENDATIONS[0:4]
try:
_, _, course_keys = get_amplitude_course_recommendations(
user.id, settings.DASHBOARD_AMPLITUDE_RECOMMENDATION_ID
)
except Exception as ex: # pylint: disable=broad-except
log.warning(f"Cannot get recommendations from Amplitude: {ex}")
return fallback_recommendations
if not course_keys:
return fallback_recommendations
filtered_courses = filter_recommended_courses(
user, course_keys, recommendation_count=4, user_country_code=user_country_code, course_fields=self.fields
)
return filtered_courses if len(filtered_courses) > 0 else fallback_recommendations
def _get_cross_product_recommendations(self, course_key, user_country_code):
"""
Helper for getting cross product recommendations
"""
associated_course_keys = get_cross_product_recommendations(course_key)
if not associated_course_keys:
return []
course_data = [get_course_data(key, self.fields) for key in associated_course_keys]
filtered_cross_product_courses = []
for course in course_data:
if (
course
and course.get("course_runs", [])
and not _has_country_restrictions(course, user_country_code)
):
active_course_run = get_active_course_run(course)
if active_course_run:
course.update({"active_course_run_key": active_course_run.get("key")})
filtered_cross_product_courses.append(course)
return filtered_cross_product_courses
def _cross_product_recommendations_response(self, course_key, user, user_country_code):
"""
Helper for collecting and forming a response for
cross product and Amplitude recommendations
"""
if is_user_enrolled_in_ut_austin_masters_program(user):
return Response(
CrossProductAndAmplitudeRecommendationsSerializer(
{
"crossProductCourses": [],
"amplitudeCourses": []
}
).data,
status=200
)
amplitude_recommendations = self._get_amplitude_recommendations(user, user_country_code)
cross_product_recommendations = self._get_cross_product_recommendations(course_key, user_country_code)
return Response(
CrossProductAndAmplitudeRecommendationsSerializer(
{
"crossProductCourses": cross_product_recommendations,
"amplitudeCourses": amplitude_recommendations
}
).data,
status=200
)
def _amplitude_recommendations_response(self, user, user_country_code):
"""
Helper for collecting and forming a response for Amplitude recommendations only
"""
if is_user_enrolled_in_ut_austin_masters_program(user):
return Response(
AmplitudeRecommendationsSerializer({
"amplitudeCourses": []
}).data,
status=200
)
amplitude_recommendations = self._get_amplitude_recommendations(user, user_country_code)
return Response(
AmplitudeRecommendationsSerializer({
"amplitudeCourses": amplitude_recommendations
}).data,
status=200
)
def get(self, request, course_id=None):
"""
Returns cross product and Amplitude recommendation courses if a course id is included,
otherwise, returns only Amplitude recommendations
"""
ip_address = get_client_ip(request)[0]
user_country_code = country_code_from_ip(ip_address).upper()
if course_id:
course_locator = CourseKey.from_string(course_id)
course_key = f'{course_locator.org}+{course_locator.course}'
return self._cross_product_recommendations_response(course_key, request.user, user_country_code)
return self._amplitude_recommendations_response(request.user, user_country_code)
class DashboardRecommendationsApiView(APIView):
"""
IMPORTANT: Please do not update or use this API. This code has been moved to edx-recommendations plugin.
Please use that plugin for further code changes. This API will be removed as part of VAN-1427.
API to get personalized recommendations from Amplitude.
**Example Request**
GET /api/learner_recommendations/courses/
"""
authentication_classes = (
JwtAuthentication,
SessionAuthenticationAllowInactiveUser,
)
permission_classes = (IsAuthenticated, NotJwtRestrictedApplication)
def get(self, request):
"""
Retrieves course recommendations details.
"""
if not enable_dashboard_recommendations():
return Response(status=404)
user_id = request.user.id
if is_user_enrolled_in_ut_austin_masters_program(request.user):
return self._recommendations_response(user_id, None, [], False)
fallback_recommendations = settings.GENERAL_RECOMMENDATIONS if show_fallback_recommendations() else []
try:
is_control, has_is_control, course_keys = get_amplitude_course_recommendations(
user_id, settings.DASHBOARD_AMPLITUDE_RECOMMENDATION_ID
)
except Exception as ex: # pylint: disable=broad-except
log.warning(f"Cannot get recommendations from Amplitude: {ex}")
return self._recommendations_response(user_id, None, fallback_recommendations, False)
is_control = is_control if has_is_control else None
if is_control or is_control is None or not course_keys:
return self._recommendations_response(user_id, is_control, fallback_recommendations, False)
ip_address = get_client_ip(request)[0]
user_country_code = country_code_from_ip(ip_address).upper()
filtered_courses = filter_recommended_courses(
request.user, course_keys, user_country_code=user_country_code, recommendation_count=5
)
# If no courses are left after filtering already enrolled courses from
# the list of amplitude recommendations, show general recommendations
# to the user.
if not filtered_courses:
return self._recommendations_response(user_id, is_control, fallback_recommendations, False)
recommended_courses = list(map(self._course_data, filtered_courses))
return self._recommendations_response(user_id, is_control, recommended_courses, True)
def _emit_recommendations_viewed_event(
self, user_id, is_control, recommended_courses, amplitude_recommendations=True
):
"""Emits an event to track Learner Home page visits."""
segment.track(
user_id,
"edx.bi.user.recommendations.viewed",
{
"is_control": is_control,
"amplitude_recommendations": amplitude_recommendations,
"course_key_array": [course["course_key"] for course in recommended_courses],
"page": "dashboard",
},
)
def _recommendations_response(self, user_id, is_control, recommended_courses, amplitude_recommendations):
""" Helper method for general recommendations response. """
self._emit_recommendations_viewed_event(
user_id, is_control, recommended_courses, amplitude_recommendations
)
return Response(
DashboardRecommendationsSerializer(
{
"courses": recommended_courses,
"is_control": is_control,
}
).data,
status=200,
)
def _course_data(self, course):
"""Helper method for personalized recommendation response"""
return {
"course_key": course.get("key"),
"title": course.get("title"),
"logo_image_url": course.get("owners")[0]["logo_image_url"] if course.get(
"owners") else "",
"marketing_url": course.get("marketing_url"),
}

6
lms/urls.py
View File

@@ -199,12 +199,6 @@ urlpatterns = [
# Learner Home
path('api/learner_home/', include('lms.djangoapps.learner_home.urls', namespace='learner_home')),
# Learner Recommendations
path(
'api/learner_recommendations/',
include('lms.djangoapps.learner_recommendations.urls', namespace='learner_recommendations')
),
path(
'api/experiments/',
include(