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

build: add support for Python 3.12 (#37160)

Browse Source 
* build: add support for Python 3.12
This commit is contained in:
Usama Sadiq
2025-12-15 12:12:29 +05:00
committed by GitHub
parent a3cc78a377
commit e1747f3844
13 changed files with 1826 additions and 1783 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.github/workflows/unit-tests.yml vendored
View File

@@ -23,6 +23,7 @@ jobs:
matrix:
python-version:
- "3.11"
- "3.12"
django-version:
- "pinned"
# When updating the shards, remember to make the same changes in

38
cms/djangoapps/contentstore/video_storage_handlers.py
View File

@@ -13,12 +13,11 @@ import requests
import shutil
import pathlib
import zipfile
import boto3
from contextlib import closing
from datetime import datetime, timedelta
from uuid import uuid4
from boto.s3.connection import S3Connection
from boto import s3
from django.conf import settings
from django.contrib.staticfiles.storage import staticfiles_storage
from django.http import FileResponse, HttpResponseNotFound, StreamingHttpResponse
@@ -826,7 +825,7 @@ def videos_post(course, request):
return {'error': error_msg}, 400
edx_video_id = str(uuid4())
key = storage_service_key(bucket, file_name=edx_video_id)
key_name = storage_service_key(bucket, file_name=edx_video_id)
metadata_list = [
('client_video_id', file_name),
@@ -846,12 +845,20 @@ def videos_post(course, request):
if transcript_preferences is not None:
metadata_list.append(('transcript_preferences', json.dumps(transcript_preferences)))
for metadata_name, value in metadata_list:
key.set_metadata(metadata_name, value)
upload_url = key.generate_url(
KEY_EXPIRATION_IN_SECONDS,
'PUT',
headers={'Content-Type': req_file['content_type']}
# Prepare metadata for presigned URL
metadata = dict(metadata_list)
# Generate presigned URL using boto3
s3_client = bucket.meta.client
upload_url = s3_client.generate_presigned_url(
'put_object',
Params={
'Bucket': bucket.name,
'Key': key_name,
'ContentType': req_file['content_type'],
'Metadata': metadata
},
ExpiresIn=KEY_EXPIRATION_IN_SECONDS
)
# persist edx_video_id in VAL
@@ -886,24 +893,19 @@ def storage_service_bucket():
'aws_secret_access_key': settings.AWS_SECRET_ACCESS_KEY
}
conn = S3Connection(**params)
conn = boto3.resource("s3", **params)
# We don't need to validate our bucket, it requires a very permissive IAM permission
# set since behind the scenes it fires a HEAD request that is equivalent to get_all_keys()
# meaning it would need ListObjects on the whole bucket, not just the path used in each
# environment (since we share a single bucket for multiple deployments in some configurations)
return conn.get_bucket(settings.VIDEO_UPLOAD_PIPELINE['VEM_S3_BUCKET'], validate=False)
return conn.Bucket(settings.VIDEO_UPLOAD_PIPELINE['VEM_S3_BUCKET'])
def storage_service_key(bucket, file_name):
"""
Returns an S3 key to the given file in the given bucket.
Returns an S3 key name for the given file in the given bucket.
"""
key_name = "{}/{}".format(
return "{}/{}".format(
settings.VIDEO_UPLOAD_PIPELINE.get("ROOT_PATH", ""),
file_name
)
return s3.key.Key(bucket, key_name)
def send_video_status_update(updates):

3539
cms/djangoapps/contentstore/views/tests/test_videos.py
View File

@@ -1,1746 +1,1793 @@
"""
Unit tests for video-related REST APIs.
"""
import csv
import json
import re
from contextlib import contextmanager
from datetime import datetime
from io import StringIO
from unittest.mock import Mock, patch
import dateutil.parser
from common.djangoapps.student.tests.factories import UserFactory
import ddt
import pytz
from django.test import TestCase
from django.conf import settings
from django.test.utils import override_settings
from django.urls import reverse
from edx_toggles.toggles.testutils import override_waffle_flag, override_waffle_switch
from edxval.api import (
create_or_update_transcript_preferences,
create_or_update_video_transcript,
create_profile,
create_video,
get_course_video_image_url,
get_transcript_preferences,
get_video_info
)
from cms.djangoapps.contentstore.models import VideoUploadConfig
from cms.djangoapps.contentstore.tests.utils import CourseTestCase
from cms.djangoapps.contentstore.utils import reverse_course_url
from openedx.core.djangoapps.profile_images.tests.helpers import make_image_file
from openedx.core.djangoapps.video_pipeline.config.waffle import (
DEPRECATE_YOUTUBE,
ENABLE_DEVSTACK_VIDEO_UPLOADS,
)
from openedx.core.djangoapps.waffle_utils.models import WaffleFlagCourseOverrideModel
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory # lint-amnesty, pylint: disable=wrong-import-order
from ..videos import (
ENABLE_VIDEO_UPLOAD_PAGINATION,
KEY_EXPIRATION_IN_SECONDS,
VIDEO_IMAGE_UPLOAD_ENABLED,
)
from cms.djangoapps.contentstore.video_storage_handlers import (
_get_default_video_image_url,
TranscriptProvider,
StatusDisplayStrings,
convert_video_status,
storage_service_bucket,
storage_service_key,
PUBLIC_VIDEO_SHARE
)
class VideoUploadTestBase:
"""
Test cases for the video upload feature
"""
def get_url_for_course_key(self, course_key, kwargs=None):
"""Return video handler URL for the given course"""
return reverse_course_url(self.VIEW_NAME, course_key, kwargs) # lint-amnesty, pylint: disable=no-member
def setUp(self):
super().setUp() # lint-amnesty, pylint: disable=no-member
self.url = self.get_url_for_course_key(self.course.id)
self.test_token = "test_token"
self.course.video_upload_pipeline = {
"course_video_upload_token": self.test_token,
}
self.save_course() # lint-amnesty, pylint: disable=no-member
# create another course for videos belonging to multiple courses
self.course2 = CourseFactory.create()
self.course2.video_upload_pipeline = {
"course_video_upload_token": self.test_token,
}
self.course2.save()
self.store.update_item(self.course2, self.user.id) # lint-amnesty, pylint: disable=no-member
# course ids for videos
course_ids = [str(self.course.id), str(self.course2.id)]
created = datetime.now(pytz.utc)
self.profiles = ["profile1", "profile2"]
self.previous_uploads = [
{
"edx_video_id": "test1",
"client_video_id": "test1.mp4",
"duration": 42.0,
"status": "upload",
"courses": course_ids,
"encoded_videos": [],
"created": created
},
{
"edx_video_id": "test2",
"client_video_id": "test2.mp4",
"duration": 128.0,
"status": "file_complete",
"courses": course_ids,
"created": created,
"encoded_videos": [
{
"profile": "profile1",
"url": "http://example.com/profile1/test2.mp4",
"file_size": 1600,
"bitrate": 100,
},
{
"profile": "profile2",
"url": "http://example.com/profile2/test2.mov",
"file_size": 16000,
"bitrate": 1000,
},
],
},
{
"edx_video_id": "non-ascii",
"client_video_id": "nón-ascii-näme.mp4",
"duration": 256.0,
"status": "transcode_active",
"courses": course_ids,
"created": created,
"encoded_videos": [
{
"profile": "profile1",
"url": "http://example.com/profile1/nón-ascii-näme.mp4",
"file_size": 3200,
"bitrate": 100,
},
]
},
]
# Ensure every status string is tested
self.previous_uploads += [
{
"edx_video_id": f"status_test_{status}",
"client_video_id": "status_test.mp4",
"duration": 3.14,
"status": status,
"courses": course_ids,
"created": created,
"encoded_videos": [],
}
for status in (
list(StatusDisplayStrings._STATUS_MAP.keys()) + # pylint:disable=protected-access
["non_existent_status"]
)
]
for profile in self.profiles:
create_profile(profile)
for video in self.previous_uploads:
create_video(video)
def _get_previous_upload(self, edx_video_id):
"""Returns the previous upload with the given video id."""
return next(
video
for video in self.previous_uploads
if video["edx_video_id"] == edx_video_id
)
class VideoStudioAccessTestsMixin:
"""
Base Access tests for studio video views
"""
def test_anon_user(self):
self.client.logout()
response = self.client.get(self.url)
self.assertEqual(response.status_code, 302)
def test_put(self):
response = self.client.put(self.url)
self.assertEqual(response.status_code, 405)
def test_invalid_course_key(self):
response = self.client.get(
self.get_url_for_course_key("Non/Existent/Course")
)
self.assertEqual(response.status_code, 404)
def test_non_staff_user(self):
client, __ = self.create_non_staff_authed_user_client()
response = client.get(self.url)
self.assertEqual(response.status_code, 403)
class VideoPipelineStudioAccessTestsMixin:
"""
Access tests for video views that rely on the video pipeline
"""
def test_video_pipeline_not_enabled(self):
settings.FEATURES["ENABLE_VIDEO_UPLOAD_PIPELINE"] = False
self.assertEqual(self.client.get(self.url).status_code, 404)
def test_video_pipeline_not_configured(self):
settings.VIDEO_UPLOAD_PIPELINE = None
self.assertEqual(self.client.get(self.url).status_code, 404)
def test_course_not_configured(self):
self.course.video_upload_pipeline = {}
self.save_course()
self.assertEqual(self.client.get(self.url).status_code, 404)
class VideoUploadPostTestsMixin:
"""
Shared test cases for video post tests.
"""
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@patch('boto.s3.key.Key')
@patch('cms.djangoapps.contentstore.video_storage_handlers.S3Connection')
def test_post_success(self, mock_conn, mock_key):
files = [
{
'file_name': 'first.mp4',
'content_type': 'video/mp4',
},
{
'file_name': 'second.mp4',
'content_type': 'video/mp4',
},
{
'file_name': 'third.mov',
'content_type': 'video/quicktime',
},
{
'file_name': 'fourth.mp4',
'content_type': 'video/mp4',
},
]
bucket = Mock()
mock_conn.return_value = Mock(get_bucket=Mock(return_value=bucket))
mock_key_instances = [
Mock(
generate_url=Mock(
return_value='http://example.com/url_{}'.format(file_info['file_name'])
)
)
for file_info in files
]
# If extra calls are made, return a dummy
mock_key.side_effect = mock_key_instances + [Mock()]
response = self.client.post(
self.url,
json.dumps({'files': files}),
content_type='application/json'
)
self.assertEqual(response.status_code, 200)
response_obj = json.loads(response.content.decode('utf-8'))
mock_conn.assert_called_once_with(
aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY
)
self.assertEqual(len(response_obj['files']), len(files))
self.assertEqual(mock_key.call_count, len(files))
for i, file_info in enumerate(files):
# Ensure Key was set up correctly and extract id
key_call_args, __ = mock_key.call_args_list[i]
self.assertEqual(key_call_args[0], bucket)
path_match = re.match(
(
settings.VIDEO_UPLOAD_PIPELINE['ROOT_PATH'] +
'/([a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12})$'
),
key_call_args[1]
)
self.assertIsNotNone(path_match)
video_id = path_match.group(1)
mock_key_instance = mock_key_instances[i]
mock_key_instance.set_metadata.assert_any_call(
'course_video_upload_token',
self.test_token
)
mock_key_instance.set_metadata.assert_any_call(
'client_video_id',
file_info['file_name']
)
mock_key_instance.set_metadata.assert_any_call('course_key', str(self.course.id))
mock_key_instance.generate_url.assert_called_once_with(
KEY_EXPIRATION_IN_SECONDS,
'PUT',
headers={'Content-Type': file_info['content_type']}
)
# Ensure VAL was updated
val_info = get_video_info(video_id)
self.assertEqual(val_info['status'], 'upload')
self.assertEqual(val_info['client_video_id'], file_info['file_name'])
self.assertEqual(val_info['status'], 'upload')
self.assertEqual(val_info['duration'], 0)
self.assertEqual(val_info['courses'], [{str(self.course.id): None}])
# Ensure response is correct
response_file = response_obj['files'][i]
self.assertEqual(response_file['file_name'], file_info['file_name'])
self.assertEqual(response_file['upload_url'], mock_key_instance.generate_url())
def test_post_non_json(self):
response = self.client.post(self.url, {"files": []})
self.assertEqual(response.status_code, 400)
def test_post_malformed_json(self):
response = self.client.post(self.url, "{", content_type="application/json")
self.assertEqual(response.status_code, 400)
def test_post_invalid_json(self):
def assert_bad(content):
"""Make request with content and assert that response is 400"""
response = self.client.post(
self.url,
json.dumps(content),
content_type="application/json"
)
self.assertEqual(response.status_code, 400)
# Top level missing files key
assert_bad({})
# Entry missing file_name
assert_bad({"files": [{"content_type": "video/mp4"}]})
# Entry missing content_type
assert_bad({"files": [{"file_name": "test.mp4"}]})
@ddt.ddt
@patch.dict("django.conf.settings.FEATURES", {"ENABLE_VIDEO_UPLOAD_PIPELINE": True})
@override_settings(VIDEO_UPLOAD_PIPELINE={
"VEM_S3_BUCKET": "vem_test_bucket", "BUCKET": "test_bucket", "ROOT_PATH": "test_root"
})
class VideosHandlerTestCase(
VideoUploadTestBase,
VideoStudioAccessTestsMixin,
VideoPipelineStudioAccessTestsMixin,
VideoUploadPostTestsMixin,
CourseTestCase
):
"""Test cases for the main video upload endpoint"""
VIEW_NAME = 'videos_handler'
def test_get_json(self):
response = self.client.get_json(self.url)
self.assertEqual(response.status_code, 200)
response_videos = json.loads(response.content.decode('utf-8'))['videos']
self.assertEqual(len(response_videos), len(self.previous_uploads))
for i, response_video in enumerate(response_videos):
# Videos should be returned by creation date descending
original_video = self.previous_uploads[-(i + 1)]
print(response_video.keys())
self.assertEqual(
set(response_video.keys()),
{
'edx_video_id',
'client_video_id',
'created',
'duration',
'status',
'status_nontranslated',
'course_video_image_url',
'file_size',
'download_link',
'transcripts',
'transcription_status',
'transcript_urls',
'error_description',
}
)
dateutil.parser.parse(response_video['created'])
for field in ['edx_video_id', 'client_video_id', 'duration']:
self.assertEqual(response_video[field], original_video[field])
self.assertEqual(
response_video['status'],
convert_video_status(original_video)
)
@ddt.data(
(
[
'edx_video_id', 'client_video_id', 'created', 'duration',
'status', 'status_nontranslated', 'course_video_image_url', 'file_size',
'download_link', 'transcripts', 'transcription_status', 'transcript_urls',
'error_description'
],
[
{
'video_id': 'test1',
'language_code': 'en',
'file_name': 'edx101.srt',
'file_format': 'srt',
'provider': 'Cielo24'
}
],
['en']
),
(
[
'edx_video_id', 'client_video_id', 'created', 'duration',
'status', 'status_nontranslated', 'course_video_image_url', 'file_size',
'download_link', 'transcripts', 'transcription_status', 'transcript_urls',
'error_description'
],
[
{
'video_id': 'test1',
'language_code': 'en',
'file_name': 'edx101_en.srt',
'file_format': 'srt',
'provider': 'Cielo24'
},
{
'video_id': 'test1',
'language_code': 'es',
'file_name': 'edx101_es.srt',
'file_format': 'srt',
'provider': 'Cielo24'
}
],
['en', 'es']
)
)
@ddt.unpack
def test_get_json_transcripts(self, expected_video_keys, uploaded_transcripts, expected_transcripts):
"""
Test that transcripts are attached based on whether the video transcript feature is enabled.
"""
for transcript in uploaded_transcripts:
create_or_update_video_transcript(
transcript['video_id'],
transcript['language_code'],
metadata={
'file_name': transcript['file_name'],
'file_format': transcript['file_format'],
'provider': transcript['provider']
}
)
response = self.client.get_json(self.url)
self.assertEqual(response.status_code, 200)
response_videos = json.loads(response.content.decode('utf-8'))['videos']
self.assertEqual(len(response_videos), len(self.previous_uploads))
for response_video in response_videos:
print(response_video)
self.assertEqual(set(response_video.keys()), set(expected_video_keys))
if response_video['edx_video_id'] == self.previous_uploads[0]['edx_video_id']:
self.assertEqual(response_video.get('transcripts', []), expected_transcripts)
def test_get_html(self):
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertRegex(response["Content-Type"], "^text/html(;.*)?$")
self.assertContains(response, _get_default_video_image_url())
# Crude check for presence of data in returned HTML
for video in self.previous_uploads:
self.assertContains(response, video["edx_video_id"])
self.assertNotContains(response, 'video_upload_pagination')
@override_waffle_flag(ENABLE_VIDEO_UPLOAD_PAGINATION, active=True)
def test_get_html_paginated(self):
"""
Tests that response is paginated.
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'video_upload_pagination')
@override_settings(AWS_ACCESS_KEY_ID="test_key_id", AWS_SECRET_ACCESS_KEY="test_secret")
@patch("boto.s3.key.Key")
@patch("cms.djangoapps.contentstore.video_storage_handlers.S3Connection")
@ddt.data(
(
[
{
"file_name": "supported-1.mp4",
"content_type": "video/mp4",
},
{
"file_name": "supported-2.mov",
"content_type": "video/quicktime",
},
],
200
),
(
[
{
"file_name": "unsupported-1.txt",
"content_type": "text/plain",
},
{
"file_name": "unsupported-2.png",
"content_type": "image/png",
},
],
400
)
)
@ddt.unpack
def test_video_supported_file_formats(self, files, expected_status, mock_conn, mock_key):
"""
Test that video upload works correctly against supported and unsupported file formats.
"""
mock_conn.get_bucket = Mock()
mock_key_instances = [
Mock(
generate_url=Mock(
return_value="http://example.com/url_{}".format(file_info["file_name"])
)
)
for file_info in files
]
# If extra calls are made, return a dummy
mock_key.side_effect = mock_key_instances + [Mock()]
# Check supported formats
response = self.client.post(
self.url,
json.dumps({"files": files}),
content_type="application/json"
)
self.assertEqual(response.status_code, expected_status)
response = json.loads(response.content.decode('utf-8'))
if expected_status == 200:
self.assertNotIn('error', response)
else:
self.assertIn('error', response)
self.assertEqual(response['error'], "Request 'files' entry contain unsupported content_type")
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@patch('cms.djangoapps.contentstore.video_storage_handlers.S3Connection')
def test_upload_with_non_ascii_charaters(self, mock_conn):
"""
Test that video uploads throws error message when file name contains special characters.
"""
mock_conn.get_bucket = Mock()
file_name = 'test\u2019_file.mp4'
files = [{'file_name': file_name, 'content_type': 'video/mp4'}]
bucket = Mock()
mock_conn.return_value = Mock(get_bucket=Mock(return_value=bucket))
response = self.client.post(
self.url,
json.dumps({'files': files}),
content_type='application/json'
)
self.assertEqual(response.status_code, 400)
response = json.loads(response.content.decode('utf-8'))
self.assertEqual(response['error'], 'The file name for %s must contain only ASCII characters.' % file_name)
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret', AWS_SECURITY_TOKEN='token')
@patch('boto.s3.key.Key')
@patch('cms.djangoapps.contentstore.video_storage_handlers.S3Connection')
@override_waffle_flag(ENABLE_DEVSTACK_VIDEO_UPLOADS, active=True)
def test_devstack_upload_connection(self, mock_conn, mock_key):
files = [{'file_name': 'first.mp4', 'content_type': 'video/mp4'}]
mock_conn.get_bucket = Mock()
mock_key_instances = [
Mock(
generate_url=Mock(
return_value='http://example.com/url_{}'.format(file_info['file_name'])
)
)
for file_info in files
]
mock_key.side_effect = mock_key_instances
response = self.client.post(
self.url,
json.dumps({'files': files}),
content_type='application/json'
)
self.assertEqual(response.status_code, 200)
mock_conn.assert_called_once_with(
aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
security_token=settings.AWS_SECURITY_TOKEN
)
@patch('boto.s3.key.Key')
@patch('cms.djangoapps.contentstore.video_storage_handlers.S3Connection')
def test_send_course_to_vem_pipeline(self, mock_conn, mock_key):
"""
Test that uploads always go to VEM S3 bucket by default.
"""
mock_conn.get_bucket = Mock()
files = [{'file_name': 'first.mp4', 'content_type': 'video/mp4'}]
mock_key_instances = [
Mock(
generate_url=Mock(
return_value='http://example.com/url_{}'.format(file_info['file_name'])
)
)
for file_info in files
]
mock_key.side_effect = mock_key_instances
response = self.client.post(
self.url,
json.dumps({'files': files}),
content_type='application/json'
)
self.assertEqual(response.status_code, 200)
mock_conn.return_value.get_bucket.assert_called_once_with(
settings.VIDEO_UPLOAD_PIPELINE['VEM_S3_BUCKET'], validate=False # pylint: disable=unsubscriptable-object
)
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@patch('boto.s3.key.Key')
@patch('cms.djangoapps.contentstore.video_storage_handlers.S3Connection')
@ddt.data(
{
'global_waffle': True,
'course_override': WaffleFlagCourseOverrideModel.ALL_CHOICES.off,
'expect_token': True
},
{
'global_waffle': False,
'course_override': WaffleFlagCourseOverrideModel.ALL_CHOICES.on,
'expect_token': False
},
{
'global_waffle': False,
'course_override': WaffleFlagCourseOverrideModel.ALL_CHOICES.off,
'expect_token': True
}
)
def test_video_upload_token_in_meta(self, data, mock_conn, mock_key):
"""
Test video upload token in s3 metadata.
"""
@contextmanager
def proxy_manager(manager, ignore_manager):
"""
This acts as proxy to the original manager in the arguments given
the original manager is not set to be ignored.
"""
if ignore_manager:
yield
else:
with manager:
yield
file_data = {
'file_name': 'first.mp4',
'content_type': 'video/mp4',
}
mock_conn.get_bucket = Mock()
mock_key_instance = Mock(
generate_url=Mock(
return_value='http://example.com/url_{}'.format(file_data['file_name'])
)
)
# If extra calls are made, return a dummy
mock_key.side_effect = [mock_key_instance]
# expected args to be passed to `set_metadata`.
expected_args = ('course_video_upload_token', self.test_token)
with patch.object(WaffleFlagCourseOverrideModel, 'override_value', return_value=data['course_override']):
with override_waffle_flag(DEPRECATE_YOUTUBE, active=data['global_waffle']):
response = self.client.post(
self.url,
json.dumps({'files': [file_data]}),
content_type='application/json'
)
self.assertEqual(response.status_code, 200)
with proxy_manager(self.assertRaises(AssertionError), data['expect_token']):
# if we're not expecting token then following should raise assertion error and
# if we're expecting token then we will be able to find the call to set the token
# in s3 metadata.
mock_key_instance.set_metadata.assert_any_call(*expected_args)
def _assert_video_removal(self, url, edx_video_id, deleted_videos):
"""
Verify that if correct video is removed from a particular course.
Arguments:
url (str): URL to get uploaded videos
edx_video_id (str): video id
deleted_videos (int): how many videos are deleted
"""
response = self.client.get_json(url)
self.assertEqual(response.status_code, 200)
response_videos = json.loads(response.content.decode('utf-8'))["videos"]
self.assertEqual(len(response_videos), len(self.previous_uploads) - deleted_videos)
if deleted_videos:
self.assertNotIn(edx_video_id, [video.get('edx_video_id') for video in response_videos])
else:
self.assertIn(edx_video_id, [video.get('edx_video_id') for video in response_videos])
def test_video_removal(self):
"""
Verifies that video removal is working as expected.
"""
edx_video_id = 'test1'
remove_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
response = self.client.delete(remove_url, HTTP_ACCEPT="application/json")
self.assertEqual(response.status_code, 204)
self._assert_video_removal(self.url, edx_video_id, 1)
def test_video_removal_multiple_courses(self):
"""
Verifies that video removal is working as expected for multiple courses.
If a video is used by multiple courses then removal from one course shouldn't effect the other course.
"""
# remove video from course1
edx_video_id = 'test1'
remove_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
response = self.client.delete(remove_url, HTTP_ACCEPT="application/json")
self.assertEqual(response.status_code, 204)
# verify that video is only deleted from course1 only
self._assert_video_removal(self.url, edx_video_id, 1)
self._assert_video_removal(self.get_url_for_course_key(self.course2.id), edx_video_id, 0)
def test_convert_video_status(self):
"""
Verifies that convert_video_status works as expected.
"""
video = self.previous_uploads[0]
# video status should be failed if it's in upload state for more than 24 hours
video['created'] = datetime(2016, 1, 1, 10, 10, 10, 0, pytz.UTC)
status = convert_video_status(video)
self.assertEqual(status, StatusDisplayStrings.get('upload_failed'))
# `invalid_token` should be converted to `youtube_duplicate`
video['created'] = datetime.now(pytz.UTC)
video['status'] = 'invalid_token'
status = convert_video_status(video)
self.assertEqual(status, StatusDisplayStrings.get('youtube_duplicate'))
# The "encode status" should be converted to `file_complete` if video encodes are complete
video['status'] = 'transcription_in_progress'
status = convert_video_status(video, is_video_encodes_ready=True)
self.assertEqual(status, StatusDisplayStrings.get('file_complete'))
# If encoding is not complete return the status as it is
video['status'] = 's3_upload_failed'
status = convert_video_status(video)
self.assertEqual(status, StatusDisplayStrings.get('s3_upload_failed'))
# for all other status, there should not be any conversion
statuses = list(StatusDisplayStrings._STATUS_MAP.keys()) # pylint: disable=protected-access
statuses.remove('invalid_token')
for status in statuses:
video['status'] = status
new_status = convert_video_status(video)
self.assertEqual(new_status, StatusDisplayStrings.get(status))
def assert_video_status(self, url, edx_video_id, status):
"""
Verifies that video with `edx_video_id` has `status`
"""
response = self.client.get_json(url)
self.assertEqual(response.status_code, 200)
videos = json.loads(response.content.decode('utf-8'))["videos"]
for video in videos:
if video['edx_video_id'] == edx_video_id:
return self.assertEqual(video['status'], status)
# Test should fail if video not found
self.assertEqual(True, False, 'Invalid edx_video_id')
@patch('cms.djangoapps.contentstore.video_storage_handlers.LOGGER')
def test_video_status_update_request(self, mock_logger):
"""
Verifies that video status update request works as expected.
"""
url = self.get_url_for_course_key(self.course.id)
edx_video_id = 'test1'
self.assert_video_status(url, edx_video_id, 'Uploading')
response = self.client.post(
url,
json.dumps([{
'edxVideoId': edx_video_id,
'status': 'upload_failed',
'message': 'server down'
}]),
content_type="application/json"
)
mock_logger.info.assert_called_with(
'VIDEOS: Video status update with id [%s], status [%s] and message [%s]',
edx_video_id,
'upload_failed',
'server down'
)
self.assertEqual(response.status_code, 204)
self.assert_video_status(url, edx_video_id, 'Failed')
@ddt.data(
('test_video_token', "Transcription in Progress"),
('', "Ready"),
)
@ddt.unpack
def test_video_transcript_status_conversion(self, course_video_upload_token, expected_video_status_text):
"""
Verifies that video status `transcription_in_progress` gets converted
correctly into the `file_complete` for the new video workflow and
stays as it is, for the old video workflow.
"""
self.course.video_upload_pipeline = {
'course_video_upload_token': course_video_upload_token
}
self.save_course()
url = self.get_url_for_course_key(self.course.id)
edx_video_id = 'test1'
self.assert_video_status(url, edx_video_id, 'Uploading')
response = self.client.post(
url,
json.dumps([{
'edxVideoId': edx_video_id,
'status': 'transcription_in_progress',
'message': 'Transcription is in progress'
}]),
content_type="application/json"
)
self.assertEqual(response.status_code, 204)
self.assert_video_status(url, edx_video_id, expected_video_status_text)
@ddt.data(True, False)
@patch('openedx.core.djangoapps.video_config.models.VideoTranscriptEnabledFlag.feature_enabled')
def test_video_index_transcript_feature_enablement(self, is_video_transcript_enabled, video_transcript_feature):
"""
Test that when video transcript is enabled/disabled, correct response is rendered.
"""
video_transcript_feature.return_value = is_video_transcript_enabled
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
# Verify that course video button is present in the response if videos transcript feature is enabled.
button_html = '<button class="button course-video-settings-button">'
if is_video_transcript_enabled:
self.assertContains(response, button_html)
else:
self.assertNotContains(response, button_html)
@ddt.ddt
@patch.dict("django.conf.settings.FEATURES", {"ENABLE_VIDEO_UPLOAD_PIPELINE": True})
@override_settings(VIDEO_UPLOAD_PIPELINE={
"VEM_S3_BUCKET": "vem_test_bucket", "BUCKET": "test_bucket", "ROOT_PATH": "test_root"
})
class GenerateVideoUploadLinkTestCase(
VideoUploadTestBase,
VideoUploadPostTestsMixin,
CourseTestCase
):
"""
Test cases for the main video upload endpoint
"""
VIEW_NAME = 'generate_video_upload_link'
def test_unsupported_requests_fail(self):
"""
The API only supports post, make sure other requests fail
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 405)
response = self.client.put(self.url)
self.assertEqual(response.status_code, 405)
response = self.client.patch(self.url)
self.assertEqual(response.status_code, 405)
@ddt.ddt
@patch.dict('django.conf.settings.FEATURES', {'ENABLE_VIDEO_UPLOAD_PIPELINE': True})
@override_settings(VIDEO_UPLOAD_PIPELINE={'BUCKET': 'test_bucket', 'ROOT_PATH': 'test_root'})
class VideoImageTestCase(VideoUploadTestBase, CourseTestCase):
"""
Tests for video image.
"""
VIEW_NAME = "video_images_handler"
def verify_image_upload_reponse(self, course_id, edx_video_id, upload_response):
"""
Verify that image is uploaded successfully.
Arguments:
course_id: ID of course
edx_video_id: ID of video
upload_response: Upload response object
Returns:
uploaded image url
"""
self.assertEqual(upload_response.status_code, 200)
response = json.loads(upload_response.content.decode('utf-8'))
val_image_url = get_course_video_image_url(course_id=course_id, edx_video_id=edx_video_id)
self.assertEqual(response['image_url'], val_image_url)
return val_image_url
def verify_error_message(self, response, error_message):
"""
Verify that image upload failure gets proper error message.
Arguments:
response: Response object.
error_message: Expected error message.
"""
self.assertEqual(response.status_code, 400)
response = json.loads(response.content.decode('utf-8'))
self.assertIn('error', response)
self.assertEqual(response['error'], error_message)
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, False)
def test_video_image_upload_disabled(self):
"""
Tests the video image upload when the feature is disabled.
"""
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': 'test_vid_id'})
response = self.client.post(video_image_upload_url, {'file': 'dummy_file'}, format='multipart')
self.assertEqual(response.status_code, 404)
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, True)
def test_video_image(self):
"""
Test video image is saved.
"""
edx_video_id = 'test1'
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
with make_image_file(
dimensions=(settings.VIDEO_IMAGE_MIN_WIDTH, settings.VIDEO_IMAGE_MIN_HEIGHT),
) as image_file:
response = self.client.post(video_image_upload_url, {'file': image_file}, format='multipart')
image_url1 = self.verify_image_upload_reponse(self.course.id, edx_video_id, response)
# upload again to verify that new image is uploaded successfully
with make_image_file(
dimensions=(settings.VIDEO_IMAGE_MIN_WIDTH, settings.VIDEO_IMAGE_MIN_HEIGHT),
) as image_file:
response = self.client.post(video_image_upload_url, {'file': image_file}, format='multipart')
image_url2 = self.verify_image_upload_reponse(self.course.id, edx_video_id, response)
self.assertNotEqual(image_url1, image_url2)
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, True)
def test_video_image_no_file(self):
"""
Test that an error error message is returned if upload request is incorrect.
"""
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': 'test1'})
response = self.client.post(video_image_upload_url, {})
self.verify_error_message(response, 'An image file is required.')
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, True)
def test_no_video_image(self):
"""
Test image url is set to None if no video image.
"""
edx_video_id = 'test1'
get_videos_url = reverse_course_url('videos_handler', self.course.id)
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
with make_image_file(
dimensions=(settings.VIDEO_IMAGE_MIN_WIDTH, settings.VIDEO_IMAGE_MIN_HEIGHT),
) as image_file:
self.client.post(video_image_upload_url, {'file': image_file}, format='multipart')
val_image_url = get_course_video_image_url(course_id=self.course.id, edx_video_id=edx_video_id)
response = self.client.get_json(get_videos_url)
self.assertEqual(response.status_code, 200)
response_videos = json.loads(response.content.decode('utf-8'))["videos"]
for response_video in response_videos:
if response_video['edx_video_id'] == edx_video_id:
self.assertEqual(response_video['course_video_image_url'], val_image_url)
else:
self.assertEqual(response_video['course_video_image_url'], None)
@ddt.data(
# Image file type validation
(
{
'extension': '.png'
},
None
),
(
{
'extension': '.gif'
},
None
),
(
{
'extension': '.bmp'
},
None
),
(
{
'extension': '.jpg'
},
None
),
(
{
'extension': '.jpeg'
},
None
),
(
{
'extension': '.PNG'
},
None
),
(
{
'extension': '.tiff'
},
'This image file type is not supported. Supported file types are {supported_file_formats}.'.format(
supported_file_formats=list(settings.VIDEO_IMAGE_SUPPORTED_FILE_FORMATS.keys())
)
),
# Image file size validation
(
{
'size': settings.VIDEO_IMAGE_SETTINGS['VIDEO_IMAGE_MAX_BYTES'] + 10
},
'This image file must be smaller than {image_max_size}.'.format(
image_max_size=settings.VIDEO_IMAGE_MAX_FILE_SIZE_MB
)
),
(
{
'size': settings.VIDEO_IMAGE_SETTINGS['VIDEO_IMAGE_MIN_BYTES'] - 10
},
'This image file must be larger than {image_min_size}.'.format(
image_min_size=settings.VIDEO_IMAGE_MIN_FILE_SIZE_KB
)
),
# Image file minimum width / height
(
{
'width': 16, # 16x9
'height': 9
},
'Recommended image resolution is {image_file_max_width}x{image_file_max_height}. The minimum resolution is {image_file_min_width}x{image_file_min_height}.'.format( # lint-amnesty, pylint: disable=line-too-long
image_file_max_width=settings.VIDEO_IMAGE_MAX_WIDTH,
image_file_max_height=settings.VIDEO_IMAGE_MAX_HEIGHT,
image_file_min_width=settings.VIDEO_IMAGE_MIN_WIDTH,
image_file_min_height=settings.VIDEO_IMAGE_MIN_HEIGHT
)
),
(
{
'width': settings.VIDEO_IMAGE_MIN_WIDTH - 10,
'height': settings.VIDEO_IMAGE_MIN_HEIGHT
},
'Recommended image resolution is {image_file_max_width}x{image_file_max_height}. The minimum resolution is {image_file_min_width}x{image_file_min_height}.'.format( # lint-amnesty, pylint: disable=line-too-long
image_file_max_width=settings.VIDEO_IMAGE_MAX_WIDTH,
image_file_max_height=settings.VIDEO_IMAGE_MAX_HEIGHT,
image_file_min_width=settings.VIDEO_IMAGE_MIN_WIDTH,
image_file_min_height=settings.VIDEO_IMAGE_MIN_HEIGHT
)
),
(
{
'width': settings.VIDEO_IMAGE_MIN_WIDTH,
'height': settings.VIDEO_IMAGE_MIN_HEIGHT - 10
},
(
'Recommended image resolution is {image_file_max_width}x{image_file_max_height}. '
'The minimum resolution is {image_file_min_width}x{image_file_min_height}.'
).format(
image_file_max_width=settings.VIDEO_IMAGE_MAX_WIDTH,
image_file_max_height=settings.VIDEO_IMAGE_MAX_HEIGHT,
image_file_min_width=settings.VIDEO_IMAGE_MIN_WIDTH,
image_file_min_height=settings.VIDEO_IMAGE_MIN_HEIGHT
)
),
(
{
'width': 1200, # not 16:9, but width/height check first.
'height': 100
},
(
'Recommended image resolution is {image_file_max_width}x{image_file_max_height}. '
'The minimum resolution is {image_file_min_width}x{image_file_min_height}.'
).format(
image_file_max_width=settings.VIDEO_IMAGE_MAX_WIDTH,
image_file_max_height=settings.VIDEO_IMAGE_MAX_HEIGHT,
image_file_min_width=settings.VIDEO_IMAGE_MIN_WIDTH,
image_file_min_height=settings.VIDEO_IMAGE_MIN_HEIGHT
)
),
# Image file aspect ratio validation
(
{
'width': settings.VIDEO_IMAGE_MAX_WIDTH, # 1280x720
'height': settings.VIDEO_IMAGE_MAX_HEIGHT
},
None
),
(
{
'width': 850, # 16:9
'height': 478
},
None
),
(
{
'width': 940, # 1.67 ratio, applicable aspect ratio margin of .01
'height': 560
},
None
),
(
{
'width': settings.VIDEO_IMAGE_MIN_WIDTH + 100,
'height': settings.VIDEO_IMAGE_MIN_HEIGHT + 200
},
'This image file must have an aspect ratio of {video_image_aspect_ratio_text}.'.format(
video_image_aspect_ratio_text=settings.VIDEO_IMAGE_ASPECT_RATIO_TEXT
)
),
# Image file name validation
(
{
'prefix': 'nøn-åßç¡¡'
},
'The image file name can only contain letters, numbers, hyphens (-), and underscores (_).'
)
)
@ddt.unpack
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, True)
def test_video_image_validation_message(self, image_data, error_message):
"""
Test video image validation gives proper error message.
Arguments:
image_data (Dict): Specific data to create image file.
error_message (String): Error message
"""
edx_video_id = 'test1'
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
with make_image_file(
dimensions=(
image_data.get('width', settings.VIDEO_IMAGE_MIN_WIDTH),
image_data.get('height', settings.VIDEO_IMAGE_MIN_HEIGHT)
),
prefix=image_data.get('prefix', 'videoimage'),
extension=image_data.get('extension', '.png'),
force_size=image_data.get('size', settings.VIDEO_IMAGE_SETTINGS['VIDEO_IMAGE_MIN_BYTES'])
) as image_file:
response = self.client.post(video_image_upload_url, {'file': image_file}, format='multipart')
if error_message:
self.verify_error_message(response, error_message)
else:
self.verify_image_upload_reponse(self.course.id, edx_video_id, response)
@ddt.ddt
@patch(
'openedx.core.djangoapps.video_config.models.VideoTranscriptEnabledFlag.feature_enabled',
Mock(return_value=True)
)
@patch.dict('django.conf.settings.FEATURES', {'ENABLE_VIDEO_UPLOAD_PIPELINE': True})
class TranscriptPreferencesTestCase(VideoUploadTestBase, CourseTestCase):
"""
Tests for video transcripts preferences.
"""
VIEW_NAME = 'transcript_preferences_handler'
def test_405_with_not_allowed_request_method(self):
"""
Verify that 405 is returned in case of not-allowed request methods.
Allowed request methods are POST and DELETE.
"""
video_transcript_url = self.get_url_for_course_key(self.course.id)
response = self.client.get(
video_transcript_url,
content_type='application/json'
)
self.assertEqual(response.status_code, 405)
@ddt.data(
# Video transcript feature disabled
(
{},
False,
'',
404,
),
# Error cases
(
{},
True,
"Invalid provider None.",
400
),
(
{
'provider': ''
},
True,
"Invalid provider .",
400
),
(
{
'provider': 'dummy-provider'
},
True,
"Invalid provider dummy-provider.",
400
),
(
{
'provider': TranscriptProvider.CIELO24
},
True,
"Invalid cielo24 fidelity None.",
400
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
},
True,
"Invalid cielo24 turnaround None.",
400
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'video_source_language': 'en'
},
True,
"Invalid languages [].",
400
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PREMIUM',
'cielo24_turnaround': 'STANDARD',
'video_source_language': 'es'
},
True,
"Unsupported source language es.",
400
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'video_source_language': 'en',
'preferred_languages': ['es', 'ur']
},
True,
"Invalid languages ['es', 'ur'].",
400
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA
},
True,
"Invalid 3play turnaround None.",
400
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA,
'three_play_turnaround': 'standard',
'video_source_language': 'zh',
},
True,
"Unsupported source language zh.",
400
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA,
'three_play_turnaround': 'standard',
'video_source_language': 'es',
'preferred_languages': ['es', 'ur']
},
True,
"Invalid languages ['es', 'ur'].",
400
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA,
'three_play_turnaround': 'standard',
'video_source_language': 'en',
'preferred_languages': ['es', 'ur']
},
True,
"Invalid languages ['es', 'ur'].",
400
),
# Success
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'video_source_language': 'es',
'preferred_languages': ['en']
},
True,
'',
200
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA,
'three_play_turnaround': 'standard',
'preferred_languages': ['en'],
'video_source_language': 'en',
},
True,
'',
200
)
)
@ddt.unpack
def test_video_transcript(self, preferences, is_video_transcript_enabled, error_message, expected_status_code):
"""
Tests that transcript handler works correctly.
"""
video_transcript_url = self.get_url_for_course_key(self.course.id)
preferences_data = {
'provider': preferences.get('provider'),
'cielo24_fidelity': preferences.get('cielo24_fidelity'),
'cielo24_turnaround': preferences.get('cielo24_turnaround'),
'three_play_turnaround': preferences.get('three_play_turnaround'),
'preferred_languages': preferences.get('preferred_languages', []),
'video_source_language': preferences.get('video_source_language'),
}
with patch(
'openedx.core.djangoapps.video_config.models.VideoTranscriptEnabledFlag.feature_enabled'
) as video_transcript_feature:
video_transcript_feature.return_value = is_video_transcript_enabled
response = self.client.post(
video_transcript_url,
json.dumps(preferences_data),
content_type='application/json'
)
status_code = response.status_code
response = json.loads(response.content.decode('utf-8')) if is_video_transcript_enabled else response
self.assertEqual(status_code, expected_status_code)
self.assertEqual(response.get('error', ''), error_message)
# Remove modified and course_id fields from the response so as to check the expected transcript preferences.
response.get('transcript_preferences', {}).pop('modified', None)
response.get('transcript_preferences', {}).pop('course_id', None)
expected_preferences = preferences_data if is_video_transcript_enabled and not error_message else {}
self.assertDictEqual(response.get('transcript_preferences', {}), expected_preferences)
def test_remove_transcript_preferences(self):
"""
Test that transcript handler removes transcript preferences correctly.
"""
# First add course wide transcript preferences.
preferences = create_or_update_transcript_preferences(str(self.course.id))
# Verify transcript preferences exist
self.assertIsNotNone(preferences)
response = self.client.delete(
self.get_url_for_course_key(self.course.id),
content_type='application/json'
)
self.assertEqual(response.status_code, 204)
# Verify transcript preferences no loger exist
preferences = get_transcript_preferences(str(self.course.id))
self.assertIsNone(preferences)
def test_remove_transcript_preferences_not_found(self):
"""
Test that transcript handler works correctly even when no preferences are found.
"""
course_id = 'course-v1:dummy+course+id'
# Verify transcript preferences do not exist
preferences = get_transcript_preferences(course_id)
self.assertIsNone(preferences)
response = self.client.delete(
self.get_url_for_course_key(course_id),
content_type='application/json'
)
self.assertEqual(response.status_code, 204)
# Verify transcript preferences do not exist
preferences = get_transcript_preferences(course_id)
self.assertIsNone(preferences)
@ddt.data(
(
None,
False
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'preferred_languages': ['en']
},
False
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'preferred_languages': ['en']
},
True
)
)
@ddt.unpack
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@patch('boto.s3.key.Key')
@patch('cms.djangoapps.contentstore.video_storage_handlers.S3Connection')
@patch('cms.djangoapps.contentstore.video_storage_handlers.get_transcript_preferences')
def test_transcript_preferences_metadata(self, transcript_preferences, is_video_transcript_enabled,
mock_transcript_preferences, mock_conn, mock_key):
"""
Tests that transcript preference metadata is only set if it is video transcript feature is enabled and
transcript preferences are already stored in the system.
"""
file_name = 'test-video.mp4'
request_data = {'files': [{'file_name': file_name, 'content_type': 'video/mp4'}]}
mock_transcript_preferences.return_value = transcript_preferences
bucket = Mock()
mock_conn.return_value = Mock(get_bucket=Mock(return_value=bucket))
mock_key_instance = Mock(
generate_url=Mock(
return_value=f'http://example.com/url_{file_name}'
)
)
# If extra calls are made, return a dummy
mock_key.side_effect = [mock_key_instance] + [Mock()]
videos_handler_url = reverse_course_url('videos_handler', self.course.id)
with patch(
'openedx.core.djangoapps.video_config.models.VideoTranscriptEnabledFlag.feature_enabled'
) as video_transcript_feature:
video_transcript_feature.return_value = is_video_transcript_enabled
response = self.client.post(videos_handler_url, json.dumps(request_data), content_type='application/json')
self.assertEqual(response.status_code, 200)
# Ensure `transcript_preferences` was set up in Key correctly if sent through request.
if is_video_transcript_enabled and transcript_preferences:
mock_key_instance.set_metadata.assert_any_call('transcript_preferences', json.dumps(transcript_preferences))
else:
with self.assertRaises(AssertionError):
mock_key_instance.set_metadata.assert_any_call(
'transcript_preferences', json.dumps(transcript_preferences)
)
@patch.dict("django.conf.settings.FEATURES", {"ENABLE_VIDEO_UPLOAD_PIPELINE": True})
@override_settings(VIDEO_UPLOAD_PIPELINE={"BUCKET": "test_bucket", "ROOT_PATH": "test_root"})
class VideoUrlsCsvTestCase(
VideoUploadTestBase,
VideoStudioAccessTestsMixin,
VideoPipelineStudioAccessTestsMixin,
CourseTestCase
):
"""Test cases for the CSV download endpoint for video uploads"""
VIEW_NAME = "video_encodings_download"
def setUp(self):
super().setUp()
VideoUploadConfig(profile_whitelist="profile1").save()
def _check_csv_response(self, expected_profiles):
"""
Check that the response is a valid CSV response containing rows
corresponding to previous_uploads and including the expected profiles.
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertEqual(
response["Content-Disposition"],
f"attachment; filename=\"{self.course.id.course}_video_urls.csv\""
)
response_content = b"".join(response.streaming_content)
response_reader = StringIO(response_content.decode())
reader = csv.DictReader(response_reader, dialect=csv.excel)
self.assertEqual(
reader.fieldnames,
(
["Name", "Duration", "Date Added", "Video ID", "Status"] +
[f"{profile} URL" for profile in expected_profiles]
)
)
rows = list(reader)
self.assertEqual(len(rows), len(self.previous_uploads))
for i, row in enumerate(rows):
response_video = dict(row.items())
# Videos should be returned by creation date descending
original_video = self.previous_uploads[-(i + 1)]
client_video_id = original_video["client_video_id"]
self.assertEqual(response_video["Name"], client_video_id)
self.assertEqual(response_video["Duration"], str(original_video["duration"]))
dateutil.parser.parse(response_video["Date Added"])
self.assertEqual(response_video["Video ID"], original_video["edx_video_id"])
self.assertEqual(response_video["Status"], convert_video_status(original_video))
for profile in expected_profiles:
response_profile_url = response_video[f"{profile} URL"]
original_encoded_for_profile = next(
(
original_encoded
for original_encoded in original_video["encoded_videos"]
if original_encoded["profile"] == profile
),
None
)
if original_encoded_for_profile:
original_encoded_for_profile_url = original_encoded_for_profile["url"]
self.assertEqual(response_profile_url, original_encoded_for_profile_url)
else:
self.assertEqual(response_profile_url, "")
def test_basic(self):
self._check_csv_response(["profile1"])
def test_profile_whitelist(self):
VideoUploadConfig(profile_whitelist="profile1,profile2").save()
self._check_csv_response(["profile1", "profile2"])
def test_non_ascii_course(self):
course = CourseFactory.create(
number="nón-äscii",
video_upload_pipeline={
"course_video_upload_token": self.test_token,
}
)
response = self.client.get(self.get_url_for_course_key(course.id))
self.assertEqual(response.status_code, 200)
self.assertEqual(
response["Content-Disposition"],
"attachment; filename*=utf-8''n%C3%B3n-%C3%A4scii_video_urls.csv"
)
@ddt.ddt
class GetVideoFeaturesTestCase(
CourseTestCase
):
"""Test cases for the get_video_features endpoint """
def setUp(self):
super().setUp()
self.url = self.get_url_for_course_key()
def get_url_for_course_key(self):
""" Helper to generate a url for a course key """
return reverse("video_features")
def test_basic(self):
""" Test for expected return keys """
response = self.client.get(self.get_url_for_course_key())
self.assertEqual(response.status_code, 200)
self.assertEqual(
set(response.json().keys()),
{
'videoSharingEnabled',
'allowThumbnailUpload',
}
)
@ddt.data(True, False)
def test_video_share_enabled(self, is_enabled):
""" Test the public video share flag """
self._test_video_feature(
PUBLIC_VIDEO_SHARE,
'videoSharingEnabled',
override_waffle_flag,
is_enabled,
)
@ddt.data(True, False)
def test_video_image_upload_enabled(self, is_enabled):
""" Test the video image upload switch """
self._test_video_feature(
VIDEO_IMAGE_UPLOAD_ENABLED,
'allowThumbnailUpload',
override_waffle_switch,
is_enabled,
)
def _test_video_feature(self, flag, key, override_fn, is_enabled):
""" Test that setting a waffle flag or switch on or off will cause the expected result """
with override_fn(flag, is_enabled):
response = self.client.get(self.get_url_for_course_key())
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()[key], is_enabled)
class GetStorageBucketTestCase(TestCase):
""" This test just check that connection works and returns the bucket.
It does not involve any mocking and triggers errors if has any import issue.
"""
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@override_settings(VIDEO_UPLOAD_PIPELINE={
"VEM_S3_BUCKET": "vem_test_bucket", "BUCKET": "test_bucket", "ROOT_PATH": "test_root"
})
def test_storage_bucket(self):
""" get bucket and generate url. It will not hit actual s3."""
bucket = storage_service_bucket()
edx_video_id = 'dummy_video'
key = storage_service_key(bucket, file_name=edx_video_id)
upload_url = key.generate_url(
KEY_EXPIRATION_IN_SECONDS,
'PUT',
headers={'Content-Type': 'mp4'}
)
self.assertIn("https://vem_test_bucket.s3.amazonaws.com:443/test_root/", upload_url)
self.assertIn(edx_video_id, upload_url)
class CourseYoutubeEdxVideoIds(ModuleStoreTestCase):
"""
This test checks youtube videos in a course
"""
VIEW_NAME = 'youtube_edx_video_ids'
def setUp(self):
super().setUp()
self.course = CourseFactory.create()
self.course_with_no_youtube_videos = CourseFactory.create()
self.store = modulestore()
self.user = UserFactory()
self.client.login(username=self.user.username, password='Password1234')
def get_url_for_course_key(self, course_key, kwargs=None):
"""Return video handler URL for the given course"""
return reverse_course_url(self.VIEW_NAME, course_key, kwargs) # lint-amnesty, pylint: disable=no-member
def test_course_with_youtube_videos(self):
course_key = self.course.id
with self.store.bulk_operations(course_key):
chapter_loc = self.store.create_child(
self.user.id, self.course.location, 'chapter', 'test_chapter'
).location
seq_loc = self.store.create_child(
self.user.id, chapter_loc, 'sequential', 'test_seq'
).location
vert_loc = self.store.create_child(self.user.id, seq_loc, 'vertical', 'test_vert').location
self.store.create_child(
self.user.id,
vert_loc,
'problem',
'test_problem',
fields={"data": "<problem>Test</problem>"}
)
self.store.create_child(
self.user.id, vert_loc, 'video', fields={
"youtube_is_available": False,
"name": "sample_video",
"edx_video_id": "youtube_193_84709099",
}
)
response = self.client.get(self.get_url_for_course_key(course_key))
self.assertEqual(response.status_code, 200)
edx_video_ids = json.loads(response.content.decode('utf-8'))['edx_video_ids']
self.assertEqual(len(edx_video_ids), 1)
def test_course_with_no_youtube_videos(self):
course_key = self.course_with_no_youtube_videos.id
with self.store.bulk_operations(course_key):
chapter_loc = self.store.create_child(
self.user.id, self.course_with_no_youtube_videos.location, 'chapter', 'test_chapter'
).location
seq_loc = self.store.create_child(
self.user.id, chapter_loc, 'sequential', 'test_seq'
).location
vert_loc = self.store.create_child(self.user.id, seq_loc, 'vertical', 'test_vert').location
self.store.create_child(
self.user.id, vert_loc, 'problem', 'test_problem', fields={"data": "<problem>Test</problem>"}
)
self.store.create_child(
self.user.id, vert_loc, 'video', fields={
"youtube_id_1_0": None,
"name": "sample_video",
"edx_video_id": "no_youtube_193_84709099",
}
)
response = self.client.get(self.get_url_for_course_key(course_key))
edx_video_ids = json.loads(response.content.decode('utf-8'))['edx_video_ids']
self.assertEqual(response.status_code, 200)
self.assertEqual(len(edx_video_ids), 0)
"""
Unit tests for video-related REST APIs.
"""
import csv
import json
import re
from contextlib import contextmanager
from datetime import datetime
from io import StringIO
from unittest.mock import Mock, patch
import dateutil.parser
from common.djangoapps.student.tests.factories import UserFactory
import ddt
import pytz
from django.test import TestCase
from django.conf import settings
from django.test.utils import override_settings
from django.urls import reverse
from edx_toggles.toggles.testutils import override_waffle_flag, override_waffle_switch
from edxval.api import (
create_or_update_transcript_preferences,
create_or_update_video_transcript,
create_profile,
create_video,
get_course_video_image_url,
get_transcript_preferences,
get_video_info
)
from cms.djangoapps.contentstore.models import VideoUploadConfig
from cms.djangoapps.contentstore.tests.utils import CourseTestCase
from cms.djangoapps.contentstore.utils import reverse_course_url
from openedx.core.djangoapps.profile_images.tests.helpers import make_image_file
from openedx.core.djangoapps.video_pipeline.config.waffle import (
DEPRECATE_YOUTUBE,
ENABLE_DEVSTACK_VIDEO_UPLOADS,
)
from openedx.core.djangoapps.waffle_utils.models import WaffleFlagCourseOverrideModel
from xmodule.modulestore.django import modulestore
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory # lint-amnesty, pylint: disable=wrong-import-order
from ..videos import (
ENABLE_VIDEO_UPLOAD_PAGINATION,
KEY_EXPIRATION_IN_SECONDS,
VIDEO_IMAGE_UPLOAD_ENABLED,
)
from cms.djangoapps.contentstore.video_storage_handlers import (
_get_default_video_image_url,
TranscriptProvider,
StatusDisplayStrings,
convert_video_status,
storage_service_bucket,
storage_service_key,
PUBLIC_VIDEO_SHARE
)
def setup_s3_mocks(mock_boto3_resource, bucket_name='test-bucket'):
"""
Helper function to set up consistent boto3 S3 mocks.
Args:
mock_boto3_resource: The patched boto3.resource mock
bucket_name: Name for the mock bucket (default: 'test-bucket')
Returns:
tuple: (mock_s3_client, mock_bucket, mock_s3_resource)
"""
mock_s3_client = Mock()
mock_bucket = Mock()
mock_bucket.name = bucket_name
mock_bucket.meta.client = mock_s3_client
mock_s3_resource = Mock()
mock_s3_resource.Bucket.return_value = mock_bucket
mock_boto3_resource.return_value = mock_s3_resource
return mock_s3_client, mock_bucket, mock_s3_resource
class VideoUploadTestBase:
"""
Test cases for the video upload feature
"""
def get_url_for_course_key(self, course_key, kwargs=None):
"""Return video handler URL for the given course"""
return reverse_course_url(self.VIEW_NAME, course_key, kwargs) # lint-amnesty, pylint: disable=no-member
def setUp(self):
super().setUp() # lint-amnesty, pylint: disable=no-member
self.url = self.get_url_for_course_key(self.course.id)
self.test_token = "test_token"
self.course.video_upload_pipeline = {
"course_video_upload_token": self.test_token,
}
self.save_course() # lint-amnesty, pylint: disable=no-member
# create another course for videos belonging to multiple courses
self.course2 = CourseFactory.create()
self.course2.video_upload_pipeline = {
"course_video_upload_token": self.test_token,
}
self.course2.save()
self.store.update_item(self.course2, self.user.id) # lint-amnesty, pylint: disable=no-member
# course ids for videos
course_ids = [str(self.course.id), str(self.course2.id)]
created = datetime.now(pytz.utc)
self.profiles = ["profile1", "profile2"]
self.previous_uploads = [
{
"edx_video_id": "test1",
"client_video_id": "test1.mp4",
"duration": 42.0,
"status": "upload",
"courses": course_ids,
"encoded_videos": [],
"created": created
},
{
"edx_video_id": "test2",
"client_video_id": "test2.mp4",
"duration": 128.0,
"status": "file_complete",
"courses": course_ids,
"created": created,
"encoded_videos": [
{
"profile": "profile1",
"url": "http://example.com/profile1/test2.mp4",
"file_size": 1600,
"bitrate": 100,
},
{
"profile": "profile2",
"url": "http://example.com/profile2/test2.mov",
"file_size": 16000,
"bitrate": 1000,
},
],
},
{
"edx_video_id": "non-ascii",
"client_video_id": "nón-ascii-näme.mp4",
"duration": 256.0,
"status": "transcode_active",
"courses": course_ids,
"created": created,
"encoded_videos": [
{
"profile": "profile1",
"url": "http://example.com/profile1/nón-ascii-näme.mp4",
"file_size": 3200,
"bitrate": 100,
},
]
},
]
# Ensure every status string is tested
self.previous_uploads += [
{
"edx_video_id": f"status_test_{status}",
"client_video_id": "status_test.mp4",
"duration": 3.14,
"status": status,
"courses": course_ids,
"created": created,
"encoded_videos": [],
}
for status in (
list(StatusDisplayStrings._STATUS_MAP.keys()) + # pylint:disable=protected-access
["non_existent_status"]
)
]
for profile in self.profiles:
create_profile(profile)
for video in self.previous_uploads:
create_video(video)
def _get_previous_upload(self, edx_video_id):
"""Returns the previous upload with the given video id."""
return next(
video
for video in self.previous_uploads
if video["edx_video_id"] == edx_video_id
)
class VideoStudioAccessTestsMixin:
"""
Base Access tests for studio video views
"""
def test_anon_user(self):
self.client.logout()
response = self.client.get(self.url)
self.assertEqual(response.status_code, 302)
def test_put(self):
response = self.client.put(self.url)
self.assertEqual(response.status_code, 405)
def test_invalid_course_key(self):
response = self.client.get(
self.get_url_for_course_key("Non/Existent/Course")
)
self.assertEqual(response.status_code, 404)
def test_non_staff_user(self):
client, __ = self.create_non_staff_authed_user_client()
response = client.get(self.url)
self.assertEqual(response.status_code, 403)
class VideoPipelineStudioAccessTestsMixin:
"""
Access tests for video views that rely on the video pipeline
"""
def test_video_pipeline_not_enabled(self):
settings.FEATURES["ENABLE_VIDEO_UPLOAD_PIPELINE"] = False
self.assertEqual(self.client.get(self.url).status_code, 404)
def test_video_pipeline_not_configured(self):
settings.VIDEO_UPLOAD_PIPELINE = None
self.assertEqual(self.client.get(self.url).status_code, 404)
def test_course_not_configured(self):
self.course.video_upload_pipeline = {}
self.save_course()
self.assertEqual(self.client.get(self.url).status_code, 404)
class VideoUploadPostTestsMixin:
"""
Shared test cases for video post tests.
"""
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@patch('cms.djangoapps.contentstore.video_storage_handlers.boto3.resource')
def test_post_success(self, mock_boto3_resource):
files = [
{
'file_name': 'first.mp4',
'content_type': 'video/mp4',
},
{
'file_name': 'second.mp4',
'content_type': 'video/mp4',
},
{
'file_name': 'third.mov',
'content_type': 'video/quicktime',
},
{
'file_name': 'fourth.mp4',
'content_type': 'video/mp4',
},
]
# Setup boto3 mocks
mock_s3_client, mock_bucket, mock_s3_resource = setup_s3_mocks(mock_boto3_resource)
# Mock generate_presigned_url to return different URLs for each file
def mock_generate_presigned_url(operation, Params=None, ExpiresIn=None):
file_name = Params['Metadata']['client_video_id']
return f'http://example.com/url_{file_name}'
mock_s3_client.generate_presigned_url.side_effect = mock_generate_presigned_url
response = self.client.post(
self.url,
json.dumps({'files': files}),
content_type='application/json'
)
self.assertEqual(response.status_code, 200)
response_obj = json.loads(response.content.decode('utf-8'))
# Verify boto3 resource was called correctly
mock_boto3_resource.assert_called_once_with(
"s3",
aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY
)
self.assertEqual(len(response_obj['files']), len(files))
self.assertEqual(mock_s3_client.generate_presigned_url.call_count, len(files))
for i, file_info in enumerate(files):
# Get the call args for this file's presigned URL generation
call_args = mock_s3_client.generate_presigned_url.call_args_list[i]
args, kwargs = call_args
# Verify the operation and params
self.assertEqual(args[0], 'put_object')
self.assertEqual(kwargs['Params']['Bucket'], 'test-bucket')
self.assertEqual(kwargs['Params']['ContentType'], file_info['content_type'])
self.assertEqual(kwargs['ExpiresIn'], KEY_EXPIRATION_IN_SECONDS)
# Extract video_id from the key
key_name = kwargs['Params']['Key']
path_match = re.match(
(
settings.VIDEO_UPLOAD_PIPELINE['ROOT_PATH'] +
'/([a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12})$'
),
key_name
)
self.assertIsNotNone(path_match)
video_id = path_match.group(1)
# Verify metadata
metadata = kwargs['Params']['Metadata']
self.assertEqual(metadata['course_video_upload_token'], self.test_token)
self.assertEqual(metadata['client_video_id'], file_info['file_name'])
self.assertEqual(metadata['course_key'], str(self.course.id))
# Ensure VAL was updated
val_info = get_video_info(video_id)
self.assertEqual(val_info['status'], 'upload')
self.assertEqual(val_info['client_video_id'], file_info['file_name'])
self.assertEqual(val_info['duration'], 0)
self.assertEqual(val_info['courses'], [{str(self.course.id): None}])
# Ensure response is correct
response_file = response_obj['files'][i]
self.assertEqual(response_file['file_name'], file_info['file_name'])
self.assertEqual(response_file['upload_url'], f'http://example.com/url_{file_info["file_name"]}')
def test_post_non_json(self):
response = self.client.post(self.url, {"files": []})
self.assertEqual(response.status_code, 400)
def test_post_malformed_json(self):
response = self.client.post(self.url, "{", content_type="application/json")
self.assertEqual(response.status_code, 400)
def test_post_invalid_json(self):
def assert_bad(content):
"""Make request with content and assert that response is 400"""
response = self.client.post(
self.url,
json.dumps(content),
content_type="application/json"
)
self.assertEqual(response.status_code, 400)
# Top level missing files key
assert_bad({})
# Entry missing file_name
assert_bad({"files": [{"content_type": "video/mp4"}]})
# Entry missing content_type
assert_bad({"files": [{"file_name": "test.mp4"}]})
@ddt.ddt
@patch.dict("django.conf.settings.FEATURES", {"ENABLE_VIDEO_UPLOAD_PIPELINE": True})
@override_settings(VIDEO_UPLOAD_PIPELINE={
"VEM_S3_BUCKET": "vem_test_bucket", "BUCKET": "test_bucket", "ROOT_PATH": "test_root"
})
class VideosHandlerTestCase(
VideoUploadTestBase,
VideoStudioAccessTestsMixin,
VideoPipelineStudioAccessTestsMixin,
VideoUploadPostTestsMixin,
CourseTestCase
):
"""Test cases for the main video upload endpoint"""
VIEW_NAME = 'videos_handler'
def test_get_json(self):
response = self.client.get_json(self.url)
self.assertEqual(response.status_code, 200)
response_videos = json.loads(response.content.decode('utf-8'))['videos']
self.assertEqual(len(response_videos), len(self.previous_uploads))
for i, response_video in enumerate(response_videos):
# Videos should be returned by creation date descending
original_video = self.previous_uploads[-(i + 1)]
print(response_video.keys())
self.assertEqual(
set(response_video.keys()),
{
'edx_video_id',
'client_video_id',
'created',
'duration',
'status',
'status_nontranslated',
'course_video_image_url',
'file_size',
'download_link',
'transcripts',
'transcription_status',
'transcript_urls',
'error_description',
}
)
dateutil.parser.parse(response_video['created'])
for field in ['edx_video_id', 'client_video_id', 'duration']:
self.assertEqual(response_video[field], original_video[field])
self.assertEqual(
response_video['status'],
convert_video_status(original_video)
)
@ddt.data(
(
[
'edx_video_id', 'client_video_id', 'created', 'duration',
'status', 'status_nontranslated', 'course_video_image_url', 'file_size',
'download_link', 'transcripts', 'transcription_status', 'transcript_urls',
'error_description'
],
[
{
'video_id': 'test1',
'language_code': 'en',
'file_name': 'edx101.srt',
'file_format': 'srt',
'provider': 'Cielo24'
}
],
['en']
),
(
[
'edx_video_id', 'client_video_id', 'created', 'duration',
'status', 'status_nontranslated', 'course_video_image_url', 'file_size',
'download_link', 'transcripts', 'transcription_status', 'transcript_urls',
'error_description'
],
[
{
'video_id': 'test1',
'language_code': 'en',
'file_name': 'edx101_en.srt',
'file_format': 'srt',
'provider': 'Cielo24'
},
{
'video_id': 'test1',
'language_code': 'es',
'file_name': 'edx101_es.srt',
'file_format': 'srt',
'provider': 'Cielo24'
}
],
['en', 'es']
)
)
@ddt.unpack
def test_get_json_transcripts(self, expected_video_keys, uploaded_transcripts, expected_transcripts):
"""
Test that transcripts are attached based on whether the video transcript feature is enabled.
"""
for transcript in uploaded_transcripts:
create_or_update_video_transcript(
transcript['video_id'],
transcript['language_code'],
metadata={
'file_name': transcript['file_name'],
'file_format': transcript['file_format'],
'provider': transcript['provider']
}
)
response = self.client.get_json(self.url)
self.assertEqual(response.status_code, 200)
response_videos = json.loads(response.content.decode('utf-8'))['videos']
self.assertEqual(len(response_videos), len(self.previous_uploads))
for response_video in response_videos:
print(response_video)
self.assertEqual(set(response_video.keys()), set(expected_video_keys))
if response_video['edx_video_id'] == self.previous_uploads[0]['edx_video_id']:
self.assertEqual(response_video.get('transcripts', []), expected_transcripts)
def test_get_html(self):
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertRegex(response["Content-Type"], "^text/html(;.*)?$")
self.assertContains(response, _get_default_video_image_url())
# Crude check for presence of data in returned HTML
for video in self.previous_uploads:
self.assertContains(response, video["edx_video_id"])
self.assertNotContains(response, 'video_upload_pagination')
@override_waffle_flag(ENABLE_VIDEO_UPLOAD_PAGINATION, active=True)
def test_get_html_paginated(self):
"""
Tests that response is paginated.
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'video_upload_pagination')
@override_settings(AWS_ACCESS_KEY_ID="test_key_id", AWS_SECRET_ACCESS_KEY="test_secret")
@patch("cms.djangoapps.contentstore.video_storage_handlers.boto3.resource")
@ddt.data(
(
[
{
"file_name": "supported-1.mp4",
"content_type": "video/mp4",
},
{
"file_name": "supported-2.mov",
"content_type": "video/quicktime",
},
],
200
),
(
[
{
"file_name": "unsupported-1.txt",
"content_type": "text/plain",
},
{
"file_name": "unsupported-2.png",
"content_type": "image/png",
},
],
400
)
)
@ddt.unpack
def test_video_supported_file_formats(self, files, expected_status, mock_boto3_resource):
"""
Test that video upload works correctly against supported and unsupported file formats.
"""
# Setup boto3 mocks
mock_s3_client, mock_bucket, mock_s3_resource = setup_s3_mocks(mock_boto3_resource)
# Mock generate_presigned_url to return different URLs for each file
def mock_generate_presigned_url(operation, Params=None, ExpiresIn=None):
file_name = Params['Metadata']['client_video_id']
return f'http://example.com/url_{file_name}'
mock_s3_client.generate_presigned_url.side_effect = mock_generate_presigned_url
# Check supported formats
response = self.client.post(
self.url,
json.dumps({"files": files}),
content_type="application/json"
)
self.assertEqual(response.status_code, expected_status)
response = json.loads(response.content.decode('utf-8'))
if expected_status == 200:
self.assertNotIn('error', response)
else:
self.assertIn('error', response)
self.assertEqual(response['error'], "Request 'files' entry contain unsupported content_type")
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@patch('cms.djangoapps.contentstore.video_storage_handlers.boto3.resource')
def test_upload_with_non_ascii_charaters(self, mock_boto3_resource):
"""
Test that video uploads throws error message when file name contains special characters.
"""
# Setup boto3 mocks
mock_s3_client, mock_bucket, mock_s3_resource = setup_s3_mocks(mock_boto3_resource)
file_name = 'test\u2019_file.mp4'
files = [{'file_name': file_name, 'content_type': 'video/mp4'}]
response = self.client.post(
self.url,
json.dumps({'files': files}),
content_type='application/json'
)
self.assertEqual(response.status_code, 400)
response = json.loads(response.content.decode('utf-8'))
self.assertEqual(response['error'], 'The file name for %s must contain only ASCII characters.' % file_name)
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret', AWS_SECURITY_TOKEN='token')
@patch('cms.djangoapps.contentstore.video_storage_handlers.boto3.resource')
@override_waffle_flag(ENABLE_DEVSTACK_VIDEO_UPLOADS, active=True)
def test_devstack_upload_connection(self, mock_boto3_resource):
files = [{'file_name': 'first.mp4', 'content_type': 'video/mp4'}]
# Setup boto3 mocks
mock_s3_client, mock_bucket, mock_s3_resource = setup_s3_mocks(mock_boto3_resource)
# Mock generate_presigned_url to return different URLs for each file
def mock_generate_presigned_url(operation, Params=None, ExpiresIn=None):
file_name = Params['Metadata']['client_video_id']
return f'http://example.com/url_{file_name}'
mock_s3_client.generate_presigned_url.side_effect = mock_generate_presigned_url
response = self.client.post(
self.url,
json.dumps({'files': files}),
content_type='application/json'
)
self.assertEqual(response.status_code, 200)
mock_boto3_resource.assert_called_once_with(
"s3",
aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
security_token=settings.AWS_SECURITY_TOKEN
)
@patch('cms.djangoapps.contentstore.video_storage_handlers.boto3.resource')
def test_send_course_to_vem_pipeline(self, mock_boto3_resource):
"""
Test that uploads always go to VEM S3 bucket by default.
"""
files = [{'file_name': 'first.mp4', 'content_type': 'video/mp4'}]
# Setup boto3 mocks
mock_s3_client, mock_bucket, mock_s3_resource = setup_s3_mocks(mock_boto3_resource)
# Mock generate_presigned_url to return different URLs for each file
def mock_generate_presigned_url(operation, Params=None, ExpiresIn=None):
file_name = Params['Metadata']['client_video_id']
return f'http://example.com/url_{file_name}'
mock_s3_client.generate_presigned_url.side_effect = mock_generate_presigned_url
response = self.client.post(
self.url,
json.dumps({'files': files}),
content_type='application/json'
)
self.assertEqual(response.status_code, 200)
mock_s3_resource.Bucket.assert_called_once_with(
settings.VIDEO_UPLOAD_PIPELINE['VEM_S3_BUCKET'] # pylint: disable=unsubscriptable-object
)
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@patch('cms.djangoapps.contentstore.video_storage_handlers.boto3.resource')
@ddt.data(
{
'global_waffle': True,
'course_override': WaffleFlagCourseOverrideModel.ALL_CHOICES.off,
'expect_token': True
},
{
'global_waffle': False,
'course_override': WaffleFlagCourseOverrideModel.ALL_CHOICES.on,
'expect_token': False
},
{
'global_waffle': False,
'course_override': WaffleFlagCourseOverrideModel.ALL_CHOICES.off,
'expect_token': True
}
)
def test_video_upload_token_in_meta(self, data, mock_boto3_resource):
"""
Test video upload token in s3 metadata.
"""
@contextmanager
def proxy_manager(manager, ignore_manager):
"""
This acts as proxy to the original manager in the arguments given
the original manager is not set to be ignored.
"""
if ignore_manager:
yield
else:
with manager:
yield
file_data = {
'file_name': 'first.mp4',
'content_type': 'video/mp4',
}
# Setup boto3 mocks
mock_s3_client, mock_bucket, mock_s3_resource = setup_s3_mocks(mock_boto3_resource)
# Track generate_presigned_url calls to inspect metadata
presigned_url_calls = []
def mock_generate_presigned_url(operation, Params=None, ExpiresIn=None):
presigned_url_calls.append((operation, Params, ExpiresIn))
file_name = Params['Metadata']['client_video_id']
return f'http://example.com/url_{file_name}'
mock_s3_client.generate_presigned_url.side_effect = mock_generate_presigned_url
with patch.object(WaffleFlagCourseOverrideModel, 'override_value', return_value=data['course_override']):
with override_waffle_flag(DEPRECATE_YOUTUBE, active=data['global_waffle']):
response = self.client.post(
self.url,
json.dumps({'files': [file_data]}),
content_type='application/json'
)
self.assertEqual(response.status_code, 200)
# Check if course_video_upload_token is in metadata based on expectation
if data['expect_token']:
# We should find the token in the metadata
self.assertEqual(len(presigned_url_calls), 1)
metadata = presigned_url_calls[0][1]['Metadata']
self.assertIn('course_video_upload_token', metadata)
self.assertEqual(metadata['course_video_upload_token'], self.test_token)
else:
# If we don't expect a token, verify it's not in metadata
if presigned_url_calls:
metadata = presigned_url_calls[0][1]['Metadata']
self.assertNotIn('course_video_upload_token', metadata)
def _assert_video_removal(self, url, edx_video_id, deleted_videos):
"""
Verify that if correct video is removed from a particular course.
Arguments:
url (str): URL to get uploaded videos
edx_video_id (str): video id
deleted_videos (int): how many videos are deleted
"""
response = self.client.get_json(url)
self.assertEqual(response.status_code, 200)
response_videos = json.loads(response.content.decode('utf-8'))["videos"]
self.assertEqual(len(response_videos), len(self.previous_uploads) - deleted_videos)
if deleted_videos:
self.assertNotIn(edx_video_id, [video.get('edx_video_id') for video in response_videos])
else:
self.assertIn(edx_video_id, [video.get('edx_video_id') for video in response_videos])
def test_video_removal(self):
"""
Verifies that video removal is working as expected.
"""
edx_video_id = 'test1'
remove_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
response = self.client.delete(remove_url, HTTP_ACCEPT="application/json")
self.assertEqual(response.status_code, 204)
self._assert_video_removal(self.url, edx_video_id, 1)
def test_video_removal_multiple_courses(self):
"""
Verifies that video removal is working as expected for multiple courses.
If a video is used by multiple courses then removal from one course shouldn't effect the other course.
"""
# remove video from course1
edx_video_id = 'test1'
remove_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
response = self.client.delete(remove_url, HTTP_ACCEPT="application/json")
self.assertEqual(response.status_code, 204)
# verify that video is only deleted from course1 only
self._assert_video_removal(self.url, edx_video_id, 1)
self._assert_video_removal(self.get_url_for_course_key(self.course2.id), edx_video_id, 0)
def test_convert_video_status(self):
"""
Verifies that convert_video_status works as expected.
"""
video = self.previous_uploads[0]
# video status should be failed if it's in upload state for more than 24 hours
video['created'] = datetime(2016, 1, 1, 10, 10, 10, 0, pytz.UTC)
status = convert_video_status(video)
self.assertEqual(status, StatusDisplayStrings.get('upload_failed'))
# `invalid_token` should be converted to `youtube_duplicate`
video['created'] = datetime.now(pytz.UTC)
video['status'] = 'invalid_token'
status = convert_video_status(video)
self.assertEqual(status, StatusDisplayStrings.get('youtube_duplicate'))
# The "encode status" should be converted to `file_complete` if video encodes are complete
video['status'] = 'transcription_in_progress'
status = convert_video_status(video, is_video_encodes_ready=True)
self.assertEqual(status, StatusDisplayStrings.get('file_complete'))
# If encoding is not complete return the status as it is
video['status'] = 's3_upload_failed'
status = convert_video_status(video)
self.assertEqual(status, StatusDisplayStrings.get('s3_upload_failed'))
# for all other status, there should not be any conversion
statuses = list(StatusDisplayStrings._STATUS_MAP.keys()) # pylint: disable=protected-access
statuses.remove('invalid_token')
for status in statuses:
video['status'] = status
new_status = convert_video_status(video)
self.assertEqual(new_status, StatusDisplayStrings.get(status))
def assert_video_status(self, url, edx_video_id, status):
"""
Verifies that video with `edx_video_id` has `status`
"""
response = self.client.get_json(url)
self.assertEqual(response.status_code, 200)
videos = json.loads(response.content.decode('utf-8'))["videos"]
for video in videos:
if video['edx_video_id'] == edx_video_id:
return self.assertEqual(video['status'], status)
# Test should fail if video not found
self.assertEqual(True, False, 'Invalid edx_video_id')
@patch('cms.djangoapps.contentstore.video_storage_handlers.LOGGER')
def test_video_status_update_request(self, mock_logger):
"""
Verifies that video status update request works as expected.
"""
url = self.get_url_for_course_key(self.course.id)
edx_video_id = 'test1'
self.assert_video_status(url, edx_video_id, 'Uploading')
response = self.client.post(
url,
json.dumps([{
'edxVideoId': edx_video_id,
'status': 'upload_failed',
'message': 'server down'
}]),
content_type="application/json"
)
mock_logger.info.assert_called_with(
'VIDEOS: Video status update with id [%s], status [%s] and message [%s]',
edx_video_id,
'upload_failed',
'server down'
)
self.assertEqual(response.status_code, 204)
self.assert_video_status(url, edx_video_id, 'Failed')
@ddt.data(
('test_video_token', "Transcription in Progress"),
('', "Ready"),
)
@ddt.unpack
def test_video_transcript_status_conversion(self, course_video_upload_token, expected_video_status_text):
"""
Verifies that video status `transcription_in_progress` gets converted
correctly into the `file_complete` for the new video workflow and
stays as it is, for the old video workflow.
"""
self.course.video_upload_pipeline = {
'course_video_upload_token': course_video_upload_token
}
self.save_course()
url = self.get_url_for_course_key(self.course.id)
edx_video_id = 'test1'
self.assert_video_status(url, edx_video_id, 'Uploading')
response = self.client.post(
url,
json.dumps([{
'edxVideoId': edx_video_id,
'status': 'transcription_in_progress',
'message': 'Transcription is in progress'
}]),
content_type="application/json"
)
self.assertEqual(response.status_code, 204)
self.assert_video_status(url, edx_video_id, expected_video_status_text)
@ddt.data(True, False)
@patch('openedx.core.djangoapps.video_config.models.VideoTranscriptEnabledFlag.feature_enabled')
def test_video_index_transcript_feature_enablement(self, is_video_transcript_enabled, video_transcript_feature):
"""
Test that when video transcript is enabled/disabled, correct response is rendered.
"""
video_transcript_feature.return_value = is_video_transcript_enabled
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
# Verify that course video button is present in the response if videos transcript feature is enabled.
button_html = '<button class="button course-video-settings-button">'
if is_video_transcript_enabled:
self.assertContains(response, button_html)
else:
self.assertNotContains(response, button_html)
@ddt.ddt
@patch.dict("django.conf.settings.FEATURES", {"ENABLE_VIDEO_UPLOAD_PIPELINE": True})
@override_settings(VIDEO_UPLOAD_PIPELINE={
"VEM_S3_BUCKET": "vem_test_bucket", "BUCKET": "test_bucket", "ROOT_PATH": "test_root"
})
class GenerateVideoUploadLinkTestCase(
VideoUploadTestBase,
VideoUploadPostTestsMixin,
CourseTestCase
):
"""
Test cases for the main video upload endpoint
"""
VIEW_NAME = 'generate_video_upload_link'
def test_unsupported_requests_fail(self):
"""
The API only supports post, make sure other requests fail
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 405)
response = self.client.put(self.url)
self.assertEqual(response.status_code, 405)
response = self.client.patch(self.url)
self.assertEqual(response.status_code, 405)
@ddt.ddt
@patch.dict('django.conf.settings.FEATURES', {'ENABLE_VIDEO_UPLOAD_PIPELINE': True})
@override_settings(VIDEO_UPLOAD_PIPELINE={'BUCKET': 'test_bucket', 'ROOT_PATH': 'test_root'})
class VideoImageTestCase(VideoUploadTestBase, CourseTestCase):
"""
Tests for video image.
"""
VIEW_NAME = "video_images_handler"
def verify_image_upload_reponse(self, course_id, edx_video_id, upload_response):
"""
Verify that image is uploaded successfully.
Arguments:
course_id: ID of course
edx_video_id: ID of video
upload_response: Upload response object
Returns:
uploaded image url
"""
self.assertEqual(upload_response.status_code, 200)
response = json.loads(upload_response.content.decode('utf-8'))
val_image_url = get_course_video_image_url(course_id=course_id, edx_video_id=edx_video_id)
self.assertEqual(response['image_url'], val_image_url)
return val_image_url
def verify_error_message(self, response, error_message):
"""
Verify that image upload failure gets proper error message.
Arguments:
response: Response object.
error_message: Expected error message.
"""
self.assertEqual(response.status_code, 400)
response = json.loads(response.content.decode('utf-8'))
self.assertIn('error', response)
self.assertEqual(response['error'], error_message)
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, False)
def test_video_image_upload_disabled(self):
"""
Tests the video image upload when the feature is disabled.
"""
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': 'test_vid_id'})
response = self.client.post(video_image_upload_url, {'file': 'dummy_file'}, format='multipart')
self.assertEqual(response.status_code, 404)
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, True)
def test_video_image(self):
"""
Test video image is saved.
"""
edx_video_id = 'test1'
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
with make_image_file(
dimensions=(settings.VIDEO_IMAGE_MIN_WIDTH, settings.VIDEO_IMAGE_MIN_HEIGHT),
) as image_file:
response = self.client.post(video_image_upload_url, {'file': image_file}, format='multipart')
image_url1 = self.verify_image_upload_reponse(self.course.id, edx_video_id, response)
# upload again to verify that new image is uploaded successfully
with make_image_file(
dimensions=(settings.VIDEO_IMAGE_MIN_WIDTH, settings.VIDEO_IMAGE_MIN_HEIGHT),
) as image_file:
response = self.client.post(video_image_upload_url, {'file': image_file}, format='multipart')
image_url2 = self.verify_image_upload_reponse(self.course.id, edx_video_id, response)
self.assertNotEqual(image_url1, image_url2)
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, True)
def test_video_image_no_file(self):
"""
Test that an error error message is returned if upload request is incorrect.
"""
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': 'test1'})
response = self.client.post(video_image_upload_url, {})
self.verify_error_message(response, 'An image file is required.')
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, True)
def test_no_video_image(self):
"""
Test image url is set to None if no video image.
"""
edx_video_id = 'test1'
get_videos_url = reverse_course_url('videos_handler', self.course.id)
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
with make_image_file(
dimensions=(settings.VIDEO_IMAGE_MIN_WIDTH, settings.VIDEO_IMAGE_MIN_HEIGHT),
) as image_file:
self.client.post(video_image_upload_url, {'file': image_file}, format='multipart')
val_image_url = get_course_video_image_url(course_id=self.course.id, edx_video_id=edx_video_id)
response = self.client.get_json(get_videos_url)
self.assertEqual(response.status_code, 200)
response_videos = json.loads(response.content.decode('utf-8'))["videos"]
for response_video in response_videos:
if response_video['edx_video_id'] == edx_video_id:
self.assertEqual(response_video['course_video_image_url'], val_image_url)
else:
self.assertEqual(response_video['course_video_image_url'], None)
@ddt.data(
# Image file type validation
(
{
'extension': '.png'
},
None
),
(
{
'extension': '.gif'
},
None
),
(
{
'extension': '.bmp'
},
None
),
(
{
'extension': '.jpg'
},
None
),
(
{
'extension': '.jpeg'
},
None
),
(
{
'extension': '.PNG'
},
None
),
(
{
'extension': '.tiff'
},
'This image file type is not supported. Supported file types are {supported_file_formats}.'.format(
supported_file_formats=list(settings.VIDEO_IMAGE_SUPPORTED_FILE_FORMATS.keys())
)
),
# Image file size validation
(
{
'size': settings.VIDEO_IMAGE_SETTINGS['VIDEO_IMAGE_MAX_BYTES'] + 10
},
'This image file must be smaller than {image_max_size}.'.format(
image_max_size=settings.VIDEO_IMAGE_MAX_FILE_SIZE_MB
)
),
(
{
'size': settings.VIDEO_IMAGE_SETTINGS['VIDEO_IMAGE_MIN_BYTES'] - 10
},
'This image file must be larger than {image_min_size}.'.format(
image_min_size=settings.VIDEO_IMAGE_MIN_FILE_SIZE_KB
)
),
# Image file minimum width / height
(
{
'width': 16, # 16x9
'height': 9
},
'Recommended image resolution is {image_file_max_width}x{image_file_max_height}. The minimum resolution is {image_file_min_width}x{image_file_min_height}.'.format( # lint-amnesty, pylint: disable=line-too-long
image_file_max_width=settings.VIDEO_IMAGE_MAX_WIDTH,
image_file_max_height=settings.VIDEO_IMAGE_MAX_HEIGHT,
image_file_min_width=settings.VIDEO_IMAGE_MIN_WIDTH,
image_file_min_height=settings.VIDEO_IMAGE_MIN_HEIGHT
)
),
(
{
'width': settings.VIDEO_IMAGE_MIN_WIDTH - 10,
'height': settings.VIDEO_IMAGE_MIN_HEIGHT
},
'Recommended image resolution is {image_file_max_width}x{image_file_max_height}. The minimum resolution is {image_file_min_width}x{image_file_min_height}.'.format( # lint-amnesty, pylint: disable=line-too-long
image_file_max_width=settings.VIDEO_IMAGE_MAX_WIDTH,
image_file_max_height=settings.VIDEO_IMAGE_MAX_HEIGHT,
image_file_min_width=settings.VIDEO_IMAGE_MIN_WIDTH,
image_file_min_height=settings.VIDEO_IMAGE_MIN_HEIGHT
)
),
(
{
'width': settings.VIDEO_IMAGE_MIN_WIDTH,
'height': settings.VIDEO_IMAGE_MIN_HEIGHT - 10
},
(
'Recommended image resolution is {image_file_max_width}x{image_file_max_height}. '
'The minimum resolution is {image_file_min_width}x{image_file_min_height}.'
).format(
image_file_max_width=settings.VIDEO_IMAGE_MAX_WIDTH,
image_file_max_height=settings.VIDEO_IMAGE_MAX_HEIGHT,
image_file_min_width=settings.VIDEO_IMAGE_MIN_WIDTH,
image_file_min_height=settings.VIDEO_IMAGE_MIN_HEIGHT
)
),
(
{
'width': 1200, # not 16:9, but width/height check first.
'height': 100
},
(
'Recommended image resolution is {image_file_max_width}x{image_file_max_height}. '
'The minimum resolution is {image_file_min_width}x{image_file_min_height}.'
).format(
image_file_max_width=settings.VIDEO_IMAGE_MAX_WIDTH,
image_file_max_height=settings.VIDEO_IMAGE_MAX_HEIGHT,
image_file_min_width=settings.VIDEO_IMAGE_MIN_WIDTH,
image_file_min_height=settings.VIDEO_IMAGE_MIN_HEIGHT
)
),
# Image file aspect ratio validation
(
{
'width': settings.VIDEO_IMAGE_MAX_WIDTH, # 1280x720
'height': settings.VIDEO_IMAGE_MAX_HEIGHT
},
None
),
(
{
'width': 850, # 16:9
'height': 478
},
None
),
(
{
'width': 940, # 1.67 ratio, applicable aspect ratio margin of .01
'height': 560
},
None
),
(
{
'width': settings.VIDEO_IMAGE_MIN_WIDTH + 100,
'height': settings.VIDEO_IMAGE_MIN_HEIGHT + 200
},
'This image file must have an aspect ratio of {video_image_aspect_ratio_text}.'.format(
video_image_aspect_ratio_text=settings.VIDEO_IMAGE_ASPECT_RATIO_TEXT
)
),
# Image file name validation
(
{
'prefix': 'nøn-åßç¡¡'
},
'The image file name can only contain letters, numbers, hyphens (-), and underscores (_).'
)
)
@ddt.unpack
@override_waffle_switch(VIDEO_IMAGE_UPLOAD_ENABLED, True)
def test_video_image_validation_message(self, image_data, error_message):
"""
Test video image validation gives proper error message.
Arguments:
image_data (Dict): Specific data to create image file.
error_message (String): Error message
"""
edx_video_id = 'test1'
video_image_upload_url = self.get_url_for_course_key(self.course.id, {'edx_video_id': edx_video_id})
with make_image_file(
dimensions=(
image_data.get('width', settings.VIDEO_IMAGE_MIN_WIDTH),
image_data.get('height', settings.VIDEO_IMAGE_MIN_HEIGHT)
),
prefix=image_data.get('prefix', 'videoimage'),
extension=image_data.get('extension', '.png'),
force_size=image_data.get('size', settings.VIDEO_IMAGE_SETTINGS['VIDEO_IMAGE_MIN_BYTES'])
) as image_file:
response = self.client.post(video_image_upload_url, {'file': image_file}, format='multipart')
if error_message:
self.verify_error_message(response, error_message)
else:
self.verify_image_upload_reponse(self.course.id, edx_video_id, response)
@ddt.ddt
@patch(
'openedx.core.djangoapps.video_config.models.VideoTranscriptEnabledFlag.feature_enabled',
Mock(return_value=True)
)
@patch.dict('django.conf.settings.FEATURES', {'ENABLE_VIDEO_UPLOAD_PIPELINE': True})
class TranscriptPreferencesTestCase(VideoUploadTestBase, CourseTestCase):
"""
Tests for video transcripts preferences.
"""
VIEW_NAME = 'transcript_preferences_handler'
def test_405_with_not_allowed_request_method(self):
"""
Verify that 405 is returned in case of not-allowed request methods.
Allowed request methods are POST and DELETE.
"""
video_transcript_url = self.get_url_for_course_key(self.course.id)
response = self.client.get(
video_transcript_url,
content_type='application/json'
)
self.assertEqual(response.status_code, 405)
@ddt.data(
# Video transcript feature disabled
(
{},
False,
'',
404,
),
# Error cases
(
{},
True,
"Invalid provider None.",
400
),
(
{
'provider': ''
},
True,
"Invalid provider .",
400
),
(
{
'provider': 'dummy-provider'
},
True,
"Invalid provider dummy-provider.",
400
),
(
{
'provider': TranscriptProvider.CIELO24
},
True,
"Invalid cielo24 fidelity None.",
400
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
},
True,
"Invalid cielo24 turnaround None.",
400
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'video_source_language': 'en'
},
True,
"Invalid languages [].",
400
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PREMIUM',
'cielo24_turnaround': 'STANDARD',
'video_source_language': 'es'
},
True,
"Unsupported source language es.",
400
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'video_source_language': 'en',
'preferred_languages': ['es', 'ur']
},
True,
"Invalid languages ['es', 'ur'].",
400
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA
},
True,
"Invalid 3play turnaround None.",
400
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA,
'three_play_turnaround': 'standard',
'video_source_language': 'zh',
},
True,
"Unsupported source language zh.",
400
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA,
'three_play_turnaround': 'standard',
'video_source_language': 'es',
'preferred_languages': ['es', 'ur']
},
True,
"Invalid languages ['es', 'ur'].",
400
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA,
'three_play_turnaround': 'standard',
'video_source_language': 'en',
'preferred_languages': ['es', 'ur']
},
True,
"Invalid languages ['es', 'ur'].",
400
),
# Success
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'video_source_language': 'es',
'preferred_languages': ['en']
},
True,
'',
200
),
(
{
'provider': TranscriptProvider.THREE_PLAY_MEDIA,
'three_play_turnaround': 'standard',
'preferred_languages': ['en'],
'video_source_language': 'en',
},
True,
'',
200
)
)
@ddt.unpack
def test_video_transcript(self, preferences, is_video_transcript_enabled, error_message, expected_status_code):
"""
Tests that transcript handler works correctly.
"""
video_transcript_url = self.get_url_for_course_key(self.course.id)
preferences_data = {
'provider': preferences.get('provider'),
'cielo24_fidelity': preferences.get('cielo24_fidelity'),
'cielo24_turnaround': preferences.get('cielo24_turnaround'),
'three_play_turnaround': preferences.get('three_play_turnaround'),
'preferred_languages': preferences.get('preferred_languages', []),
'video_source_language': preferences.get('video_source_language'),
}
with patch(
'openedx.core.djangoapps.video_config.models.VideoTranscriptEnabledFlag.feature_enabled'
) as video_transcript_feature:
video_transcript_feature.return_value = is_video_transcript_enabled
response = self.client.post(
video_transcript_url,
json.dumps(preferences_data),
content_type='application/json'
)
status_code = response.status_code
response = json.loads(response.content.decode('utf-8')) if is_video_transcript_enabled else response
self.assertEqual(status_code, expected_status_code)
self.assertEqual(response.get('error', ''), error_message)
# Remove modified and course_id fields from the response so as to check the expected transcript preferences.
response.get('transcript_preferences', {}).pop('modified', None)
response.get('transcript_preferences', {}).pop('course_id', None)
expected_preferences = preferences_data if is_video_transcript_enabled and not error_message else {}
self.assertDictEqual(response.get('transcript_preferences', {}), expected_preferences)
def test_remove_transcript_preferences(self):
"""
Test that transcript handler removes transcript preferences correctly.
"""
# First add course wide transcript preferences.
preferences = create_or_update_transcript_preferences(str(self.course.id))
# Verify transcript preferences exist
self.assertIsNotNone(preferences)
response = self.client.delete(
self.get_url_for_course_key(self.course.id),
content_type='application/json'
)
self.assertEqual(response.status_code, 204)
# Verify transcript preferences no loger exist
preferences = get_transcript_preferences(str(self.course.id))
self.assertIsNone(preferences)
def test_remove_transcript_preferences_not_found(self):
"""
Test that transcript handler works correctly even when no preferences are found.
"""
course_id = 'course-v1:dummy+course+id'
# Verify transcript preferences do not exist
preferences = get_transcript_preferences(course_id)
self.assertIsNone(preferences)
response = self.client.delete(
self.get_url_for_course_key(course_id),
content_type='application/json'
)
self.assertEqual(response.status_code, 204)
# Verify transcript preferences do not exist
preferences = get_transcript_preferences(course_id)
self.assertIsNone(preferences)
@ddt.data(
(
None,
False
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'preferred_languages': ['en']
},
False
),
(
{
'provider': TranscriptProvider.CIELO24,
'cielo24_fidelity': 'PROFESSIONAL',
'cielo24_turnaround': 'STANDARD',
'preferred_languages': ['en']
},
True
)
)
@ddt.unpack
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@patch('cms.djangoapps.contentstore.video_storage_handlers.get_transcript_preferences')
@patch('cms.djangoapps.contentstore.video_storage_handlers.boto3.resource')
def test_transcript_preferences_metadata(self, transcript_preferences, is_video_transcript_enabled,
mock_boto3_resource, mock_transcript_preferences):
"""
Tests that transcript preference metadata is only set if it is video transcript feature is enabled and
transcript preferences are already stored in the system.
"""
file_name = 'test-video.mp4'
request_data = {'files': [{'file_name': file_name, 'content_type': 'video/mp4'}]}
mock_transcript_preferences.return_value = transcript_preferences
# Setup boto3 mocks
mock_s3_client, mock_bucket, mock_s3_resource = setup_s3_mocks(mock_boto3_resource)
# Track generate_presigned_url calls to inspect metadata
presigned_url_calls = []
def mock_generate_presigned_url(operation, Params=None, ExpiresIn=None):
presigned_url_calls.append((operation, Params, ExpiresIn))
file_name = Params['Metadata']['client_video_id']
return f'http://example.com/url_{file_name}'
mock_s3_client.generate_presigned_url.side_effect = mock_generate_presigned_url
videos_handler_url = reverse_course_url('videos_handler', self.course.id)
with patch(
'openedx.core.djangoapps.video_config.models.VideoTranscriptEnabledFlag.feature_enabled'
) as video_transcript_feature:
video_transcript_feature.return_value = is_video_transcript_enabled
response = self.client.post(videos_handler_url, json.dumps(request_data), content_type='application/json')
self.assertEqual(response.status_code, 200)
# Ensure `transcript_preferences` was set up in metadata correctly if sent through request.
if is_video_transcript_enabled and transcript_preferences:
self.assertEqual(len(presigned_url_calls), 1)
metadata = presigned_url_calls[0][1]['Metadata']
self.assertIn('transcript_preferences', metadata)
self.assertEqual(metadata['transcript_preferences'], json.dumps(transcript_preferences))
else:
# If conditions aren't met, verify transcript_preferences is not in metadata
if presigned_url_calls:
metadata = presigned_url_calls[0][1]['Metadata']
self.assertNotIn('transcript_preferences', metadata)
@patch.dict("django.conf.settings.FEATURES", {"ENABLE_VIDEO_UPLOAD_PIPELINE": True})
@override_settings(VIDEO_UPLOAD_PIPELINE={"BUCKET": "test_bucket", "ROOT_PATH": "test_root"})
class VideoUrlsCsvTestCase(
VideoUploadTestBase,
VideoStudioAccessTestsMixin,
VideoPipelineStudioAccessTestsMixin,
CourseTestCase
):
"""Test cases for the CSV download endpoint for video uploads"""
VIEW_NAME = "video_encodings_download"
def setUp(self):
super().setUp()
VideoUploadConfig(profile_whitelist="profile1").save()
def _check_csv_response(self, expected_profiles):
"""
Check that the response is a valid CSV response containing rows
corresponding to previous_uploads and including the expected profiles.
"""
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertEqual(
response["Content-Disposition"],
f"attachment; filename=\"{self.course.id.course}_video_urls.csv\""
)
response_content = b"".join(response.streaming_content)
response_reader = StringIO(response_content.decode())
reader = csv.DictReader(response_reader, dialect=csv.excel)
self.assertEqual(
reader.fieldnames,
(
["Name", "Duration", "Date Added", "Video ID", "Status"] +
[f"{profile} URL" for profile in expected_profiles]
)
)
rows = list(reader)
self.assertEqual(len(rows), len(self.previous_uploads))
for i, row in enumerate(rows):
response_video = dict(row.items())
# Videos should be returned by creation date descending
original_video = self.previous_uploads[-(i + 1)]
client_video_id = original_video["client_video_id"]
self.assertEqual(response_video["Name"], client_video_id)
self.assertEqual(response_video["Duration"], str(original_video["duration"]))
dateutil.parser.parse(response_video["Date Added"])
self.assertEqual(response_video["Video ID"], original_video["edx_video_id"])
self.assertEqual(response_video["Status"], convert_video_status(original_video))
for profile in expected_profiles:
response_profile_url = response_video[f"{profile} URL"]
original_encoded_for_profile = next(
(
original_encoded
for original_encoded in original_video["encoded_videos"]
if original_encoded["profile"] == profile
),
None
)
if original_encoded_for_profile:
original_encoded_for_profile_url = original_encoded_for_profile["url"]
self.assertEqual(response_profile_url, original_encoded_for_profile_url)
else:
self.assertEqual(response_profile_url, "")
def test_basic(self):
self._check_csv_response(["profile1"])
def test_profile_whitelist(self):
VideoUploadConfig(profile_whitelist="profile1,profile2").save()
self._check_csv_response(["profile1", "profile2"])
def test_non_ascii_course(self):
course = CourseFactory.create(
number="nón-äscii",
video_upload_pipeline={
"course_video_upload_token": self.test_token,
}
)
response = self.client.get(self.get_url_for_course_key(course.id))
self.assertEqual(response.status_code, 200)
self.assertEqual(
response["Content-Disposition"],
"attachment; filename*=utf-8''n%C3%B3n-%C3%A4scii_video_urls.csv"
)
@ddt.ddt
class GetVideoFeaturesTestCase(
CourseTestCase
):
"""Test cases for the get_video_features endpoint """
def setUp(self):
super().setUp()
self.url = self.get_url_for_course_key()
def get_url_for_course_key(self):
""" Helper to generate a url for a course key """
return reverse("video_features")
def test_basic(self):
""" Test for expected return keys """
response = self.client.get(self.get_url_for_course_key())
self.assertEqual(response.status_code, 200)
self.assertEqual(
set(response.json().keys()),
{
'videoSharingEnabled',
'allowThumbnailUpload',
}
)
@ddt.data(True, False)
def test_video_share_enabled(self, is_enabled):
""" Test the public video share flag """
self._test_video_feature(
PUBLIC_VIDEO_SHARE,
'videoSharingEnabled',
override_waffle_flag,
is_enabled,
)
@ddt.data(True, False)
def test_video_image_upload_enabled(self, is_enabled):
""" Test the video image upload switch """
self._test_video_feature(
VIDEO_IMAGE_UPLOAD_ENABLED,
'allowThumbnailUpload',
override_waffle_switch,
is_enabled,
)
def _test_video_feature(self, flag, key, override_fn, is_enabled):
""" Test that setting a waffle flag or switch on or off will cause the expected result """
with override_fn(flag, is_enabled):
response = self.client.get(self.get_url_for_course_key())
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()[key], is_enabled)
class GetStorageBucketTestCase(TestCase):
""" This test just check that connection works and returns the bucket.
It does not involve any mocking and triggers errors if has any import issue.
"""
@override_settings(AWS_ACCESS_KEY_ID='test_key_id', AWS_SECRET_ACCESS_KEY='test_secret')
@override_settings(VIDEO_UPLOAD_PIPELINE={
"VEM_S3_BUCKET": "vem_test_bucket", "BUCKET": "test_bucket", "ROOT_PATH": "test_root"
})
@patch('cms.djangoapps.contentstore.video_storage_handlers.boto3.resource')
def test_storage_bucket(self, mock_boto3_resource):
""" Test that storage service functions work correctly with boto3."""
# Setup boto3 mocks
mock_s3_client, mock_bucket, mock_s3_resource = setup_s3_mocks(mock_boto3_resource, 'vem_test_bucket')
# Test storage_service_bucket function
bucket = storage_service_bucket()
self.assertEqual(bucket.name, 'vem_test_bucket')
mock_s3_resource.Bucket.assert_called_once_with('vem_test_bucket')
# Test storage_service_key function
edx_video_id = 'dummy_video'
key_name = storage_service_key(bucket, file_name=edx_video_id)
expected_key = 'test_root/dummy_video'
self.assertEqual(key_name, expected_key)
# Test that we can generate presigned URL using the bucket's client
mock_s3_client.generate_presigned_url.return_value = (
'https://vem_test_bucket.s3.amazonaws.com:443/test_root/dummy_video?signature=test'
)
upload_url = mock_s3_client.generate_presigned_url(
'put_object',
Params={
'Bucket': bucket.name,
'Key': key_name,
'ContentType': 'video/mp4'
},
ExpiresIn=KEY_EXPIRATION_IN_SECONDS
)
self.assertIn("vem_test_bucket.s3.amazonaws.com", upload_url)
self.assertIn("test_root/dummy_video", upload_url)
class CourseYoutubeEdxVideoIds(ModuleStoreTestCase):
"""
This test checks youtube videos in a course
"""
VIEW_NAME = 'youtube_edx_video_ids'
def setUp(self):
super().setUp()
self.course = CourseFactory.create()
self.course_with_no_youtube_videos = CourseFactory.create()
self.store = modulestore()
self.user = UserFactory()
self.client.login(username=self.user.username, password='Password1234')
def get_url_for_course_key(self, course_key, kwargs=None):
"""Return video handler URL for the given course"""
return reverse_course_url(self.VIEW_NAME, course_key, kwargs) # lint-amnesty, pylint: disable=no-member
def test_course_with_youtube_videos(self):
course_key = self.course.id
with self.store.bulk_operations(course_key):
chapter_loc = self.store.create_child(
self.user.id, self.course.location, 'chapter', 'test_chapter'
).location
seq_loc = self.store.create_child(
self.user.id, chapter_loc, 'sequential', 'test_seq'
).location
vert_loc = self.store.create_child(self.user.id, seq_loc, 'vertical', 'test_vert').location
self.store.create_child(
self.user.id,
vert_loc,
'problem',
'test_problem',
fields={"data": "<problem>Test</problem>"}
)
self.store.create_child(
self.user.id, vert_loc, 'video', fields={
"youtube_is_available": False,
"name": "sample_video",
"edx_video_id": "youtube_193_84709099",
}
)
response = self.client.get(self.get_url_for_course_key(course_key))
self.assertEqual(response.status_code, 200)
edx_video_ids = json.loads(response.content.decode('utf-8'))['edx_video_ids']
self.assertEqual(len(edx_video_ids), 1)
def test_course_with_no_youtube_videos(self):
course_key = self.course_with_no_youtube_videos.id
with self.store.bulk_operations(course_key):
chapter_loc = self.store.create_child(
self.user.id, self.course_with_no_youtube_videos.location, 'chapter', 'test_chapter'
).location
seq_loc = self.store.create_child(
self.user.id, chapter_loc, 'sequential', 'test_seq'
).location
vert_loc = self.store.create_child(self.user.id, seq_loc, 'vertical', 'test_vert').location
self.store.create_child(
self.user.id, vert_loc, 'problem', 'test_problem', fields={"data": "<problem>Test</problem>"}
)
self.store.create_child(
self.user.id, vert_loc, 'video', fields={
"youtube_id_1_0": None,
"name": "sample_video",
"edx_video_id": "no_youtube_193_84709099",
}
)
response = self.client.get(self.get_url_for_course_key(course_key))
edx_video_ids = json.loads(response.content.decode('utf-8'))['edx_video_ids']
self.assertEqual(response.status_code, 200)
self.assertEqual(len(edx_video_ids), 0)

2
lms/djangoapps/commerce/tests/test_utils.py
View File

@@ -50,7 +50,7 @@ class AuditLogTests(TestCase):
# Verify that the logged message contains comma-separated
# key-value pairs ordered alphabetically by key.
message = 'foo: bar="baz", qux="quux"'
assert mock_log.info.called_with(message)
mock_log.info.assert_called_with(message)
@ddt.ddt

4
lms/djangoapps/course_home_api/progress/tests/test_api.py
View File

@@ -109,7 +109,7 @@ class ProgressApiTests(TestCase):
}
results = calculate_progress_for_learner_in_course("some_course", "some_user")
assert mock_get_summary.called_once_with("some_course", "some_user")
mock_get_summary.assert_called_once_with("some_course", "some_user")
assert results == expected_data
@patch("lms.djangoapps.course_home_api.progress.api.get_course_blocks_completion_summary")
@@ -134,7 +134,7 @@ class ProgressApiTests(TestCase):
}
results = calculate_progress_for_learner_in_course("some_course", "some_user")
assert mock_get_summary.called_once_with("some_course", "some_user")
mock_get_summary.assert_called_once_with("some_course", "some_user")
assert results == expected_data
@patch("lms.djangoapps.course_home_api.progress.api.get_course_blocks_completion_summary")

3
openedx/core/djangoapps/api_admin/tests/test_views.py
View File

@@ -358,10 +358,11 @@ class CatalogEditViewTest(CatalogTest):
@httpretty.activate
def test_edit(self):
# Mock both PATCH and GET endpoints before making the POST request
self.mock_catalog_endpoint(self.catalog.attributes, method=httpretty.PATCH, catalog_id=self.catalog.id)
new_attributes = dict(self.catalog.attributes, **{'delete-catalog': 'off', 'name': 'changed'})
response = self.client.post(self.url, new_attributes)
self.mock_catalog_endpoint(new_attributes, catalog_id=self.catalog.id)
response = self.client.post(self.url, new_attributes)
self.assertRedirects(response, reverse('api_admin:catalog-edit', kwargs={'catalog_id': self.catalog.id}))
@httpretty.activate

9
openedx/core/djangoapps/catalog/tests/test_utils.py
View File

@@ -456,9 +456,12 @@ class TestGetCourseRuns(CatalogIntegrationMixin, CacheIsolationTestCase):
catalog_integration = self.create_catalog_integration(service_username="nonexistent-user")
data = get_course_runs()
mock_log_error.any_call(
"Catalog service user with username [%s] does not exist. Course runs will not be retrieved.",
catalog_integration.service_username,
mock_log_error.assert_any_call(
"Catalog service user with username [{username}] does not exist. "
"{field} will not be retrieved.".format(
username=catalog_integration.service_username,
field="Course runs",
)
)
assert not mock_get_edx_api_data.called
assert data == []

2
openedx/core/djangoapps/content_libraries/tests/test_api.py
View File

@@ -102,7 +102,7 @@ class EdxModulestoreImportClientTest(TestCase):
self.client.import_blocks_from_course('foobar', lambda *_: None)
mock_get_library_block.assert_called_once()
mock_get_library_block_static_asset_files.called_once()
mock_get_library_block_static_asset_files.assert_called_once()
mock_set_library_block_olx.assert_called_once_with(
mock.ANY, 'fake-olx')
mock_publish_changes.assert_called_once()

2
requirements/edx/base.txt
View File

@@ -67,8 +67,6 @@ bleach[css]==6.3.0
# ora2
# xblock-drag-and-drop-v2
# xblock-poll
boto==2.49.0
# via -r requirements/edx/kernel.in
boto3==1.42.5
# via
# -r requirements/edx/kernel.in

4
requirements/edx/development.txt
View File

@@ -137,10 +137,6 @@ bleach[css]==6.3.0
# ora2
# xblock-drag-and-drop-v2
# xblock-poll
boto==2.49.0
# via
# -r requirements/edx/doc.txt
# -r requirements/edx/testing.txt
boto3==1.42.5
# via
# -r requirements/edx/doc.txt

2
requirements/edx/doc.txt
View File

@@ -102,8 +102,6 @@ bleach[css]==6.3.0
# ora2
# xblock-drag-and-drop-v2
# xblock-poll
boto==2.49.0
# via -r requirements/edx/base.txt
boto3==1.42.5
# via
# -r requirements/edx/base.txt

1
requirements/edx/kernel.in
View File

@@ -23,7 +23,6 @@ acid-xblock # This XBlock is used for unit tests as well
analytics-python # Used for Segment analytics
attrs # Reduces boilerplate code involving class attributes
Babel # Internationalization utilities, used for date formatting in a few places
boto # Deprecated version of the AWS SDK; we should stop using this
boto3 # Amazon Web Services SDK for Python
botocore # via boto3, s3transfer
bridgekeeper # Used for determining permissions for courseware. Use of django-rules is preferred.

2
requirements/edx/testing.txt
View File

@@ -101,8 +101,6 @@ bleach[css]==6.3.0
# ora2
# xblock-drag-and-drop-v2
# xblock-poll
boto==2.49.0
# via -r requirements/edx/base.txt
boto3==1.42.5
# via
# -r requirements/edx/base.txt