<commit_before>"""Tests for items views."""

from __future__ import absolute_import

import copy
import json
import tempfile
import textwrap
from codecs import BOM_UTF8
from uuid import uuid4

import ddt
import six
from django.conf import settings
from django.test.utils import override_settings
from django.urls import reverse
from edxval.api import create_video
from mock import Mock, patch
from opaque_keys.edx.keys import UsageKey

from contentstore.tests.utils import CourseTestCase, mock_requests_get
from openedx.core.djangoapps.contentserver.caching import del_cached_content
from xmodule.contentstore.content import StaticContent
from xmodule.contentstore.django import contentstore
from xmodule.exceptions import NotFoundError
from xmodule.modulestore.django import modulestore
from xmodule.video_module import VideoBlock
from xmodule.video_module.transcripts_utils import (
    GetTranscriptsFromYouTubeException,
    Transcript,
    get_video_transcript_content,
    remove_subs_from_store
)

TEST_DATA_CONTENTSTORE = copy.deepcopy(settings.CONTENTSTORE)
TEST_DATA_CONTENTSTORE['DOC_STORE_CONFIG']['db'] = 'test_xcontent_%s' % uuid4().hex

SRT_TRANSCRIPT_CONTENT = b"""0
00:00:10,500 --> 00:00:13,000
Elephant's Dream

1
00:00:15,000 --> 00:00:18,000
At the left we can see...

"""

SJSON_TRANSCRIPT_CONTENT = Transcript.convert(
    SRT_TRANSCRIPT_CONTENT,
    Transcript.SRT,
    Transcript.SJSON,
)


@override_settings(CONTENTSTORE=TEST_DATA_CONTENTSTORE)
class BaseTranscripts(CourseTestCase):
    """Base test class for transcripts tests."""

    def clear_subs_content(self):
        """Remove, if transcripts content exists."""
        for youtube_id in self.get_youtube_ids().values():
            filename = 'subs_{0}.srt.sjson'.format(youtube_id)
            content_location = StaticContent.compute_location(self.course.id, filename)
            try:
                content = contentstore().find(content_location)
                contentstore().delete(content.get_id())
            except NotFoundError:
                pass

    def save_subs_to_store(self, subs, subs_id):
        """
        Save transcripts into `StaticContent`.
        """
        filedata = json.dumps(subs, indent=2)
        mime_type = 'application/json'
        filename = 'subs_{0}.srt.sjson'.format(subs_id)

        content_location = StaticContent.compute_location(self.course.id, filename)
        content = StaticContent(content_location, filename, mime_type, filedata)
        contentstore().save(content)
        del_cached_content(content_location)
        return content_location

    def setUp(self):
        """Create initial data."""
        super(BaseTranscripts, self).setUp()

        # Add video module
        data = {
            'parent_locator': six.text_type(self.course.location),
            'category': 'video',
            'type': 'video'
        }
        resp = self.client.ajax_post('/xblock/', data)
        self.assertEqual(resp.status_code, 200)

        self.video_usage_key = self._get_usage_key(resp)
        self.item = modulestore().get_item(self.video_usage_key)
        # hI10vDNYz4M - valid Youtube ID with transcripts.
        # JMD_ifUUfsU, AKqURZnYqpk, DYpADpL7jAY - valid Youtube IDs without transcripts.
        self.set_fields_from_xml(
            self.item, '<video youtube="0.75:JMD_ifUUfsU,1.0:hI10vDNYz4M,1.25:AKqURZnYqpk,1.50:DYpADpL7jAY" />'
        )
        modulestore().update_item(self.item, self.user.id)

        self.item = modulestore().get_item(self.video_usage_key)
        # Remove all transcripts for current module.
        self.clear_subs_content()

    def _get_usage_key(self, resp):
        """ Returns the usage key from the response returned by a create operation. """
        usage_key_string = json.loads(resp.content.decode('utf-8')).get('locator')
        return UsageKey.from_string(usage_key_string)

    def get_youtube_ids(self):
        """Return youtube speeds and ids."""
        item = modulestore().get_item(self.video_usage_key)

        return {
            0.75: item.youtube_id_0_75,
            1: item.youtube_id_1_0,
            1.25: item.youtube_id_1_25,
            1.5: item.youtube_id_1_5
        }

    def create_non_video_module(self):
        """
        Setup non video module for tests.
        """
        data = {
            'parent_locator': six.text_type(self.course.location),
            'category': 'non_video',
            'type': 'non_video'
        }
        response = self.client.ajax_post('/xblock/', data)
        usage_key = self._get_usage_key(response)
        item = modulestore().get_item(usage_key)
        self.set_fields_from_xml(self.item, '<non_video youtube="0.75:JMD_ifUUfsU,1.0:hI10vDNYz4M" />')
        modulestore().update_item(item, self.user.id)

        return usage_key

    def assert_response(self, response, expected_status_code, expected_message):
        response_content = json.loads(response.content.decode('utf-8'))
        self.assertEqual(response.status_code, expected_status_code)
        self.assertEqual(response_content['status'], expected_message)

    def set_fields_from_xml(self, item, xml):
        fields_data = VideoBlock.parse_video_xml(xml)
        for key, value in fields_data.items():
            setattr(item, key, value)


