core/domain/voiceover_services_test.py - develop - githubmirror/oppia

core/domain/voiceover_services_test.py
2 154 строки76 KB
Nikhil
Adds beam job for voiceover synthesis by language accent (#25759)
15 апр 2026, 08:22
15 апр 2026, 08:228429520
# coding: utf-8
#
# Copyright 2024 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 voiecover service methods."""

from __future__ import annotations

import json
import os
import uuid

from core import feconf, schema_utils
from core.constants import constants
from core.domain import exp_domain, exp_services
from core.domain import platform_parameter_list as param_list
from core.domain import (
    state_domain,
    suggestion_services,
    taskqueue_services,
    translation_domain,
    translation_fetchers,
    voiceover_domain,
    voiceover_regeneration_services,
    voiceover_services,
)
from core.platform import models
from core.tests import test_utils

from typing import Dict, List, Sequence, Tuple

MYPY = False
if MYPY:  # pragma: no cover
    from mypy_imports import (
        cloud_task_models,
        email_models,
        exp_models,
        translation_models,
        voiceover_models,
    )

(
    cloud_task_models,
    exp_models,
    email_models,
    voiceover_models,
    translation_models,
) = models.Registry.import_models(
    [
        models.Names.CLOUD_TASK,
        models.Names.EXPLORATION,
        models.Names.EMAIL,
        models.Names.VOICEOVER,
        models.Names.TRANSLATION,
    ]
)


class EntityVoiceoversServicesTests(test_utils.GenericTestBase):
    def test_get_entity_voiceovers_returns_correctly(self) -> None:
        dummy_manual_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename1.mp3',
            'file_size_bytes': 3000,
            'needs_update': False,
            'duration_secs': 6.1,
        }
        dummy_autogenerated_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename2.mp3',
            'file_size_bytes': 3500,
            'needs_update': False,
            'duration_secs': 5.9,
        }
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id',
            1,
            'en-US',
            {
                'content_0': {
                    'manual': (dummy_manual_voiceover_dict),
                    'auto': (dummy_autogenerated_voiceover_dict),
                }
            },
            {},
        ).put()

        entity_voiceovers_instance = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                'exploration', 'exp_id', 1, 'en-US'
            )
        )

        self.assertEqual(entity_voiceovers_instance.entity_id, 'exp_id')
        self.assertEqual(entity_voiceovers_instance.entity_type, 'exploration')
        self.assertEqual(entity_voiceovers_instance.entity_version, 1)
        self.assertEqual(
            entity_voiceovers_instance.language_accent_code, 'en-US'
        )
        automatic_voiceover = entity_voiceovers_instance.voiceovers_mapping[
            'content_0'
        ]['auto']
        manual_voiceover = entity_voiceovers_instance.voiceovers_mapping[
            'content_0'
        ]['manual']
        assert isinstance(automatic_voiceover, state_domain.Voiceover)
        assert isinstance(manual_voiceover, state_domain.Voiceover)
        self.assertDictEqual(
            manual_voiceover.to_dict(), dummy_manual_voiceover_dict
        )
        self.assertDictEqual(
            automatic_voiceover.to_dict(), dummy_autogenerated_voiceover_dict
        )

    def test_get_entity_voiceovers_creates_new_entity(self) -> None:
        entity_voiceovers_instance = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                'exploration', 'exp_id', 1, 'en-US'
            )
        )

        self.assertEqual(entity_voiceovers_instance.entity_id, 'exp_id')
        self.assertEqual(entity_voiceovers_instance.entity_type, 'exploration')
        self.assertEqual(entity_voiceovers_instance.entity_version, 1)
        self.assertEqual(
            entity_voiceovers_instance.language_accent_code, 'en-US'
        )
        self.assertDictEqual(entity_voiceovers_instance.voiceovers_mapping, {})

    def test_should_convert_domain_instance_to_model_instance(self) -> None:
        dummy_manual_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename1.mp3',
            'file_size_bytes': 3000,
            'needs_update': False,
            'duration_secs': 6.1,
        }
        dummy_autogenerated_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename2.mp3',
            'file_size_bytes': 3500,
            'needs_update': False,
            'duration_secs': 5.9,
        }
        entity_voiceovers = voiceover_domain.EntityVoiceovers(
            entity_id='exp_id',
            entity_type=feconf.ENTITY_TYPE_EXPLORATION,
            entity_version=1,
            language_accent_code='en-IN',
            voiceovers_mapping={
                'content_0': {
                    'manual': (
                        state_domain.Voiceover.from_dict(
                            dummy_manual_voiceover_dict
                        )
                    ),
                    'auto': (
                        state_domain.Voiceover.from_dict(
                            dummy_autogenerated_voiceover_dict
                        )
                    ),
                }
            },
            automated_voiceovers_audio_offsets_msecs={},
        )
        entity_voiceovers_model = (
            voiceover_services.create_entity_voiceovers_model(entity_voiceovers)
        )

        self.assertEqual(
            entity_voiceovers.entity_id, entity_voiceovers_model.entity_id
        )
        self.assertEqual(
            entity_voiceovers.entity_type, entity_voiceovers_model.entity_type
        )
        self.assertEqual(
            entity_voiceovers.entity_version,
            entity_voiceovers_model.entity_version,
        )
        self.assertEqual(
            entity_voiceovers.language_accent_code,
            entity_voiceovers_model.language_accent_code,
        )
        manual_voiceover = entity_voiceovers.voiceovers_mapping['content_0'][
            'manual'
        ]
        automatic_voiceover = entity_voiceovers.voiceovers_mapping['content_0'][
            'auto'
        ]
        assert isinstance(automatic_voiceover, state_domain.Voiceover)
        assert isinstance(manual_voiceover, state_domain.Voiceover)
        self.assertDictEqual(
            manual_voiceover.to_dict(),
            entity_voiceovers_model.voiceovers_mapping['content_0']['manual'],
        )
        self.assertDictEqual(
            automatic_voiceover.to_dict(),
            entity_voiceovers_model.voiceovers_mapping['content_0']['auto'],
        )

    def test_should_get_entity_voiceovers_by_exploration(self) -> None:
        dummy_manual_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename1.mp3',
            'file_size_bytes': 3000,
            'needs_update': False,
            'duration_secs': 6.1,
        }
        dummy_autogenerated_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename2.mp3',
            'file_size_bytes': 3500,
            'needs_update': False,
            'duration_secs': 5.9,
        }
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id_1',
            1,
            'en-US',
            {
                'content_0': {
                    'manual': dummy_manual_voiceover_dict,
                    'auto': dummy_autogenerated_voiceover_dict,
                }
            },
            {},
        ).put()
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id_2',
            1,
            'en-US',
            {
                'content_0': {
                    'manual': dummy_manual_voiceover_dict,
                    'auto': dummy_autogenerated_voiceover_dict,
                }
            },
            {},
        ).put()
        retrieved_entity_voiceovers = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                'exp_id_1', 'exploration', 1
            )
        )

        self.assertEqual(len(retrieved_entity_voiceovers), 1)

    def test_should_get_entity_voiceovers_by_exploration_and_language_code(
        self,
    ) -> None:
        dummy_manual_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename1.mp3',
            'file_size_bytes': 3000,
            'needs_update': False,
            'duration_secs': 6.1,
        }
        dummy_autogenerated_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename2.mp3',
            'file_size_bytes': 3500,
            'needs_update': False,
            'duration_secs': 5.9,
        }
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id_1',
            1,
            'en-US',
            {
                'content_0': {
                    'manual': dummy_manual_voiceover_dict,
                    'auto': dummy_autogenerated_voiceover_dict,
                }
            },
            {},
        ).put()
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id_1',
            1,
            'en-IN',
            {
                'content_0': {
                    'manual': dummy_manual_voiceover_dict,
                    'auto': dummy_autogenerated_voiceover_dict,
                }
            },
            {},
        ).put()
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id_1',
            1,
            'en-GB',
            {
                'content_0': {
                    'manual': dummy_manual_voiceover_dict,
                    'auto': dummy_autogenerated_voiceover_dict,
                }
            },
            {},
        ).put()
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION, 'exp_id_1', 1, 'en-NG', {}, {}
        ).put()
        language_codes_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True, 'en-IN': True, 'en-NG': True},
            'hi': {'hi-IN': False},
        }
        voiceover_services.save_language_accent_support(
            language_codes_mapping=language_codes_mapping
        )

        retrieved_entity_voiceovers = (
            voiceover_services.fetch_entity_voiceovers_by_language_code(
                'exp_id_1', 'exploration', 1, 'en'
            )
        )

        self.assertEqual(len(retrieved_entity_voiceovers), 2)

    def test_should_compute_entity_voiceovers_model_successfully(self) -> None:
        retrieved_entity_voiceovers = (
            voiceover_services.fetch_entity_voiceovers_by_language_code(
                'exp_id_1', 'exploration', 1, 'en'
            )
        )
        self.assertEqual(len(retrieved_entity_voiceovers), 0)

        manual_voiceover_1: state_domain.VoiceoverDict = {
            'filename': 'filename1.mp3',
            'file_size_bytes': 3000,
            'needs_update': False,
            'duration_secs': 6.1,
        }
        manual_voiceover_2: state_domain.VoiceoverDict = {
            'filename': 'filename2.mp3',
            'file_size_bytes': 3500,
            'needs_update': False,
            'duration_secs': 5.9,
        }

        exploration = exp_domain.Exploration.create_default_exploration(
            'exp_id_1', title='Title', category='Category'
        )
        exploration.add_states(['New state'])

        voiceover_changes = [
            exp_domain.ExplorationChange(
                {
                    'cmd': 'update_voiceovers',
                    'language_accent_code': 'en-US',
                    'content_id': 'content_0',
                    'voiceovers': {'manual': manual_voiceover_1},
                }
            ),
            exp_domain.ExplorationChange(
                {
                    'cmd': 'update_voiceovers',
                    'language_accent_code': 'en-US',
                    'content_id': 'default_outcome_1',
                    'voiceovers': {'manual': manual_voiceover_2},
                }
            ),
        ]

        entity_voiceovers_models = (
            voiceover_services.compute_voiceover_related_change(
                updated_exploration=exploration,
                voiceover_changes=voiceover_changes,
            )
        )

        voiceover_models.EntityVoiceoversModel.update_timestamps_multi(
            entity_voiceovers_models
        )
        voiceover_models.EntityVoiceoversModel.put_multi(
            entity_voiceovers_models
        )

        exploration.version += 1

        self.assertEqual(len(entity_voiceovers_models), 1)

        retrieved_voiceover_1 = entity_voiceovers_models[0].voiceovers_mapping[
            'content_0'
        ]['manual']
        retrieved_voiceover_2 = entity_voiceovers_models[0].voiceovers_mapping[
            'default_outcome_1'
        ]['manual']

        self.assertDictEqual(retrieved_voiceover_1, manual_voiceover_1)
        self.assertDictEqual(retrieved_voiceover_2, manual_voiceover_2)

        manual_voiceover_3: state_domain.VoiceoverDict = {
            'filename': 'filename1.mp3',
            'file_size_bytes': 3000,
            'needs_update': True,
            'duration_secs': 6.1,
        }

        voiceover_changes = [
            exp_domain.ExplorationChange(
                {
                    'cmd': 'update_voiceovers',
                    'language_accent_code': 'en-US',
                    'content_id': 'content_0',
                    'voiceovers': {'manual': manual_voiceover_3},
                }
            ),
            exp_domain.ExplorationChange(
                {
                    'cmd': 'update_voiceovers',
                    'language_accent_code': 'en-US',
                    'content_id': 'default_outcome_1',
                    'voiceovers': {},
                }
            ),
        ]

        entity_voiceovers_models = (
            voiceover_services.compute_voiceover_related_change(
                updated_exploration=exploration,
                voiceover_changes=voiceover_changes,
            )
        )

        self.assertEqual(len(entity_voiceovers_models), 1)

        retrieved_voiceover_3 = entity_voiceovers_models[0].voiceovers_mapping[
            'content_0'
        ]['manual']

        self.assertDictEqual(retrieved_voiceover_3, manual_voiceover_3)

    def test_should_mark_entity_voiceovers_as_needing_update(self) -> None:
        exploration = exp_domain.Exploration.create_default_exploration(
            'exp_id_1', title='Title', category='Category'
        )
        exploration.version += 1
        exploration.add_states(['New state'])
        exploration.version += 1

        dummy_manual_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename1.mp3',
            'file_size_bytes': 3000,
            'needs_update': False,
            'duration_secs': 6.1,
        }
        dummy_autogenerated_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename2.mp3',
            'file_size_bytes': 3500,
            'needs_update': False,
            'duration_secs': 5.9,
        }
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id_1',
            1,
            'en-US',
            {
                'content_0': {
                    'manual': (dummy_manual_voiceover_dict),
                    'auto': (dummy_autogenerated_voiceover_dict),
                }
            },
            {},
        ).put()

        language_codes_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True},
            'hi': {'hi-IN': False},
        }

        voiceover_services.save_language_accent_support(
            language_codes_mapping=language_codes_mapping
        )

        voiceover_change_list = [
            exp_domain.ExplorationChange(
                {
                    'cmd': 'mark_voiceovers_needs_update',
                    'language_code': 'en',
                    'content_id': 'content_0',
                }
            ),
            exp_domain.ExplorationChange(
                {
                    'cmd': 'mark_voiceovers_needs_update',
                    'language_code': 'hi',
                    'content_id': 'content_0',
                }
            ),
        ]

        entity_voiceovers_models = (
            voiceover_services.compute_voiceover_related_change(
                updated_exploration=exploration,
                voiceover_changes=voiceover_change_list,
            )
        )

        voiceover_models.EntityVoiceoversModel.update_timestamps_multi(
            entity_voiceovers_models
        )
        voiceover_models.EntityVoiceoversModel.put_multi(
            entity_voiceovers_models
        )

        retrieved_entity_voiceovers = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                'exploration', 'exp_id_1', 2, 'en-US'
            )
        )

        manual_voiceover = retrieved_entity_voiceovers.voiceovers_mapping[
            'content_0'
        ]['manual']
        # Ruling out the possibility of None for mypy type checking.
        assert manual_voiceover is not None

        auto_voiceover = retrieved_entity_voiceovers.voiceovers_mapping[
            'content_0'
        ]['auto']
        # Ruling out the possibility of None for mypy type checking.
        assert auto_voiceover is not None

        self.assertTrue(manual_voiceover.needs_update)
        self.assertTrue(auto_voiceover.needs_update)

    def test_should_remove_entity_voiceovers(self) -> None:
        exploration = exp_domain.Exploration.create_default_exploration(
            'exp_id_1', title='Title', category='Category'
        )
        exploration.version += 1
        exploration.add_states(['New state'])
        exploration.version += 1

        dummy_manual_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename1.mp3',
            'file_size_bytes': 3000,
            'needs_update': False,
            'duration_secs': 6.1,
        }
        dummy_autogenerated_voiceover_dict: state_domain.VoiceoverDict = {
            'filename': 'filename2.mp3',
            'file_size_bytes': 3500,
            'needs_update': False,
            'duration_secs': 5.9,
        }
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id_1',
            1,
            'en-US',
            {
                'content_0': {
                    'manual': (dummy_manual_voiceover_dict),
                    'auto': (dummy_autogenerated_voiceover_dict),
                }
            },
            {},
        ).put()

        language_codes_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True},
            'hi': {'hi-IN': False},
        }

        voiceover_services.save_language_accent_support(
            language_codes_mapping=language_codes_mapping
        )

        voiceover_change_list = [
            exp_domain.ExplorationChange(
                {
                    'cmd': 'remove_voiceovers',
                    'language_code': 'en',
                    'content_id': 'content_0',
                }
            ),
            exp_domain.ExplorationChange(
                {
                    'cmd': 'remove_voiceovers',
                    'language_code': 'hi',
                    'content_id': 'content_0',
                }
            ),
        ]

        entity_voiceovers_models = (
            voiceover_services.compute_voiceover_related_change(
                updated_exploration=exploration,
                voiceover_changes=voiceover_change_list,
            )
        )

        voiceover_models.EntityVoiceoversModel.update_timestamps_multi(
            entity_voiceovers_models
        )
        voiceover_models.EntityVoiceoversModel.put_multi(
            entity_voiceovers_models
        )

        retrieved_entity_voiceovers = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                'exploration', 'exp_id_1', 2, 'en-US'
            )
        )
        self.assertNotIn(
            'content_0', retrieved_entity_voiceovers.voiceovers_mapping
        )

    def test_should_get_entity_voiceovers_for_reverted_version(self) -> None:
        exploration = exp_domain.Exploration.create_default_exploration(
            'exp_id_1', title='Title', category='Category'
        )
        exploration.add_states(['New state'])
        exploration.version += 1

        dummy_manual_voiceover_dict_1: state_domain.VoiceoverDict = {
            'filename': 'filename1.mp3',
            'file_size_bytes': 3000,
            'needs_update': False,
            'duration_secs': 6.1,
        }
        dummy_manual_voiceover_dict_2: state_domain.VoiceoverDict = {
            'filename': 'filename2.mp3',
            'file_size_bytes': 3500,
            'needs_update': False,
            'duration_secs': 5.9,
        }
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id_1',
            1,
            'en-US',
            {
                'content_0': {
                    'manual': (dummy_manual_voiceover_dict_1),
                    'auto': (None),
                }
            },
            {},
        ).put()

        # Updating the version of exploration and entity voiceovers.
        exploration.version += 1
        voiceover_models.EntityVoiceoversModel.create_new(
            feconf.ENTITY_TYPE_EXPLORATION,
            'exp_id_1',
            2,
            'en-US',
            {
                'content_0': {
                    'manual': (dummy_manual_voiceover_dict_2),
                    'auto': (None),
                }
            },
            {},
        ).put()

        entity_voiceovers_id = (
            voiceover_models.EntityVoiceoversModel.generate_id(
                feconf.ENTITY_TYPE_EXPLORATION, 'exp_id_1', 2, 'en-US'
            )
        )

        entity_voicovers_model = voiceover_models.EntityVoiceoversModel.get(
            entity_voiceovers_id
        )
        self.assertEqual(entity_voicovers_model.entity_version, 2)

        # Reverting exploration from version 2 to version 1 and the new
        # version is 3.
        exploration.version += 1
        new_entity_voiceovers_models = (
            voiceover_services.compute_voiceover_related_changes_upon_revert(
                exploration, 1
            )
        )
        new_entity_voiceovers_models[0].update_timestamps()
        new_entity_voiceovers_models[0].put()

        entity_voiceovers_id = (
            voiceover_models.EntityVoiceoversModel.generate_id(
                feconf.ENTITY_TYPE_EXPLORATION, 'exp_id_1', 3, 'en-US'
            )
        )
        entity_voicovers_model = voiceover_models.EntityVoiceoversModel.get(
            entity_voiceovers_id
        )

        self.assertEqual(entity_voicovers_model.entity_version, 3)
        self.assertEqual(
            entity_voicovers_model.voiceovers_mapping['content_0']['manual'][
                'filename'
            ],
            'filename1.mp3',
        )


