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

core/domain/topic_services_test.py
5 013 строк186 KB
Rohan Unbeg
[GSoC 2026] M 2.3 - Fix part of #24933: Implement skill opportunity lifecycle in domain services (#26931)
06 авг 2026, 14:51
06 авг 2026, 14:51ab0672b
# coding: utf-8
#
# Copyright 2018 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 topic services."""

from __future__ import annotations

import logging
import os

from core import feature_flag_list, feconf, utils
from core.constants import constants
from core.domain import (
    exp_services,
    feature_flag_services,
    fs_services,
    question_domain,
    question_services,
    rights_manager,
    skill_domain,
    skill_fetchers,
    skill_services,
    story_domain,
    story_fetchers,
    story_services,
    study_guide_domain,
    study_guide_services,
    subtopic_page_domain,
    subtopic_page_services,
    suggestion_services,
    topic_domain,
    topic_fetchers,
    topic_services,
    translation_domain,
    user_services,
)
from core.platform import models
from core.tests import test_utils

from typing import Dict, List, Optional, Union, cast

MYPY = False
if MYPY:  # pragma: no cover
    from mypy_imports import opportunity_models, topic_models

opportunity_models, topic_models, story_models = models.Registry.import_models(
    [models.Names.OPPORTUNITY, models.Names.TOPIC, models.Names.STORY]
)


