core/controllers/resources_test.py - develop - githubmirror/oppia

core/controllers/resources_test.py
1 105 строк38 KB
Sean Lip
Remove dependencies from common.py and from scripts on core (#23993)
09 дек 2025, 20:34
09 дек 2025, 20:342c53071
# Copyright 2014 The Oppia Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS-IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for Oppia resource handling (e.g. templates, images)."""

from __future__ import annotations

import os

from core import feconf
from core.constants import constants
from core.domain import (
    exp_services,
    fs_services,
    rights_manager,
    skill_services,
    story_services,
    topic_domain,
    topic_fetchers,
    user_services,
)
from core.tests import test_utils

from typing import Final


class AssetDevHandlerImageTests(test_utils.GenericTestBase):

    ASSET_HANDLER_URL_PREFIX = '/assetsdevhandler'

    def _get_image_url(
        self, entity_type: str, entity_id: str, filename: str
    ) -> str:
        """Gets the image URL."""
        return '%s/%s/%s/assets/image/%s' % (
            self.ASSET_HANDLER_URL_PREFIX,
            entity_type,
            entity_id,
            filename,
        )

    def setUp(self) -> None:
        """Load a demo exploration and register self.EDITOR_EMAIL."""
        super().setUp()

        exp_services.delete_demo('0')
        self.system_user = user_services.get_system_user()
        exp_services.load_demo('0')

        rights_manager.release_ownership_of_exploration(self.system_user, '0')
        self.signup(self.EDITOR_EMAIL, self.EDITOR_USERNAME)

    def test_image_upload_with_no_filename_raises_error(self) -> None:
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {},
            csrf_token=csrf_token,
            upload_files=[('image', 'unused_filename', raw_image)],
            expected_status_int=400,
        )

        self.assertEqual(
            response_dict['error'],
            'At \'http://localhost/createhandler/imageupload/exploration/0\' '
            'these errors are happening:\n'
            'Missing key in handler args: filename.',
        )

        self.logout()

    def test_get_image_with_invalid_page_context_raises_error(self) -> None:
        self.login(self.EDITOR_EMAIL)

        # Only 404 is raised here due to the try - except block in the
        # controller.
        self.get_json(
            self._get_image_url('invalid_context', '0', 'filename'),
            expected_status_int=400,
        )
        self.logout()

    def test_image_upload_with_invalid_filename_raises_error(self) -> None:
        """Test that filenames starting with dot are rejected."""
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': '.png'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('image', 'unused_filename', raw_image)],
        )

        error_msg = (
            'At \'http://localhost/createhandler/imageupload/exploration/0\' '
            'these errors are happening:\n'
            'Schema validation for \'filename\' failed: Validation failed: '
            'is_regex_matched ({\'regex_pattern\': '
            '\'^[a-zA-Z0-9\\\\-_]+\\\\.(jpg|jpeg|png|gif|svg)$\'}'
            ') for object .png'
        )
        self.assertEqual(response_dict['error'], error_msg)

        self.logout()

    def test_cannot_upload_duplicate_image(self) -> None:
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            upload_files=[('image', 'unused_filename', raw_image)],
        )

        filename = response_dict['filename']

        response = self.get_custom_response(
            self._get_image_url('exploration', '0', filename), 'image/png'
        )
        self.assertEqual(response.body, raw_image)

        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            upload_files=[('image', 'unused_filename', raw_image)],
            expected_status_int=400,
        )

        self.assertEqual(
            response_dict['error'],
            'A file with the name test.png already exists. Please choose a '
            'different name.',
        )

    def test_image_upload_and_download(self) -> None:
        """Test image uploading and downloading."""
        self.signup(self.CURRICULUM_ADMIN_EMAIL, self.CURRICULUM_ADMIN_USERNAME)
        admin_id = self.get_user_id_from_email(self.CURRICULUM_ADMIN_EMAIL)
        self.set_curriculum_admins([self.CURRICULUM_ADMIN_USERNAME])

        subtopic = topic_domain.Subtopic.create_default_subtopic(
            1, 'Subtopic Title', 'url-frag'
        )
        story_id = story_services.get_new_story_id()
        topic_id = topic_fetchers.get_new_topic_id()
        skill_id = skill_services.get_new_skill_id()
        self.save_new_story(story_id, admin_id, topic_id)
        self.save_new_topic(
            topic_id,
            admin_id,
            name='Name',
            description='Description',
            canonical_story_ids=[story_id],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[subtopic],
            next_subtopic_id=2,
        )
        self.save_new_skill(skill_id, admin_id, description='Description')

        # Page context: Exploration.
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            upload_files=[('image', 'unused_filename', raw_image)],
        )
        filename = response_dict['filename']

        self.logout()

        response = self.get_custom_response(
            self._get_image_url('exploration', '0', filename), 'image/png'
        )
        self.assertEqual(response.body, raw_image)

        # Page context: Topic.
        self.login(self.CURRICULUM_ADMIN_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/topic/%s' % (feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX, topic_id),
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            upload_files=[('image', 'unused_filename', raw_image)],
        )
        filename = response_dict['filename']

        self.logout()

        response = self.get_custom_response(
            self._get_image_url('topic', topic_id, filename), 'image/png'
        )
        self.assertEqual(response.body, raw_image)

        # Page context: Story.
        self.login(self.CURRICULUM_ADMIN_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/story/%s' % (feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX, story_id),
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            upload_files=[('image', 'unused_filename', raw_image)],
        )
        filename = response_dict['filename']

        self.logout()

        response = self.get_custom_response(
            self._get_image_url('story', story_id, filename), 'image/png'
        )
        self.assertEqual(response.body, raw_image)

        # Page context: Skill.
        self.login(self.CURRICULUM_ADMIN_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/skill/%s' % (feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX, skill_id),
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            upload_files=[('image', 'unused_filename', raw_image)],
        )
        filename = response_dict['filename']

        self.logout()

        response = self.get_custom_response(
            self._get_image_url('skill', skill_id, filename), 'image/png'
        )
        self.assertEqual(response.body, raw_image)

        # Image context: Question Suggestions.
        self.login(self.CURRICULUM_ADMIN_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/question_suggestions/%s'
            % (feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX, skill_id),
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            upload_files=[('image', 'unused_filename', raw_image)],
        )
        filename = response_dict['filename']

        self.logout()

        response = self.get_custom_response(
            self._get_image_url('skill', skill_id, filename), 'image/png'
        )
        self.assertEqual(response.body, raw_image)

    def test_non_matching_extensions_are_detected(self) -> None:
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        filename_without_extension = 'test'
        supplied_filename = '%s.jpg' % filename_without_extension
        filename_with_correct_extension = '%s.png' % filename_without_extension

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        # Pass JPG extension even though raw_image data is PNG.
        # This test verifies that, when the filename extension differs from what
        # the raw data 'appears' to be, the image is rejected.
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': supplied_filename},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('image', 'unused_filename', raw_image)],
        )
        self.assertEqual(response_dict['status_code'], 400)
        self.assertEqual(
            response_dict['error'],
            'Expected a filename ending in .png, received test.jpg',
        )
        self.logout()

        # Test that neither form of the image is stored.
        self.get_json(
            self._get_image_url('exploration', '0', supplied_filename),
            expected_status_int=404,
        )
        self.get_json(
            self._get_image_url(
                'exploration', '0', filename_with_correct_extension
            ),
            expected_status_int=404,
        )

    def test_upload_empty_image(self) -> None:
        """Test upload of an empty image."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        # Upload an empty image.
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('image', 'unused_filename', b'')],
        )
        self.assertEqual(response_dict['status_code'], 400)
        self.assertEqual(response_dict['error'], 'No image supplied')

        self.logout()

    def test_upload_unsupported_image_format(self) -> None:
        """Test upload of a valid image with an unsupported format."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        # Simulate valid image data of a different type (e.g., BMP).
        bmp_image_data = (
            b'BM\x1a\x00\x00\x00\x00\x00\x00\x00\x1a\x00\x00\x00'
            b'\x0c\x00\x00\x00\x01\x00\x01\x00\x01\x00\x18\x00\xff\xff\xff'
            b'\x00\x00\x00'
        )

        # Upload the BMP image.
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('image', 'unused_filename', bmp_image_data)],
        )

        # Assert the response.
        self.assertEqual(response_dict['status_code'], 400)
        self.assertEqual(
            response_dict['error'], 'Image uses unsupported format'
        )

        self.logout()

    def test_upload_corrupted_image(self) -> None:
        """Test upload of a corrupted image."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        # Simulate corrupted image data.
        corrupted_image_data = (
            b'\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\xcc\xdd\xee\xff'
        )

        # Upload the corrupted image.
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test.png'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('image', 'unused_filename', corrupted_image_data)],
        )

        # Assert the response.
        self.assertEqual(response_dict['status_code'], 400)
        self.assertEqual(response_dict['error'], 'Image not recognized')

        self.logout()

    def test_upload_an_invalid_svg_image(self) -> None:
        """Test upload of an invalid SVG image."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        # Upload an invalid SVG image.
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test.svg'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('image', 'unused_filename', b'<badsvg></badsvg>')],
        )
        self.assertEqual(response_dict['status_code'], 400)
        self.assertEqual(
            response_dict['error'],
            'Unsupported tags/attributes found in the SVG:\ntags: '
            '[\'badsvg\']\n',
        )

        self.logout()

    def test_upload_a_large_svg(self) -> None:
        """Test upload of an SVG image that exceeds the 100 KB size limit."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        # Upload an SVG image that exceeds the file size limit of 100 KB.
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test.svg'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[
                (
                    'image',
                    'unused_filename',
                    b'<svg><path d="%s" /></svg>'
                    % (b'M150 0 L75 200 L225 200 Z ' * 4000),
                )
            ],
        )
        self.assertEqual(response_dict['status_code'], 400)
        self.assertEqual(
            response_dict['error'], 'Image exceeds file size limit of 100 KB.'
        )

        self.logout()

    def test_get_invalid_image(self) -> None:
        """Test retrieval of invalid images."""

        self.get_json(
            self._get_image_url('exploration', '0', 'bad_image'),
            expected_status_int=400,
        )

    def test_bad_filenames_are_detected(self) -> None:
        """Test that filenames with invalid characters are rejected."""
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test/a.png'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('image', 'unused_filename', raw_image)],
        )
        self.assertEqual(response_dict['status_code'], 400)

        error_msg = (
            'At \'http://localhost/createhandler/imageupload/exploration/0\' '
            'these errors are happening:\n'
            'Schema validation for \'filename\' failed: Validation failed: '
            'is_regex_matched ({\'regex_pattern\': '
            '\'^[a-zA-Z0-9\\\\-_]+\\\\.(jpg|jpeg|png|gif|svg)$\'}'
            ') for object test/a.png'
        )
        self.assertEqual(response_dict['error'], error_msg)

        self.logout()

    def test_missing_extensions_are_detected(self) -> None:
        """Test that filenames without extensions are rejected."""
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()
        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('image', 'unused_filename', raw_image)],
        )
        self.assertEqual(response_dict['status_code'], 400)

        error_msg = (
            'At \'http://localhost/createhandler/imageupload/exploration/0\' '
            'these errors are happening:\n'
            'Schema validation for \'filename\' failed: Validation failed: '
            'is_regex_matched ({\'regex_pattern\': '
            '\'^[a-zA-Z0-9\\\\-_]+\\\\.(jpg|jpeg|png|gif|svg)$\'}'
            ') for object test'
        )
        self.assertEqual(error_msg, response_dict['error'])

        self.logout()

    def test_bad_extensions_are_detected(self) -> None:
        """Test that invalid file extensions are rejected."""
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_image = f.read()
        response_dict = self.post_json(
            '%s/exploration/0' % feconf.EXPLORATION_IMAGE_UPLOAD_PREFIX,
            {'filename': 'test.pdf'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('image', 'unused_filename', raw_image)],
        )
        self.assertEqual(response_dict['status_code'], 400)

        error_msg = (
            'At \'http://localhost/createhandler/imageupload/exploration/0\' '
            'these errors are happening:\n'
            'Schema validation for \'filename\' failed: Validation failed: '
            'is_regex_matched ({\'regex_pattern\': '
            '\'^[a-zA-Z0-9\\\\-_]+\\\\.(jpg|jpeg|png|gif|svg)$\'}'
            ') for object test.pdf'
        )
        self.assertEqual(response_dict['error'], error_msg)

        self.logout()

    def test_request_invalid_asset_type(self) -> None:
        """Test that requests for invalid asset type is rejected with a 404."""
        self.login(self.EDITOR_EMAIL)

        self.get_html_response(
            '/assetsdevhandler/exploration/0/assets/unknowntype/myfile',
            expected_status_int=404,
        )
        self.logout()

    def test_get_response_with_dev_mode_disabled_returns_404_status(
        self,
    ) -> None:
        self.login(self.EDITOR_EMAIL)
        with self.swap(constants, 'EMULATOR_MODE', False):
            self.get_json(
                '/assetsdevhandler/exploration/0/assets/image/myfile.png',
                expected_status_int=404,
            )
        self.logout()


class AssetDevHandlerAudioTest(test_utils.GenericTestBase):
    """Test the upload of audio files to GCS."""

    TEST_AUDIO_FILE_MP3: Final = 'cafe.mp3'
    TEST_AUDIO_FILE_FLAC: Final = 'cafe.flac'
    TEST_AUDIO_FILE_OVER_MAX_LENGTH: Final = 'cafe-over-five-minutes.mp3'
    TEST_AUDIO_FILE_MPEG_CONTAINER: Final = 'test-mpeg-container.mp3'
    AUDIO_UPLOAD_URL_PREFIX: Final = '/createhandler/audioupload'

    def setUp(self) -> None:
        super().setUp()
        exp_services.delete_demo('0')
        self.system_user = user_services.get_system_user()
        exp_services.load_demo('0')

        rights_manager.release_ownership_of_exploration(self.system_user, '0')
        self.signup(self.EDITOR_EMAIL, self.EDITOR_USERNAME)

    def test_guest_can_not_upload(self) -> None:
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_MP3),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()
        response = self.post_json(
            '%s/0' % (self.AUDIO_UPLOAD_URL_PREFIX),
            {'filename': self.TEST_AUDIO_FILE_MP3},
            csrf_token=csrf_token,
            upload_files=[('raw_audio_file', 'unused_filename', raw_audio)],
            expected_status_int=401,
        )
        self.assertEqual(
            response['error'], 'You must be logged in to access this resource.'
        )

    def test_cannot_upload_audio_with_invalid_exp_id(self) -> None:
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_MP3),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()
        self.post_json(
            '%s/invalid_exp_id' % (self.AUDIO_UPLOAD_URL_PREFIX),
            {'filename': self.TEST_AUDIO_FILE_MP3},
            csrf_token=csrf_token,
            upload_files=[('raw_audio_file', 'unused_filename', raw_audio)],
            expected_status_int=400,
        )
        self.logout()

    def test_audio_upload(self) -> None:
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_MP3),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()
        self.post_json(
            '%s/0' % (self.AUDIO_UPLOAD_URL_PREFIX),
            {'filename': self.TEST_AUDIO_FILE_MP3},
            csrf_token=csrf_token,
            upload_files=[
                ('raw_audio_file', self.TEST_AUDIO_FILE_MP3, raw_audio)
            ],
            expected_status_int=200,
        )
        self.logout()

    def test_audio_upload_with_non_mp3_file(self) -> None:
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        fs = fs_services.GcsFileSystem(feconf.ENTITY_TYPE_EXPLORATION, '0')

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_FLAC),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()

        self.assertFalse(fs.isfile('audio/%s' % self.TEST_AUDIO_FILE_FLAC))

        response_dict = self.post_json(
            '%s/0' % self.AUDIO_UPLOAD_URL_PREFIX,
            {'filename': self.TEST_AUDIO_FILE_FLAC},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[
                ('raw_audio_file', self.TEST_AUDIO_FILE_FLAC, raw_audio)
            ],
        )
        error_msg = (
            'At \'http://localhost/createhandler/audioupload/0\' '
            'these errors are happening:\n'
            'Schema validation for \'raw_audio_file\' failed: '
            'Audio not recognized as a mp3 file\n'
            'Schema validation for \'filename\' failed: Validation failed: '
            'is_regex_matched ({\'regex_pattern\': '
            '\'[^\\\\s]+(\\\\.(?i)(mp3))$\'}) for object cafe.flac'
        )
        self.assertEqual(response_dict['error'], error_msg)

        self.logout()

    def test_detect_non_matching_extensions(self) -> None:
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        # Use an accepted audio extension in mismatched_filename
        # that differs from the uploaded file's audio type.
        mismatched_filename = 'test.mp3'
        with open(
            os.path.join(feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_FLAC),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()

        response_dict = self.post_json(
            '%s/0' % self.AUDIO_UPLOAD_URL_PREFIX,
            {'filename': mismatched_filename},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('raw_audio_file', mismatched_filename, raw_audio)],
        )
        self.logout()
        error_msg = (
            'At \'http://localhost/createhandler/audioupload/0\' '
            'these errors are happening:\n'
            'Schema validation for \'raw_audio_file\' failed: '
            'Audio not recognized as a mp3 file'
        )
        self.assertEqual(response_dict['error'], error_msg)

    def test_detect_non_audio_file(self) -> None:
        """Test that filenames with extensions that don't match the audio are
        detected.
        """

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'img.png'), 'rb', encoding=None
        ) as f:
            raw_audio = f.read()

        response_dict = self.post_json(
            '%s/0' % self.AUDIO_UPLOAD_URL_PREFIX,
            {'filename': self.TEST_AUDIO_FILE_FLAC},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('raw_audio_file', 'unused_filename', raw_audio)],
        )
        self.logout()

        error_msg = (
            'At \'http://localhost/createhandler/audioupload/0\' '
            'these errors are happening:\n'
            'Schema validation for \'raw_audio_file\' failed: '
            'Audio not recognized as a mp3 file\n'
            'Schema validation for \'filename\' failed: Validation failed: '
            'is_regex_matched ({\'regex_pattern\': '
            '\'[^\\\\s]+(\\\\.(?i)(mp3))$\'}) for object cafe.flac'
        )
        self.assertEqual(response_dict['error'], error_msg)

    def test_audio_upload_mpeg_container(self) -> None:
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(
                feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_MPEG_CONTAINER
            ),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()
        self.post_json(
            '%s/0' % (self.AUDIO_UPLOAD_URL_PREFIX),
            {'filename': self.TEST_AUDIO_FILE_MPEG_CONTAINER},
            csrf_token=csrf_token,
            upload_files=[('raw_audio_file', 'unused_filename', raw_audio)],
            expected_status_int=200,
        )
        self.logout()

    def test_invalid_extension_is_detected(self) -> None:
        """Test that invalid extensions are caught."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        filename_without_extension = 'test'
        invalid_extension = 'wav'
        supplied_filename = '%s.%s' % (
            filename_without_extension,
            invalid_extension,
        )

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_MP3),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()
        response_dict = self.post_json(
            '%s/0' % (self.AUDIO_UPLOAD_URL_PREFIX),
            {'filename': supplied_filename},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('raw_audio_file', 'unused_filename', raw_audio)],
        )
        self.logout()
        self.assertEqual(response_dict['status_code'], 400)
        error_msg = (
            'At \'http://localhost/createhandler/audioupload/0\' '
            'these errors are happening:\n'
            'Schema validation for \'filename\' failed: Validation failed: '
            'is_regex_matched ({\'regex_pattern\': '
            '\'[^\\\\s]+(\\\\.(?i)(mp3))$\'}) for object test.wav'
        )
        self.assertEqual(response_dict['error'], error_msg)

    def test_upload_empty_audio(self) -> None:
        """Test upload of empty audio."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()
        # Upload empty audio.
        response_dict = self.post_json(
            '%s/0' % self.AUDIO_UPLOAD_URL_PREFIX,
            {'filename': 'test.mp3'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('raw_audio_file', 'unused_filename', b'')],
        )
        self.logout()
        self.assertEqual(response_dict['status_code'], 400)
        self.assertEqual(
            response_dict['error'],
            'At \'http://localhost/createhandler/audioupload/0\' '
            'these errors are happening:\n'
            'Schema validation for '
            '\'raw_audio_file\' failed: No audio supplied',
        )

    def test_upload_bad_audio(self) -> None:
        """Test upload of malformed audio."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()
        response_dict = self.post_json(
            '%s/0' % self.AUDIO_UPLOAD_URL_PREFIX,
            {'filename': 'test.mp3'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[
                ('raw_audio_file', 'unused_filename', b'non_audio_data')
            ],
        )
        self.logout()
        self.assertEqual(response_dict['status_code'], 400)
        self.assertEqual(
            response_dict['error'],
            'At \'http://localhost/createhandler/audioupload/0\' '
            'these errors are happening:\n'
            'Schema validation for \'raw_audio_file\''
            ' failed: Audio not recognized as a mp3 file',
        )

    def test_missing_extensions_are_detected(self) -> None:
        """Test upload of filenames with no extensions are caught."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        missing_extension_filename = 'test'
        with open(
            os.path.join(feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_MP3),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()
        response_dict = self.post_json(
            '%s/0' % (self.AUDIO_UPLOAD_URL_PREFIX),
            {'filename': missing_extension_filename},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('raw_audio_file', 'unused_filename', raw_audio)],
        )
        self.logout()
        self.assertEqual(response_dict['status_code'], 400)
        error_msg = (
            'At \'http://localhost/createhandler/audioupload/0\' these '
            'errors are happening:\n'
            'Schema validation for \'filename\' failed: Validation failed: '
            'is_regex_matched ({\'regex_pattern\': '
            '\'[^\\\\s]+(\\\\.(?i)(mp3))$\'}) for object test'
        )
        self.assertEqual(response_dict['error'], error_msg)

    def test_exceed_max_length_detected(self) -> None:
        """Test that audio file is less than max playback length."""

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(
                feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_OVER_MAX_LENGTH
            ),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()
        response_dict = self.post_json(
            '%s/0' % self.AUDIO_UPLOAD_URL_PREFIX,
            {'filename': 'test.mp3'},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('raw_audio_file', 'unused_filename', raw_audio)],
        )
        self.logout()
        self.assertEqual(response_dict['status_code'], 400)
        self.assertIn(
            'Audio files must be under %s seconds in length'
            % feconf.MAX_AUDIO_FILE_LENGTH_SEC,
            response_dict['error'],
        )

    def test_non_matching_extensions_are_detected(self) -> None:
        """Test that filenames with extensions that don't match the audio are
        detected.
        """

        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        # Use an accepted audio extension in mismatched_filename
        # that differs from the uploaded file's audio type.
        mismatched_filename = 'test.mp3'
        with open(
            os.path.join(feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_FLAC),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()
        response_dict = self.post_json(
            '%s/0' % self.AUDIO_UPLOAD_URL_PREFIX,
            {'filename': mismatched_filename},
            csrf_token=csrf_token,
            expected_status_int=400,
            upload_files=[('raw_audio_file', 'unused_filename', raw_audio)],
        )
        self.logout()
        self.assertEqual(response_dict['status_code'], 400)
        self.assertEqual(
            response_dict['error'],
            'At \'http://localhost/createhandler/audioupload/0\' these '
            'errors are happening:\n'
            'Schema validation for \'raw_audio_file\' '
            'failed: Audio not recognized as a mp3 file',
        )

    def test_upload_check_for_duration_sec_as_response(self) -> None:
        """Tests the file upload and trying to confirm the
        audio file duration_secs is accurate.
        """
        self.login(self.EDITOR_EMAIL)
        csrf_token = self.get_new_csrf_token()

        with open(
            os.path.join(feconf.TESTS_DATA_DIR, self.TEST_AUDIO_FILE_MP3),
            'rb',
            encoding=None,
        ) as f:
            raw_audio = f.read()
        response_dict = self.post_json(
            '%s/0' % self.AUDIO_UPLOAD_URL_PREFIX,
            {'filename': self.TEST_AUDIO_FILE_MP3},
            csrf_token=csrf_token,
            expected_status_int=200,
            upload_files=[('raw_audio_file', 'unused_filename', raw_audio)],
        )
        self.logout()
        expected_value = {
            'filename': self.TEST_AUDIO_FILE_MP3,
            'duration_secs': 15.255510204081633,
        }
        self.assertEqual(response_dict, expected_value)


class PromoBarHandlerTest(test_utils.GenericTestBase):
    """Test for the PromoBarHandler."""

    def setUp(self) -> None:
        super().setUp()
        self.signup(
            self.RELEASE_COORDINATOR_EMAIL, self.RELEASE_COORDINATOR_USERNAME
        )

        self.add_user_role(
            self.RELEASE_COORDINATOR_USERNAME,
            feconf.ROLE_ID_RELEASE_COORDINATOR,
        )

    def test_get_promo_bar_data(self) -> None:
        response = self.get_json('/promo_bar_handler')
        self.assertEqual(
            response, {'promo_bar_enabled': False, 'promo_bar_message': ''}
        )

    def test_release_coordinator_able_to_update_promo_bar_config(self) -> None:
        self.login(self.RELEASE_COORDINATOR_EMAIL)

        csrf_token = self.get_new_csrf_token()

        response = self.put_json(
            '/promo_bar_handler',
            {
                'promo_bar_enabled': True,
                'promo_bar_message': 'New promo bar message.',
            },
            csrf_token=csrf_token,
        )
        self.assertEqual(response, {})

        response = self.get_json('/promo_bar_handler')
        self.assertEqual(
            response,
            {
                'promo_bar_enabled': True,
                'promo_bar_message': 'New promo bar message.',
            },
        )

        self.logout()


class ValueGeneratorHandlerTests(test_utils.GenericTestBase):

    def test_value_generated_error(self) -> None:
        dummy_id = 'someID'
        response = self.get_json(
            '/value_generator_handler/%s' % dummy_id, expected_status_int=400
        )
        self.assertIn(
            'Schema validation for \'generator_id\' failed: Received someID '
            'which is not in the allowed range of choices: [\'Copier\', '
            '\'RandomSelector\']',
            response['error'],
        )

    def test_html_response(self) -> None:
        copier_id = 'Copier'
        response = self.get_html_response(
            '/value_generator_handler/%s' % copier_id
        )
        self.assertIn(b'<object-editor [objType]="objType"', response.body)


class FaviconHandlerTest(test_utils.GenericTestBase):
    """Test for the FaviconHandler."""

    def test_redirect_to_assetsstatic(self) -> None:
        response = self.get_html_response(
            '/favicon.ico', expected_status_int=302
        )
        self.assertEqual(
            'http://localhost:8181/assetsstatic/favicon.ico',
            response.headers['location'],
        )


class RobotsTxtHandlerTest(test_utils.GenericTestBase):
    """Test for the RobotsTxtHandler."""

    def test_redirect_to_assetsstatic(self) -> None:
        response = self.get_html_response(
            '/robots.txt', expected_status_int=302
        )
        self.assertEqual(
            'http://localhost:8181/assetsstatic/robots.txt',
            response.headers['location'],
        )


class CopyrightImagesHandlerTest(test_utils.GenericTestBase):
    """Test for the CopyrightImagesHandler."""

    def test_redirect_to_assetsstatic(self) -> None:
        response = self.get_html_response(
            '/assets/copyrighted-images/general/mascot.svg',
            expected_status_int=302,
        )
        self.assertEqual(
            'http://localhost:8181/assetsstatic/copyrighted-images/'
            'general/mascot.svg',
            response.headers['location'],
        )