class VoiceoverAutogenerationPolicyTests(test_utils.GenericTestBase):
    """Unit tests to validate the saving and fetching behavior in the
    VoiceoverAutogenerationPolicyModel.
    """

    def test_save_and_get_language_accent_codes_works_correctly(self) -> None:
        language_codes_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True},
            'hi': {'hi-IN': False},
        }
        retrieved_language_codes_mapping: Dict[str, Dict[str, bool]] = (
            voiceover_services.get_all_language_accent_codes_for_voiceovers()
        )
        expected_language_codes_mapping: Dict[str, Dict[str, bool]] = {}
        self.assertDictEqual(
            retrieved_language_codes_mapping, expected_language_codes_mapping
        )

        voiceover_services.save_language_accent_support(
            language_codes_mapping=language_codes_mapping
        )

        retrieved_language_codes_mapping = (
            voiceover_services.get_all_language_accent_codes_for_voiceovers()
        )
        self.assertDictEqual(
            retrieved_language_codes_mapping, language_codes_mapping
        )

    def test_get_and_set_azure_config_for_automatic_voiceovers(self) -> None:
        voiceover_autogeneration_policy_model = (
            voiceover_models.VoiceoverAutogenerationPolicyModel(
                id=voiceover_models.VOICEOVER_AUTOGENERATION_POLICY_ID
            )
        )
        voiceover_autogeneration_policy_model.language_codes_mapping = {}
        (
            voiceover_autogeneration_policy_model.autogenerated_voiceovers_are_enabled
        ) = True
        voiceover_autogeneration_policy_model.update_timestamps()
        voiceover_autogeneration_policy_model.put()

        self.assertTrue(
            voiceover_services.is_voiceover_autogeneration_using_cloud_service_enabled()
        )

        voiceover_services.update_admin_config_for_voiceover_autogeneration(
            False
        )

        self.assertFalse(
            voiceover_services.is_voiceover_autogeneration_using_cloud_service_enabled()
        )

    def test_should_successfully_get_autogeneratable_accents(self) -> None:
        language_codes_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True, 'en-IN': True, 'en-NG': False},
            'hi': {'hi-IN': False},
        }
        voiceover_services.save_language_accent_support(
            language_codes_mapping=language_codes_mapping
        )

        autogeneratable_accents_for_english = voiceover_services.get_supported_autogeneratable_accents_by_language(
            'en'
        )
        self.assertItemsEqual(
            autogeneratable_accents_for_english, ['en-US', 'en-IN']
        )

        autogeneratable_accents_for_hindi = voiceover_services.get_supported_autogeneratable_accents_by_language(
            'hi'
        )
        self.assertItemsEqual(autogeneratable_accents_for_hindi, [])

    def test_get_new_auto_voiceover_accent_returns_new_enabled_accent(
        self,
    ) -> None:
        existing_language_accent_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True},
            'hi': {'hi-IN': False},
        }
        voiceover_services.save_language_accent_support(
            language_codes_mapping=existing_language_accent_mapping
        )
        updated_language_accent_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True, 'en-IN': True},
            'hi': {'hi-IN': False},
        }

        new_accent_code = voiceover_services.get_new_auto_voiceover_accent(
            updated_language_accent_mapping
        )

        self.assertEqual(new_accent_code, 'en-IN')

    def test_get_new_auto_voiceover_accent_returns_enabled_existing_accent(
        self,
    ) -> None:
        existing_language_accent_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True},
            'hi': {'hi-IN': False},
        }
        voiceover_services.save_language_accent_support(
            language_codes_mapping=existing_language_accent_mapping
        )
        updated_language_accent_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True},
            'hi': {'hi-IN': True},
        }

        new_accent_code = voiceover_services.get_new_auto_voiceover_accent(
            updated_language_accent_mapping
        )

        self.assertEqual(new_accent_code, 'hi-IN')

    def test_get_new_auto_voiceover_accent_returns_none_when_no_new_enabled_accent(
        self,
    ) -> None:
        existing_language_accent_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True},
        }
        voiceover_services.save_language_accent_support(
            language_codes_mapping=existing_language_accent_mapping
        )
        updated_language_accent_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True, 'en-IN': False},
        }

        new_accent_code = voiceover_services.get_new_auto_voiceover_accent(
            updated_language_accent_mapping
        )

        self.assertIsNone(new_accent_code)