class TopicServicesUnitTests(test_utils.GenericTestBase):
    """Tests for topic services."""

    user_id: str = 'user_id'
    story_id_1: str = 'story_1'
    story_id_2: str = 'story_2'
    story_id_3: str = 'story_3'
    subtopic_id: int = 1
    skill_id_1: str = 'skill_1'
    skill_id_2: str = 'skill_2'
    skill_id_3: str = 'skill_3'

    def setUp(self) -> None:
        self.test_list: List[str] = []
        super().setUp()
        self.TOPIC_ID = topic_fetchers.get_new_topic_id()
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title',
                    'subtopic_id': 1,
                    'url_fragment': 'fragment-one',
                }
            )
        ]
        self.save_new_topic(
            self.TOPIC_ID,
            self.user_id,
            name='Name',
            description='Description',
            canonical_story_ids=[self.story_id_1, self.story_id_2],
            additional_story_ids=[self.story_id_3],
            uncategorized_skill_ids=[self.skill_id_1, self.skill_id_2],
            subtopics=[],
            next_subtopic_id=1,
        )
        self.save_new_story(self.story_id_1, self.user_id, self.TOPIC_ID)
        self.save_new_story(
            self.story_id_3,
            self.user_id,
            self.TOPIC_ID,
            title='Title 3',
            description='Description 3',
        )
        self.save_new_story(
            self.story_id_2,
            self.user_id,
            self.TOPIC_ID,
            title='Title 2',
            description='Description 2',
        )
        self.signup('a@example.com', 'A')
        self.signup('b@example.com', 'B')
        self.signup(self.CURRICULUM_ADMIN_EMAIL, self.CURRICULUM_ADMIN_USERNAME)
        self.signup(self.OWNER_EMAIL, self.OWNER_USERNAME)

        self.user_id_a = self.get_user_id_from_email('a@example.com')
        self.user_id_b = self.get_user_id_from_email('b@example.com')
        self.user_id_admin = self.get_user_id_from_email(
            self.CURRICULUM_ADMIN_EMAIL
        )
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin, self.TOPIC_ID, changelist, 'Added a subtopic'
        )
        self.owner_id = self.get_user_id_from_email(self.OWNER_EMAIL)

        self.topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.set_curriculum_admins([self.CURRICULUM_ADMIN_USERNAME])
        self.set_topic_managers(
            [user_services.get_username(self.user_id_a)], self.TOPIC_ID
        )
        self.user_a = user_services.get_user_actions_info(self.user_id_a)
        self.user_b = user_services.get_user_actions_info(self.user_id_b)
        self.user_admin = user_services.get_user_actions_info(
            self.user_id_admin
        )

    def test_raises_error_if_guest_user_trying_to_deassign_roles_from_topic(
        self,
    ) -> None:
        guest_user = user_services.get_user_actions_info(None)
        with self.assertRaisesRegex(
            Exception,
            'Guest users are not allowed to deassing users from all topics.',
        ):
            topic_services.deassign_user_from_all_topics(guest_user, 'user_id')

        with self.assertRaisesRegex(
            Exception,
            'Guest users are not allowed to deassing manager role from topic.',
        ):
            topic_services.deassign_manager_role_from_topic(
                guest_user, 'user_id', 'topic_id'
            )

    def test_get_story_titles_in_topic(self) -> None:
        story_titles = topic_services.get_story_titles_in_topic(self.topic)
        self.assertEqual(len(story_titles), 2)
        self.assertIn('Title', story_titles)
        self.assertIn('Title 2', story_titles)

    def test_update_story_and_topic_summary(self) -> None:
        change_list = [
            story_domain.StoryChange(
                {
                    'cmd': story_domain.CMD_UPDATE_STORY_PROPERTY,
                    'property_name': story_domain.STORY_PROPERTY_TITLE,
                    'old_value': 'Title',
                    'new_value': 'New Title',
                }
            )
        ]
        topic_services.update_story_and_topic_summary(
            self.user_id,
            self.story_id_1,
            change_list,
            'Updated story title',
            self.TOPIC_ID,
        )
        story_titles = topic_services.get_story_titles_in_topic(self.topic)
        self.assertIn('New Title', story_titles)

    def test_compute_summary(self) -> None:
        topic_summary = topic_services.compute_summary_of_topic(self.topic)

        self.assertEqual(topic_summary.id, self.topic.id)
        self.assertEqual(topic_summary.name, self.topic.name)
        self.assertEqual(
            topic_summary.canonical_name, self.topic.canonical_name
        )
        self.assertEqual(topic_summary.url_fragment, self.topic.url_fragment)
        self.assertEqual(topic_summary.description, self.topic.description)
        self.assertEqual(topic_summary.language_code, self.topic.language_code)
        self.assertEqual(topic_summary.version, self.topic.version)
        self.assertEqual(topic_summary.canonical_story_count, 0)
        self.assertEqual(topic_summary.additional_story_count, 0)
        self.assertEqual(topic_summary.uncategorized_skill_count, 2)
        self.assertEqual(topic_summary.subtopic_count, 1)
        self.assertEqual(topic_summary.total_skill_count, 2)
        self.assertEqual(topic_summary.total_published_node_count, 0)
        self.assertEqual(
            topic_summary.thumbnail_filename, self.topic.thumbnail_filename
        )
        self.assertEqual(
            topic_summary.thumbnail_bg_color, self.topic.thumbnail_bg_color
        )
        self.assertEqual(
            topic_summary.topic_model_created_on, self.topic.created_on
        )
        self.assertEqual(
            topic_summary.topic_model_last_updated, self.topic.last_updated
        )

    def test_compute_summary_when_serial_chapter_launch_disabled(self) -> None:
        story_exp_ids = ['exp-1', 'exp-2']
        no_exp_ids: List[str] = []
        self._create_linked_explorations(
            self.TOPIC_ID, self.story_id_1, story_exp_ids
        )
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_2, self.user_id_admin
        )
        updated_topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)

        topic_summary = topic_services.compute_summary_of_topic(updated_topic)

        self.assertEqual(
            topic_summary.total_published_node_count, len(story_exp_ids)
        )
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping,
            {self.story_id_1: story_exp_ids, self.story_id_2: no_exp_ids},
        )

    @test_utils.enable_feature_flags(
        [
            feature_flag_list.FeatureNames.SERIAL_CHAPTER_LAUNCH_CURRICULUM_ADMIN_VIEW
        ]
    )
    def test_compute_summary_when_serial_chapter_launch_enabled(self) -> None:
        story_1_published_chapter_exp_ids = ['exp-1', 'exp-2']
        story_2_exp_ids = ['exp-3', 'exp-4']
        story_2_published_chapter_exp_ids = story_2_exp_ids[:1]
        self._publish_story_chapters_with_explorations(
            self.TOPIC_ID, self.story_id_1, story_1_published_chapter_exp_ids
        )
        self._publish_story_chapters_with_explorations(
            self.TOPIC_ID,
            self.story_id_2,
            story_2_exp_ids,
            story_2_published_chapter_exp_ids,
        )
        updated_topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)

        topic_summary = topic_services.compute_summary_of_topic(updated_topic)

        self.assertEqual(
            topic_summary.total_published_node_count,
            len(story_1_published_chapter_exp_ids)
            + len(story_2_published_chapter_exp_ids),
        )
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping,
            {
                self.story_id_1: story_1_published_chapter_exp_ids,
                self.story_id_2: story_2_published_chapter_exp_ids,
            },
        )

    def test_compute_summary_does_not_map_unpublished_stories(self) -> None:
        story_exp_ids = ['exp-1', 'exp-2']
        self._create_linked_explorations(
            self.TOPIC_ID, self.story_id_1, story_exp_ids
        )
        updated_topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)

        topic_summary = topic_services.compute_summary_of_topic(updated_topic)

        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping, {}
        )

    def _publish_story_chapters_with_explorations(
        self,
        topic_id: str,
        story_id: str,
        exp_ids: List[str],
        chapter_exp_ids: Optional[List[str]] = None,
    ) -> None:
        """Creates explorations with exp_ids. Links them to the story given
        by story_id. Publishes the story under topic with topic_id, along
        with its nodes who has linked chapter_exp_ids. If chapter_exp_ids
        is not provided, then the nodes published will be based on exp_ids.

        Args:
            topic_id: str. The id of the topic containing the story.
            story_id: str. The id of the story to publish and to link
                explorations to.
            exp_ids: list(str). A list of exploration ids to create
                explorations with.
            chapter_exp_ids: list(str)|None. A list of exploration ids that
                correspond with each story node that will be published.
        """
        chapter_exp_ids = chapter_exp_ids or exp_ids
        self._create_linked_explorations(topic_id, story_id, exp_ids)
        topic_services.publish_story(topic_id, story_id, self.user_id_admin)
        self._publish_story_chapters(topic_id, story_id, chapter_exp_ids)

    def _publish_story_chapters(
        self, topic_id: str, story_id: str, chapter_exp_ids: List[str]
    ) -> None:
        """Publishes the story chapters linked to each exploration id in
        chapter_exp_ids. The chapters belong to story_id, and the story
        belongs to topic_id.

        Args:
            topic_id: str. Topic id containing the story.
            story_id: str. Story id containing the chapters.
            chapter_exp_ids: list(str). List of exp ids linked to each chapter
                to be published.
        """
        story = story_fetchers.get_story_by_id(story_id)
        change_list = [
            story_domain.StoryChange(
                {
                    'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                    'property_name': story_domain.STORY_NODE_PROPERTY_STATUS,
                    'node_id': story.story_contents.nodes[i].id,
                    'old_value': constants.STORY_NODE_STATUS_DRAFT,
                    'new_value': constants.STORY_NODE_STATUS_PUBLISHED,
                }
            )
            for i in range(len(chapter_exp_ids))
            if story.story_contents.nodes[i].exploration_id
            == chapter_exp_ids[i]
        ]
        topic_services.update_story_and_topic_summary(
            self.user_id_admin,
            story_id,
            change_list,
            'Publish these story chapters only.',
            topic_id,
        )

    def _create_linked_explorations(
        self, topic_id: str, story_id: str, exp_ids: List[str]
    ) -> None:
        """Creates explorations with exp_ids. Links them to the story given
        by story_id that is in the topic given by topic_id.

        Args:
            topic_id: str. The id of the topic containing the story.
            story_id: str. The id of the story to publish and to link
                explorations to.
            exp_ids: list(str). A list of exploration ids to create
                explorations with.
        """
        for exp_id in exp_ids:
            self.save_new_valid_exploration(
                exp_id, self.user_id_admin, end_state_name='end'
            )
            self.publish_exploration(self.user_id_admin, exp_id)
        self.add_explorations_to_story(topic_id, story_id, exp_ids)

    def test_raises_error_while_computing_topic_summary_with_invalid_data(
        self,
    ) -> None:
        test_topic = self.topic
        test_topic.created_on = None
        with self.assertRaisesRegex(
            Exception, 'No data available for when the topic was last updated.'
        ):
            topic_services.compute_summary_of_topic(self.topic)

    def test_generate_topic_summary_when_publishing_story(self) -> None:
        with self.swap_with_checks(
            topic_services,
            'generate_topic_summary',
            topic_services.generate_topic_summary,
            expected_args=[(self.TOPIC_ID,)],
        ):
            topic_services.publish_story(
                self.TOPIC_ID, self.story_id_1, self.user_id_admin
            )

        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        self.assertEqual(topic_summary.id, self.TOPIC_ID)
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping,
            {self.story_id_1: []},
        )

    def test_generate_topic_summary_when_unpublishing_story(self) -> None:
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )

        with self.swap_with_checks(
            topic_services,
            'generate_topic_summary',
            topic_services.generate_topic_summary,
            expected_args=[(self.TOPIC_ID,)],
        ):
            topic_services.unpublish_story(
                self.TOPIC_ID, self.story_id_1, self.user_id_admin
            )

        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        self.assertEqual(topic_summary.id, self.TOPIC_ID)
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping, {}
        )

    def test_generate_topic_summary_when_deleting_published_story(self) -> None:
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_3, self.user_id_admin
        )

        # Keep the swap assertions separate to ensure exactly 1 call to
        # generate_topic_summary occurs per delete call.
        with self.swap_with_checks(
            topic_services,
            'generate_topic_summary',
            topic_services.generate_topic_summary,
            expected_args=[(self.TOPIC_ID,)],
        ):
            topic_services.delete_canonical_story(
                self.user_id_admin, self.TOPIC_ID, self.story_id_1
            )

        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        self.assertEqual(topic_summary.id, self.TOPIC_ID)
        self.assertEqual(topic_summary.canonical_story_count, 0)
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping,
            {self.story_id_3: []},
        )

        with self.swap_with_checks(
            topic_services,
            'generate_topic_summary',
            topic_services.generate_topic_summary,
            expected_args=[(self.TOPIC_ID,)],
        ):
            topic_services.delete_additional_story(
                self.user_id_admin, self.TOPIC_ID, self.story_id_3
            )

        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        self.assertEqual(topic_summary.id, self.TOPIC_ID)
        self.assertEqual(topic_summary.additional_story_count, 0)
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping, {}
        )

    def test_generate_topic_summary_when_publishing_story_chapter(self) -> None:
        linked_exp_ids = ['exp_1']
        self._create_linked_explorations(
            self.TOPIC_ID, self.story_id_1, linked_exp_ids
        )
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        story = story_fetchers.get_story_by_id(self.story_id_1)

        with self.swap_to_always_return(
            feature_flag_services, 'is_feature_flag_enabled', True
        ), self.swap_with_checks(
            topic_services,
            'generate_topic_summary',
            topic_services.generate_topic_summary,
            expected_args=[(self.TOPIC_ID,)],
        ):
            topic_services.update_story_and_topic_summary(
                self.user_id_admin,
                self.story_id_1,
                [
                    story_domain.StoryChange(
                        {
                            'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                            'property_name': story_domain.STORY_NODE_PROPERTY_STATUS,
                            'node_id': story.story_contents.nodes[0].id,
                            'old_value': constants.STORY_NODE_STATUS_DRAFT,
                            'new_value': constants.STORY_NODE_STATUS_PUBLISHED,
                        }
                    )
                ],
                'Publish story chapter.',
                self.TOPIC_ID,
            )

        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        self.assertEqual(topic_summary.id, self.TOPIC_ID)
        self.assertEqual(
            topic_summary.total_published_node_count, len(linked_exp_ids)
        )
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping,
            {self.story_id_1: linked_exp_ids},
        )

    @test_utils.enable_feature_flags(
        [
            feature_flag_list.FeatureNames.SERIAL_CHAPTER_LAUNCH_CURRICULUM_ADMIN_VIEW
        ]
    )
    def test_generate_topic_summary_when_unpublishing_story_chapter(
        self,
    ) -> None:
        linked_exp_ids = ['exp_1']
        self._create_linked_explorations(
            self.TOPIC_ID, self.story_id_1, linked_exp_ids
        )
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        story = story_fetchers.get_story_by_id(self.story_id_1)

        topic_services.update_story_and_topic_summary(
            self.user_id_admin,
            self.story_id_1,
            [
                story_domain.StoryChange(
                    {
                        'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                        'property_name': story_domain.STORY_NODE_PROPERTY_STATUS,
                        'node_id': story.story_contents.nodes[0].id,
                        'old_value': constants.STORY_NODE_STATUS_DRAFT,
                        'new_value': constants.STORY_NODE_STATUS_PUBLISHED,
                    }
                )
            ],
            'Publish story chapter.',
            self.TOPIC_ID,
        )

        with self.swap_with_checks(
            topic_services,
            'generate_topic_summary',
            topic_services.generate_topic_summary,
            expected_args=[(self.TOPIC_ID,)],
        ):
            topic_services.update_story_and_topic_summary(
                self.user_id_admin,
                self.story_id_1,
                [
                    story_domain.StoryChange(
                        {
                            'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                            'property_name': story_domain.STORY_NODE_PROPERTY_STATUS,
                            'node_id': story.story_contents.nodes[0].id,
                            'old_value': constants.STORY_NODE_STATUS_PUBLISHED,
                            'new_value': constants.STORY_NODE_STATUS_DRAFT,
                        }
                    )
                ],
                'Unpublish story chapter.',
                self.TOPIC_ID,
            )

        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        self.assertEqual(topic_summary.id, self.TOPIC_ID)
        self.assertEqual(
            topic_summary.total_published_node_count, len(linked_exp_ids) - 1
        )
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping,
            {self.story_id_1: []},
        )

    @test_utils.enable_feature_flags(
        [
            feature_flag_list.FeatureNames.SERIAL_CHAPTER_LAUNCH_CURRICULUM_ADMIN_VIEW
        ]
    )
    def test_generate_topic_summary_when_deleting_published_story_chapter(
        self,
    ) -> None:
        linked_exp_ids = ['exp_1']
        self._create_linked_explorations(
            self.TOPIC_ID, self.story_id_1, linked_exp_ids
        )
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        story = story_fetchers.get_story_by_id(self.story_id_1)
        topic_services.update_story_and_topic_summary(
            self.user_id_admin,
            self.story_id_1,
            [
                story_domain.StoryChange(
                    {
                        'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                        'property_name': story_domain.STORY_NODE_PROPERTY_STATUS,
                        'node_id': story.story_contents.nodes[0].id,
                        'old_value': constants.STORY_NODE_STATUS_DRAFT,
                        'new_value': constants.STORY_NODE_STATUS_PUBLISHED,
                    }
                )
            ],
            'Publish story chapter.',
            self.TOPIC_ID,
        )

        with self.swap_with_checks(
            topic_services,
            'generate_topic_summary',
            topic_services.generate_topic_summary,
            expected_args=[(self.TOPIC_ID,)],
        ):
            topic_services.update_story_and_topic_summary(
                self.user_id_admin,
                self.story_id_1,
                [
                    story_domain.StoryChange(
                        {
                            'cmd': story_domain.CMD_DELETE_STORY_NODE,
                            'node_id': story.story_contents.nodes[0].id,
                        }
                    )
                ],
                'Delete story chapter.',
                self.TOPIC_ID,
            )

        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        self.assertEqual(topic_summary.id, self.TOPIC_ID)
        self.assertEqual(
            topic_summary.total_published_node_count, len(linked_exp_ids) - 1
        )
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping,
            {self.story_id_1: []},
        )

    @test_utils.enable_feature_flags(
        [
            feature_flag_list.FeatureNames.SERIAL_CHAPTER_LAUNCH_CURRICULUM_ADMIN_VIEW
        ]
    )
    def test_generate_topic_summary_when_changing_exp_id_linked_to_published_story_chapter(  # pylint: disable=line-too-long
        self,
    ) -> None:
        exp_id_1 = 'exp_1'
        exp_id_2 = 'exp_2'
        self._create_linked_explorations(
            self.TOPIC_ID, self.story_id_1, [exp_id_1]
        )
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        story = story_fetchers.get_story_by_id(self.story_id_1)
        topic_services.update_story_and_topic_summary(
            self.user_id_admin,
            self.story_id_1,
            [
                story_domain.StoryChange(
                    {
                        'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                        'property_name': story_domain.STORY_NODE_PROPERTY_STATUS,
                        'node_id': story.story_contents.nodes[0].id,
                        'old_value': constants.STORY_NODE_STATUS_DRAFT,
                        'new_value': constants.STORY_NODE_STATUS_PUBLISHED,
                    }
                )
            ],
            'Publish story chapter.',
            self.TOPIC_ID,
        )
        self.save_new_valid_exploration(
            exp_id_2, self.user_id_admin, end_state_name='end'
        )
        self.publish_exploration(self.user_id_admin, exp_id_2)

        with self.swap_with_checks(
            topic_services,
            'generate_topic_summary',
            topic_services.generate_topic_summary,
            expected_args=[(self.TOPIC_ID,)],
        ):
            topic_services.update_story_and_topic_summary(
                self.user_id_admin,
                self.story_id_1,
                [
                    story_domain.StoryChange(
                        {
                            'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                            'property_name': (
                                story_domain.STORY_NODE_PROPERTY_EXPLORATION_ID
                            ),
                            'node_id': story.story_contents.nodes[0].id,
                            'old_value': exp_id_1,
                            'new_value': exp_id_2,
                        }
                    )
                ],
                'Change exploration of chapter.',
                self.TOPIC_ID,
            )

        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        self.assertEqual(topic_summary.id, self.TOPIC_ID)
        self.assertDictEqual(
            topic_summary.published_story_exploration_mapping,
            {self.story_id_1: [exp_id_2]},
        )

    def test_get_topic_from_model(self) -> None:
        topic_model = topic_models.TopicModel.get(self.TOPIC_ID)
        topic = topic_fetchers.get_topic_from_model(topic_model)
        self.assertEqual(topic.to_dict(), self.topic.to_dict())

    def test_cannot_get_topic_from_model_with_invalid_schema_version(
        self,
    ) -> None:
        topic_services.create_new_topic_rights('topic_id', self.user_id_a)
        commit_cmd = topic_domain.TopicChange(
            {'cmd': topic_domain.CMD_CREATE_NEW, 'name': 'name'}
        )
        subtopic_dict = {'id': 1, 'title': 'subtopic_title', 'skill_ids': []}
        model = topic_models.TopicModel(
            id='topic_id',
            name='name',
            abbreviated_name='abbrev',
            url_fragment='name-one',
            description='description1',
            canonical_name='canonical_name',
            next_subtopic_id=1,
            language_code='en',
            subtopics=[subtopic_dict],
            subtopic_schema_version=0,
            story_reference_schema_version=0,
            page_title_fragment_for_web='fragm',
        )
        commit_cmd_dicts = [commit_cmd.to_dict()]
        model.commit(self.user_id_a, 'topic model created', commit_cmd_dicts)

        with self.assertRaisesRegex(
            Exception,
            'Sorry, we can only process v1-v%d subtopic schemas at '
            'present.' % feconf.CURRENT_SUBTOPIC_SCHEMA_VERSION,
        ):
            topic_fetchers.get_topic_from_model(model)

        topic_services.create_new_topic_rights('topic_id_2', self.user_id_a)
        model = topic_models.TopicModel(
            id='topic_id_2',
            name='name 2',
            abbreviated_name='abbrev',
            url_fragment='name-two',
            description='description',
            canonical_name='canonical_name_2',
            next_subtopic_id=1,
            language_code='en',
            subtopics=[subtopic_dict],
            subtopic_schema_version=1,
            story_reference_schema_version=0,
            page_title_fragment_for_web='fragm',
        )
        commit_cmd_dicts = [commit_cmd.to_dict()]
        model.commit(self.user_id_a, 'topic model created', commit_cmd_dicts)

        with self.assertRaisesRegex(
            Exception,
            'Sorry, we can only process v1-v%d story reference schemas at '
            'present.' % feconf.CURRENT_STORY_REFERENCE_SCHEMA_VERSION,
        ):
            topic_fetchers.get_topic_from_model(model)

    def test_cannot_create_topic_change_class_with_invalid_changelist(
        self,
    ) -> None:
        with self.assertRaisesRegex(
            Exception, 'Missing cmd key in change dict'
        ):
            topic_domain.TopicChange(
                {
                    'invalid_cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_DESCRIPTION,
                    'old_value': 'Description',
                    'new_value': 'New Description',
                }
            )

    def test_cannot_rearrange_story_with_missing_index_values(self) -> None:
        with self.assertRaisesRegex(
            Exception,
            (
                'The following required attributes are missing: '
                'from_index, to_index'
            ),
        ):
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_REARRANGE_CANONICAL_STORY,
                }
            )

    def test_cannot_rearrange_story_with_missing_from_index_value(self) -> None:
        with self.assertRaisesRegex(
            Exception,
            'The following required attributes are missing: from_index',
        ):
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_REARRANGE_CANONICAL_STORY,
                    'to_index': 1,
                }
            )

    def test_cannot_rearrange_story_with_missing_to_index_value(self) -> None:
        with self.assertRaisesRegex(
            Exception,
            ('The following required attributes are missing: to_index'),
        ):
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_REARRANGE_CANONICAL_STORY,
                    'from_index': 1,
                }
            )

    def test_rearrange_canonical_stories_in_topic(self) -> None:
        story_id_new = 'story_id_new'
        topic_services.add_canonical_story(
            self.user_id_admin, self.TOPIC_ID, 'story_id_new'
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.canonical_story_references), 3)
        self.assertEqual(
            topic.canonical_story_references[0].story_id, self.story_id_1
        )
        self.assertEqual(
            topic.canonical_story_references[1].story_id, self.story_id_2
        )
        self.assertEqual(
            topic.canonical_story_references[2].story_id, story_id_new
        )

        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_REARRANGE_CANONICAL_STORY,
                    'from_index': 2,
                    'to_index': 0,
                }
            )
        ]

        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Rearranged canonical story on index 2 to index 0.',
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.canonical_story_references), 3)
        self.assertEqual(
            topic.canonical_story_references[0].story_id, story_id_new
        )
        self.assertEqual(
            topic.canonical_story_references[1].story_id, self.story_id_1
        )
        self.assertEqual(
            topic.canonical_story_references[2].story_id, self.story_id_2
        )
        topic_commit_log_entry = (
            topic_models.TopicCommitLogEntryModel.get_commit(self.TOPIC_ID, 4)
        )
        # Ruling out the possibility of None for mypy type checking.
        assert topic_commit_log_entry is not None
        self.assertEqual(topic_commit_log_entry.commit_type, 'edit')
        self.assertEqual(topic_commit_log_entry.topic_id, self.TOPIC_ID)
        self.assertEqual(topic_commit_log_entry.user_id, self.user_id_admin)
        self.assertEqual(
            topic_commit_log_entry.commit_message,
            'Rearranged canonical story on index 2 to index 0.',
        )

    def test_rearrange_skill_in_subtopic(self) -> None:
        topic_services.add_uncategorized_skill(
            self.user_id_admin, self.TOPIC_ID, self.skill_id_3
        )
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': 1,
                    'skill_id': self.skill_id_1,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': 1,
                    'skill_id': self.skill_id_2,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': 1,
                    'skill_id': self.skill_id_3,
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Added skills to the subtopic.',
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics[0].skill_ids), 3)
        skill_ids = topic.subtopics[0].skill_ids
        self.assertEqual(skill_ids[0], self.skill_id_1)
        self.assertEqual(skill_ids[1], self.skill_id_2)
        self.assertEqual(skill_ids[2], self.skill_id_3)

        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_REARRANGE_SKILL_IN_SUBTOPIC,
                    'subtopic_id': 1,
                    'from_index': 2,
                    'to_index': 0,
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Rearranged skill from index 2 to index 0 for subtopic with id 1.',
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics[0].skill_ids), 3)
        skill_ids = topic.subtopics[0].skill_ids
        self.assertEqual(skill_ids[0], self.skill_id_3)
        self.assertEqual(skill_ids[1], self.skill_id_1)
        self.assertEqual(skill_ids[2], self.skill_id_2)

        topic_commit_log_entry = (
            topic_models.TopicCommitLogEntryModel.get_commit(self.TOPIC_ID, 5)
        )
        # Ruling out the possibility of None for mypy type checking.
        assert topic_commit_log_entry is not None
        self.assertEqual(topic_commit_log_entry.commit_type, 'edit')
        self.assertEqual(topic_commit_log_entry.topic_id, self.TOPIC_ID)
        self.assertEqual(topic_commit_log_entry.user_id, self.user_id_admin)
        self.assertEqual(
            topic_commit_log_entry.commit_message,
            'Rearranged skill from index 2 to index 0 for subtopic with id 1.',
        )

    def test_rearrange_subtopic(self) -> None:
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title2',
                    'subtopic_id': 2,
                    'url_fragment': 'fragment-two',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': topic_domain.SUBTOPIC_PROPERTY_URL_FRAGMENT,
                    'new_value': 'title-two',
                    'old_value': '',
                    'subtopic_id': 2,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title3',
                    'subtopic_id': 3,
                    'url_fragment': 'fragment-three',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': topic_domain.SUBTOPIC_PROPERTY_URL_FRAGMENT,
                    'new_value': 'title-three',
                    'old_value': '',
                    'subtopic_id': 3,
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Added subtopics to the topic.',
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics), 3)
        subtopics = topic.subtopics
        self.assertEqual(subtopics[0].id, 1)
        self.assertEqual(subtopics[1].id, 2)
        self.assertEqual(subtopics[2].id, 3)

        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_REARRANGE_SUBTOPIC,
                    'from_index': 2,
                    'to_index': 0,
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Rearranged subtopic from index 2 to index 0.',
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics), 3)
        subtopics = topic.subtopics
        self.assertEqual(subtopics[0].id, 3)
        self.assertEqual(subtopics[1].id, 1)
        self.assertEqual(subtopics[2].id, 2)

        topic_commit_log_entry = (
            topic_models.TopicCommitLogEntryModel.get_commit(self.TOPIC_ID, 4)
        )
        # Ruling out the possibility of None for mypy type checking.
        assert topic_commit_log_entry is not None
        self.assertEqual(topic_commit_log_entry.commit_type, 'edit')
        self.assertEqual(topic_commit_log_entry.topic_id, self.TOPIC_ID)
        self.assertEqual(topic_commit_log_entry.user_id, self.user_id_admin)
        self.assertEqual(
            topic_commit_log_entry.commit_message,
            'Rearranged subtopic from index 2 to index 0.',
        )

    def test_cannot_update_topic_property_with_invalid_changelist(self) -> None:
        with self.assertRaisesRegex(
            Exception,
            (
                'Value for property_name in cmd update_topic_property: '
                'invalid property is not allowed'
            ),
        ):
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': 'invalid property',
                    'old_value': 'Description',
                    'new_value': 'New Description',
                }
            )

    def test_cannot_update_subtopic_property_with_invalid_changelist(
        self,
    ) -> None:
        with self.assertRaisesRegex(
            Exception,
            'The following required attributes are missing: subtopic_id',
        ):
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': 'invalid property',
                    'old_value': 'Description',
                    'new_value': 'New Description',
                }
            )

    def test_update_subtopic_property_without_study_guide(self) -> None:
        # We create a topic and subtopic in this way so as to not create a
        # study guide.
        topic_id_1 = topic_fetchers.get_new_topic_id()
        topic_1 = topic_domain.Topic.create_default_topic(
            topic_id_1,
            'Dummy Topic 1',
            'dummy-topic-one',
            'description',
            'fragm',
        )
        topic_1.add_subtopic(1, 'Dummy Subtopic Title', 'dummysubtopic')
        topic_1.update_meta_tag_content('dummy-meta')
        subtopic_page = (
            subtopic_page_domain.SubtopicPage.create_default_subtopic_page(
                1, topic_id_1
            )
        )
        topic_services.save_new_topic(self.user_id, topic_1)
        subtopic_page_services.save_subtopic_page(
            self.user_id,
            subtopic_page,
            'Added subtopic',
            [
                topic_domain.TopicChange(
                    {
                        'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                        'subtopic_id': 1,
                        'title': 'Dummy Subtopic Title',
                        'url_fragment': 'dummy-fragment',
                    }
                )
            ],
        )

        # Check that study guide does not exist.
        with self.assertRaisesRegex(
            Exception,
            'Entity for class StudyGuideModel with id %s not found'
            % (f'{topic_id_1}-1'),
        ):
            study_guide_services.get_study_guide_by_id(topic_id_1, 1)

        # Test whether a subtopic without corresponding study guide
        # already existing in datastore can be edited.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': 'title',
                    'subtopic_id': 1,
                    'old_value': 'Dummy Subtopic Title',
                    'new_value': 'New Title',
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            topic_id_1,
            changelist,
            'Update title of subtopic.',
        )
        topic = topic_fetchers.get_topic_by_id(topic_id_1)

        self.assertEqual(len(topic.subtopics), 1)
        self.assertEqual(topic.subtopics[0].title, 'New Title')

    def test_update_subtopic_property(self) -> None:
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)

        self.assertEqual(len(topic.subtopics), 1)
        self.assertEqual(topic.subtopics[0].title, 'Title')

        # Store a dummy image in filesystem.
        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'test_svg.svg'),
            'rb',
            encoding=None,
        ) as f:
            raw_image = f.read()
        fs = fs_services.GcsFileSystem(feconf.ENTITY_TYPE_TOPIC, self.TOPIC_ID)
        fs.commit(
            '%s/image.svg' % (constants.ASSET_TYPE_THUMBNAIL),
            raw_image,
            mimetype='image/svg+xml',
        )

        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': 'title',
                    'subtopic_id': 1,
                    'old_value': 'Title',
                    'new_value': 'New Title',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': 'thumbnail_filename',
                    'subtopic_id': 1,
                    'old_value': None,
                    'new_value': 'image.svg',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': 'thumbnail_bg_color',
                    'subtopic_id': 1,
                    'old_value': None,
                    'new_value': constants.ALLOWED_THUMBNAIL_BG_COLORS[
                        'subtopic'
                    ][0],
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Update title of subtopic.',
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)

        self.assertEqual(len(topic.subtopics), 1)
        self.assertEqual(topic.subtopics[0].title, 'New Title')
        self.assertEqual(topic.subtopics[0].thumbnail_filename, 'image.svg')
        self.assertEqual(
            topic.subtopics[0].thumbnail_bg_color,
            constants.ALLOWED_THUMBNAIL_BG_COLORS['subtopic'][0],
        )

    def test_cannot_create_topic_change_class_with_invalid_cmd(self) -> None:
        with self.assertRaisesRegex(
            Exception, 'Command invalid cmd is not allowed'
        ):
            topic_domain.TopicChange(
                {
                    'cmd': 'invalid cmd',
                    'property_name': 'title',
                    'subtopic_id': 1,
                    'old_value': 'Description',
                    'new_value': 'New Description',
                }
            )

    def test_publish_and_unpublish_story(self) -> None:
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(
            topic.canonical_story_references[0].story_is_published, False
        )
        self.assertEqual(
            topic.additional_story_references[0].story_is_published, False
        )

        with self.swap_with_call_counter(
            topic_services, 'generate_topic_summary'
        ) as (generate_topic_summary):
            topic_services.publish_story(
                self.TOPIC_ID, self.story_id_1, self.user_id_admin
            )
            self.assertGreaterEqual(generate_topic_summary.times_called, 1)
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_3, self.user_id_admin
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        # Ruling out the possibility of None for mypy type checking.
        assert topic_summary is not None
        self.assertEqual(
            topic.canonical_story_references[0].story_is_published, True
        )
        self.assertEqual(
            topic.additional_story_references[0].story_is_published, True
        )
        self.assertEqual(topic_summary.canonical_story_count, 1)
        self.assertEqual(topic_summary.additional_story_count, 1)

        with self.swap_with_call_counter(
            topic_services, 'generate_topic_summary'
        ) as (generate_topic_summary):
            topic_services.unpublish_story(
                self.TOPIC_ID, self.story_id_1, self.user_id_admin
            )
            self.assertGreaterEqual(generate_topic_summary.times_called, 1)
        topic_services.unpublish_story(
            self.TOPIC_ID, self.story_id_3, self.user_id_admin
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        # Ruling out the possibility of None for mypy type checking.
        assert topic_summary is not None
        self.assertEqual(
            topic.canonical_story_references[0].story_is_published, False
        )
        self.assertEqual(
            topic.additional_story_references[0].story_is_published, False
        )
        self.assertEqual(topic_summary.canonical_story_count, 0)
        self.assertEqual(topic_summary.additional_story_count, 0)

    def test_unpublish_story_temporarily_sets_unpublish_type(self) -> None:
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        topic_services.unpublish_story(
            self.TOPIC_ID,
            self.story_id_1,
            self.user_id_admin,
            topic_domain.STORY_PUBLICATION_ACTION_TEMPORARY_UNPUBLISH,
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        for reference in topic.canonical_story_references:
            if reference.story_id == self.story_id_1:
                self.assertEqual(reference.story_is_published, False)
                self.assertEqual(
                    reference.story_unpublish_type,
                    topic_domain.STORY_PUBLICATION_ACTION_TEMPORARY_UNPUBLISH,
                )

    def test_unpublish_story_permanently_sets_unpublish_type(self) -> None:
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        topic_services.unpublish_story(
            self.TOPIC_ID,
            self.story_id_1,
            self.user_id_admin,
            topic_domain.STORY_PUBLICATION_ACTION_PERMANENT_UNPUBLISH,
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        for reference in topic.canonical_story_references:
            if reference.story_id == self.story_id_1:
                self.assertEqual(reference.story_is_published, False)
                self.assertEqual(
                    reference.story_unpublish_type,
                    topic_domain.STORY_PUBLICATION_ACTION_PERMANENT_UNPUBLISH,
                )

    def test_invalid_publish_and_unpublish_story(self) -> None:
        with self.assertRaisesRegex(
            Exception,
            'Entity for class TopicModel with id invalid_topic not found',
        ):
            topic_services.publish_story(
                'invalid_topic', 'story_id_new', self.user_id_admin
            )

        with self.assertRaisesRegex(
            Exception, 'A topic with the given ID doesn\'t exist'
        ):
            topic_services.unpublish_story(
                'invalid_topic', 'story_id_new', self.user_id_admin
            )

        with self.assertRaisesRegex(
            Exception,
            'The user does not have enough rights to publish the story.',
        ):
            topic_services.publish_story(
                self.TOPIC_ID, self.story_id_3, self.user_id_b
            )

        with self.assertRaisesRegex(
            Exception,
            'The user does not have enough rights to unpublish the story.',
        ):
            topic_services.unpublish_story(
                self.TOPIC_ID, self.story_id_3, self.user_id_b
            )

        with self.assertRaisesRegex(
            Exception, 'A story with the given ID doesn\'t exist'
        ):
            topic_services.publish_story(
                self.TOPIC_ID, 'invalid_story', self.user_id_admin
            )

        with self.assertRaisesRegex(
            Exception, 'A story with the given ID doesn\'t exist'
        ):
            topic_services.unpublish_story(
                self.TOPIC_ID, 'invalid_story', self.user_id_admin
            )

        self.save_new_story(
            'story_10',
            self.user_id,
            self.TOPIC_ID,
            title='Title 2',
            description='Description 2',
        )
        with self.assertRaisesRegex(
            Exception, 'Story with given id doesn\'t exist in the topic'
        ):
            topic_services.publish_story(
                self.TOPIC_ID, 'story_10', self.user_id_admin
            )

        with self.assertRaisesRegex(
            Exception, 'Story with given id doesn\'t exist in the topic'
        ):
            topic_services.unpublish_story(
                self.TOPIC_ID, 'story_10', self.user_id_admin
            )

        # Throw error if a story node doesn't have an exploration.
        self.save_new_story(
            'story_id_new',
            self.user_id,
            self.TOPIC_ID,
            title='Title 2',
            description='Description 2',
        )
        topic_services.add_canonical_story(
            self.user_id_admin, self.TOPIC_ID, 'story_id_new'
        )

        self.save_new_default_exploration(
            'exp_id', self.user_id_admin, title='title'
        )
        self.publish_exploration(self.user_id_admin, 'exp_id')

        change_list = [
            story_domain.StoryChange(
                {
                    'cmd': story_domain.CMD_ADD_STORY_NODE,
                    'node_id': 'node_1',
                    'title': 'Title 1',
                }
            ),
            story_domain.StoryChange(
                {
                    'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                    'property_name': (
                        story_domain.STORY_NODE_PROPERTY_EXPLORATION_ID
                    ),
                    'node_id': 'node_1',
                    'old_value': None,
                    'new_value': 'exp_id',
                }
            ),
        ]
        story_services.update_story(
            self.user_id_admin,
            'story_id_new',
            change_list,
            'Added story node with linked exploration.',
        )

        self.set_moderators([self.CURRICULUM_ADMIN_USERNAME])
        self.user_admin = user_services.get_user_actions_info(
            self.user_id_admin
        )
        rights_manager.unpublish_exploration(self.user_admin, 'exp_id')
        with self.assertRaisesRegex(
            Exception,
            'Exploration with ID exp_id is not public. Please '
            'publish explorations before adding them to a story.',
        ):
            topic_services.publish_story(
                self.TOPIC_ID, 'story_id_new', self.user_id_admin
            )

        # Throws error if exploration doesn't exist.
        exp_services.delete_exploration(self.user_id_admin, 'exp_id')

        with self.assertRaisesRegex(
            Exception,
            'Expected story to only reference valid explorations, '
            'but found a reference to an invalid exploration with ID: exp_id',
        ):
            topic_services.publish_story(
                self.TOPIC_ID, 'story_id_new', self.user_id_admin
            )

    def _create_story_with_node_and_exploration(
        self,
        story_id: str,
        story_title: str,
        exp_id: str,
    ) -> str:
        """Creates a story with a single node linked to an exploration.

        Args:
            story_id: str. The story ID.
            story_title: str. The story title.
            exp_id: str. The exploration ID.

        Returns:
            str. The node ID of the created node.
        """
        self.save_new_story(
            story_id, self.user_id, self.TOPIC_ID, title=story_title
        )
        topic_services.add_canonical_story(
            self.user_id_admin, self.TOPIC_ID, story_id
        )
        change_list = [
            story_domain.StoryChange(
                {
                    'cmd': story_domain.CMD_ADD_STORY_NODE,
                    'node_id': 'node_1',
                    'title': 'Chapter 1',
                }
            ),
            story_domain.StoryChange(
                {
                    'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                    'property_name': (
                        story_domain.STORY_NODE_PROPERTY_EXPLORATION_ID
                    ),
                    'node_id': 'node_1',
                    'old_value': None,
                    'new_value': exp_id,
                }
            ),
        ]
        story_services.update_story(
            self.user_id_admin, story_id, change_list, 'Added story node.'
        )
        self.save_new_default_exploration(
            exp_id, self.user_id_admin, title='title'
        )
        self.publish_exploration(self.user_id_admin, exp_id)
        return 'node_1'

    def test_publish_story_fails_when_skill_has_less_than_10_questions(
        self,
    ) -> None:
        story_id = 'storskill1'
        exp_id = 'skill_check1'
        node_id = self._create_story_with_node_and_exploration(
            story_id, 'Story With Skill', exp_id
        )

        skill_id = 'skill_1'
        change_list = [
            story_domain.StoryChange(
                {
                    'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                    'property_name': (
                        story_domain.STORY_NODE_PROPERTY_ACQUIRED_SKILL_IDS
                    ),
                    'node_id': node_id,
                    # Here we use cast because mypy expects List[str] but [] is List[<nothing>].
                    'old_value': cast(List[str], []),
                    'new_value': [skill_id],
                }
            ),
        ]
        story_services.update_story(
            self.user_id_admin,
            story_id,
            change_list,
            'Added skill to node.',
        )

        with self.swap_to_always_return(
            question_services, 'get_total_question_count_for_skill_ids', 9
        ):
            with self.assertRaisesRegex(
                utils.ValidationError,
                'Skill %s has only 9 questions. Each skill linked to a '
                'story node must have at least 10 questions before '
                'publishing.' % skill_id,
            ):
                topic_services.publish_story(
                    self.TOPIC_ID, story_id, self.user_id_admin
                )

    def test_publish_story_succeeds_when_skill_has_at_least_10_questions(
        self,
    ) -> None:
        story_id = 'stor10qids'
        exp_id = 'exp10q'
        node_id = self._create_story_with_node_and_exploration(
            story_id, 'Story With Enough Questions', exp_id
        )

        skill_id = 'skill_1'
        change_list = [
            story_domain.StoryChange(
                {
                    'cmd': story_domain.CMD_UPDATE_STORY_NODE_PROPERTY,
                    'property_name': (
                        story_domain.STORY_NODE_PROPERTY_ACQUIRED_SKILL_IDS
                    ),
                    'node_id': node_id,
                    # Here we use cast because mypy expects List[str] but [] is List[<nothing>].
                    'old_value': cast(List[str], []),
                    'new_value': [skill_id],
                }
            ),
        ]
        story_services.update_story(
            self.user_id_admin,
            story_id,
            change_list,
            'Added skill to node.',
        )

        with self.swap_to_always_return(
            question_services, 'get_total_question_count_for_skill_ids', 10
        ):
            topic_services.publish_story(
                self.TOPIC_ID, story_id, self.user_id_admin
            )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertTrue(
            any(
                ref.story_id == story_id and ref.story_is_published
                for ref in topic.canonical_story_references
            )
        )

    def test_publish_story_succeeds_when_no_acquired_skill_ids(self) -> None:
        story_id = 'stornoskls'
        exp_id = 'expnoskls'
        self._create_story_with_node_and_exploration(
            story_id, 'Story Without Skills', exp_id
        )

        topic_services.publish_story(
            self.TOPIC_ID, story_id, self.user_id_admin
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertTrue(
            any(
                ref.story_id == story_id and ref.story_is_published
                for ref in topic.canonical_story_references
            )
        )

    def test_update_topic(self) -> None:
        # Save a dummy image on filesystem, to be used as thumbnail.
        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'test_svg.svg'),
            'rb',
            encoding=None,
        ) as f:
            raw_image = f.read()
        fs = fs_services.GcsFileSystem(feconf.ENTITY_TYPE_TOPIC, self.TOPIC_ID)
        fs.commit(
            '%s/thumbnail.svg' % (constants.ASSET_TYPE_THUMBNAIL),
            raw_image,
            mimetype='image/svg+xml',
        )

        # Test whether an admin can edit a topic.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_DESCRIPTION,
                    'old_value': 'Description',
                    'new_value': 'New Description',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_ABBREVIATED_NAME,
                    'old_value': '',
                    'new_value': 'short-name',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_URL_FRAGMENT,
                    'old_value': '',
                    'new_value': 'url-name',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_THUMBNAIL_FILENAME,
                    'old_value': '',
                    'new_value': 'thumbnail.svg',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_THUMBNAIL_BG_COLOR,
                    'old_value': '',
                    'new_value': '#C6DCDA',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_META_TAG_CONTENT,
                    'old_value': '',
                    'new_value': 'topic meta tag content',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_PRACTICE_TAB_IS_DISPLAYED
                    ),
                    'old_value': False,
                    'new_value': True,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_PAGE_TITLE_FRAGMENT_FOR_WEB
                    ),
                    'old_value': '',
                    'new_value': 'topic page title',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_SKILL_IDS_FOR_DIAGNOSTIC_TEST
                    ),
                    'old_value': ['test_skill_id'],
                    'new_value': self.test_list,
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Updated Description.',
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        # Ruling out the possibility of None for mypy type checking.
        assert topic_summary is not None
        self.assertEqual(topic.description, 'New Description')
        self.assertEqual(topic.abbreviated_name, 'short-name')
        self.assertEqual(topic.url_fragment, 'url-name')
        self.assertEqual(topic.thumbnail_filename, 'thumbnail.svg')
        self.assertEqual(topic.thumbnail_size_in_bytes, len(raw_image))
        self.assertEqual(topic.thumbnail_bg_color, '#C6DCDA')
        self.assertEqual(topic.version, 3)
        self.assertEqual(topic.practice_tab_is_displayed, True)
        self.assertEqual(topic.meta_tag_content, 'topic meta tag content')
        self.assertEqual(topic.page_title_fragment_for_web, 'topic page title')
        self.assertEqual(topic.skill_ids_for_diagnostic_test, [])

        self.assertEqual(topic_summary.version, 3)
        self.assertEqual(topic_summary.thumbnail_filename, 'thumbnail.svg')
        self.assertEqual(topic_summary.thumbnail_bg_color, '#C6DCDA')

        # Test whether a topic_manager can update a dummy thumbnail_filename.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_THUMBNAIL_FILENAME,
                    'old_value': '',
                    'new_value': 'dummy_thumbnail.svg',
                }
            )
        ]
        with self.assertRaisesRegex(
            Exception,
            (
                'The thumbnail dummy_thumbnail.svg for topic with id '
                '%s does not exist in the filesystem.' % self.TOPIC_ID
            ),
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin,
                self.TOPIC_ID,
                changelist,
                'Updated thumbnail filename.',
            )

        # Test whether a topic_manager can edit a topic.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_NAME,
                    'old_value': 'Name',
                    'new_value': 'New Name',
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_a, self.TOPIC_ID, changelist, 'Updated Name.'
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        topic_summary = topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID)
        # Ruling out the possibility of None for mypy type checking.
        assert topic_summary is not None
        self.assertEqual(topic.name, 'New Name')
        self.assertEqual(topic.canonical_name, 'new name')
        self.assertEqual(topic.version, 4)
        self.assertEqual(topic_summary.name, 'New Name')
        self.assertEqual(topic_summary.version, 4)

    def test_simultaneous_subtopic_and_subtopic_page_changes(self) -> None:
        # Change the subtopic title first and then the subtopic page contents.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': 'title',
                    'subtopic_id': 1,
                    'old_value': 'Title',
                    'new_value': 'New Title',
                }
            ),
            subtopic_page_domain.SubtopicPageChange(
                {
                    'cmd': subtopic_page_domain.CMD_UPDATE_SUBTOPIC_PAGE_PROPERTY,
                    'property_name': (
                        subtopic_page_domain.SUBTOPIC_PAGE_PROPERTY_PAGE_CONTENTS_HTML
                    ),
                    'old_value': '',
                    'subtopic_id': 1,
                    'new_value': {
                        'html': '<p>New Value</p>',
                        'content_id': 'content',
                    },
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Update title and content of subtopic.',
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics), 1)
        self.assertEqual(topic.subtopics[0].title, 'New Title')
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 1
        )
        self.assertEqual(
            subtopic_page.page_contents.subtitled_html.html, '<p>New Value</p>'
        )
        study_guide = study_guide_services.get_study_guide_by_id(
            self.TOPIC_ID, 1
        )
        study_guide_section = study_guide.sections[0]
        self.assertEqual(study_guide_section.heading.unicode_str, 'New Title')
        self.assertEqual(study_guide_section.content.html, '<p>New Value</p>')

        # Change the subtopic page contents first and then the subtopic title.
        changelist = [
            subtopic_page_domain.SubtopicPageChange(
                {
                    'cmd': subtopic_page_domain.CMD_UPDATE_SUBTOPIC_PAGE_PROPERTY,
                    'property_name': (
                        (
                            subtopic_page_domain.SUBTOPIC_PAGE_PROPERTY_PAGE_CONTENTS_HTML
                        )
                    ),
                    'old_value': '<p>New Value</p>',
                    'subtopic_id': 1,
                    'new_value': {
                        'html': '<p>Another New Value</p>',
                        'content_id': 'content',
                    },
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': 'title',
                    'subtopic_id': 1,
                    'old_value': 'New Title',
                    'new_value': 'Another New Title',
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Again update title and content of subtopic.',
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics), 1)
        self.assertEqual(topic.subtopics[0].title, 'Another New Title')
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 1
        )
        self.assertEqual(
            subtopic_page.page_contents.subtitled_html.html,
            '<p>Another New Value</p>',
        )
        study_guide = study_guide_services.get_study_guide_by_id(
            self.TOPIC_ID, 1
        )
        study_guide_section = study_guide.sections[0]
        self.assertEqual(
            study_guide_section.heading.unicode_str, 'Another New Title'
        )
        self.assertEqual(
            study_guide_section.content.html, '<p>Another New Value</p>'
        )

    def test_update_topic_and_subtopic_page(self) -> None:
        changelist: List[
            Union[
                topic_domain.TopicChange,
                subtopic_page_domain.SubtopicPageChange,
                study_guide_domain.StudyGuideChange,
            ]
        ] = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title3',
                    'subtopic_id': 3,
                    'url_fragment': 'fragment-three',
                }
            )
        ]
        with self.assertRaisesRegex(
            Exception,
            'The given new subtopic id 3 is not equal to '
            'the expected next subtopic id: 2',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin, self.TOPIC_ID, changelist, 'Added subtopic.'
            )

        # Test whether the subtopic page was created for the above failed
        # attempt.
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 3, strict=False
        )
        self.assertIsNone(subtopic_page)

        # Test exception raised for simultaneous adding and removing of
        # subtopics.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title2',
                    'subtopic_id': 2,
                    'url_fragment': 'fragment-two',
                }
            ),
            topic_domain.TopicChange(
                {'cmd': topic_domain.CMD_DELETE_SUBTOPIC, 'subtopic_id': 2}
            ),
        ]
        with self.assertRaisesRegex(
            Exception,
            'The incoming changelist had simultaneous'
            ' creation and deletion of subtopics.',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin,
                self.TOPIC_ID,
                changelist,
                'Added and deleted a subtopic.',
            )

        # Test whether a subtopic page already existing in datastore can be
        # edited.
        changelist = [
            subtopic_page_domain.SubtopicPageChange(
                {
                    'cmd': subtopic_page_domain.CMD_UPDATE_SUBTOPIC_PAGE_PROPERTY,
                    'property_name': (
                        subtopic_page_domain.SUBTOPIC_PAGE_PROPERTY_PAGE_CONTENTS_HTML
                    ),
                    'old_value': '',
                    'subtopic_id': 1,
                    'new_value': {
                        'html': '<p>New Value</p>',
                        'content_id': 'content',
                    },
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin, self.TOPIC_ID, changelist, 'Updated html data'
        )
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 1
        )
        self.assertEqual(
            subtopic_page.page_contents.subtitled_html.html, '<p>New Value</p>'
        )

        # Test a sequence of changes with both topic and subtopic page changes.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title2',
                    'subtopic_id': 2,
                    'url_fragment': 'fragment-two',
                }
            ),
            topic_domain.TopicChange(
                {'cmd': topic_domain.CMD_DELETE_SUBTOPIC, 'subtopic_id': 1}
            ),
            subtopic_page_domain.SubtopicPageChange(
                {
                    'cmd': subtopic_page_domain.CMD_UPDATE_SUBTOPIC_PAGE_PROPERTY,
                    'property_name': (
                        subtopic_page_domain.SUBTOPIC_PAGE_PROPERTY_PAGE_CONTENTS_HTML
                    ),
                    'old_value': {'html': '', 'content_id': 'content'},
                    'subtopic_id': 2,
                    'new_value': {
                        'html': '<p>New Value</p>',
                        'content_id': 'content',
                    },
                }
            ),
            subtopic_page_domain.SubtopicPageChange(
                {
                    'cmd': subtopic_page_domain.CMD_UPDATE_SUBTOPIC_PAGE_PROPERTY,
                    'property_name': (
                        subtopic_page_domain.SUBTOPIC_PAGE_PROPERTY_PAGE_CONTENTS_AUDIO
                    ),
                    'old_value': {'voiceovers_mapping': {'content': {}}},
                    'new_value': {
                        'voiceovers_mapping': {
                            'content': {
                                'en': {
                                    'filename': 'test.mp3',
                                    'file_size_bytes': 100,
                                    'needs_update': False,
                                    'duration_secs': 0.3,
                                }
                            }
                        }
                    },
                    'subtopic_id': 2,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': 2,
                    'skill_id': self.skill_id_1,
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Added and removed a subtopic.',
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics), 1)
        self.assertEqual(topic.next_subtopic_id, 3)
        self.assertEqual(topic.subtopics[0].title, 'Title2')
        self.assertEqual(topic.subtopics[0].skill_ids, [self.skill_id_1])

        # Test whether the subtopic page corresponding to the deleted subtopic
        # was also deleted.
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 1, strict=False
        )
        self.assertIsNone(subtopic_page)
        # Validate the newly created subtopic page.
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 2, strict=False
        )
        # Ruling out the possibility of None for mypy type checking.
        assert subtopic_page is not None
        self.assertEqual(
            subtopic_page.page_contents.subtitled_html.html, '<p>New Value</p>'
        )
        self.assertEqual(
            subtopic_page.page_contents.recorded_voiceovers.to_dict(),
            {
                'voiceovers_mapping': {
                    'content': {
                        'en': {
                            'filename': 'test.mp3',
                            'file_size_bytes': 100,
                            'needs_update': False,
                            'duration_secs': 0.3,
                        }
                    }
                }
            },
        )

        # Making sure everything resets when an error is encountered anywhere.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title3',
                    'subtopic_id': 3,
                    'url_fragment': 'fragment-three',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title4',
                    'subtopic_id': 4,
                    'url_fragment': 'fragment-four',
                }
            ),
            topic_domain.TopicChange(
                {'cmd': topic_domain.CMD_DELETE_SUBTOPIC, 'subtopic_id': 2}
            ),
            # The following is an invalid command as subtopic with id 2 was
            # deleted in previous step.
            subtopic_page_domain.SubtopicPageChange(
                {
                    'cmd': subtopic_page_domain.CMD_UPDATE_SUBTOPIC_PAGE_PROPERTY,
                    'property_name': (
                        subtopic_page_domain.SUBTOPIC_PAGE_PROPERTY_PAGE_CONTENTS_HTML
                    ),
                    'old_value': '',
                    'subtopic_id': 2,
                    'new_value': {
                        'html': '<p>New Value</p>',
                        'content_id': 'content',
                    },
                }
            ),
        ]
        with self.assertRaisesRegex(
            Exception, 'The subtopic with id 2 doesn\'t exist'
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin,
                self.TOPIC_ID,
                changelist,
                'Done some changes.',
            )

        # Make sure the topic object in datastore is not affected.
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics), 1)
        self.assertEqual(topic.next_subtopic_id, 3)
        self.assertEqual(topic.subtopics[0].title, 'Title2')
        self.assertEqual(topic.subtopics[0].skill_ids, [self.skill_id_1])

        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 3, strict=False
        )
        self.assertIsNone(subtopic_page)
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 4, strict=False
        )
        self.assertIsNone(subtopic_page)
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 2, strict=False
        )
        self.assertIsNotNone(subtopic_page)

    @test_utils.enable_feature_flags(
        [feature_flag_list.FeatureNames.SHOW_RESTRUCTURED_STUDY_GUIDES]
    )
    def test_update_topic_and_study_guide(self) -> None:
        changelist: List[
            Union[topic_domain.TopicChange, study_guide_domain.StudyGuideChange]
        ] = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title3',
                    'subtopic_id': 3,
                    'url_fragment': 'fragment-three',
                }
            )
        ]
        with self.assertRaisesRegex(
            Exception,
            'The given new subtopic id 3 is not equal to '
            'the expected next subtopic id: 2',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin, self.TOPIC_ID, changelist, 'Added subtopic.'
            )

        # Test whether the study guide was created for the above failed
        # attempt.
        study_guide = study_guide_services.get_study_guide_by_id(
            self.TOPIC_ID, 3, strict=False
        )
        self.assertIsNone(study_guide)

        # Test exception raised for simultaneous adding and removing of
        # subtopics.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title2',
                    'subtopic_id': 2,
                    'url_fragment': 'fragment-two',
                }
            ),
            topic_domain.TopicChange(
                {'cmd': topic_domain.CMD_DELETE_SUBTOPIC, 'subtopic_id': 2}
            ),
        ]
        with self.assertRaisesRegex(
            Exception,
            'The incoming changelist had simultaneous'
            ' creation and deletion of subtopics.',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin,
                self.TOPIC_ID,
                changelist,
                'Added and deleted a subtopic.',
            )

        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title2',
                    'subtopic_id': 2,
                    'url_fragment': 'fragment-two',
                }
            ),
            study_guide_domain.StudyGuideChange(
                {
                    'cmd': study_guide_domain.CMD_UPDATE_STUDY_GUIDE_PROPERTY,
                    'property_name': (
                        study_guide_domain.STUDY_GUIDE_PROPERTY_SECTIONS
                    ),
                    'subtopic_id': 2,
                    'old_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': '',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '',
                            },
                        }
                    ],
                    'new_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': 'Heading',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '<p>Content</p>',
                            },
                        }
                    ],
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin, self.TOPIC_ID, changelist, 'Added a subtopic'
        )

        # Test whether a study guide already existing in datastore can be
        # edited.
        changelist = [
            study_guide_domain.StudyGuideChange(
                {
                    'cmd': study_guide_domain.CMD_UPDATE_STUDY_GUIDE_PROPERTY,
                    'property_name': (
                        study_guide_domain.STUDY_GUIDE_PROPERTY_SECTIONS
                    ),
                    'subtopic_id': 2,
                    'old_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': 'Heading',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '<p>Content</p>',
                            },
                        }
                    ],
                    'new_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': 'New Heading',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '<p>New content</p>',
                            },
                        }
                    ],
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin, self.TOPIC_ID, changelist, 'Updated html data'
        )
        study_guide = study_guide_services.get_study_guide_by_id(
            self.TOPIC_ID, 2
        )
        self.assertEqual(
            study_guide.sections[0].to_dict(),
            {
                'heading': {
                    'content_id': 'section_heading_0',
                    'unicode_str': 'New Heading',
                },
                'content': {
                    'content_id': 'section_content_1',
                    'html': '<p>New content</p>',
                },
            },
        )

        # Test a sequence of changes with both topic and study guide changes.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title3',
                    'subtopic_id': 3,
                    'url_fragment': 'fragment-three',
                }
            ),
            topic_domain.TopicChange(
                {'cmd': topic_domain.CMD_DELETE_SUBTOPIC, 'subtopic_id': 2}
            ),
            study_guide_domain.StudyGuideChange(
                {
                    'cmd': study_guide_domain.CMD_UPDATE_STUDY_GUIDE_PROPERTY,
                    'property_name': (
                        study_guide_domain.STUDY_GUIDE_PROPERTY_SECTIONS
                    ),
                    'subtopic_id': 3,
                    'old_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': 'heading',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': 'content',
                            },
                        }
                    ],
                    'new_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': 'New Heading',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '<p>New content</p>',
                            },
                        }
                    ],
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': 3,
                    'skill_id': self.skill_id_1,
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Added and removed a subtopic.',
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics), 2)
        self.assertEqual(topic.next_subtopic_id, 4)
        self.assertEqual(topic.subtopics[1].title, 'Title3')
        self.assertEqual(topic.subtopics[1].skill_ids, [self.skill_id_1])

        # Test whether the study guide corresponding to the deleted subtopic
        # was also deleted.
        study_guide = study_guide_services.get_study_guide_by_id(
            self.TOPIC_ID, 2, strict=False
        )
        self.assertIsNone(study_guide)
        # Validate the newly created study guide.
        study_guide = study_guide_services.get_study_guide_by_id(
            self.TOPIC_ID, 3, strict=False
        )
        # Ruling out the possibility of None for mypy type checking.
        assert study_guide is not None
        self.assertEqual(
            study_guide.sections[0].to_dict(),
            {
                'heading': {
                    'content_id': 'section_heading_0',
                    'unicode_str': 'New Heading',
                },
                'content': {
                    'content_id': 'section_content_1',
                    'html': '<p>New content</p>',
                },
            },
        )

        # Making sure everything resets when an error is encountered anywhere.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title4',
                    'subtopic_id': 4,
                    'url_fragment': 'fragment-four',
                }
            ),
            study_guide_domain.StudyGuideChange(
                {
                    'cmd': study_guide_domain.CMD_UPDATE_STUDY_GUIDE_PROPERTY,
                    'property_name': (
                        study_guide_domain.STUDY_GUIDE_PROPERTY_SECTIONS
                    ),
                    'subtopic_id': 4,
                    'old_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': '',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '',
                            },
                        }
                    ],
                    'new_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': 'Heading',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '<p>Content</p>',
                            },
                        }
                    ],
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title5',
                    'subtopic_id': 5,
                    'url_fragment': 'fragment-five',
                }
            ),
            study_guide_domain.StudyGuideChange(
                {
                    'cmd': study_guide_domain.CMD_UPDATE_STUDY_GUIDE_PROPERTY,
                    'property_name': (
                        study_guide_domain.STUDY_GUIDE_PROPERTY_SECTIONS
                    ),
                    'subtopic_id': 5,
                    'old_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': '',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '',
                            },
                        }
                    ],
                    'new_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': 'Heading',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '<p>Content</p>',
                            },
                        }
                    ],
                }
            ),
            topic_domain.TopicChange(
                {'cmd': topic_domain.CMD_DELETE_SUBTOPIC, 'subtopic_id': 3}
            ),
            # The following is an invalid command as subtopic with id 3 was
            # deleted in previous step.
            study_guide_domain.StudyGuideChange(
                {
                    'cmd': study_guide_domain.CMD_UPDATE_STUDY_GUIDE_PROPERTY,
                    'property_name': (
                        study_guide_domain.STUDY_GUIDE_PROPERTY_SECTIONS
                    ),
                    'old_value': '',
                    'subtopic_id': 3,
                    'new_value': [
                        {
                            'heading': {
                                'content_id': 'section_heading_0',
                                'unicode_str': 'New Heading',
                            },
                            'content': {
                                'content_id': 'section_content_1',
                                'html': '<p>New content</p>',
                            },
                        }
                    ],
                }
            ),
        ]
        with self.assertRaisesRegex(
            Exception, 'The subtopic with id 3 doesn\'t exist'
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin,
                self.TOPIC_ID,
                changelist,
                'Done some changes.',
            )

        # Make sure the topic object in datastore is not affected.
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.subtopics), 2)
        self.assertEqual(topic.next_subtopic_id, 4)
        self.assertEqual(topic.subtopics[1].title, 'Title3')

        study_guide = study_guide_services.get_study_guide_by_id(
            self.TOPIC_ID, 2, strict=False
        )
        self.assertIsNone(study_guide)
        study_guide = study_guide_services.get_study_guide_by_id(
            self.TOPIC_ID, 4, strict=False
        )
        self.assertIsNone(study_guide)
        study_guide = study_guide_services.get_study_guide_by_id(
            self.TOPIC_ID, 3, strict=False
        )
        self.assertIsNotNone(study_guide)

    def test_update_subtopic_page_without_study_guide(self) -> None:
        # We create a topic and subtopic in this way so as to not create a
        # study guide.
        topic_id_1 = topic_fetchers.get_new_topic_id()
        topic_1 = topic_domain.Topic.create_default_topic(
            topic_id_1,
            'Dummy Topic 1',
            'dummy-topic-one',
            'description',
            'fragm',
        )
        topic_1.add_subtopic(1, 'Dummy Subtopic Title', 'dummysubtopic')
        topic_1.update_meta_tag_content('dummy-meta')
        subtopic_page = (
            subtopic_page_domain.SubtopicPage.create_default_subtopic_page(
                1, topic_id_1
            )
        )
        topic_services.save_new_topic(self.user_id, topic_1)
        subtopic_page_services.save_subtopic_page(
            self.user_id,
            subtopic_page,
            'Added subtopic',
            [
                topic_domain.TopicChange(
                    {
                        'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                        'subtopic_id': 1,
                        'title': 'Dummy Subtopic Title',
                        'url_fragment': 'dummy-fragment',
                    }
                )
            ],
        )

        # Check that study guide does not exist.
        with self.assertRaisesRegex(
            Exception,
            'Entity for class StudyGuideModel with id %s not found'
            % (f'{topic_id_1}-1'),
        ):
            study_guide_services.get_study_guide_by_id(topic_id_1, 1)

        # Test whether a subtopic page without corresponding study guide
        # already existing in datastore can be edited.
        changelist = [
            subtopic_page_domain.SubtopicPageChange(
                {
                    'cmd': subtopic_page_domain.CMD_UPDATE_SUBTOPIC_PAGE_PROPERTY,
                    'property_name': (
                        subtopic_page_domain.SUBTOPIC_PAGE_PROPERTY_PAGE_CONTENTS_HTML
                    ),
                    'old_value': '',
                    'subtopic_id': 1,
                    'new_value': {
                        'html': '<p>New Value</p>',
                        'content_id': 'content',
                    },
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin, topic_id_1, changelist, 'Updated html data'
        )
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            topic_id_1, 1
        )
        self.assertEqual(
            subtopic_page.page_contents.subtitled_html.html, '<p>New Value</p>'
        )

    def test_update_topic_schema(self) -> None:
        orig_topic_dict = topic_fetchers.get_topic_by_id(
            self.TOPIC_ID
        ).to_dict()

        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MIGRATE_SUBTOPIC_SCHEMA_TO_LATEST_VERSION,
                    'from_version': 2,
                    'to_version': 3,
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin, self.TOPIC_ID, changelist, 'Update schema.'
        )

        new_topic_dict = topic_fetchers.get_topic_by_id(self.TOPIC_ID).to_dict()

        # Check version is updated.
        self.assertEqual(new_topic_dict['version'], 3)

        # Delete version and check that the two dicts are the same.
        del orig_topic_dict['version']
        del new_topic_dict['version']
        self.assertEqual(orig_topic_dict, new_topic_dict)

    def test_add_uncategorized_skill(self) -> None:
        topic_services.add_uncategorized_skill(
            self.user_id_admin, self.TOPIC_ID, 'skill_id_3'
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(
            topic.uncategorized_skill_ids,
            [self.skill_id_1, self.skill_id_2, 'skill_id_3'],
        )
        topic_commit_log_entry = (
            topic_models.TopicCommitLogEntryModel.get_commit(self.TOPIC_ID, 3)
        )
        # Ruling out the possibility of None for mypy type checking.
        assert topic_commit_log_entry is not None
        self.assertEqual(topic_commit_log_entry.commit_type, 'edit')
        self.assertEqual(topic_commit_log_entry.topic_id, self.TOPIC_ID)
        self.assertEqual(topic_commit_log_entry.user_id, self.user_id_admin)
        self.assertEqual(
            topic_commit_log_entry.commit_message,
            'Added skill_id_3 to uncategorized skill ids',
        )

    def test_delete_uncategorized_skill(self) -> None:
        topic_services.delete_uncategorized_skill(
            self.user_id_admin, self.TOPIC_ID, self.skill_id_1
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(topic.uncategorized_skill_ids, [self.skill_id_2])
        topic_commit_log_entry = (
            topic_models.TopicCommitLogEntryModel.get_commit(self.TOPIC_ID, 3)
        )
        # Ruling out the possibility of None for mypy type checking.
        assert topic_commit_log_entry is not None
        self.assertEqual(topic_commit_log_entry.commit_type, 'edit')
        self.assertEqual(topic_commit_log_entry.topic_id, self.TOPIC_ID)
        self.assertEqual(topic_commit_log_entry.user_id, self.user_id_admin)
        self.assertEqual(
            topic_commit_log_entry.commit_message,
            'Removed %s from uncategorized skill ids' % self.skill_id_1,
        )

    @test_utils.enable_feature_flags(
        [
            feature_flag_list.FeatureNames.ENABLE_TRANSLATION_OPPORTUNITIES_WITH_NEW_OPP_MODELS
        ]
    )
    def test_add_and_delete_uncategorized_skill_with_new_models(self) -> None:
        self.save_new_skill(
            'skill_id_3', self.user_id_admin, description='Skill 3'
        )
        topic_services.add_uncategorized_skill(
            self.user_id_admin, self.TOPIC_ID, 'skill_id_3'
        )
        model = opportunity_models.TranslationOpportunityModel.get(
            'skill.skill_id_3', strict=False
        )
        self.assertIsNotNone(model)
        assert model is not None
        self.assertEqual(model.topic_ids, [self.TOPIC_ID])
        self.assertEqual(model.content_count, 1)
        self.assertEqual(model.translation_counts, {})

        topic_services.delete_uncategorized_skill(
            self.user_id_admin, self.TOPIC_ID, 'skill_id_3'
        )
        model = opportunity_models.TranslationOpportunityModel.get(
            'skill.skill_id_3', strict=False
        )
        assert model is not None
        self.assertEqual(model.topic_ids, [])
        self.assertEqual(model.content_count, 1)
        self.assertEqual(model.translation_counts, {})

    @test_utils.enable_feature_flags(
        [
            feature_flag_list.FeatureNames.ENABLE_TRANSLATION_OPPORTUNITIES_WITH_NEW_OPP_MODELS
        ]
    )
    def test_delete_uncategorized_skill_when_skill_in_another_topic_with_new_models(
        self,
    ) -> None:
        topic_id_2 = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            topic_id_2,
            self.user_id,
            name='Topic Two',
            description='Description',
            url_fragment='topic-two',
        )
        self.save_new_skill(
            'skill_id_4', self.user_id_admin, description='Skill 4'
        )
        topic_services.add_uncategorized_skill(
            self.user_id_admin, self.TOPIC_ID, 'skill_id_4'
        )
        topic_services.add_uncategorized_skill(
            self.user_id_admin, topic_id_2, 'skill_id_4'
        )
        model = opportunity_models.TranslationOpportunityModel.get(
            'skill.skill_id_4', strict=False
        )
        self.assertIsNotNone(model)
        assert model is not None
        self.assertEqual(
            sorted(model.topic_ids), sorted([self.TOPIC_ID, topic_id_2])
        )
        self.assertEqual(model.content_count, 1)
        self.assertEqual(model.translation_counts, {})

        topic_services.delete_uncategorized_skill(
            self.user_id_admin, self.TOPIC_ID, 'skill_id_4'
        )
        model = opportunity_models.TranslationOpportunityModel.get(
            'skill.skill_id_4', strict=False
        )
        self.assertIsNotNone(model)
        assert model is not None
        self.assertEqual(model.topic_ids, [topic_id_2])
        self.assertEqual(model.content_count, 1)
        self.assertEqual(model.translation_counts, {})

    def test_delete_canonical_story(self) -> None:
        topic_services.delete_canonical_story(
            self.user_id_admin, self.TOPIC_ID, self.story_id_1
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.canonical_story_references), 1)
        self.assertEqual(
            topic.canonical_story_references[0].story_id, self.story_id_2
        )
        topic_commit_log_entry = (
            topic_models.TopicCommitLogEntryModel.get_commit(self.TOPIC_ID, 3)
        )
        # Ruling out the possibility of None for mypy type checking.
        assert topic_commit_log_entry is not None
        self.assertEqual(topic_commit_log_entry.commit_type, 'edit')
        self.assertEqual(topic_commit_log_entry.topic_id, self.TOPIC_ID)
        self.assertEqual(topic_commit_log_entry.user_id, self.user_id_admin)
        self.assertEqual(
            topic_commit_log_entry.commit_message,
            'Removed %s from canonical story ids' % self.story_id_1,
        )

    def test_add_canonical_story(self) -> None:
        topic_services.add_canonical_story(
            self.user_id_admin, self.TOPIC_ID, 'story_id'
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.canonical_story_references), 3)
        self.assertEqual(
            topic.canonical_story_references[2].story_id, 'story_id'
        )
        topic_commit_log_entry = (
            topic_models.TopicCommitLogEntryModel.get_commit(self.TOPIC_ID, 3)
        )
        # Ruling out the possibility of None for mypy type checking.
        assert topic_commit_log_entry is not None
        self.assertEqual(topic_commit_log_entry.commit_type, 'edit')
        self.assertEqual(topic_commit_log_entry.topic_id, self.TOPIC_ID)
        self.assertEqual(topic_commit_log_entry.user_id, self.user_id_admin)
        self.assertEqual(
            topic_commit_log_entry.commit_message,
            'Added %s to canonical story ids' % 'story_id',
        )

    def test_delete_additional_story(self) -> None:
        topic_services.delete_additional_story(
            self.user_id_admin, self.TOPIC_ID, self.story_id_3
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.additional_story_references), 0)

        topic_commit_log_entry = (
            topic_models.TopicCommitLogEntryModel.get_commit(self.TOPIC_ID, 3)
        )
        # Ruling out the possibility of None for mypy type checking.
        assert topic_commit_log_entry is not None
        self.assertEqual(topic_commit_log_entry.commit_type, 'edit')
        self.assertEqual(topic_commit_log_entry.topic_id, self.TOPIC_ID)
        self.assertEqual(topic_commit_log_entry.user_id, self.user_id_admin)
        self.assertEqual(
            topic_commit_log_entry.commit_message,
            'Removed %s from additional story ids' % self.story_id_3,
        )

    def test_add_additional_story(self) -> None:
        topic_services.add_additional_story(
            self.user_id_admin, self.TOPIC_ID, 'story_id_4'
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(len(topic.additional_story_references), 2)
        self.assertEqual(
            topic.additional_story_references[1].story_id, 'story_id_4'
        )
        topic_commit_log_entry = (
            topic_models.TopicCommitLogEntryModel.get_commit(self.TOPIC_ID, 3)
        )
        # Ruling out the possibility of None for mypy type checking.
        assert topic_commit_log_entry is not None
        self.assertEqual(topic_commit_log_entry.commit_type, 'edit')
        self.assertEqual(topic_commit_log_entry.topic_id, self.TOPIC_ID)
        self.assertEqual(topic_commit_log_entry.user_id, self.user_id_admin)
        self.assertEqual(
            topic_commit_log_entry.commit_message,
            'Added story_id_4 to additional story ids',
        )

    def test_delete_topic(self) -> None:
        # Add suggestion for the topic to test if it is deleted too.
        content_id_generator = translation_domain.ContentIdGenerator()
        question = self.save_new_question(
            'question_id',
            self.user_id_admin,
            self._create_valid_question_data('dest', content_id_generator),
            [self.skill_id_1],
            content_id_generator.next_content_id_index,
        )
        suggestion = suggestion_services.create_suggestion(
            feconf.SUGGESTION_TYPE_ADD_QUESTION,
            feconf.ENTITY_TYPE_TOPIC,
            self.TOPIC_ID,
            1,
            self.user_id_admin,
            {
                'cmd': question_domain.CMD_CREATE_NEW_FULLY_SPECIFIED_QUESTION,
                'skill_difficulty': 0.3,
                'skill_id': self.skill_id_1,
                'question_dict': question.to_dict(),
            },
            'change',
        )

        self.assertIsNotNone(
            suggestion_services.get_suggestion_by_id(suggestion.suggestion_id)
        )

        topic_services.delete_topic(self.user_id_admin, self.TOPIC_ID)
        self.assertIsNone(
            topic_fetchers.get_topic_by_id(self.TOPIC_ID, strict=False)
        )
        self.assertIsNone(
            topic_fetchers.get_topic_summary_by_id(self.TOPIC_ID, strict=False)
        )
        self.assertIsNone(
            subtopic_page_services.get_subtopic_page_by_id(
                self.TOPIC_ID, 1, strict=False
            )
        )
        self.assertIsNone(
            suggestion_services.get_suggestion_by_id(
                suggestion.suggestion_id, strict=False
            )
        )

    @test_utils.enable_feature_flags(
        [
            feature_flag_list.FeatureNames.ENABLE_TRANSLATION_OPPORTUNITIES_WITH_NEW_OPP_MODELS
        ]
    )
    def test_delete_topic_with_new_models(self) -> None:
        topic_id_2 = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            topic_id_2,
            self.user_id,
            name='Topic Two',
            description='Description',
            url_fragment='topic-two',
        )
        self.save_new_skill(
            'skill_id_5', self.user_id_admin, description='Skill 5'
        )
        self.save_new_skill(
            'skill_id_6', self.user_id_admin, description='Skill 6'
        )
        topic_services.add_uncategorized_skill(
            self.user_id_admin, self.TOPIC_ID, 'skill_id_5'
        )
        topic_services.add_uncategorized_skill(
            self.user_id_admin, self.TOPIC_ID, 'skill_id_6'
        )
        topic_services.add_uncategorized_skill(
            self.user_id_admin, topic_id_2, 'skill_id_6'
        )
        model_5_before = opportunity_models.TranslationOpportunityModel.get(
            'skill.skill_id_5', strict=False
        )
        self.assertIsNotNone(model_5_before)
        assert model_5_before is not None
        self.assertEqual(model_5_before.topic_ids, [self.TOPIC_ID])
        self.assertEqual(model_5_before.content_count, 1)

        model_6_before = opportunity_models.TranslationOpportunityModel.get(
            'skill.skill_id_6', strict=False
        )
        self.assertIsNotNone(model_6_before)
        assert model_6_before is not None
        self.assertEqual(
            sorted(model_6_before.topic_ids),
            sorted([self.TOPIC_ID, topic_id_2]),
        )
        self.assertEqual(model_6_before.content_count, 1)

        topic_services.delete_topic(self.user_id_admin, self.TOPIC_ID)

        model_5 = opportunity_models.TranslationOpportunityModel.get(
            'skill.skill_id_5', strict=False
        )
        assert model_5 is not None
        self.assertEqual(model_5.topic_ids, [])
        self.assertEqual(model_5.content_count, 1)

        model_6 = opportunity_models.TranslationOpportunityModel.get(
            'skill.skill_id_6', strict=False
        )
        assert model_6 is not None
        self.assertEqual(model_6.topic_ids, [topic_id_2])
        self.assertEqual(model_6.content_count, 1)

    def test_delete_subtopic_with_skill_ids(self) -> None:
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_DELETE_SUBTOPIC,
                    'subtopic_id': self.subtopic_id,
                }
            )
        ]
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            self.TOPIC_ID, 1, strict=True
        )
        self.assertEqual(subtopic_page.id, self.TOPIC_ID + '-1')
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin, self.TOPIC_ID, changelist, 'Removed 1 subtopic.'
        )
        subtopic_page_with_none = (
            subtopic_page_services.get_subtopic_page_by_id(
                self.TOPIC_ID, 1, strict=False
            )
        )
        self.assertIsNone(subtopic_page_with_none)
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(
            topic.uncategorized_skill_ids, [self.skill_id_1, self.skill_id_2]
        )
        self.assertEqual(topic.subtopics, [])

    def test_update_subtopic_skill_ids(self) -> None:
        # Adds a subtopic and moves skill id from one to another.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': self.subtopic_id,
                    'skill_id': self.skill_id_1,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': self.subtopic_id,
                    'skill_id': self.skill_id_2,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title2',
                    'subtopic_id': 2,
                    'url_fragment': 'fragment-two',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': self.subtopic_id,
                    'new_subtopic_id': 2,
                    'skill_id': self.skill_id_2,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': topic_domain.SUBTOPIC_PROPERTY_URL_FRAGMENT,
                    'new_value': 'new-subtopic',
                    'old_value': '',
                    'subtopic_id': 2,
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Updated subtopic skill ids.',
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        subtopic_page = subtopic_page_services.get_subtopic_page_by_id(
            topic.id, 2
        )
        self.assertEqual(topic.uncategorized_skill_ids, [])
        self.assertEqual(topic.subtopics[0].skill_ids, [self.skill_id_1])
        self.assertEqual(topic.subtopics[1].skill_ids, [self.skill_id_2])
        self.assertEqual(topic.subtopics[1].id, 2)
        self.assertEqual(topic.next_subtopic_id, 3)
        self.assertEqual(subtopic_page.topic_id, topic.id)
        self.assertEqual(subtopic_page.id, self.TOPIC_ID + '-2')

        # Tests invalid case where skill id is not present in the old subtopic.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': self.subtopic_id,
                    'new_subtopic_id': 2,
                    'skill_id': self.skill_id_2,
                }
            )
        ]
        with self.assertRaisesRegex(
            Exception,
            'Skill id %s is not present in the given old subtopic'
            % self.skill_id_2,
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin,
                self.TOPIC_ID,
                changelist,
                'Updated subtopic skill ids.',
            )

        # Tests invalid case where skill id is not an uncategorized skill id.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': 2,
                    'skill_id': 'skill_10',
                }
            )
        ]
        with self.assertRaisesRegex(
            Exception, 'Skill id skill_10 is not an uncategorized skill id'
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin,
                self.TOPIC_ID,
                changelist,
                'Updated subtopic skill ids.',
            )

        # Tests invalid case where target subtopic doesn't exist.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': self.subtopic_id,
                    'new_subtopic_id': None,
                    'skill_id': self.skill_id_1,
                }
            )
        ]
        with self.assertRaisesRegex(
            Exception, 'The subtopic with id None does not exist.'
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin,
                self.TOPIC_ID,
                changelist,
                'Updated subtopic skill ids.',
            )

        # Tests valid case skill id removal case.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_REMOVE_SKILL_ID_FROM_SUBTOPIC,
                    'subtopic_id': 2,
                    'skill_id': self.skill_id_2,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_REMOVE_SKILL_ID_FROM_SUBTOPIC,
                    'subtopic_id': self.subtopic_id,
                    'skill_id': self.skill_id_1,
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Updated subtopic skill ids.',
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(
            topic.uncategorized_skill_ids, [self.skill_id_2, self.skill_id_1]
        )
        self.assertEqual(topic.subtopics[1].skill_ids, [])
        self.assertEqual(topic.subtopics[0].skill_ids, [])

        # Tests invalid case where skill id is not present in the subtopic
        # from which it is to be removed.
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_REMOVE_SKILL_ID_FROM_SUBTOPIC,
                    'subtopic_id': self.subtopic_id,
                    'skill_id': 'skill_10',
                }
            )
        ]
        with self.assertRaisesRegex(
            Exception, 'Skill id skill_10 is not present in the old subtopic'
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id_admin,
                self.TOPIC_ID,
                changelist,
                'Updated subtopic skill ids.',
            )

    def test_admin_can_manage_topic(self) -> None:
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)

        self.assertTrue(
            topic_services.check_can_edit_topic(self.user_admin, topic_rights)
        )

    def test_admin_can_edit_questions_in_topic(self) -> None:
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)

        self.assertTrue(
            topic_services.check_can_edit_question(
                self.user_admin, topic_rights
            )
        )

    def test_filter_published_topic_ids(self) -> None:
        published_topic_ids = topic_services.filter_published_topic_ids(
            [self.TOPIC_ID, 'invalid_id']
        )
        self.assertEqual(len(published_topic_ids), 0)
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': self.subtopic_id,
                    'skill_id': 'skill_1',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_SKILL_IDS_FOR_DIAGNOSTIC_TEST
                    ),
                    'old_value': self.test_list,
                    'new_value': ['skill_1'],
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Updated subtopic skill ids.',
        )
        topic_services.publish_topic(self.TOPIC_ID, self.user_id_admin)
        published_topic_ids = topic_services.filter_published_topic_ids(
            [self.TOPIC_ID, 'invalid_id']
        )
        self.assertEqual(len(published_topic_ids), 1)
        self.assertEqual(published_topic_ids[0], self.TOPIC_ID)

    def test_publish_and_unpublish_topic(self) -> None:
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertFalse(topic_rights.topic_is_published)
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': self.subtopic_id,
                    'skill_id': 'skill_1',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_SKILL_IDS_FOR_DIAGNOSTIC_TEST
                    ),
                    'old_value': self.test_list,
                    'new_value': ['skill_1'],
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Updated subtopic skill ids.',
        )
        topic_services.publish_topic(self.TOPIC_ID, self.user_id_admin)

        with self.assertRaisesRegex(
            Exception,
            'The user does not have enough rights to unpublish the topic.',
        ):
            topic_services.unpublish_topic(self.TOPIC_ID, self.user_id_a)

        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertTrue(topic_rights.topic_is_published)

        topic_services.unpublish_topic(self.TOPIC_ID, self.user_id_admin)
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertFalse(topic_rights.topic_is_published)

        with self.assertRaisesRegex(
            Exception,
            'The user does not have enough rights to publish the topic.',
        ):
            topic_services.publish_topic(self.TOPIC_ID, self.user_id_a)

    def test_create_new_topic_rights(self) -> None:
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)

        self.assertTrue(
            topic_services.check_can_edit_topic(self.user_a, topic_rights)
        )
        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_b, topic_rights)
        )

    def test_non_admin_cannot_assign_roles(self) -> None:
        self.signup('x@example.com', 'X')
        self.signup('y@example.com', 'Y')

        user_id_x = self.get_user_id_from_email('x@example.com')
        user_id_y = self.get_user_id_from_email('y@example.com')

        user_x = user_services.get_user_actions_info(user_id_x)
        user_y = user_services.get_user_actions_info(user_id_y)
        with self.assertRaisesRegex(
            Exception, 'UnauthorizedUserException: Could not assign new role.'
        ):
            topic_services.assign_role(
                user_y, user_x, topic_domain.ROLE_MANAGER, self.TOPIC_ID
            )

        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertFalse(
            topic_services.check_can_edit_topic(user_x, topic_rights)
        )
        self.assertFalse(
            topic_services.check_can_edit_topic(user_y, topic_rights)
        )

    def test_guest_user_cannot_assign_roles(self) -> None:
        guest_user = user_services.get_user_actions_info(None)
        with self.assertRaisesRegex(
            Exception, 'Guest user is not allowed to assign roles to a user.'
        ):
            topic_services.assign_role(
                guest_user,
                self.user_b,
                topic_domain.ROLE_MANAGER,
                self.TOPIC_ID,
            )

    def test_roles_of_guest_user_cannot_be_changed_until_guest_is_logged_in(
        self,
    ) -> None:
        guest_user = user_services.get_user_actions_info(None)
        with self.assertRaisesRegex(
            Exception, 'Cannot change the role of the Guest user.'
        ):
            topic_services.assign_role(
                self.user_admin,
                guest_user,
                topic_domain.ROLE_MANAGER,
                self.TOPIC_ID,
            )

    def test_role_cannot_be_assigned_to_non_topic_manager(self) -> None:
        with self.assertRaisesRegex(
            Exception,
            'The assignee doesn\'t have enough rights to become a manager.',
        ):
            topic_services.assign_role(
                self.user_admin,
                self.user_b,
                topic_domain.ROLE_MANAGER,
                self.TOPIC_ID,
            )

    def test_manager_cannot_assign_roles(self) -> None:
        with self.assertRaisesRegex(
            Exception, 'UnauthorizedUserException: Could not assign new role.'
        ):
            topic_services.assign_role(
                self.user_a,
                self.user_b,
                topic_domain.ROLE_MANAGER,
                self.TOPIC_ID,
            )

        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertTrue(
            topic_services.check_can_edit_topic(self.user_a, topic_rights)
        )
        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_b, topic_rights)
        )

    def test_cannot_save_new_topic_with_existing_name(self) -> None:
        with self.assertRaisesRegex(
            Exception, 'Topic with name \'Name\' already exists'
        ):
            self.save_new_topic(
                'topic_2',
                self.user_id,
                name='Name',
                description='Description 2',
                canonical_story_ids=[],
                additional_story_ids=[],
                uncategorized_skill_ids=[],
                subtopics=[],
                next_subtopic_id=1,
            )

    def test_does_not_update_subtopic_url_fragment_if_it_already_exists(
        self,
    ) -> None:
        topic_id = topic_fetchers.get_new_topic_id()
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title',
                    'subtopic_id': 1,
                    'url_fragment': 'fragment-one',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': topic_domain.SUBTOPIC_PROPERTY_URL_FRAGMENT,
                    'new_value': 'original',
                    'old_value': '',
                    'subtopic_id': 1,
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_ADD_SUBTOPIC,
                    'title': 'Title',
                    'subtopic_id': 2,
                    'url_fragment': 'fragment-two',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_SUBTOPIC_PROPERTY,
                    'property_name': topic_domain.SUBTOPIC_PROPERTY_URL_FRAGMENT,
                    'new_value': 'original',
                    'old_value': '',
                    'subtopic_id': 2,
                }
            ),
        ]
        self.save_new_topic(
            topic_id,
            self.user_id,
            name='topic-with-duplicate-subtopic',
            description='Description',
            canonical_story_ids=[],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
            url_fragment='frag-dup-subtopic',
        )
        with self.assertRaisesRegex(
            Exception,
            'Subtopic url fragments are not unique across subtopics '
            'in the topic',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id, topic_id, changelist, 'Update url fragment'
            )

    def test_does_not_create_topic_url_fragment_if_it_already_exists(
        self,
    ) -> None:
        topic_id_1 = topic_fetchers.get_new_topic_id()
        topic_id_2 = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            topic_id_1,
            self.user_id,
            name='topic 1',
            description='Description',
            canonical_story_ids=[],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
            url_fragment='topic-frag-one',
        )
        with self.assertRaisesRegex(
            Exception,
            'Topic with URL Fragment \'topic-frag-one\' already exists',
        ):
            self.save_new_topic(
                topic_id_2,
                self.user_id,
                name='topic 2',
                description='Description',
                canonical_story_ids=[],
                additional_story_ids=[],
                uncategorized_skill_ids=[],
                subtopics=[],
                next_subtopic_id=1,
                url_fragment='topic-frag-one',
            )

    def test_does_not_update_topic_if_url_fragment_already_exists(self) -> None:
        topic_id_1 = topic_fetchers.get_new_topic_id()
        topic_id_2 = topic_fetchers.get_new_topic_id()
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_URL_FRAGMENT,
                    'new_value': 'topic-frag-one',
                    'old_value': 'topic-frag-two',
                }
            )
        ]
        self.save_new_topic(
            topic_id_1,
            self.user_id,
            name='topic name 1',
            description='Description',
            canonical_story_ids=[],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
            url_fragment='topic-frag-one',
        )
        self.save_new_topic(
            topic_id_2,
            self.user_id,
            name='topic name 2',
            description='Description',
            canonical_story_ids=[],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
            url_fragment='topic-frag-two',
        )
        with self.assertRaisesRegex(
            Exception,
            'Topic with URL Fragment \'topic-frag-one\' already exists',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id, topic_id_2, changelist, 'Update url fragment'
            )

    def test_does_not_update_topic_if_name_already_exists(self) -> None:
        topic_id_1 = topic_fetchers.get_new_topic_id()
        topic_id_2 = topic_fetchers.get_new_topic_id()
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_NAME,
                    'new_value': 'topic 1',
                    'old_value': 'topic 2',
                }
            )
        ]
        self.save_new_topic(
            topic_id_1,
            self.user_id,
            name='topic 1',
            description='Description',
            canonical_story_ids=[],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
            url_fragment='topic-frag-one',
        )
        self.save_new_topic(
            topic_id_2,
            self.user_id,
            name='topic 2',
            description='Description',
            canonical_story_ids=[],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
            url_fragment='topic-frag-two',
        )
        with self.assertRaisesRegex(
            Exception, 'Topic with name \'topic 1\' already exists'
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id, topic_id_2, changelist, 'Update name'
            )

    def test_does_not_create_topic_if_name_is_non_string(self) -> None:
        topic_id = topic_fetchers.get_new_topic_id()
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_NAME,
                    'new_value': 123,
                    'old_value': 'topic name',
                }
            )
        ]
        self.save_new_topic(
            topic_id,
            self.user_id,
            name='topic name',
            description='Description',
            canonical_story_ids=[],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
            url_fragment='topic-frag',
        )
        with self.assertRaisesRegex(Exception, 'Name should be a string.'):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id, topic_id, changelist, 'Update topic name'
            )

    def test_does_not_save_topic_with_superseding_skill_in_subtopic(
        self,
    ) -> None:
        self.save_new_skill('supersede2', self.user_id)
        self.save_new_skill('has_superseding2', self.user_id)
        topic_id = topic_fetchers.get_new_topic_id()
        subtopic = topic_domain.Subtopic(
            1,
            'Test Subtopic',
            [],
            'image.svg',
            '#FFFFFF',
            None,
            'test-subtopic',
        )
        self.save_new_topic(
            topic_id,
            self.user_id,
            name='topic-supersede2',
            description='desc',
            url_fragment='topic-url-supersede',
            uncategorized_skill_ids=['has_superseding2'],
            subtopics=[subtopic],
            next_subtopic_id=2,
        )
        move_changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': 1,
                    'skill_id': 'has_superseding2',
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id, topic_id, move_changelist, 'Move skill to subtopic.'
        )
        skill_changelist = [
            skill_domain.SkillChange(
                {
                    'cmd': skill_domain.CMD_UPDATE_SKILL_PROPERTY,
                    'property_name': (
                        skill_domain.SKILL_PROPERTY_SUPERSEDING_SKILL_ID
                    ),
                    'old_value': '',
                    'new_value': 'supersede2',
                }
            )
        ]
        skill_services.update_skill(
            self.user_id,
            'has_superseding2',
            skill_changelist,
            'Merging skill.',
        )
        update_changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_DESCRIPTION,
                    'old_value': 'desc',
                    'new_value': 'updated desc',
                }
            )
        ]
        with self.assertRaisesRegex(
            Exception,
            'The skill \'has_superseding2\' in subtopic \'Test Subtopic\' '
            'has a superseding skill \'supersede2\'',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id, topic_id, update_changelist, 'Update topic.'
            )

    def test_does_not_add_skill_with_superseding_skill_to_topic(self) -> None:
        self.save_new_skill('supersede', self.user_id)
        self.save_new_skill('has_superseding', self.user_id)
        changelist = [
            skill_domain.SkillChange(
                {
                    'cmd': skill_domain.CMD_UPDATE_SKILL_PROPERTY,
                    'property_name': (
                        skill_domain.SKILL_PROPERTY_SUPERSEDING_SKILL_ID
                    ),
                    'old_value': '',
                    'new_value': 'supersede',
                }
            )
        ]
        skill_services.update_skill(
            self.user_id, 'has_superseding', changelist, 'Merging skill.'
        )
        topic_id = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            topic_id,
            self.user_id,
            name='topic-supersede',
            description='desc',
            url_fragment='topic-url-frag',
            uncategorized_skill_ids=[],
        )
        with self.assertRaisesRegex(
            Exception,
            'The skill \'has_superseding\' in uncategorized skills has a superseding skill \'supersede\'',
        ):  # pylint:disable=line-too-long
            topic_services.add_uncategorized_skill(
                self.user_id, topic_id, 'has_superseding'
            )

    def test_find_superseded_skill_in_topic_returns_skill(
        self,
    ) -> None:
        self.save_new_skill('superseding_skill', self.user_id)
        self.save_new_skill('skill_to_merge', self.user_id)
        changelist = [
            skill_domain.SkillChange(
                {
                    'cmd': skill_domain.CMD_UPDATE_SKILL_PROPERTY,
                    'property_name': (
                        skill_domain.SKILL_PROPERTY_SUPERSEDING_SKILL_ID
                    ),
                    'old_value': '',
                    'new_value': 'superseding_skill',
                }
            )
        ]
        skill_services.update_skill(
            self.user_id, 'skill_to_merge', changelist, 'Merging skill.'
        )
        topic_id = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            topic_id,
            self.user_id,
            name='Topic With Superseding',
            description='desc',
            url_fragment='topic-with-super',
            uncategorized_skill_ids=['skill_to_merge'],
        )
        topic = topic_fetchers.get_topic_by_id(topic_id)
        result = topic_services.find_superseded_skill_in_topic(topic)
        self.assertIsNotNone(result)
        assert result is not None
        self.assertEqual(result.id, 'skill_to_merge')

    def test_find_superseded_skill_in_topic_returns_none(
        self,
    ) -> None:
        self.save_new_skill('regular_skill', self.user_id)
        topic_id = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            topic_id,
            self.user_id,
            name='Topic Without Superseding',
            description='desc',
            url_fragment='topic-no-super',
            uncategorized_skill_ids=['regular_skill'],
        )
        topic = topic_fetchers.get_topic_by_id(topic_id)
        result = topic_services.find_superseded_skill_in_topic(topic)
        self.assertIsNone(result)

    def test_find_superseded_skill_in_topic_skips_none_skills(
        self,
    ) -> None:
        topic_id = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            topic_id,
            self.user_id,
            name='Topic For None Skill Test',
            description='desc',
            url_fragment='topic-none-test',
            uncategorized_skill_ids=['dummy_skill_id'],
        )
        topic = topic_fetchers.get_topic_by_id(topic_id)
        with self.swap(
            skill_fetchers,
            'get_multi_skills',
            lambda skill_ids, strict=True: [None],
        ):
            result = topic_services.find_superseded_skill_in_topic(topic)
        self.assertIsNone(result)

    # TODO(#13059): Here we use MyPy ignore because after we fully type the
    # codebase we plan to get rid of the tests that intentionally test wrong
    # inputs that we can normally catch by typing.
    def test_url_fragment_existence_fails_for_non_string_url_fragment(
        self,
    ) -> None:
        with self.assertRaisesRegex(
            Exception, 'Topic URL fragment should be a string.'
        ):
            topic_services.does_topic_with_url_fragment_exist(123)  # type: ignore[arg-type]

    # TODO(#13059): Here we use MyPy ignore because after we fully type the
    # codebase we plan to get rid of the tests that intentionally test wrong
    # inputs that we can normally catch by typing.
    def test_name_existence_fails_for_non_string_name(self) -> None:
        with self.assertRaisesRegex(Exception, 'Name should be a string.'):
            topic_services.does_topic_with_name_exist(123)  # type: ignore[arg-type]

    def test_update_topic_language_code(self) -> None:
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(topic.language_code, 'en')

        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_LANGUAGE_CODE,
                    'old_value': 'en',
                    'new_value': 'bn',
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id, self.TOPIC_ID, changelist, 'Change language code'
        )

        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        self.assertEqual(topic.language_code, 'bn')

    def test_cannot_update_topic_and_subtopic_pages_with_empty_changelist(
        self,
    ) -> None:
        with self.assertRaisesRegex(
            Exception,
            'Unexpected error: received an invalid change list when trying to '
            'save topic',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id, self.TOPIC_ID, [], 'commit message'
            )

    def test_cannot_update_topic_and_subtopic_pages_with_mismatch_of_versions(
        self,
    ) -> None:
        topic_model = topic_models.TopicModel.get(self.TOPIC_ID)
        topic_model.version = 0
        topic_model.commit(self.user_id, 'changed version', [])

        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': topic_domain.TOPIC_PROPERTY_LANGUAGE_CODE,
                    'old_value': 'en',
                    'new_value': 'bn',
                }
            )
        ]

        with self.assertRaisesRegex(
            Exception,
            'Unexpected error: trying to update version 1 of topic '
            'from version 2. Please reload the page and try again.',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id, self.TOPIC_ID, changelist, 'change language_code'
            )

        topic_model = topic_models.TopicModel.get(self.TOPIC_ID)
        topic_model.version = 100
        topic_model.commit(self.user_id, 'changed version', [])

        with self.assertRaisesRegex(
            Exception,
            'Trying to update version 101 of topic from version 2, '
            'which is too old. Please reload the page and try again.',
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id, self.TOPIC_ID, changelist, 'change language_code'
            )

    def test_cannot_update_topic_and_subtopic_pages_with_empty_commit_message(
        self,
    ) -> None:
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': self.subtopic_id,
                    'skill_id': 'skill_1',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_SKILL_IDS_FOR_DIAGNOSTIC_TEST
                    ),
                    'old_value': self.test_list,
                    'new_value': ['skill_1'],
                }
            ),
        ]
        # Test can have an empty commit message when not published.
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin, self.TOPIC_ID, changelist, None
        )
        topic_services.publish_topic(self.TOPIC_ID, self.user_id_admin)
        # Test must have a commit message when published.
        with self.assertRaisesRegex(
            Exception, 'Expected a commit message, received none.'
        ):
            topic_services.update_topic_and_subtopic_pages(
                self.user_id, self.TOPIC_ID, [], None
            )

    def test_cannot_publish_topic_with_no_topic_rights(self) -> None:
        with self.assertRaisesRegex(
            Exception, 'The given topic does not exist'
        ):
            topic_services.publish_topic('invalid_topic_id', self.user_id_admin)

    def test_cannot_publish_a_published_topic(self) -> None:
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertFalse(topic_rights.topic_is_published)
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_MOVE_SKILL_ID_TO_SUBTOPIC,
                    'old_subtopic_id': None,
                    'new_subtopic_id': self.subtopic_id,
                    'skill_id': 'skill_1',
                }
            ),
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_SKILL_IDS_FOR_DIAGNOSTIC_TEST
                    ),
                    'old_value': self.test_list,
                    'new_value': ['skill_1'],
                }
            ),
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            self.TOPIC_ID,
            changelist,
            'Updated subtopic skill ids.',
        )
        topic_services.publish_topic(self.TOPIC_ID, self.user_id_admin)
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertTrue(topic_rights.topic_is_published)

        with self.assertRaisesRegex(
            Exception, 'The topic is already published.'
        ):
            topic_services.publish_topic(self.TOPIC_ID, self.user_id_admin)

    def test_cannot_unpublish_topic_with_no_topic_rights(self) -> None:
        with self.assertRaisesRegex(
            Exception, 'The given topic does not exist'
        ):
            topic_services.unpublish_topic(
                'invalid_topic_id', self.user_id_admin
            )

    def test_cannot_unpublish_an_unpublished_topic(self) -> None:
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertFalse(topic_rights.topic_is_published)

        with self.assertRaisesRegex(
            Exception, 'The topic is already unpublished.'
        ):
            topic_services.unpublish_topic(self.TOPIC_ID, self.user_id_admin)

    def test_cannot_edit_topic_with_no_topic_rights(self) -> None:
        self.assertFalse(topic_services.check_can_edit_topic(self.user_a, None))

    def test_cannot_assign_role_with_invalid_role(self) -> None:
        with self.assertRaisesRegex(Exception, 'Invalid role'):
            topic_services.assign_role(
                self.user_admin, self.user_a, 'invalid_role', self.TOPIC_ID
            )

    def test_deassign_user_from_all_topics(self) -> None:
        self.save_new_topic(
            'topic_2',
            self.user_id,
            name='Name 2',
            abbreviated_name='name-two',
            url_fragment='name-six',
            description='Description 2',
            canonical_story_ids=[],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
        )
        self.save_new_topic(
            'topic_3',
            self.user_id,
            name='Name 3',
            abbreviated_name='name-three',
            url_fragment='name-seven',
            description='Description 3',
            canonical_story_ids=[],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
        )

        topic_services.assign_role(
            self.user_admin, self.user_a, topic_domain.ROLE_MANAGER, 'topic_2'
        )
        topic_rights = topic_fetchers.get_topic_rights_with_user(self.user_id_a)
        self.assertEqual(len(topic_rights), 2)

        topic_services.deassign_user_from_all_topics(
            self.user_admin, self.user_id_a
        )
        topic_rights = topic_fetchers.get_topic_rights_with_user(self.user_id_a)
        self.assertEqual(len(topic_rights), 0)

    def test_reassigning_manager_role_to_same_user(self) -> None:
        with self.assertRaisesRegex(
            Exception, 'This user already is a manager for this topic'
        ):
            topic_services.assign_role(
                self.user_admin,
                self.user_a,
                topic_domain.ROLE_MANAGER,
                self.TOPIC_ID,
            )

        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertTrue(
            topic_services.check_can_edit_topic(self.user_a, topic_rights)
        )
        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_b, topic_rights)
        )

    def test_assigning_none_role(self) -> None:
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)

        self.assertTrue(
            topic_services.check_can_edit_topic(self.user_a, topic_rights)
        )
        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_b, topic_rights)
        )
        # Assigning None role to manager.
        topic_services.assign_role(
            self.user_admin, self.user_a, topic_domain.ROLE_NONE, self.TOPIC_ID
        )

        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_a, topic_rights)
        )
        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_b, topic_rights)
        )

        # Assigning None role to another role.
        topic_services.assign_role(
            self.user_admin, self.user_a, topic_domain.ROLE_NONE, self.TOPIC_ID
        )

        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_b, topic_rights)
        )

    def test_deassigning_manager_role(self) -> None:
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)

        self.assertTrue(
            topic_services.check_can_edit_topic(self.user_a, topic_rights)
        )
        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_b, topic_rights)
        )

        topic_services.deassign_manager_role_from_topic(
            self.user_admin, self.user_id_a, self.TOPIC_ID
        )

        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_a, topic_rights)
        )
        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_b, topic_rights)
        )

    def test_deassigning_an_unassigned_user_from_topic_raise_exception(
        self,
    ) -> None:
        topic_rights = topic_fetchers.get_topic_rights(self.TOPIC_ID)
        self.assertFalse(
            topic_services.check_can_edit_topic(self.user_b, topic_rights)
        )

        with self.assertRaisesRegex(
            Exception, 'User does not have manager rights in topic.'
        ):
            topic_services.deassign_manager_role_from_topic(
                self.user_admin, self.user_id_b, self.TOPIC_ID
            )

    def test_update_thumbnail_filename(self) -> None:
        self.assertEqual(self.topic.thumbnail_filename, 'topic.svg')
        # Test exception when thumbnail is not found on filesystem.
        with self.assertRaisesRegex(
            Exception,
            'The thumbnail img.svg for topic with id %s does not exist'
            ' in the filesystem.' % (self.TOPIC_ID),
        ):
            topic_services.update_thumbnail_filename(self.topic, 'img.svg')

        # Save the dummy image to the filesystem to be used as thumbnail.
        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'test_svg.svg'),
            'rb',
            encoding=None,
        ) as f:
            raw_image = f.read()
        fs = fs_services.GcsFileSystem(feconf.ENTITY_TYPE_TOPIC, self.TOPIC_ID)
        fs.commit(
            '%s/img.svg' % (constants.ASSET_TYPE_THUMBNAIL),
            raw_image,
            mimetype='image/svg+xml',
        )
        # Test successful update of thumbnail present in the filesystem.
        topic_services.update_thumbnail_filename(self.topic, 'img.svg')
        self.assertEqual(self.topic.thumbnail_filename, 'img.svg')
        self.assertEqual(self.topic.thumbnail_size_in_bytes, len(raw_image))

    def test_update_subtopic_thumbnail_filename(self) -> None:
        self.assertEqual(len(self.topic.subtopics), 1)
        self.assertEqual(self.topic.subtopics[0].thumbnail_filename, None)

        # Test Exception when the thumbnail is not found in filesystem.
        with self.assertRaisesRegex(
            Exception,
            'The thumbnail %s for subtopic with topic_id %s does not exist '
            'in the filesystem.' % ('new_image.svg', self.TOPIC_ID),
        ):
            topic_services.update_subtopic_thumbnail_filename(
                self.topic, 1, 'new_image.svg'
            )

        # Test successful update of thumbnail_filename when the thumbnail
        # is found in the filesystem.
        with open(
            os.path.join(feconf.TESTS_DATA_DIR, 'test_svg.svg'),
            'rb',
            encoding=None,
        ) as f:
            raw_image = f.read()
        fs = fs_services.GcsFileSystem(feconf.ENTITY_TYPE_TOPIC, self.TOPIC_ID)
        fs.commit(
            'thumbnail/new_image.svg', raw_image, mimetype='image/svg+xml'
        )
        topic_services.update_subtopic_thumbnail_filename(
            self.topic, 1, 'new_image.svg'
        )
        self.assertEqual(
            self.topic.subtopics[0].thumbnail_filename, 'new_image.svg'
        )
        self.assertEqual(
            self.topic.subtopics[0].thumbnail_size_in_bytes, len(raw_image)
        )

    def test_get_topic_id_to_diagnostic_test_skill_ids(self) -> None:
        fractions_id = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            fractions_id,
            self.user_id,
            name='Fractions',
            url_fragment='fractions',
            description='Description of fraction',
            canonical_story_ids=[self.story_id_1, self.story_id_2],
            additional_story_ids=[self.story_id_3],
            uncategorized_skill_ids=[self.skill_id_1, self.skill_id_2],
            subtopics=[],
            next_subtopic_id=1,
        )
        old_value: List[str] = []
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_SKILL_IDS_FOR_DIAGNOSTIC_TEST
                    ),
                    'old_value': old_value,
                    'new_value': [self.skill_id_1],
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            fractions_id,
            changelist,
            'Adds diagnostic test.',
        )

        additions_id = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            additions_id,
            self.user_id,
            name='Additions',
            url_fragment='additions',
            description='Description of addition.',
            canonical_story_ids=[self.story_id_1, self.story_id_2],
            additional_story_ids=[self.story_id_3],
            uncategorized_skill_ids=[self.skill_id_1, self.skill_id_2],
            subtopics=[],
            next_subtopic_id=1,
        )
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_SKILL_IDS_FOR_DIAGNOSTIC_TEST
                    ),
                    'old_value': old_value,
                    'new_value': [self.skill_id_2],
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            additions_id,
            changelist,
            'Adds diagnostic test.',
        )

        expected_dict = {
            fractions_id: [self.skill_id_1],
            additions_id: [self.skill_id_2],
        }
        self.assertEqual(
            topic_services.get_topic_id_to_diagnostic_test_skill_ids(
                [fractions_id, additions_id]
            ),
            expected_dict,
        )

        error_msg = (
            'No corresponding topic models exist for these topic IDs: %s.'
            % (', '.join(['']))
        )
        with self.assertRaisesRegex(Exception, error_msg):
            topic_services.get_topic_id_to_diagnostic_test_skill_ids(
                [additions_id, 'incorrect_topic_id']
            )

    def test_get_topic_id_to_topic_name_dict(self) -> None:
        fractions_id = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            fractions_id,
            self.user_id,
            name='Fractions',
            url_fragment='fractions',
            description='Description of fraction',
            canonical_story_ids=[self.story_id_1, self.story_id_2],
            additional_story_ids=[self.story_id_3],
            uncategorized_skill_ids=[self.skill_id_1, self.skill_id_2],
            subtopics=[],
            next_subtopic_id=1,
        )
        old_value: List[str] = []
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_SKILL_IDS_FOR_DIAGNOSTIC_TEST
                    ),
                    'old_value': old_value,
                    'new_value': [self.skill_id_1],
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            fractions_id,
            changelist,
            'Adds diagnostic test.',
        )

        additions_id = topic_fetchers.get_new_topic_id()
        self.save_new_topic(
            additions_id,
            self.user_id,
            name='Additions',
            url_fragment='additions',
            description='Description of addition.',
            canonical_story_ids=[self.story_id_1, self.story_id_2],
            additional_story_ids=[self.story_id_3],
            uncategorized_skill_ids=[self.skill_id_1, self.skill_id_2],
            subtopics=[],
            next_subtopic_id=1,
        )
        changelist = [
            topic_domain.TopicChange(
                {
                    'cmd': topic_domain.CMD_UPDATE_TOPIC_PROPERTY,
                    'property_name': (
                        topic_domain.TOPIC_PROPERTY_SKILL_IDS_FOR_DIAGNOSTIC_TEST
                    ),
                    'old_value': old_value,
                    'new_value': [self.skill_id_2],
                }
            )
        ]
        topic_services.update_topic_and_subtopic_pages(
            self.user_id_admin,
            additions_id,
            changelist,
            'Adds diagnostic test.',
        )

        expected_dict = {fractions_id: 'Fractions', additions_id: 'Additions'}
        self.assertEqual(
            topic_services.get_topic_id_to_topic_name_dict(
                [fractions_id, additions_id]
            ),
            expected_dict,
        )

        error_msg = (
            'No corresponding topic models exist for these topic IDs: %s.'
            % (', '.join(['']))
        )
        with self.assertRaisesRegex(Exception, error_msg):
            topic_services.get_topic_id_to_topic_name_dict(
                [additions_id, 'incorrect_topic_id']
            )

    def test_populate_topic_model_fields(self) -> None:
        model = topic_models.TopicModel(
            id='dummy_topic_id',
            name='dummy_name',
            abbreviated_name='dn',
            url_fragment='name-one',
            description='dummy description1',
            canonical_name='dummy_canonical_name',
            next_subtopic_id=1,
            language_code='en',
            subtopic_schema_version=1,
            story_reference_schema_version=2,
            page_title_fragment_for_web='fragm1',
        )
        topic = topic_fetchers.get_topic_by_id(self.TOPIC_ID)
        populated_model = topic_services.populate_topic_model_fields(
            model, topic
        )
        self.assertEqual(populated_model.description, topic.description)
        self.assertEqual(populated_model.name, topic.name)
        self.assertEqual(populated_model.canonical_name, topic.canonical_name)
        self.assertEqual(
            populated_model.abbreviated_name, topic.abbreviated_name
        )
        self.assertEqual(populated_model.url_fragment, topic.url_fragment)
        self.assertEqual(
            populated_model.thumbnail_bg_color, topic.thumbnail_bg_color
        )
        self.assertEqual(
            populated_model.thumbnail_filename, topic.thumbnail_filename
        )
        self.assertEqual(
            populated_model.thumbnail_size_in_bytes,
            topic.thumbnail_size_in_bytes,
        )
        for model_reference, topic_reference in zip(
            populated_model.canonical_story_references,
            topic.canonical_story_references,
        ):
            self.assertEqual(model_reference, topic_reference.to_dict())

        for model_reference, topic_reference in zip(
            populated_model.additional_story_references,
            topic.additional_story_references,
        ):
            self.assertEqual(model_reference, topic_reference.to_dict())
        self.assertEqual(
            populated_model.uncategorized_skill_ids,
            topic.uncategorized_skill_ids,
        )
        for model_subtopic, topic_subtopic in zip(
            populated_model.subtopics, topic.subtopics
        ):
            self.assertEqual(model_subtopic, topic_subtopic.to_dict())
        self.assertEqual(
            populated_model.subtopic_schema_version,
            topic.subtopic_schema_version,
        )
        self.assertEqual(
            populated_model.story_reference_schema_version,
            topic.story_reference_schema_version,
        )
        self.assertEqual(
            populated_model.next_subtopic_id, topic.next_subtopic_id
        )
        self.assertEqual(populated_model.language_code, topic.language_code)
        self.assertEqual(
            populated_model.meta_tag_content, topic.meta_tag_content
        )
        self.assertEqual(
            populated_model.practice_tab_is_displayed,
            topic.practice_tab_is_displayed,
        )
        self.assertEqual(
            populated_model.page_title_fragment_for_web,
            topic.page_title_fragment_for_web,
        )
        self.assertEqual(
            populated_model.skill_ids_for_diagnostic_test,
            topic.skill_ids_for_diagnostic_test,
        )

    def test_populate_topic_summary_model_fields(self) -> None:
        model = topic_models.TopicSummaryModel(
            id=self.TOPIC_ID,
            name='dummy topic summary',
            canonical_name='dummy topic summary',
            language_code='cs',
            description=' dummy description',
            url_fragment='/fragm',
            canonical_story_count=0,
            additional_story_count=0,
            total_skill_count=0,
            total_published_node_count=0,
            uncategorized_skill_count=0,
            subtopic_count=0,
            version=1,
            published_story_exploration_mapping={
                self.story_id_1: [],
                self.story_id_2: [],
                self.story_id_3: [],
            },
        )
        topic_summary = topic_services.compute_summary_of_topic(self.topic)
        populated_model = topic_services.populate_topic_summary_model_fields(
            model, topic_summary
        )
        self.assertEqual(populated_model.name, topic_summary.name)
        self.assertEqual(populated_model.description, topic_summary.description)
        self.assertEqual(
            populated_model.canonical_name, topic_summary.canonical_name
        )
        self.assertEqual(
            populated_model.language_code, topic_summary.language_code
        )
        self.assertEqual(populated_model.version, topic_summary.version)
        self.assertEqual(
            populated_model.additional_story_count,
            topic_summary.additional_story_count,
        )
        self.assertEqual(
            populated_model.canonical_story_count,
            topic_summary.canonical_story_count,
        )
        self.assertEqual(
            populated_model.uncategorized_skill_count,
            topic_summary.uncategorized_skill_count,
        )
        self.assertEqual(
            populated_model.subtopic_count, topic_summary.subtopic_count
        )
        self.assertEqual(
            populated_model.total_skill_count, topic_summary.total_skill_count
        )
        self.assertEqual(
            populated_model.total_published_node_count,
            topic_summary.total_published_node_count,
        )
        self.assertEqual(
            populated_model.thumbnail_filename, topic_summary.thumbnail_filename
        )
        self.assertEqual(
            populated_model.thumbnail_bg_color, topic_summary.thumbnail_bg_color
        )
        self.assertEqual(
            populated_model.topic_model_last_updated,
            topic_summary.topic_model_last_updated,
        )
        self.assertEqual(
            populated_model.topic_model_created_on,
            topic_summary.topic_model_created_on,
        )
        self.assertEqual(
            populated_model.url_fragment, topic_summary.url_fragment
        )
        self.assertDictEqual(
            populated_model.published_story_exploration_mapping,
            topic_summary.published_story_exploration_mapping,
        )

    def test_get_chapter_counts_in_topic_summaries(self) -> None:
        canonical_story_id_1 = story_services.get_new_story_id()
        story = story_domain.Story.create_default_story(
            canonical_story_id_1,
            'title',
            'description',
            self.TOPIC_ID,
            'url-fragment',
        )
        story.meta_tag_content = 'story meta content'
        node_1: story_domain.StoryNodeDict = {
            'outline': 'outline',
            'exploration_id': 'exp-1',
            'destination_node_ids': [],
            'outline_is_finalized': False,
            'acquired_skill_ids': [],
            'id': 'node_1',
            'title': 'Chapter 1',
            'description': '',
            'prerequisite_skill_ids': [],
            'thumbnail_filename': 'image.svg',
            'thumbnail_bg_color': constants.ALLOWED_THUMBNAIL_BG_COLORS[
                'chapter'
            ][0],
            'thumbnail_size_in_bytes': 21131,
            'status': constants.STORY_NODE_STATUS_PUBLISHED,
            'planned_publication_date_msecs': 1690800486000,
            'first_publication_date_msecs': 1672684200000,
            'last_modified_msecs': 1672684200000,
            'unpublishing_reason': None,
        }
        node_2: story_domain.StoryNodeDict = {
            'outline': 'outline',
            'exploration_id': 'exp-2',
            'destination_node_ids': [],
            'outline_is_finalized': False,
            'acquired_skill_ids': [],
            'id': 'node_2',
            'title': 'Chapter 2',
            'description': '',
            'prerequisite_skill_ids': [],
            'thumbnail_filename': 'image.svg',
            'thumbnail_bg_color': constants.ALLOWED_THUMBNAIL_BG_COLORS[
                'chapter'
            ][0],
            'thumbnail_size_in_bytes': 21131,
            'status': constants.STORY_NODE_STATUS_DRAFT,
            'planned_publication_date_msecs': 1659264486000,
            'first_publication_date_msecs': None,
            'last_modified_msecs': 1672684200000,
            'unpublishing_reason': None,
        }
        node_3: story_domain.StoryNodeDict = {
            'outline': 'outline',
            'exploration_id': 'exp-3',
            'destination_node_ids': [],
            'outline_is_finalized': False,
            'acquired_skill_ids': [],
            'id': 'node_3',
            'title': 'Chapter 3',
            'description': '',
            'prerequisite_skill_ids': [],
            'thumbnail_filename': 'image.svg',
            'thumbnail_bg_color': constants.ALLOWED_THUMBNAIL_BG_COLORS[
                'chapter'
            ][0],
            'thumbnail_size_in_bytes': 21131,
            'status': constants.STORY_NODE_STATUS_READY_TO_PUBLISH,
            'planned_publication_date_msecs': 1690800486000,
            'first_publication_date_msecs': None,
            'last_modified_msecs': 1672684200000,
            'unpublishing_reason': None,
        }
        node_4: story_domain.StoryNodeDict = {
            'outline': 'outline',
            'exploration_id': 'exp-4',
            'destination_node_ids': [],
            'outline_is_finalized': False,
            'acquired_skill_ids': [],
            'id': 'node_4',
            'title': 'Chapter 4',
            'description': '',
            'prerequisite_skill_ids': [],
            'thumbnail_filename': 'image.svg',
            'thumbnail_bg_color': constants.ALLOWED_THUMBNAIL_BG_COLORS[
                'chapter'
            ][0],
            'thumbnail_size_in_bytes': 21131,
            'status': constants.STORY_NODE_STATUS_READY_TO_PUBLISH,
            'planned_publication_date_msecs': 1693478886000,
            'first_publication_date_msecs': None,
            'last_modified_msecs': 1672684200000,
            'unpublishing_reason': None,
        }
        story.story_contents.nodes = [
            story_domain.StoryNode.from_dict(node_1),
            story_domain.StoryNode.from_dict(node_2),
            story_domain.StoryNode.from_dict(node_3),
            story_domain.StoryNode.from_dict(node_4),
        ]
        story.story_contents.initial_node_id = 'node_1'
        story.story_contents.next_node_id = 'node_5'

        story_services.save_new_story(self.user_id, story)
        topic_services.add_canonical_story(
            self.user_id, self.TOPIC_ID, canonical_story_id_1
        )

        def mock_get_current_time_in_millisecs() -> int:
            return 1690555400000

        with self.swap(
            utils,
            'get_current_time_in_millisecs',
            mock_get_current_time_in_millisecs,
        ):
            topic_summary = topic_fetchers.get_topic_summary_by_id(
                self.TOPIC_ID
            ).to_dict()
            frontend_topic_summary: topic_domain.FrontendTopicSummaryDict = {
                'id': topic_summary['id'],
                'name': topic_summary['name'],
                'url_fragment': topic_summary['url_fragment'],
                'language_code': topic_summary['language_code'],
                'description': topic_summary['description'],
                'version': topic_summary['version'],
                'canonical_story_count': topic_summary['canonical_story_count'],
                'additional_story_count': topic_summary[
                    'canonical_story_count'
                ],
                'uncategorized_skill_count': topic_summary[
                    'additional_story_count'
                ],
                'subtopic_count': topic_summary['subtopic_count'],
                'total_skill_count': topic_summary['total_skill_count'],
                'total_published_node_count': topic_summary[
                    'total_published_node_count'
                ],
                'thumbnail_filename': topic_summary['thumbnail_filename'],
                'thumbnail_bg_color': topic_summary['thumbnail_bg_color'],
                'published_story_exploration_mapping': topic_summary[
                    'published_story_exploration_mapping'
                ],
                'topic_model_created_on': topic_summary[
                    'topic_model_created_on'
                ],
                'topic_model_last_updated': topic_summary[
                    'topic_model_last_updated'
                ],
                'is_published': True,
                'can_edit_topic': True,
                'can_edit_question': True,
                'classroom': None,
                'total_upcoming_chapters_count': 0,
                'total_overdue_chapters_count': 0,
                'total_chapter_counts_for_each_story': [],
                'published_chapter_counts_for_each_story': [],
            }
            chapter_counts = (
                topic_services.get_chapter_counts_in_topic_summaries(
                    [frontend_topic_summary]
                )
            )
            self.assertEqual(
                chapter_counts[self.TOPIC_ID].total_upcoming_chapters_count, 1
            )
            self.assertEqual(
                chapter_counts[self.TOPIC_ID].total_overdue_chapters_count, 1
            )
            self.assertEqual(
                chapter_counts[
                    self.TOPIC_ID
                ].total_chapter_counts_for_each_story,
                [0, 0, 4],
            )
            self.assertEqual(
                chapter_counts[
                    self.TOPIC_ID
                ].published_chapter_counts_for_each_story,
                [0, 0, 1],
            )

    def test_get_chapter_counts_in_topic_summaries_logs_error(self) -> None:
        topic_model = topic_models.TopicModel.get_by_id(self.TOPIC_ID)
        invalid_story_reference_dict = {
            'story_id': 'invalid_story_id',
            'story_is_published': False,
        }
        topic_model.canonical_story_references = [invalid_story_reference_dict]
        topic_model.commit(self.user_id, 'reference invalid story id', [])

        def mock_get_current_time_in_millisecs() -> int:
            return 1690555400000

        with self.swap(
            utils,
            'get_current_time_in_millisecs',
            mock_get_current_time_in_millisecs,
        ):
            with self.capture_logging(min_level=logging.ERROR) as logs:
                topic_summary = topic_fetchers.get_topic_summary_by_id(
                    self.TOPIC_ID
                ).to_dict()
                frontend_topic_summary: (
                    topic_domain.FrontendTopicSummaryDict
                ) = {
                    'id': topic_summary['id'],
                    'name': topic_summary['name'],
                    'url_fragment': topic_summary['url_fragment'],
                    'language_code': topic_summary['language_code'],
                    'description': topic_summary['description'],
                    'version': topic_summary['version'],
                    'canonical_story_count': topic_summary[
                        'canonical_story_count'
                    ],
                    'additional_story_count': topic_summary[
                        'canonical_story_count'
                    ],
                    'uncategorized_skill_count': topic_summary[
                        'additional_story_count'
                    ],
                    'subtopic_count': topic_summary['subtopic_count'],
                    'total_skill_count': topic_summary['total_skill_count'],
                    'total_published_node_count': topic_summary[
                        'total_published_node_count'
                    ],
                    'thumbnail_filename': topic_summary['thumbnail_filename'],
                    'thumbnail_bg_color': topic_summary['thumbnail_bg_color'],
                    'published_story_exploration_mapping': topic_summary[
                        'published_story_exploration_mapping'
                    ],
                    'topic_model_created_on': topic_summary[
                        'topic_model_created_on'
                    ],
                    'topic_model_last_updated': topic_summary[
                        'topic_model_last_updated'
                    ],
                    'is_published': True,
                    'can_edit_topic': True,
                    'can_edit_question': True,
                    'classroom': None,
                    'total_upcoming_chapters_count': 0,
                    'total_overdue_chapters_count': 0,
                    'total_chapter_counts_for_each_story': [],
                    'published_chapter_counts_for_each_story': [],
                }
                topic_services.get_chapter_counts_in_topic_summaries(
                    [frontend_topic_summary]
                )
                self.assertEqual(
                    logs,
                    [
                        'Topic %s has an invalid story reference ID '
                        'invalid_story_id' % self.TOPIC_ID
                    ],
                )

    def test_get_published_story_exploration_ids_ignores_exp_ids_from_unpublished_stories(  # pylint: disable=line-too-long
        self,
    ) -> None:
        topic_exp_ids = ['exp_1', 'exp_2', 'exp_3', 'exp_4']
        self._create_linked_explorations(
            self.TOPIC_ID, self.story_id_1, topic_exp_ids
        )

        story_exp_ids = topic_services.get_all_published_story_exploration_ids(
            self.TOPIC_ID
        )

        self.assertItemsEqual(story_exp_ids, [])

    def test_get_published_story_exploration_ids_from_published_stories_when_serial_chapter_feature_disabled(  # pylint: disable=line-too-long
        self,
    ) -> None:
        topic_exp_ids = ['exp_1', 'exp_2', 'exp_3', 'exp_4']
        self._create_linked_explorations(
            self.TOPIC_ID, self.story_id_1, topic_exp_ids
        )
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        topic_services.generate_topic_summary(self.TOPIC_ID)

        story_exp_ids = topic_services.get_all_published_story_exploration_ids(
            self.TOPIC_ID
        )

        self.assertItemsEqual(story_exp_ids, topic_exp_ids)

    @test_utils.enable_feature_flags(
        [
            feature_flag_list.FeatureNames.SERIAL_CHAPTER_LAUNCH_CURRICULUM_ADMIN_VIEW
        ]
    )
    def test_get_published_story_exploration_ids_from_published_chapters_when_serial_chapter_feature_enabled(  # pylint: disable=line-too-long
        self,
    ) -> None:
        topic_exp_ids = ['exp_1', 'exp_2', 'exp_3', 'exp_4']
        topic_published_chapters_exp_ids = topic_exp_ids[:2]
        self._publish_story_chapters_with_explorations(
            self.TOPIC_ID,
            self.story_id_1,
            topic_exp_ids,
            chapter_exp_ids=topic_published_chapters_exp_ids,
        )
        topic_services.generate_topic_summary(self.TOPIC_ID)

        story_exp_ids = topic_services.get_all_published_story_exploration_ids(
            self.TOPIC_ID
        )

        self.assertItemsEqual(story_exp_ids, topic_published_chapters_exp_ids)

    @test_utils.enable_feature_flags(
        [
            feature_flag_list.FeatureNames.SERIAL_CHAPTER_LAUNCH_CURRICULUM_ADMIN_VIEW
        ]
    )
    def test_get_published_story_exploration_ids_in_all_topics_when_topic_id_not_given(  # pylint: disable=line-too-long
        self,
    ) -> None:
        topic_id_2 = topic_fetchers.get_new_topic_id()
        story_id_4 = 'story_4'
        self.save_new_topic(
            topic_id_2,
            self.user_id,
            name='Name 2',
            abbreviated_name='name2',
            url_fragment='name-two',
            description='Description',
            canonical_story_ids=[story_id_4],
            additional_story_ids=[],
            uncategorized_skill_ids=[],
            subtopics=[],
            next_subtopic_id=1,
        )
        self.save_new_story(story_id_4, self.user_id, topic_id_2)

        exp_ids = ['exp_1', 'exp_2', 'exp_3', 'exp_4']
        self._publish_story_chapters_with_explorations(
            self.TOPIC_ID, self.story_id_1, exp_ids[:1]
        )
        self._publish_story_chapters_with_explorations(
            self.TOPIC_ID, self.story_id_3, exp_ids[1:3]
        )
        self._publish_story_chapters_with_explorations(
            topic_id_2, story_id_4, exp_ids[3:]
        )

        topic_services.generate_topic_summary(self.TOPIC_ID)

        story_exp_ids = topic_services.get_all_published_story_exploration_ids()

        self.assertItemsEqual(story_exp_ids, exp_ids)

        topic_services.generate_topic_summary(self.TOPIC_ID)

        story_exp_ids = topic_services.get_all_published_story_exploration_ids()

        self.assertItemsEqual(story_exp_ids, exp_ids)

    def test_get_published_story_exploration_ids_when_mapping_not_precomputed(
        self,
    ) -> None:
        topic_exp_ids = ['exp_1', 'exp_2', 'exp_3', 'exp_4']
        self._create_linked_explorations(
            self.TOPIC_ID, self.story_id_1, topic_exp_ids
        )
        topic_services.publish_story(
            self.TOPIC_ID, self.story_id_1, self.user_id_admin
        )
        topic_summary_without_exp_ids = topic_services.compute_summary_of_topic(
            self.topic
        )
        # -> Here we use MyPy ignore because we introduce the mapping
        # value of None.
        topic_summary_without_exp_ids.published_story_exploration_mapping = (
            None  # type: ignore[assignment]
        )

        with self.swap_to_always_return(
            topic_fetchers,
            'get_topic_summary_by_id',
            topic_summary_without_exp_ids,
        ):
            story_exp_ids = (
                topic_services.get_all_published_story_exploration_ids(
                    self.TOPIC_ID
                )
            )

        self.assertItemsEqual(story_exp_ids, topic_exp_ids)