@ddt.ddt
class TestUploadTranscripts(BaseTranscripts):
    """
    Tests for '/transcripts/upload' endpoint.
    """
    def setUp(self):
        super(TestUploadTranscripts, self).setUp()
        self.contents = {
            'good': SRT_TRANSCRIPT_CONTENT,
            'bad': 'Some BAD data',
        }
        # Create temporary transcript files
        self.good_srt_file = self.create_transcript_file(content=self.contents['good'], suffix='.srt')
        self.bad_data_srt_file = self.create_transcript_file(content=self.contents['bad'], suffix='.srt')
        self.bad_name_srt_file = self.create_transcript_file(content=self.contents['good'], suffix='.bad')
        self.bom_srt_file = self.create_transcript_file(content=self.contents['good'], suffix='.srt', include_bom=True)

        # Setup a VEDA produced video and persist `edx_video_id` in VAL.
        create_video({
            'edx_video_id': u'123-456-789',
            'status': 'upload',
            'client_video_id': u'Test Video',
            'duration': 0,
            'encoded_videos': [],
            'courses': [six.text_type(self.course.id)]
        })

        # Add clean up handler
        self.addCleanup(self.clean_temporary_transcripts)

    def create_transcript_file(self, content, suffix, include_bom=False):
        """
        Setup a transcript file with suffix and content.
        """
        transcript_file = tempfile.NamedTemporaryFile(suffix=suffix)
        wrapped_content = textwrap.dedent(content)
        if include_bom:
            wrapped_content = wrapped_content.encode('utf-8-sig')
            # Verify that ufeff(BOM) character is in content.
            self.assertIn(BOM_UTF8, wrapped_content)

        transcript_file.write(wrapped_content)
        transcript_file.seek(0)

        return transcript_file

    def clean_temporary_transcripts(self):
        """
        Close transcript files gracefully.
        """
        self.good_srt_file.close()
        self.bad_data_srt_file.close()
        self.bad_name_srt_file.close()
        self.bom_srt_file.close()

    def upload_transcript(self, locator, transcript_file, edx_video_id=None):
        """
        Uploads a transcript for a video
        """
        payload = {}
        if locator:
            payload.update({'locator': locator})

        if edx_video_id is not None:
            payload.update({'edx_video_id': edx_video_id})

        if transcript_file:
            payload.update({'transcript-file': transcript_file})

        upload_url = reverse('upload_transcripts')
        response = self.client.post(upload_url, payload)
<commit_msg>BOM-357 Fix failing test<commit_after>"""Tests for items views."""

from __future__ import absolute_import

import copy
import json
import tempfile
import textwrap
from codecs import BOM_UTF8
from uuid import uuid4

