diff --git a/core/domain/question_services.py b/core/domain/question_services.py
index 2d2ee655b2535..16161a57c4179 100644
--- a/core/domain/question_services.py
+++ b/core/domain/question_services.py
@@ -726,11 +726,12 @@ def apply_change_list(
question.update_inapplicable_skill_misconception_ids(
update_skill_misconception_ids_cmd.new_value
)
- elif (
- change.property_name
- == question_domain.QUESTION_PROPERTY_NEXT_CONTENT_ID_INDEX
- ):
- # Here we use cast because this 'if' condition forces
+ else:
+ assert (
+ change.property_name
+ == question_domain.QUESTION_PROPERTY_NEXT_CONTENT_ID_INDEX
+ )
+ # Here we use cast because this 'else' branch forces
# change to have type
# UpdateQuestionPropertyNextContentIdIndexCmd.
cmd = cast(
diff --git a/core/domain/question_services_test.py b/core/domain/question_services_test.py
index 3f03c9019e7f7..65f372792faaf 100644
--- a/core/domain/question_services_test.py
+++ b/core/domain/question_services_test.py
@@ -709,6 +709,46 @@ def test_delete_question_model_with_deleted_summary_model(self) -> None:
None,
)
+ def test_delete_question_when_question_model_does_not_exist(
+ self,
+ ) -> None:
+ # Hard-delete the underlying model so get_by_id returns None,
+ # covering the `if question_model is not None` False branch.
+ question_models.QuestionModel.delete_multi(
+ [self.question_id],
+ self.editor_id,
+ feconf.COMMIT_MESSAGE_QUESTION_DELETED,
+ force_deletion=True,
+ )
+ self.assertIsNone(
+ question_models.QuestionModel.get_by_id(self.question_id)
+ )
+ question_services.delete_question(
+ self.editor_id, self.question_id, force_deletion=True
+ )
+ self.assertIsNone(
+ question_models.QuestionModel.get_by_id(self.question_id)
+ )
+
+ def test_apply_change_list_with_non_update_property_cmd(self) -> None:
+ # A change with cmd != CMD_UPDATE_QUESTION_PROPERTY should be
+ # silently skipped, covering the False branch of that `if`.
+ change_list = [
+ question_domain.QuestionChange(
+ {
+ 'cmd': (
+ question_domain.CMD_MIGRATE_STATE_SCHEMA_TO_LATEST_VERSION
+ ),
+ 'from_version': 44,
+ 'to_version': 45,
+ }
+ )
+ ]
+ question = question_services.apply_change_list(
+ self.question_id, change_list
+ )
+ self.assertEqual(question.id, self.question_id)
+
def test_update_question(self) -> None:
new_question_data = self._create_valid_question_data(
'DEF', self.content_id_generator
diff --git a/core/domain/rights_manager.py b/core/domain/rights_manager.py
index 915eeb600b6ab..a78575e077f3f 100644
--- a/core/domain/rights_manager.py
+++ b/core/domain/rights_manager.py
@@ -164,7 +164,8 @@ def _save_activity_rights(
committer_id, activity_rights, commit_message, commit_cmds
)
return
- elif activity_type == constants.ACTIVITY_TYPE_COLLECTION:
+ else:
+ assert activity_type == constants.ACTIVITY_TYPE_COLLECTION
model = collection_models.CollectionRightsModel.get(
activity_rights.id, strict=True
)
@@ -276,7 +277,8 @@ def _update_activity_summary(
if activity_type == constants.ACTIVITY_TYPE_EXPLORATION:
assert isinstance(activity_rights, exp_rights_domain.ExplorationRights)
_update_exploration_summary(activity_rights)
- elif activity_type == constants.ACTIVITY_TYPE_COLLECTION:
+ else:
+ assert activity_type == constants.ACTIVITY_TYPE_COLLECTION
_update_collection_summary(activity_rights)
@@ -418,7 +420,8 @@ def _get_activity_rights_where_user_is_owner(
exp_models.ExplorationRightsModel.owner_ids == user_id
)
).fetch()
- elif activity_type == constants.ACTIVITY_TYPE_COLLECTION:
+ else:
+ assert activity_type == constants.ACTIVITY_TYPE_COLLECTION
activity_rights_models = collection_models.CollectionRightsModel.query(
datastore_services.any_of(
collection_models.CollectionRightsModel.owner_ids == user_id
@@ -1155,7 +1158,8 @@ def _assign_role(
activity_rights.viewer_ids.remove(assignee_id)
old_role = rights_domain.ROLE_VIEWER
- elif new_role == rights_domain.ROLE_VIEWER:
+ else:
+ assert new_role == rights_domain.ROLE_VIEWER
if (
activity_rights.is_owner(assignee_id)
@@ -1382,7 +1386,8 @@ def _change_activity_status(
activity_rights.status = new_status
if activity_type == constants.ACTIVITY_TYPE_EXPLORATION:
cmd_type = rights_domain.CMD_CHANGE_EXPLORATION_STATUS
- elif activity_type == constants.ACTIVITY_TYPE_COLLECTION:
+ else:
+ assert activity_type == constants.ACTIVITY_TYPE_COLLECTION
cmd_type = rights_domain.CMD_CHANGE_COLLECTION_STATUS
commit_cmds = [
{'cmd': cmd_type, 'old_status': old_status, 'new_status': new_status}
diff --git a/core/domain/rights_manager_test.py b/core/domain/rights_manager_test.py
index cbce4612fb6b0..7b5c167fab909 100644
--- a/core/domain/rights_manager_test.py
+++ b/core/domain/rights_manager_test.py
@@ -218,6 +218,19 @@ def test_check_can_modify_core_activity_roles_for_none_activity(
)
)
+ def test_check_can_modify_core_activity_roles_owned_action_but_not_owner(
+ self,
+ ) -> None:
+ # user_a has ACTION_MODIFY_CORE_ROLES_FOR_OWNED_ACTIVITY but is not
+ # the owner of exp_id, so the function should return False.
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_b)
+ exp_rights = rights_manager.get_exploration_rights(self.EXP_ID)
+ self.assertFalse(
+ rights_manager.check_can_modify_core_activity_roles(
+ self.user_a, exp_rights
+ )
+ )
+
def test_non_splash_page_demo_exploration(self) -> None:
# Note: there is no difference between permissions for demo
# explorations, whether or not they are on the splash page.
@@ -1287,6 +1300,188 @@ def test_get_activity_rights_where_user_is_owner_for_exploration(
self.assertEqual(activity_rights_list[0].id, 'exp1')
self.assertTrue(activity_rights_list[0].is_owner(owner_id))
+ def test_get_exploration_rights_where_user_is_owner(self) -> None:
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_a)
+ owned = rights_manager.get_exploration_rights_where_user_is_owner(
+ self.user_id_a
+ )
+ self.assertEqual(len(owned), 1)
+ self.assertEqual(owned[0].id, self.EXP_ID)
+
+ def test_exploration_status_helpers(self) -> None:
+ exp = exp_domain.Exploration.create_default_exploration(
+ self.EXP_ID, title='A title', category='A category'
+ )
+ exp_services.save_new_exploration(self.user_id_a, exp)
+ self.assertTrue(rights_manager.is_exploration_private(self.EXP_ID))
+ self.assertFalse(rights_manager.is_exploration_public(self.EXP_ID))
+ self.assertFalse(rights_manager.is_exploration_cloned(self.EXP_ID))
+ rights_manager.publish_exploration(self.user_a, self.EXP_ID)
+ self.assertFalse(rights_manager.is_exploration_private(self.EXP_ID))
+ self.assertTrue(rights_manager.is_exploration_public(self.EXP_ID))
+
+ def test_check_can_functions_return_false_for_none_activity_rights(
+ self,
+ ) -> None:
+ self.assertFalse(
+ rights_manager.check_can_access_activity(self.user_a, None)
+ )
+ self.assertFalse(
+ rights_manager.check_can_edit_activity(self.user_a, None)
+ )
+ self.assertFalse(
+ rights_manager.check_can_voiceover_activity(self.user_a, None)
+ )
+ self.assertFalse(
+ rights_manager.check_can_delete_activity(self.user_a, None)
+ )
+ self.assertFalse(
+ rights_manager.check_can_release_ownership(self.user_a, None)
+ )
+ self.assertFalse(
+ rights_manager.check_can_publish_activity(self.user_a, None)
+ )
+ self.assertFalse(
+ rights_manager.check_can_unpublish_activity(self.user_a, None)
+ )
+
+ def test_check_can_edit_activity_returns_false_without_edit_action(
+ self,
+ ) -> None:
+ guest_user = user_services.get_user_actions_info(None)
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_a)
+ exp_rights = rights_manager.get_exploration_rights(self.EXP_ID)
+ self.assertFalse(
+ rights_manager.check_can_edit_activity(guest_user, exp_rights)
+ )
+
+ def test_check_can_voiceover_activity_returns_false_without_edit_action(
+ self,
+ ) -> None:
+ guest_user = user_services.get_user_actions_info(None)
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_a)
+ exp_rights = rights_manager.get_exploration_rights(self.EXP_ID)
+ self.assertFalse(
+ rights_manager.check_can_voiceover_activity(guest_user, exp_rights)
+ )
+
+ def test_check_can_modify_core_roles_returns_false_for_community_owned(
+ self,
+ ) -> None:
+ exp = exp_domain.Exploration.create_default_exploration(
+ self.EXP_ID, title='A title', category='A category'
+ )
+ exp_services.save_new_exploration(self.user_id_a, exp)
+ rights_manager.publish_exploration(self.user_a, self.EXP_ID)
+ rights_manager.release_ownership_of_exploration(
+ self.user_a, self.EXP_ID
+ )
+ exp_rights = rights_manager.get_exploration_rights(self.EXP_ID)
+ self.assertFalse(
+ rights_manager.check_can_modify_core_activity_roles(
+ self.user_a, exp_rights
+ )
+ )
+
+ def test_check_can_modify_core_roles_returns_false_without_owned_action(
+ self,
+ ) -> None:
+ # Use a mock user with no actions to force the outer `if` at the
+ # ACTION_MODIFY_CORE_ROLES_FOR_OWNED_ACTIVITY check to be False.
+ mock_user = unittest.mock.MagicMock(spec=user_domain.UserActionsInfo)
+ mock_user.user_id = self.user_id_a
+ mock_user.actions = []
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_a)
+ exp_rights = rights_manager.get_exploration_rights(self.EXP_ID)
+ self.assertFalse(
+ rights_manager.check_can_modify_core_activity_roles(
+ mock_user, exp_rights
+ )
+ )
+
+ def test_check_can_publish_returns_false_for_cloned_activity(
+ self,
+ ) -> None:
+ mock_rights = unittest.mock.MagicMock()
+ mock_rights.cloned_from = 'some_exploration_id'
+ self.assertFalse(
+ rights_manager.check_can_publish_activity(self.user_a, mock_rights)
+ )
+
+ def test_check_can_publish_returns_false_without_publish_action(
+ self,
+ ) -> None:
+ # Use a mock user with no actions to force the outer `if` at the
+ # ACTION_PUBLISH_OWNED_ACTIVITY check to be False.
+ mock_user = unittest.mock.MagicMock(spec=user_domain.UserActionsInfo)
+ mock_user.user_id = self.user_id_b
+ mock_user.actions = []
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_a)
+ exp_rights = rights_manager.get_exploration_rights(self.EXP_ID)
+ self.assertFalse(
+ rights_manager.check_can_publish_activity(mock_user, exp_rights)
+ )
+
+ def test_check_can_unpublish_returns_false_for_community_owned(
+ self,
+ ) -> None:
+ exp = exp_domain.Exploration.create_default_exploration(
+ self.EXP_ID, title='A title', category='A category'
+ )
+ exp_services.save_new_exploration(self.user_id_a, exp)
+ rights_manager.publish_exploration(self.user_a, self.EXP_ID)
+ rights_manager.release_ownership_of_exploration(
+ self.user_a, self.EXP_ID
+ )
+ exp_rights = rights_manager.get_exploration_rights(self.EXP_ID)
+ self.assertFalse(
+ rights_manager.check_can_unpublish_activity(
+ self.user_moderator, exp_rights
+ )
+ )
+
+ def test_release_ownership_raises_for_unauthorized_user(self) -> None:
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_a)
+ rights_manager.publish_exploration(self.user_a, self.EXP_ID)
+ with self.assertRaisesRegex(
+ Exception,
+ 'The ownership of this exploration cannot be released.',
+ ):
+ rights_manager.release_ownership_of_exploration(
+ self.user_b, self.EXP_ID
+ )
+
+ def test_publish_raises_for_unauthorized_user(self) -> None:
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_a)
+ with self.assertRaisesRegex(
+ Exception, 'This exploration cannot be published.'
+ ):
+ rights_manager.publish_exploration(self.user_b, self.EXP_ID)
+
+ def test_unpublish_raises_for_unauthorized_user(self) -> None:
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_a)
+ rights_manager.publish_exploration(self.user_a, self.EXP_ID)
+ with self.assertRaisesRegex(
+ Exception, 'This exploration cannot be unpublished.'
+ ):
+ rights_manager.unpublish_exploration(self.user_a, self.EXP_ID)
+
+ def test_republish_does_not_reset_first_published_msec(self) -> None:
+ self.save_new_valid_exploration(self.EXP_ID, self.user_id_a)
+ rights_manager.publish_exploration(self.user_a, self.EXP_ID)
+ rights_manager.unpublish_exploration(self.user_moderator, self.EXP_ID)
+ first_published = rights_manager.get_exploration_rights(
+ self.EXP_ID
+ ).first_published_msec
+ self.assertIsNotNone(first_published)
+ rights_manager.publish_exploration(self.user_a, self.EXP_ID)
+ self.assertEqual(
+ rights_manager.get_exploration_rights(
+ self.EXP_ID
+ ).first_published_msec,
+ first_published,
+ )
+
class CollectionRightsTests(test_utils.GenericTestBase):
"""Test that rights for actions on collections work as expected."""
@@ -2132,6 +2327,20 @@ def test_get_collection_rights_where_user_is_owner(self) -> None:
self.assertEqual(owned_rights[0].id, 'col1')
self.assertTrue(owned_rights[0].is_owner(self.user_id_a))
+ def test_collection_status_helpers(self) -> None:
+ self.save_new_valid_collection(self.COLLECTION_ID, self.user_id_a)
+ self.assertTrue(
+ rights_manager.is_collection_private(self.COLLECTION_ID)
+ )
+ self.assertFalse(
+ rights_manager.is_collection_public(self.COLLECTION_ID)
+ )
+ rights_manager.publish_collection(self.user_a, self.COLLECTION_ID)
+ self.assertFalse(
+ rights_manager.is_collection_private(self.COLLECTION_ID)
+ )
+ self.assertTrue(rights_manager.is_collection_public(self.COLLECTION_ID))
+
class CheckCanReleaseOwnershipTest(test_utils.GenericTestBase):
"""Tests for check_can_release_ownership function."""
diff --git a/core/domain/skill_domain_test.py b/core/domain/skill_domain_test.py
index 7977c823e729f..c5b4594ee798e 100644
--- a/core/domain/skill_domain_test.py
+++ b/core/domain/skill_domain_test.py
@@ -17,6 +17,7 @@
from __future__ import annotations
import datetime
+import json
from core import feconf, utils
from core.constants import constants
@@ -106,6 +107,9 @@ def test_skill_id_validation_fails_with_invalid_skill_id_length(
with self.assertRaisesRegex(utils.ValidationError, 'Invalid skill id'):
skill_domain.Skill.require_valid_skill_id('abc')
+ def test_skill_id_validation_passes_with_valid_skill_id(self) -> None:
+ skill_domain.Skill.require_valid_skill_id('skillid12345')
+
# 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.
@@ -371,6 +375,24 @@ def test_update_explanation(self) -> None:
self.skill.validate()
self.assertEqual(self.skill.skill_contents.explanation, new_explanation)
+ def test_update_explanation_with_none_existing_explanation(self) -> None:
+ # Here we use MyPy ignore because the explanation field is typed as
+ # SubtitledHtml (not Optional), but update_explanation handles a falsy
+ # explanation at runtime to cover that branch.
+ self.skill.skill_contents.explanation = None # type: ignore[assignment]
+ new_explanation = state_domain.SubtitledHtml(
+ '4', 'New Explanation
'
+ )
+ self.skill.update_explanation(new_explanation)
+ self.assertEqual(self.skill.skill_contents.explanation, new_explanation)
+
+ def test_update_explanation_with_different_content_id(self) -> None:
+ new_explanation = state_domain.SubtitledHtml(
+ '4', 'New Explanation
'
+ )
+ self.skill.update_explanation(new_explanation)
+ self.assertEqual(self.skill.skill_contents.explanation, new_explanation)
+
def test_update_rubric(self) -> None:
difficulty = constants.SKILL_DIFFICULTIES[0]
explanations = ['explanation1']
@@ -837,6 +859,25 @@ def test_serialize_and_deserialize_returns_unchanged_skill(self) -> None:
skill_domain.Skill.deserialize(self.skill.serialize()).to_dict(),
)
+ def test_serialize_with_none_created_on_and_last_updated(self) -> None:
+ self.skill.created_on = None
+ self.skill.last_updated = None
+ serialized = self.skill.serialize()
+ skill_dict = json.loads(serialized)
+ self.assertNotIn('created_on', skill_dict)
+ self.assertNotIn('last_updated', skill_dict)
+
+ def test_convert_skill_contents_v4_dict_to_v5_dict(self) -> None:
+ skill_contents_dict: skill_domain.SkillContentsDict = (
+ self.skill.skill_contents.to_dict()
+ )
+ result = skill_domain.Skill._convert_skill_contents_v4_dict_to_v5_dict( # pylint: disable=protected-access
+ skill_contents_dict
+ )
+ self.assertIn('explanation', result)
+ self.assertIn('recorded_voiceovers', result)
+ self.assertIn('written_translations', result)
+
def test_generate_skill_misconception_id(self) -> None:
"""Checks that skill misconception id is generated correctly."""
self.assertEqual(