# TODO(#7009): Remove this mock class and the SubtopicMigrationTests class
# once the actual functions for subtopic migrations are implemented.
class MockTopicObject(topic_domain.Topic):
    """Mocks Topic domain object."""

    @classmethod
    def _convert_story_reference_v1_dict_to_v2_dict(
        cls, story_reference: topic_domain.StoryReferenceDict
    ) -> topic_domain.StoryReferenceDict:
        """Converts v1 story reference dict to v2."""
        return story_reference


class SubtopicMigrationTests(test_utils.GenericTestBase):

    def test_migrate_subtopic_to_latest_schema(self) -> None:
        topic_services.create_new_topic_rights('topic_id', 'user_id_admin')
        commit_cmd = topic_domain.TopicChange(
            {'cmd': topic_domain.CMD_CREATE_NEW, 'name': 'name'}
        )
        subtopic_v1_dict = {'id': 1, 'title': 'subtopic_title', 'skill_ids': []}
        subtopic_v4_dict: Dict[str, Union[str, int, Optional[List[str]]]] = {
            'id': 1,
            'thumbnail_filename': None,
            'thumbnail_bg_color': None,
            'thumbnail_size_in_bytes': None,
            'title': 'subtopic_title',
            'skill_ids': [],
            'url_fragment': 'subtopictitle',
        }
        model = topic_models.TopicModel(
            id='topic_id',
            name='name',
            abbreviated_name='abbrev',
            url_fragment='name-eight',
            canonical_name='Name',
            description='description1',
            next_subtopic_id=1,
            language_code='en',
            subtopics=[subtopic_v1_dict],
            subtopic_schema_version=1,
            story_reference_schema_version=1,
            page_title_fragment_for_web='fragm',
        )
        commit_cmd_dicts = [commit_cmd.to_dict()]
        model.commit('user_id_admin', 'topic model created', commit_cmd_dicts)

        swap_topic_object = self.swap(topic_domain, 'Topic', MockTopicObject)
        current_schema_version_swap = self.swap(
            feconf, 'CURRENT_SUBTOPIC_SCHEMA_VERSION', 4
        )

        with swap_topic_object, current_schema_version_swap:
            topic = topic_fetchers.get_topic_from_model(model)

        self.assertEqual(topic.subtopic_schema_version, 4)
        self.assertEqual(topic.name, 'name')
        self.assertEqual(topic.canonical_name, 'name')
        self.assertEqual(topic.next_subtopic_id, 1)
        self.assertEqual(topic.language_code, 'en')
        self.assertEqual(len(topic.subtopics), 1)
        self.assertEqual(topic.subtopics[0].to_dict(), subtopic_v4_dict)