import ddt
import six
from django.conf import settings
from django.test.utils import override_settings
from django.urls import reverse
from edxval.api import create_video
from mock import Mock, patch
from opaque_keys.edx.keys import UsageKey

from contentstore.tests.utils import CourseTestCase, mock_requests_get
from openedx.core.djangoapps.contentserver.caching import del_cached_content
from xmodule.contentstore.content import StaticContent
from xmodule.contentstore.django import contentstore
from xmodule.exceptions import NotFoundError
from xmodule.modulestore.django import modulestore
from xmodule.video_module import VideoBlock
from xmodule.video_module.transcripts_utils import (
    GetTranscriptsFromYouTubeException,
    Transcript,
    get_video_transcript_content,
    remove_subs_from_store
)

TEST_DATA_CONTENTSTORE = copy.deepcopy(settings.CONTENTSTORE)
TEST_DATA_CONTENTSTORE['DOC_STORE_CONFIG']['db'] = 'test_xcontent_%s' % uuid4().hex

SRT_TRANSCRIPT_CONTENT = b"""0
00:00:10,500 --> 00:00:13,000
Elephant's Dream

1
00:00:15,000 --> 00:00:18,000
At the left we can see...

"""

SJSON_TRANSCRIPT_CONTENT = Transcript.convert(
    SRT_TRANSCRIPT_CONTENT,
    Transcript.SRT,
    Transcript.SJSON,
)


@override_settings(CONTENTSTORE=TEST_DATA_CONTENTSTORE)
class BaseTranscripts(CourseTestCase):
    """Base test class for transcripts tests."""

    def clear_subs_content(self):
        """Remove, if transcripts content exists."""
        for youtube_id in self.get_youtube_ids().values():
            filename = 'subs_{0}.srt.sjson'.format(youtube_id)
            content_location = StaticContent.compute_location(self.course.id, filename)
            try:
                content = contentstore().find(content_location)
                contentstore().delete(content.get_id())
            except NotFoundError:
                pass

    def save_subs_to_store(self, subs, subs_id):
        """
        Save transcripts into `StaticContent`.
        """
        filedata = json.dumps(subs, indent=2)
        mime_type = 'application/json'
        filename = 'subs_{0}.srt.sjson'.format(subs_id)

        content_location = StaticContent.compute_location(self.course.id, filename)
        content = StaticContent(content_location, filename, mime_type, filedata)
        contentstore().save(content)
        del_cached_content(content_location)
        return content_location

    def setUp(self):
        """Create initial data."""
        super(BaseTranscripts, self).setUp()

        # Add video module
        data = {
            'parent_locator': six.text_type(self.course.location),
            'category': 'video',
            'type': 'video'
        }
        resp = self.client.ajax_post('/xblock/', data)
        self.assertEqual(resp.status_code, 200)

        self.video_usage_key = self._get_usage_key(resp)
        self.item = modulestore().get_item(self.video_usage_key)
        # hI10vDNYz4M - valid Youtube ID with transcripts.
        # JMD_ifUUfsU, AKqURZnYqpk, DYpADpL7jAY - valid Youtube IDs without transcripts.
        self.set_fields_from_xml(
            self.item, '<video youtube="0.75:JMD_ifUUfsU,1.0:hI10vDNYz4M,1.25:AKqURZnYqpk,1.50:DYpADpL7jAY" />'
        )
        modulestore().update_item(self.item, self.user.id)

        self.item = modulestore().get_item(self.video_usage_key)
        # Remove all transcripts for current module.
        self.clear_subs_content()

    def _get_usage_key(self, resp):
        """ Returns the usage key from the response returned by a create operation. """
        usage_key_string = json.loads(resp.content.decode('utf-8')).get('locator')
        return UsageKey.from_string(usage_key_string)

    def get_youtube_ids(self):
        """Return youtube speeds and ids."""
        item = modulestore().get_item(self.video_usage_key)

        return {
            0.75: item.youtube_id_0_75,
            1: item.youtube_id_1_0,
            1.25: item.youtube_id_1_25,
            1.5: item.youtube_id_1_5
        }

    def create_non_video_module(self):
        """
        Setup non video module for tests.
        """
        data = {
            'parent_locator': six.text_type(self.course.location),
            'category': 'non_video',
            'type': 'non_video'
        }
        response = self.client.ajax_post('/xblock/', data)
        usage_key = self._get_usage_key(response)
        item = modulestore().get_item(usage_key)
        self.set_fields_from_xml(self.item, '<non_video youtube="0.75:JMD_ifUUfsU,1.0:hI10vDNYz4M" />')
        modulestore().update_item(item, self.user.id)

        return usage_key

    def assert_response(self, response, expected_status_code, expected_message):
        response_content = json.loads(response.content.decode('utf-8'))
        self.assertEqual(response.status_code, expected_status_code)
        self.assertEqual(response_content['status'], expected_message)

    def set_fields_from_xml(self, item, xml):
        fields_data = VideoBlock.parse_video_xml(xml)
        for key, value in fields_data.items():
            setattr(item, key, value)