class VoiceoversLanguageAccentConstantsTests(test_utils.GenericTestBase):
    """Unit tests to validate the language-accent information saved as
    constants in the JSON files.
    """

    def test_get_language_accent_master_list_works_correctly(self) -> None:
        file_path = os.path.join('assets', 'language_accent_master_list.json')

        with open(file_path, 'r', encoding='utf-8') as f:
            language_accent_master_list: Dict[str, Dict[str, str]] = json.loads(
                f.read()
            )
            expected_language_accent_master_list = (
                voiceover_services.get_language_accent_master_list()
            )

            self.assertDictEqual(
                language_accent_master_list,
                expected_language_accent_master_list,
            )

    def test_get_autogeneratable_language_accent_list_works_correctly(
        self,
    ) -> None:
        file_path = os.path.join(
            'assets',
            'autogeneratable_language_accent_list.json',
        )
        with open(file_path, 'r', encoding='utf-8') as f:
            autogeneratable_language_accent_list: Dict[str, Dict[str, str]] = (
                json.loads(f.read())
            )
            expected_autogeneratable_language_accent_list = (
                voiceover_services.get_autogeneratable_language_accent_list()
            )

            self.assertDictEqual(
                autogeneratable_language_accent_list,
                expected_autogeneratable_language_accent_list,
            )

    def test_should_get_available_autogeneratable_language_accent_list(
        self,
    ) -> None:
        available_language_accent_list = (
            voiceover_services.get_autogeneratable_language_accent_codes()
        )
        # All language accent codes supported for autogenerated voiceovers are
        # listed in: data/voiceovers/autogeneratable_language_accent_list.json.
        number_of_available_accent_codes = 143
        self.assertEqual(
            len(available_language_accent_list),
            number_of_available_accent_codes,
        )

    def test_validate_data_format_stored_in_master_list_is_correct(
        self,
    ) -> None:
        valid_schema = {
            'type': 'variable_keys_dict',
            'keys': {'schema': {'type': 'basestring'}},
            'values': {
                'schema': {
                    'type': 'variable_keys_dict',
                    'keys': {'schema': {'type': 'basestring'}},
                    'values': {'schema': {'type': 'basestring'}},
                }
            },
        }
        try:
            schema_utils.normalize_against_schema(
                voiceover_services.get_language_accent_master_list(),
                valid_schema,
            )
        except Exception:
            self.fail(
                'get_language_accent_master_list() raised '
                'ExceptionType unexpectedly!'
            )

    def test_validate_data_format_stored_in_autogeneratable_list_is_correct(
        self,
    ) -> None:
        valid_schema = {
            'type': 'variable_keys_dict',
            'keys': {'schema': {'type': 'basestring'}},
            'values': {
                'schema': {
                    'type': 'dict',
                    'properties': [
                        {'name': 'service', 'schema': {'type': 'basestring'}},
                        {
                            'name': 'voice_code',
                            'schema': {'type': 'basestring'},
                        },
                    ],
                }
            },
        }
        try:
            schema_utils.normalize_against_schema(
                voiceover_services.get_autogeneratable_language_accent_list(),
                valid_schema,
            )
        except Exception:
            self.fail(
                'get_autogeneratable_language_accent_list() raised '
                'ExceptionType unexpectedly!'
            )

    def test_validate_autogeneratable_list_is_subset_of_master_list(
        self,
    ) -> None:
        language_accent_master_list = []
        for (
            accent_code_to_description
        ) in voiceover_services.get_language_accent_master_list().values():
            for lang_accent_code in accent_code_to_description.keys():
                language_accent_master_list.append(lang_accent_code)
        autogeneratable_language_accent_codes = list(
            voiceover_services.get_autogeneratable_language_accent_list().keys()
        )

        self.assertTrue(
            set(language_accent_master_list).issuperset(
                set(autogeneratable_language_accent_codes)
            )
        )

    def test_should_get_correct_language_code_for_given_accent(self) -> None:
        language_accent_code = 'en-US'
        expected_language_code = 'en'

        self.assertEqual(
            voiceover_services.get_language_code_from_language_accent_code(
                language_accent_code
            ),
            expected_language_code,
        )

        language_accent_code = 'hi-IN'
        expected_language_code = 'hi'

        self.assertEqual(
            voiceover_services.get_language_code_from_language_accent_code(
                language_accent_code
            ),
            expected_language_code,
        )

    def test_validate_language_accent_code_for_autogeneration(self) -> None:
        invalid_accent_code = 'en-XX'
        self.assertFalse(
            voiceover_services.is_accent_code_valid_for_autogeneration(
                invalid_accent_code
            )
        )

        # Here we use MyPy ignore because here we assign type int to
        # type str. This is done to test the validation of the
        # is_accent_code_valid_for_autogeneration method.
        invalid_accent_code = 5  # type: ignore[assignment]
        self.assertFalse(
            voiceover_services.is_accent_code_valid_for_autogeneration(
                invalid_accent_code
            )
        )

        valid_accent_code = 'en-US'
        self.assertTrue(
            voiceover_services.is_accent_code_valid_for_autogeneration(
                valid_accent_code
            )
        )


