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

feat: add tests for recommendation api and utils (#31679)

Browse Source
This commit is contained in:
Zainab Amir
2023-02-01 21:10:59 +05:00
committed by GitHub
parent 201499ae84
commit 2e4a20eded
4 changed files with 271 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -8,6 +8,7 @@ from common.djangoapps.student.tests.factories import (
UserFactory,
)
from lms.djangoapps.learner_recommendations.utils import (
_has_country_restrictions,
filter_recommended_courses,
get_amplitude_course_recommendations,
)
@@ -55,6 +56,27 @@ class TestRecommendationsHelper(TestCase):
)
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."""
@@ -62,12 +84,10 @@ class TestFilterRecommendedCourses(ModuleStoreTestCase):
def setUp(self):
super().setUp()
self.user = UserFactory()
self.course_data = {
"course_key": "Mocked course key",
"title": "Mocked course title",
"owners": [{"logo_image_url": "https://www.logo_image_url.com"}],
"marketing_url": "https://www.marketing_url.com",
}
self.unrestricted_course_keys = [
"MITx+6.00.1x",
"IBM+PY0101EN",
]
self.recommended_course_keys = [
"MITx+6.00.1x",
"IBM+PY0101EN",
@@ -93,6 +113,29 @@ class TestFilterRecommendedCourses(ModuleStoreTestCase):
"course-v1:MichinX+101x+Run_0",
]
def _mock_get_course_data(self, course_id, fields=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"],
}
}
)
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
@@ -103,11 +146,49 @@ class TestFilterRecommendedCourses(ModuleStoreTestCase):
"""
total_enrolled_courses = 6
total_recommendations = len(self.recommended_course_keys)
mocked_get_course_data.return_value = self.course_data
mocked_get_course_data.side_effect = self._mock_get_course_data
for course_run_key in self.course_run_keys[:total_enrolled_courses]:
CourseEnrollmentFactory(course_id=course_run_key, user=self.user)
filtered_courses = filter_recommended_courses(
self.user, self.recommended_course_keys, total_recommendations
)
self.assertEqual(len(filtered_courses), total_recommendations - total_enrolled_courses)
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.recommended_course_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=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

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