@ddt.ddt
class TestUploadTranscripts(BaseTranscripts):
    """
    Tests for '/transcripts/upload' endpoint.
    """
    def setUp(self):
        super(TestUploadTranscripts, self).setUp()
        self.contents = {
            'good': SRT_TRANSCRIPT_CONTENT,
            'bad': b'Some BAD data',
        }
        # Create temporary transcript files
        self.good_srt_file = self.create_transcript_file(content=self.contents['good'], suffix='.srt')
        self.bad_data_srt_file = self.create_transcript_file(content=self.contents['bad'], suffix='.srt')
        self.bad_name_srt_file = self.create_transcript_file(content=self.contents['good'], suffix='.bad')
        self.bom_srt_file = self.create_transcript_file(content=self.contents['good'], suffix='.srt', include_bom=True)

        # Setup a VEDA produced video and persist `edx_video_id` in VAL.
        create_video({
            'edx_video_id': u'123-456-789',
            'status': 'upload',
            'client_video_id': u'Test Video',
            'duration': 0,
            'encoded_videos': [],
            'courses': [six.text_type(self.course.id)]
        })

        # Add clean up handler
        self.addCleanup(self.clean_temporary_transcripts)

    def create_transcript_file(self, content, suffix, include_bom=False):
        """
        Setup a transcript file with suffix and content.
        """
        transcript_file = tempfile.NamedTemporaryFile(suffix=suffix)
        wrapped_content = textwrap.dedent(content.decode('utf-8'))
        if include_bom:
            wrapped_content = wrapped_content.encode('utf-8-sig')
            # Verify that ufeff(BOM) character is in content.
            self.assertIn(BOM_UTF8, wrapped_content)
            transcript_file.write(wrapped_content)
        else:
            transcript_file.write(wrapped_content.encode('utf-8'))
        transcript_file.seek(0)

        return transcript_file

    def clean_temporary_transcripts(self):
        """
        Close transcript files gracefully.
        """
        self.good_srt_file.close()
        self.bad_data_srt_file.close()
        self.bad_name_srt_file.close()
        self.bom_srt_file.close()

    def upload_transcript(self, locator, transcript_file, edx_video_id=None):
        """
        Uploads a transcript for a video
        """
        payload = {}
        if locator:
            payload.update({'locator': locator})

        if edx_video_id is not None:
            payload.update({'edx_video_id': edx_video_id})

        if transcript_file:
            payload.update({'transcript-file': transcript_file})

        upload_url = reverse('upload_transcripts')
        response = self.client.post(upload_url, payload)

{'fraction_non_alphanumeric': 0.10350238513683153, 'fraction_numerical': 0.013871453678132062, 'mean_word_length': 2.950657079097446, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 12}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7148410', 'n_tokens_mistral': 5145, 'n_tokens_neox': 4856, 'n_words': 921}