class StoryReferenceMigrationTests(test_utils.GenericTestBase):

    def test_migrate_story_reference_to_latest_schema(self) -> None:
        topic_services.create_new_topic_rights('topic_id', 'user_id_admin')
        commit_cmd = topic_domain.TopicChange(
            {'cmd': topic_domain.CMD_CREATE_NEW, 'name': 'name'}
        )
        story_reference_dict = {
            'story_id': 'story_id',
            'story_is_published': False,
            'story_unpublish_type': None,
        }
        model = topic_models.TopicModel(
            id='topic_id',
            name='name',
            abbreviated_name='abbrev',
            url_fragment='name-nine',
            canonical_name='Name',
            description='description1',
            next_subtopic_id=1,
            language_code='en',
            subtopics=[],
            subtopic_schema_version=1,
            story_reference_schema_version=1,
            canonical_story_references=[story_reference_dict],
            page_title_fragment_for_web='fragm',
        )
        commit_cmd_dicts = [commit_cmd.to_dict()]
        model.commit('user_id_admin', 'topic model created', commit_cmd_dicts)

        swap_topic_object = self.swap(topic_domain, 'Topic', MockTopicObject)
        current_schema_version_swap = self.swap(
            feconf, 'CURRENT_STORY_REFERENCE_SCHEMA_VERSION', 2
        )

        with swap_topic_object, current_schema_version_swap:
            topic = topic_fetchers.get_topic_from_model(model)

        self.assertEqual(topic.story_reference_schema_version, 2)
        self.assertEqual(topic.name, 'name')
        self.assertEqual(topic.canonical_name, 'name')
        self.assertEqual(topic.next_subtopic_id, 1)
        self.assertEqual(topic.language_code, 'en')
        self.assertEqual(len(topic.canonical_story_references), 1)
        self.assertEqual(
            topic.canonical_story_references[0].to_dict(), story_reference_dict
        )