@@ -3,12 +3,15 @@ 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
from django.urls import reverse_lazy
from rest_framework.test import APITestCase
from common.djangoapps.student.tests.factories import UserFactory
from lms.djangoapps.learner_recommendations.toggles import (
ENABLE_COURSE_ABOUT_PAGE_RECOMMENDATIONS,
)
class TestAlgoliaCoursesSearchView(APITestCase):
@@ -132,3 +135,127 @@ class TestAlgoliaCoursesSearchView(APITestCase):
response_content = json.loads(response.content)
self.assertEqual(response_content.get("courses"), self.expected_courses_recommendation["hits"])
self.assertEqual(response_content.get("count"), self.expected_courses_recommendation["nbHits"])
@override_waffle_flag(ENABLE_COURSE_ABOUT_PAGE_RECOMMENDATIONS, active=True)
class TestAmplitudeRecommendationsView(APITestCase):
"""Unit tests for the Amplitude recommendations API"""
url = reverse_lazy(
"learner_recommendations:amplitude_recommendations",
kwargs={'course_id': 'course-v1:test+TestX+Test_Course'}
)
def setUp(self):
super().setUp()
self.user = UserFactory()
self.client.login(username=self.user.username, password="test")
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",
]
def _get_filtered_courses(self):
"""
Returns the filtered course data
"""
filtered_course = []
for course_key in self.recommended_courses:
filtered_course.append({
"key": course_key,
"uuid": "4f8cb2c9-589b-4d1e-88c1-b01a02db3a9c",
"title": f"Title for {course_key}",
"image": "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.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
):
"""
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,
]
response = self.client.get(self.url)
response_content = json.loads(response.content)
self.assertEqual(response.status_code, 200)
self.assertEqual(response_content.get("is_control"), False)
self.assertEqual(
len(response_content.get("courses")), expected_recommendations_length
)

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

@@ -23,6 +23,7 @@ COURSE_LEVELS = [
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
@@ -78,10 +79,10 @@ def _has_country_restrictions(product, user_country):
countries = location_restriction.get("countries")
if restriction_type == "allowlist":
allow_list = countries
if restriction_type == "blocklist":
elif restriction_type == "blocklist":
block_list = countries
return user_country in block_list or (allow_list and user_country not in allow_list)
return user_country in block_list or (bool(allow_list) and user_country not in allow_list)
def _parse_course_owner_data(owner):
@@ -91,17 +92,21 @@ def _parse_course_owner_data(owner):
return {
"key": owner.get("key"),
"name": owner.get("name"),
"logo_image_url": owner.get("logo_image_url")
"logo_image_url": owner.get("logo_image_url"),
}
def course_data_for_discovery_card(course_data):
"""Helper method to prepare data for prospectus course card"""
recommended_course_data = {}
active_course_run = [course_run for course_run in course_data.get("course_runs", [])
if course_run.get("availability") == "Current"][0]
if active_course_run:
owners = map(_parse_course_owner_data, course_data.get("owners"))
active_course_runs = [
course_run
for course_run in course_data.get("course_runs", [])
if course_run.get("availability") == "Current"
]
if active_course_runs:
owners = list(map(_parse_course_owner_data, course_data.get("owners")))
recommended_course_data.update({
"uuid": course_data.get("uuid"),
"title": course_data.get("title"),
@@ -109,11 +114,12 @@ def course_data_for_discovery_card(course_data):
"owners": owners,
"prospectus_path": f"courses/{course_data.get('url_slug')}",
"active_course_run": {
"key": active_course_run.get("key"),
"key": active_course_runs[0].get("key"),
"type": "Active",
"marketing_url": active_course_run.get("marketing_url"),
"marketing_url": active_course_runs[0].get("marketing_url"),
}
})
return recommended_course_data
@@ -196,7 +202,11 @@ def get_amplitude_course_recommendations(user_id, recommendation_id):
def filter_recommended_courses(
user, unfiltered_course_keys, recommendation_count=10, user_country_code=None, request_course=None
user,
unfiltered_course_keys,
recommendation_count=10,
user_country_code=None,
request_course=None,
):
"""
Returns the filtered course recommendations. The unfiltered course keys
@@ -215,12 +225,18 @@ def filter_recommended_courses(
# Remove the course keys a user is already enrolled in
enrollable_course_keys = _remove_user_enrolled_course_keys(user, unfiltered_course_keys)
# If user is seeing the recommendations on a course about pages, filter that course out of recommendations
recommended_course_keys = [course_key for course_key in enrollable_course_keys if course_key != request_course]
# If user is seeing the recommendations on a course about page, filter that course out of recommendations
recommended_course_keys = [
course_key
for course_key in enrollable_course_keys
if course_key != request_course
]
for course_id in recommended_course_keys:
if len(filtered_recommended_courses) >= recommendation_count:
break
course_data = get_course_data(course_id, fields)
if course_data and not _has_country_restrictions(course_data, user_country_code):
filtered_recommended_courses.append(course_data)

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

@@ -10,13 +10,14 @@ from edx_rest_framework_extensions.auth.session.authentication import (
SessionAuthenticationAllowInactiveUser,
)
from django.core.exceptions import PermissionDenied
from opaque_keys.edx.keys import CourseKey
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from openedx.core.djangoapps.catalog.utils import (
get_course_data,
get_course_run_details
get_course_run_details,
)
from openedx.core.djangoapps.geoinfo.api import country_code_from_ip
from openedx.features.enterprise_support.utils import is_enterprise_learner
@@ -71,39 +72,47 @@ class AmplitudeRecommendationsView(APIView):
authentication_classes = (JwtAuthentication, SessionAuthenticationAllowInactiveUser,)
permission_classes = (IsAuthenticated,)
recommendations_count = 4
def get(self, request, course_id):
"""
Recommend courses based on amplitude recommendations.
Recommend program if user is enrolled-in any other course of program.
Returns
- Amplitude course recommendations
- Upsell program if the requesting course has a related program
"""
if not enable_course_about_page_recommendations():
return Response("Recommendations not found", status=404)
return Response(status=404)
if is_enterprise_learner(request.user):
raise PermissionDenied()
recommendation_count = 4
course_key = course_id
split_course_key = course_key.split(":")[-1]
split_course_id = split_course_key.split("+")[:2]
course_id = f"{split_course_id[0]}+{split_course_id[1]}"
user = request.user
course_locator = CourseKey.from_string(course_id)
course_key = f'{course_locator.org}+{course_locator.course}'
try:
is_control, has_is_control, course_keys = get_amplitude_course_recommendations(
request.user.id, settings.COURSE_ABOUT_PAGE_AMPLITUDE_RECOMMENDATION_ID
user.id, settings.COURSE_ABOUT_PAGE_AMPLITUDE_RECOMMENDATION_ID
)
except Exception as err: # pylint: disable=broad-except
log.info(f"Amplitude API failed due to: {err}")
return Response("Recommendations not found", status=404)
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
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, request_course=course_id,
request.user, course_keys, user_country_code=user_country_code, request_course=course_key,
)
recommended_courses = map(course_data_for_discovery_card, filtered_courses)
recommended_courses = recommended_courses[:recommendation_count]
recommended_courses = []
for course in filtered_courses:
course_data = course_data_for_discovery_card(course)
if course_data:
recommended_courses.append(course_data)
if len(recommended_courses) == self.recommendations_count:
break
return Response({"is_control": is_control, "courses": recommended_courses}, status=200)