class VoiceoverRegenerationTests(test_utils.GenericTestBase):
    """Test class to verify voiceover regeneration across various scenarios,
    such as exploration updates or translation updates.
    """

    def setUp(self) -> None:
        super().setUp()
        self.signup(self.VIEWER_EMAIL, self.VIEWER_USERNAME)
        self.committer_1_id = self.get_user_id_from_email(self.VIEWER_EMAIL)

        self.old_content_html = '<p>old content html</p>'
        self.new_content_html = '<p>new content html</p>'

        self.voiceover_autogeneration_policy_model = (
            voiceover_models.VoiceoverAutogenerationPolicyModel(
                id=voiceover_models.VOICEOVER_AUTOGENERATION_POLICY_ID
            )
        )
        self.voiceover_autogeneration_policy_model.language_codes_mapping = {
            'en': {'en-US': True, 'en-IN': True, 'en-NG': False},
            'hi': {'hi-IN': False},
            'ar': {
                'ar-AE': True,
            },
        }
        (
            self.voiceover_autogeneration_policy_model.autogenerated_voiceovers_are_enabled
        ) = True
        self.voiceover_autogeneration_policy_model.update_timestamps()
        self.voiceover_autogeneration_policy_model.put()

    def mock_send_email_to_voiceover_admins_and_tech_leads(
        self,
        exploration_id: str,
        exploration_title: str,
        date_time: str,
        language_accents_used_for_voiceover_regeneration: List[str],
        error_collections_during_voiceover_regeneration: List[
            Dict[str, List[Tuple[str, str]] | str]
        ],
        number_of_contents_for_voiceover_regeneration: int,
        number_of_contents_failed_to_regenerate: int,
        author_id: str,
    ) -> None:
        """Mock function to simulate sending email to voiceover admins and
        tech leads after voiceover regeneration.
        """
        pass

    def test_should_regenerate_voiceover_for_curated_exploration_content_update(
        self,
    ) -> None:
        exploration_id = 'exp_id_1'
        exploration_version = 2
        self.signup('tester@org.com', 'tester')

        commit1 = exp_models.ExplorationCommitLogEntryModel.create(
            exploration_id,
            2,
            self.committer_1_id,
            'msg',
            'create',
            [
                {
                    'cmd': exp_domain.CMD_EDIT_STATE_PROPERTY,
                    'property_name': exp_domain.STATE_PROPERTY_CONTENT,
                    'state_name': 'State 1',
                    'old_value': {
                        'content_id': 'content_5',
                        'html': self.old_content_html,
                    },
                    'new_value': {
                        'content_id': 'content_5',
                        'html': self.new_content_html,
                    },
                }
            ],
            constants.ACTIVITY_STATUS_PRIVATE,
            False,
        )

        commit1.exploration_id = exploration_id
        commit1.update_timestamps()
        commit1.put()

        entity_voiceovers_models = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id, 'exploration', exploration_version
            )
        )
        self.assertEqual(len(entity_voiceovers_models), 0)

        parent_cloud_task_model_id = 'cloud_task_model_id'
        task_name = 'projects/%s/locations/%s/queues/%s/tasks/%s' % (
            'dev-project-id',
            'us-central',
            'voiceover-regeneration',
            uuid.uuid4().hex,
        )
        function_id = 'regenerate_voiceovers_on_exploration_update'
        taskqueue_services.create_new_cloud_task_model(
            parent_cloud_task_model_id, task_name, function_id
        )
        voiceover_services.regenerate_voiceovers_on_exploration_update(
            exploration_id, exploration_version, parent_cloud_task_model_id
        )
        updated_cloud_task_runs = sorted(
            taskqueue_services.get_all_cloud_task_runs(),
            key=lambda task_run: task_run.created_on,
        )
        for cloud_run in updated_cloud_task_runs:
            if (
                cloud_run.function_id
                == 'regenerate_voiceovers_for_batch_contents'
            ):
                voiceover_services.regenerate_voiceovers_for_batch_contents(
                    exploration_id,
                    parent_cloud_task_model_id,
                    cloud_run.task_run_id,
                )

        entity_voiceovers_models = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id, 'exploration', exploration_version
            )
        )
        self.assertEqual(len(entity_voiceovers_models), 2)
        self.assertItemsEqual(
            [
                entity_voiceovers_models[0].language_accent_code,
                entity_voiceovers_models[1].language_accent_code,
            ],
            ['en-US', 'en-IN'],
        )

    def test_should_regenerate_voiceover_for_translation_update(self) -> None:
        exploration_id = 'exp_id_1'
        exploration_version = 2
        self.signup('tester@org.com', 'tester')

        commit1 = exp_models.ExplorationCommitLogEntryModel.create(
            exploration_id,
            2,
            self.committer_1_id,
            'msg',
            'create',
            [
                {
                    'cmd': exp_domain.CMD_EDIT_TRANSLATION,
                    'content_id': 'content_5',
                    'language_code': 'ar',
                    'translation': translation_domain.TranslatedContent(
                        'Updated translations in Hindi!',
                        translation_domain.TranslatableContentFormat.HTML,
                        False,
                    ).to_dict(),
                }
            ],
            constants.ACTIVITY_STATUS_PRIVATE,
            False,
        )

        commit1.exploration_id = exploration_id
        commit1.update_timestamps()
        commit1.put()

        entity_voiceovers_models = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id, 'exploration', exploration_version
            )
        )
        self.assertEqual(len(entity_voiceovers_models), 0)

        parent_cloud_task_model_id = 'cloud_task_model_id'
        task_name = 'projects/%s/locations/%s/queues/%s/tasks/%s' % (
            'dev-project-id',
            'us-central',
            'voiceover-regeneration',
            uuid.uuid4().hex,
        )
        function_id = 'regenerate_voiceovers_on_exploration_update'
        taskqueue_services.create_new_cloud_task_model(
            parent_cloud_task_model_id, task_name, function_id
        )
        voiceover_services.regenerate_voiceovers_on_exploration_update(
            exploration_id, exploration_version, parent_cloud_task_model_id
        )
        updated_cloud_task_runs = sorted(
            taskqueue_services.get_all_cloud_task_runs(),
            key=lambda task_run: task_run.created_on,
        )
        for cloud_run in updated_cloud_task_runs:
            if (
                cloud_run.function_id
                == 'regenerate_voiceovers_for_batch_contents'
            ):
                voiceover_services.regenerate_voiceovers_for_batch_contents(
                    exploration_id,
                    parent_cloud_task_model_id,
                    cloud_run.task_run_id,
                )

        entity_voiceovers_models = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id, 'exploration', exploration_version
            )
        )
        self.assertEqual(len(entity_voiceovers_models), 1)
        self.assertItemsEqual(
            [entity_voiceovers_models[0].language_accent_code], ['ar-AE']
        )

    def test_should_raise_exception_when_change_diff_is_not_accessible(
        self,
    ) -> None:
        exploration_id = 'exp_id_1'
        exploration_version = 2

        error = (
            'Could not fetch change diff for exploration %s, version %s during '
            'voiceover regeneration.'
            % (exploration_id, str(exploration_version))
        )

        parent_cloud_task_model_id = 'cloud_task_model_id'
        task_name = 'projects/%s/locations/%s/queues/%s/tasks/%s' % (
            'dev-project-id',
            'us-central',
            'voiceover-regeneration',
            uuid.uuid4().hex,
        )
        function_id = 'regenerate_voiceovers_on_exploration_update'
        taskqueue_services.create_new_cloud_task_model(
            parent_cloud_task_model_id, task_name, function_id
        )

        with self.assertRaisesRegex(Exception, error):
            voiceover_services.regenerate_voiceovers_on_exploration_update(
                exploration_id, exploration_version, parent_cloud_task_model_id
            )

    def test_should_not_regenerate_voiceover_for_non_supported_accents(
        self,
    ) -> None:
        exploration_id = 'exp_id_1'
        exploration_version = 2
        self.signup('tester@org.com', 'tester')

        commit1 = exp_models.ExplorationCommitLogEntryModel.create(
            exploration_id,
            2,
            self.committer_1_id,
            'msg',
            'create',
            [
                {
                    'cmd': exp_domain.CMD_EDIT_STATE_PROPERTY,
                    'property_name': exp_domain.STATE_PROPERTY_CONTENT,
                    'state_name': 'State 1',
                    'old_value': {
                        'content_id': 'content_5',
                        'html': self.old_content_html,
                    },
                    'new_value': {
                        'content_id': 'content_5',
                        'html': self.new_content_html,
                    },
                }
            ],
            constants.ACTIVITY_STATUS_PRIVATE,
            False,
        )

        commit1.exploration_id = exploration_id
        commit1.update_timestamps()
        commit1.put()

        self.voiceover_autogeneration_policy_model.language_codes_mapping = {
            'en': {'en-US': False}
        }

        entity_voiceovers_models = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id, 'exploration', exploration_version
            )
        )
        self.assertEqual(len(entity_voiceovers_models), 0)

        parent_cloud_task_model_id = 'cloud_task_model_id'
        task_name = 'projects/%s/locations/%s/queues/%s/tasks/%s' % (
            'dev-project-id',
            'us-central',
            'voiceover-regeneration',
            uuid.uuid4().hex,
        )
        function_id = 'regenerate_voiceovers_on_exploration_update'
        taskqueue_services.create_new_cloud_task_model(
            parent_cloud_task_model_id, task_name, function_id
        )
        voiceover_services.regenerate_voiceovers_on_exploration_update(
            exploration_id, exploration_version, parent_cloud_task_model_id
        )
        updated_cloud_task_runs = sorted(
            taskqueue_services.get_all_cloud_task_runs(),
            key=lambda task_run: task_run.created_on,
        )
        for cloud_run in updated_cloud_task_runs:
            if (
                cloud_run.function_id
                == 'regenerate_voiceovers_for_batch_contents'
            ):
                voiceover_services.regenerate_voiceovers_for_batch_contents(
                    exploration_id,
                    parent_cloud_task_model_id,
                    cloud_run.task_run_id,
                )

        entity_voiceovers_models = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id, 'exploration', exploration_version
            )
        )

        # Since the autogeneration is disabled for 'en-US', no voiceovers
        # should be regenerated.
        self.assertEqual(len(entity_voiceovers_models), 0)

    @test_utils.set_platform_parameters(
        [
            (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@system.com'),
            (param_list.ParamName.EMAIL_FOOTER, 'dummy_footer'),
            (param_list.ParamName.EMAIL_SENDER_NAME, 'admin'),
            (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'),
            (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True),
            (param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, 'subject'),
            (param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, 'body'),
            (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'dummy@system.com'),
        ]
    )
    def test_should_send_emails_to_voiceover_admins_and_tech_leads(
        self,
    ) -> None:
        exploration_id = 'exp_id_1'
        exploration_title = 'Test Exploration'
        date_time = '2025-07-27T21:42:29.864077'
        number_of_contents_for_voiceover_regeneration = 10
        number_of_contents_failed_to_regenerate = 2
        language_accents_used_for_voiceover_regeneration = [
            'en-US',
            'en-IN',
            'ar-AE',
        ]
        self.signup('tester@org.com', 'tester')
        author_id = self.get_user_id_from_email('tester@org.com')

        error_collections_during_voiceover_regeneration: List[
            Dict[str, List[Tuple[str, str]] | str]
        ] = [
            {
                'exploration_id': exploration_id,
                'language_accent_code': 'en-US',
                'error_messages': [
                    ('content0', 'Error 1 occurred'),
                    ('content1', 'Error 2 occurred'),
                ],
            },
            {
                'exploration_id': exploration_id,
                'language_accent_code': 'en-IN',
                'error_messages': [('content2', 'Error 3 occurred')],
            },
        ]

        (
            voiceover_services.send_email_to_voiceover_admins_and_tech_leads_after_regeneration(
                exploration_id,
                exploration_title,
                date_time,
                language_accents_used_for_voiceover_regeneration,
                error_collections_during_voiceover_regeneration,
                number_of_contents_for_voiceover_regeneration,
                number_of_contents_failed_to_regenerate,
                author_id=author_id,
            )
        )

        # Make sure correct email model is stored.
        all_models: Sequence[email_models.SentEmailModel] = (
            email_models.SentEmailModel.get_all().fetch()
        )
        self.assertEqual(len(all_models), 3)

    @test_utils.set_platform_parameters(
        [
            (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@system.com'),
            (param_list.ParamName.EMAIL_FOOTER, 'dummy_footer'),
            (param_list.ParamName.EMAIL_SENDER_NAME, 'admin'),
            (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'),
            (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True),
            (param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, 'subject'),
            (param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, 'body'),
            (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'dummy@system.com'),
        ]
    )
    def test_should_raise_error_while_regenerating_voiceover(self) -> None:
        exploration_id = 'exp_id_1'
        exploration_version = 2
        commit1 = exp_models.ExplorationCommitLogEntryModel.create(
            exploration_id,
            2,
            self.committer_1_id,
            'msg',
            'create',
            [
                {
                    'cmd': exp_domain.CMD_EDIT_STATE_PROPERTY,
                    'property_name': exp_domain.STATE_PROPERTY_CONTENT,
                    'state_name': 'State 1',
                    'old_value': {
                        'content_id': 'content_5',
                        'html': self.old_content_html,
                    },
                    'new_value': {
                        'content_id': 'content_5',
                        'html': self.new_content_html,
                    },
                }
            ],
            constants.ACTIVITY_STATUS_PRIVATE,
            False,
        )

        commit1.exploration_id = exploration_id
        commit1.update_timestamps()
        commit1.put()

        self.voiceover_autogeneration_policy_model.language_codes_mapping = {
            'en': {'en-US': True}
        }

        entity_voiceovers_models = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id, 'exploration', exploration_version
            )
        )
        self.assertEqual(len(entity_voiceovers_models), 0)

        def mock_regenerate_voiceovers_of_exploration(
            _exploration_id: str,
            _exploration_version: int,
            _content_id_to_content_html: Dict[str, str],
            _language_accent_code: str,
        ) -> List[Tuple[str, str]]:
            errors_while_voiceover_regeneration = [
                ('content_5', 'Error 1 occurred'),
            ]
            return errors_while_voiceover_regeneration

        parent_cloud_task_model_id = 'cloud_task_model_id'
        task_name = 'projects/%s/locations/%s/queues/%s/tasks/%s' % (
            'dev-project-id',
            'us-central',
            'voiceover-regeneration',
            uuid.uuid4().hex,
        )
        function_id = 'regenerate_voiceovers_on_exploration_update'
        taskqueue_services.create_new_cloud_task_model(
            parent_cloud_task_model_id, task_name, function_id
        )

        voiceover_services.regenerate_voiceovers_on_exploration_update(
            exploration_id, exploration_version, parent_cloud_task_model_id
        )

        updated_cloud_task_runs = sorted(
            taskqueue_services.get_all_cloud_task_runs(),
            key=lambda task_run: task_run.created_on,
        )

        with self.swap(
            voiceover_regeneration_services,
            'regenerate_voiceovers_of_exploration',
            mock_regenerate_voiceovers_of_exploration,
        ):
            for cloud_run in updated_cloud_task_runs:
                if (
                    cloud_run.function_id
                    == 'regenerate_voiceovers_for_batch_contents'
                ):
                    voiceover_services.regenerate_voiceovers_for_batch_contents(
                        exploration_id,
                        parent_cloud_task_model_id,
                        cloud_run.task_run_id,
                    )

        all_models: Sequence[email_models.SentEmailModel] = (
            email_models.SentEmailModel.get_all().fetch()
        )

        self.assertEqual(len(all_models), 2)

        expected_html_body = (
            'Hi Voiceover Admins,<br><br>tester has initiated the generation '
            'of automatic voiceovers for the exploration titled '
            '<a href="https://www.oppia.org/create/exp_id_1">Test Exploration'
            '</a> in language(s) English (United States).<br>This generation '
            'was completed at 2025-08-01 08:35:05. Below is a summary of the '
            'voiceover synthesis status.<br><br><ul><li>Total contents for '
            'speech synthesis: 1.</li><li>Successfully generated voiceovers: '
            '0.</li><li>Failed contents for voiceover generation: 1.</li>'
            '</ul><br>You have also been cc’d on a separate email, sent to the '
            'voiceover tech lead, to address the failed voiceover synthesis. '
            'Please follow up on that email as needed.<br><br>Thank you.'
        )
        for email_model in all_models:
            if (
                email_model.subject
                == 'Report on Automatic Voiceovers Generated for Test Exploration'
            ):
                self.assertEqual(email_model.html_body, expected_html_body)

        updated_cloud_task_run_model = cloud_task_models.CloudTaskRunModel.get(
            parent_cloud_task_model_id
        )
        assert updated_cloud_task_run_model is not None
        self.assertEqual(
            updated_cloud_task_run_model.latest_job_state, 'PERMANENTLY_FAILED'
        )

    def test_should_raise_exception_while_regenerating_voiceovers_in_batch(
        self,
    ) -> None:
        exploration_id = 'exp_id_1'
        exploration_version = 2
        commit1 = exp_models.ExplorationCommitLogEntryModel.create(
            exploration_id,
            2,
            self.committer_1_id,
            'msg',
            'create',
            [
                {
                    'cmd': exp_domain.CMD_EDIT_STATE_PROPERTY,
                    'property_name': exp_domain.STATE_PROPERTY_CONTENT,
                    'state_name': 'State 1',
                    'old_value': {
                        'content_id': 'content_5',
                        'html': self.old_content_html,
                    },
                    'new_value': {
                        'content_id': 'content_5',
                        'html': self.new_content_html,
                    },
                }
            ],
            constants.ACTIVITY_STATUS_PRIVATE,
            False,
        )

        commit1.exploration_id = exploration_id
        commit1.update_timestamps()
        commit1.put()

        self.voiceover_autogeneration_policy_model.language_codes_mapping = {
            'en': {'en-US': True}
        }

        entity_voiceovers_models = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id, 'exploration', exploration_version
            )
        )
        self.assertEqual(len(entity_voiceovers_models), 0)

        def mock_regenerate_voiceovers_of_exploration(
            _exploration_id: str,
            _exploration_version: int,
            _content_id_to_content_html: Dict[str, str],
            _language_accent_code: str,
        ) -> List[Tuple[str, str]]:
            raise Exception(
                'Expected error raised during voiceover regeneration!'
            )

        parent_cloud_task_model_id = 'cloud_task_model_id'
        task_name = 'projects/%s/locations/%s/queues/%s/tasks/%s' % (
            'dev-project-id',
            'us-central',
            'voiceover-regeneration',
            uuid.uuid4().hex,
        )
        function_id = 'regenerate_voiceovers_on_exploration_update'
        taskqueue_services.create_new_cloud_task_model(
            parent_cloud_task_model_id, task_name, function_id
        )

        voiceover_services.regenerate_voiceovers_on_exploration_update(
            exploration_id, exploration_version, parent_cloud_task_model_id
        )

        updated_cloud_task_runs = sorted(
            taskqueue_services.get_all_cloud_task_runs(),
            key=lambda task_run: task_run.created_on,
        )
        with self.swap(
            voiceover_regeneration_services,
            'regenerate_voiceovers_of_exploration',
            mock_regenerate_voiceovers_of_exploration,
        ):
            for cloud_run in updated_cloud_task_runs:
                if (
                    cloud_run.function_id
                    == 'regenerate_voiceovers_for_batch_contents'
                ):
                    voiceover_services.regenerate_voiceovers_for_batch_contents(
                        exploration_id,
                        parent_cloud_task_model_id,
                        cloud_run.task_run_id,
                    )

    def _create_exploration_and_arabic_translation(
        self, exploration_id: str, language_code: str
    ) -> None:
        """Creates a new exploration with a translation in Arabic.

        Args:
            exploration_id: str. The ID of the exploration to create.
            language_code: str. The language code for the translation.
        """
        self.signup(self.OWNER_EMAIL, self.OWNER_USERNAME)
        owner_id = self.get_user_id_from_email(self.OWNER_EMAIL)

        exploration = exp_domain.Exploration.create_default_exploration(
            exploration_id,
            title='A Title',
            category='A Category',
            objective='An Objective',
        )
        exploration.states['Introduction'].content.html = 'First Card!'
        content_id_generator = translation_domain.ContentIdGenerator(
            exploration.next_content_id_index
        )
        self.set_interaction_for_state(
            exploration.states['Introduction'],
            'TextInput',
            content_id_generator,
        )
        exploration.states[
            feconf.DEFAULT_INIT_STATE_NAME
        ].interaction.answer_groups = [
            state_domain.AnswerGroup(
                state_domain.Outcome(
                    'Second',
                    None,
                    state_domain.SubtitledHtml(
                        'feedback_1', '<p>state outcome html</p>'
                    ),
                    False,
                    [],
                    None,
                    None,
                ),
                [
                    state_domain.RuleSpec(
                        'Equals',
                        {
                            'x': {
                                'contentId': 'rule_input_Equals',
                                'normalizedStrSet': ['Test'],
                            }
                        },
                    )
                ],
                [],
                None,
            )
        ]
        exploration.add_state('Second', 'content_2', 'content-3')

        exp_services.save_new_exploration(owner_id, exploration)

        arabic_translation = 'المحتوى المترجم'
        translations_mapping: Dict[str, feconf.TranslatedContentDict] = {
            'content_0': {
                'content_value': arabic_translation,
                'content_format': 'html',
                'needs_update': False,
            },
            'content_1': {
                'content_value': arabic_translation,
                'content_format': 'html',
                'needs_update': True,
            },
            'rule_input_Equals': {
                'content_value': arabic_translation,
                'content_format': 'html',
                'needs_update': False,
            },
            'feedback_1': {
                'content_value': arabic_translation,
                'content_format': 'html',
                'needs_update': False,
            },
        }

        translation_models.EntityTranslationsModel.create_new(
            feconf.TranslatableEntityType.EXPLORATION.value,
            exploration_id,
            exploration.version,
            language_code,
            translations_mapping,
        ).put()

    def test_should_regenerate_voiceover_for_arabic_language(self) -> None:
        language_accent_code = 'ar-AE'
        language_code = 'ar'
        exploration_id = 'exp_id_1'
        exploration_version = 1
        self.signup('tester@org.com', 'tester')

        self._create_exploration_and_arabic_translation(
            exploration_id, language_code
        )

        entity_translation = translation_fetchers.get_entity_translation(
            feconf.TranslatableEntityType.EXPLORATION,
            exploration_id,
            exploration_version,
            language_code,
        )
        entity_voiceovers = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                feconf.ENTITY_TYPE_EXPLORATION,
                exploration_id,
                exploration_version,
                language_accent_code,
            )
        )

        self.assertEqual(entity_translation.language_code, language_code)
        self.assertEqual(entity_voiceovers.voiceovers_mapping, {})

        parent_cloud_task_model_id = 'cloud_task_model_id'
        task_name = 'projects/%s/locations/%s/queues/%s/tasks/%s' % (
            'dev-project-id',
            'us-central',
            'voiceover-regeneration',
            uuid.uuid4().hex,
        )
        function_id = 'regenerate_voiceovers_on_exploration_update'
        taskqueue_services.create_new_cloud_task_model(
            parent_cloud_task_model_id, task_name, function_id
        )
        voiceover_services.regenerate_voiceovers_of_exploration_for_given_language_accent(
            exploration_id, language_accent_code, parent_cloud_task_model_id
        )
        updated_cloud_task_runs = sorted(
            taskqueue_services.get_all_cloud_task_runs(),
            key=lambda task_run: task_run.created_on,
        )
        for cloud_run in updated_cloud_task_runs:
            if (
                cloud_run.function_id
                == 'regenerate_voiceovers_for_batch_contents'
            ):
                voiceover_services.regenerate_voiceovers_for_batch_contents(
                    exploration_id,
                    parent_cloud_task_model_id,
                    cloud_run.task_run_id,
                )

        entity_voiceovers = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                feconf.ENTITY_TYPE_EXPLORATION,
                exploration_id,
                exploration_version,
                language_accent_code,
            )
        )

        self.assertEqual(
            entity_voiceovers.language_accent_code, language_accent_code
        )
        self.assertNotEqual(entity_voiceovers.voiceovers_mapping, {})
        self.assertDictEqual(
            entity_voiceovers.automated_voiceovers_audio_offsets_msecs,
            {
                'content_0': [
                    {'token': 'This', 'audio_offset_msecs': 0.0},
                    {'token': 'is', 'audio_offset_msecs': 100.0},
                    {'token': 'a', 'audio_offset_msecs': 200.0},
                    {'token': 'test', 'audio_offset_msecs': 300.0},
                    {'token': 'text', 'audio_offset_msecs': 400.0},
                ],
                'feedback_1': [
                    {'token': 'This', 'audio_offset_msecs': 0.0},
                    {'token': 'is', 'audio_offset_msecs': 100.0},
                    {'token': 'a', 'audio_offset_msecs': 200.0},
                    {'token': 'test', 'audio_offset_msecs': 300.0},
                    {'token': 'text', 'audio_offset_msecs': 400.0},
                ],
            },
        )

    def test_should_raise_exception_when_language_accent_code_is_not_supported(
        self,
    ) -> None:
        language_accent_code = 'ar-non-existent'
        exploration_id = 'exp_id_1'
        self.signup('tester@org.com', 'tester')

        with self.assertRaisesRegex(
            Exception, 'Invalid language accent code: %s' % language_accent_code
        ):
            (
                voiceover_services.regenerate_voiceovers_of_exploration_for_given_language_accent(
                    exploration_id, language_accent_code, 'cloud_task_run_id'
                )
            )

    def test_should_regenerate_voiceover_for_english_language(self) -> None:
        language_accent_code = 'en-US'
        language_code = 'en'
        exploration_id = 'exp_id_1'
        exploration_version = 1
        self.signup('tester@org.com', 'tester')

        self._create_exploration_and_arabic_translation(
            exploration_id, language_code
        )

        entity_voiceovers = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                feconf.ENTITY_TYPE_EXPLORATION,
                exploration_id,
                exploration_version,
                language_accent_code,
            )
        )
        self.assertEqual(entity_voiceovers.voiceovers_mapping, {})

        parent_cloud_task_model_id = 'cloud_task_model_id'
        task_name = 'projects/%s/locations/%s/queues/%s/tasks/%s' % (
            'dev-project-id',
            'us-central',
            'voiceover-regeneration',
            uuid.uuid4().hex,
        )
        function_id = 'regenerate_voiceovers_on_exploration_update'
        taskqueue_services.create_new_cloud_task_model(
            parent_cloud_task_model_id, task_name, function_id
        )
        voiceover_services.regenerate_voiceovers_of_exploration_for_given_language_accent(
            exploration_id, language_accent_code, parent_cloud_task_model_id
        )
        updated_cloud_task_runs = sorted(
            taskqueue_services.get_all_cloud_task_runs(),
            key=lambda task_run: task_run.created_on,
        )
        for cloud_run in updated_cloud_task_runs:
            if (
                cloud_run.function_id
                == 'regenerate_voiceovers_for_batch_contents'
            ):
                voiceover_services.regenerate_voiceovers_for_batch_contents(
                    exploration_id,
                    parent_cloud_task_model_id,
                    cloud_run.task_run_id,
                )

        entity_voiceovers = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                feconf.ENTITY_TYPE_EXPLORATION,
                exploration_id,
                exploration_version,
                language_accent_code,
            )
        )

        self.assertEqual(
            entity_voiceovers.language_accent_code, language_accent_code
        )
        self.assertNotEqual(entity_voiceovers.voiceovers_mapping, {})
        default_audio_offset = [
            {'token': 'This', 'audio_offset_msecs': 0.0},
            {'token': 'is', 'audio_offset_msecs': 100.0},
            {'token': 'a', 'audio_offset_msecs': 200.0},
            {'token': 'test', 'audio_offset_msecs': 300.0},
            {'token': 'text', 'audio_offset_msecs': 400.0},
        ]
        self.assertDictEqual(
            entity_voiceovers.automated_voiceovers_audio_offsets_msecs,
            {
                'content_0': default_audio_offset,
                'feedback_1': default_audio_offset,
                'content_2': default_audio_offset,
                'default_outcome_1': default_audio_offset,
                'content-3': default_audio_offset,
                'ca_placeholder_2': default_audio_offset,
            },
        )

    def test_should_regenerate_voiceover_when_exploration_is_curated(
        self,
    ) -> None:
        language_code = 'ar'
        exploration_id = 'exp_id_1'
        exploration_version = 1
        self.signup('tester@org.com', 'tester')

        self._create_exploration_and_arabic_translation(
            exploration_id, language_code
        )
        entity_voiceovers_list = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id,
                feconf.ENTITY_TYPE_EXPLORATION,
                exploration_version,
            )
        )

        self.assertEqual(len(entity_voiceovers_list), 0)
        parent_cloud_task_run_model_id = (
            cloud_task_models.CloudTaskRunModel.get_new_id()
        )
        cloud_task_models.CloudTaskRunModel.create_cloud_task_run_model(
            cloud_task_run_model_id=parent_cloud_task_run_model_id,
            cloud_task_name=(
                'projects/dev-project-id/locations/us-central1/queues/'
                'voiceover-regeneration/tasks/task1'
            ),
            latest_job_state='RUNNING',
            function_id='update_stats',
            current_retry_attempt=1,
        )

        voiceover_regeneration_task_mapping_model_id = '%s:%s' % (
            exploration_id,
            parent_cloud_task_run_model_id,
        )
        voiceover_regeneration_task_mapping_model = (
            cloud_task_models.VoiceoverRegenerationJobModel.get(
                voiceover_regeneration_task_mapping_model_id, strict=False
            )
        )

        self.assertIsNone(voiceover_regeneration_task_mapping_model)

        with self.swap(
            voiceover_services,
            'send_email_to_voiceover_admins_and_tech_leads_after_regeneration',
            self.mock_send_email_to_voiceover_admins_and_tech_leads,
        ):
            (
                voiceover_services.regenerate_voiceovers_on_exploration_added_to_topic(
                    exploration_id,
                    parent_cloud_task_run_model_id,
                )
            )
            updated_cloud_task_runs = sorted(
                taskqueue_services.get_all_cloud_task_runs(),
                key=lambda task_run: task_run.created_on,
            )
            for cloud_run in updated_cloud_task_runs:
                if (
                    cloud_run.function_id
                    == 'regenerate_voiceovers_for_batch_contents'
                ):
                    voiceover_services.regenerate_voiceovers_for_batch_contents(
                        exploration_id,
                        parent_cloud_task_run_model_id,
                        cloud_run.task_run_id,
                    )

        entity_voiceovers_list = (
            voiceover_services.get_entity_voiceovers_for_given_exploration(
                exploration_id,
                feconf.ENTITY_TYPE_EXPLORATION,
                exploration_version,
            )
        )
        # The entity_voiceovers_list should have a length of 3 because the setup
        # method supports three language accents for Oppia's voiceover
        # autogeneration: 'en-US', 'en-IN', and 'ar-AE'.
        self.assertEqual(len(entity_voiceovers_list), 3)

        voiceover_regeneration_task_mapping_model = (
            cloud_task_models.VoiceoverRegenerationJobModel.get(
                voiceover_regeneration_task_mapping_model_id, strict=False
            )
        )
        expected_language_accent_to_content_status_map = {
            'en-US': {
                'ca_placeholder_2': 'SUCCEEDED',
                'content-3': 'SUCCEEDED',
                'content_0': 'SUCCEEDED',
                'content_2': 'SUCCEEDED',
                'default_outcome_1': 'SUCCEEDED',
                'feedback_1': 'SUCCEEDED',
            },
            'en-IN': {
                'ca_placeholder_2': 'SUCCEEDED',
                'content-3': 'SUCCEEDED',
                'content_0': 'SUCCEEDED',
                'content_2': 'SUCCEEDED',
                'default_outcome_1': 'SUCCEEDED',
                'feedback_1': 'SUCCEEDED',
            },
            'ar-AE': {'content_0': 'SUCCEEDED', 'feedback_1': 'SUCCEEDED'},
        }

        self.assertIsNotNone(voiceover_regeneration_task_mapping_model)

        # Ruling out the possibility of None for mypy type checking.
        assert voiceover_regeneration_task_mapping_model is not None

        self.assertEqual(
            voiceover_regeneration_task_mapping_model.language_accent_to_content_status_map,
            expected_language_accent_to_content_status_map,
        )
        self.assertEqual(
            voiceover_regeneration_task_mapping_model.exploration_id,
            exploration_id,
        )
        self.assertEqual(
            voiceover_regeneration_task_mapping_model.cloud_task_run_id,
            parent_cloud_task_run_model_id,
        )

    def test_should_generate_voiceover_for_translated_content(self) -> None:
        language_code = 'hi'
        exploration_id = 'exp_id_1'
        language_codes_mapping: Dict[str, Dict[str, bool]] = {
            'en': {'en-US': True},
            'hi': {'hi-IN': True},
        }
        voiceover_services.save_language_accent_support(
            language_codes_mapping=language_codes_mapping
        )
        exploration_version = 1
        language_accent_code = 'hi-IN'

        self.signup(self.OWNER_EMAIL, self.OWNER_USERNAME)
        owner_id = self.get_user_id_from_email(self.OWNER_EMAIL)

        exploration = exp_domain.Exploration.create_default_exploration(
            exploration_id,
            title='A Title',
            category='A Category',
            objective='An Objective',
        )
        exploration.states['Introduction'].content.html = 'First Card!'
        exp_services.save_new_exploration(owner_id, exploration)

        add_translation_change_dict = {
            'cmd': exp_domain.CMD_ADD_WRITTEN_TRANSLATION,
            'state_name': 'Introduction',
            'content_id': 'content_0',
            'language_code': language_code,
            'content_html': 'First Card!',
            'translation_html': 'पहला कार्ड',
            'data_format': 'html',
        }

        translation_suggestion = suggestion_services.create_suggestion(
            feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT,
            feconf.ENTITY_TYPE_EXPLORATION,
            exploration_id,
            exploration.version,
            owner_id,
            add_translation_change_dict,
            'test description',
        )

        entity_voiceovers = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                feconf.ENTITY_TYPE_EXPLORATION,
                exploration_id,
                exploration_version,
                language_accent_code,
            )
        )
        self.assertEqual(entity_voiceovers.voiceovers_mapping, {})

        parent_cloud_task_model_id = 'cloud_task_model_id'
        task_name = 'projects/%s/locations/%s/queues/%s/tasks/%s' % (
            'dev-project-id',
            'us-central',
            'voiceover-regeneration',
            uuid.uuid4().hex,
        )
        function_id = 'regenerate_voiceovers_on_exploration_update'
        taskqueue_services.create_new_cloud_task_model(
            parent_cloud_task_model_id, task_name, function_id
        )

        with self.swap(
            voiceover_services,
            'send_email_to_voiceover_admins_and_tech_leads_after_regeneration',
            self.mock_send_email_to_voiceover_admins_and_tech_leads,
        ):
            (
                voiceover_services.regenerate_voiceovers_after_accepting_suggestion(
                    translation_suggestion.suggestion_id,
                    parent_cloud_task_model_id,
                )
            )

            updated_cloud_task_runs = sorted(
                taskqueue_services.get_all_cloud_task_runs(),
                key=lambda task_run: task_run.created_on,
            )
            for cloud_run in updated_cloud_task_runs:
                if (
                    cloud_run.function_id
                    == 'regenerate_voiceovers_for_batch_contents'
                ):
                    voiceover_services.regenerate_voiceovers_for_batch_contents(
                        exploration_id,
                        parent_cloud_task_model_id,
                        cloud_run.task_run_id,
                    )

        entity_voiceovers = (
            voiceover_services.get_voiceovers_for_given_language_accent_code(
                feconf.ENTITY_TYPE_EXPLORATION,
                exploration_id,
                exploration_version,
                language_accent_code,
            )
        )