Merge pull request #703 from SUNET/alex-split-language-name-api

Create 2 APIs specific for user name and user language

src/eduid/webapp/personal_data/schemas.py

@@ -28,6 +28,28 @@ class PersonalDataSchema(EduidSchema):
     language = fields.String(required=True, attribute="preferredLanguage")
 
 
+class UserNameRequestSchema(EduidSchema, CSRFRequestMixin):
+    given_name = fields.String(required=True, validate=[validate_nonempty])
+    chosen_given_name = fields.String(required=False)
+    surname = fields.String(required=True, validate=[validate_nonempty])
+    legal_name = fields.String(required=False)
+
+
+class UserNameSchema(EduidSchema):
+    given_name = fields.String(required=True, attribute="givenName")
+    chosen_given_name = fields.String(required=False)
+    surname = fields.String(required=True)
+    legal_name = fields.String(required=False)
+
+
+class UserLanguageRequestSchema(EduidSchema, CSRFRequestMixin):
+    language = fields.String(required=True, default="en", validate=validate_language)
+
+
+class UserLanguageSchema(EduidSchema):
+    language = fields.String(required=True, attribute="preferredLanguage")
+
+
 class UserPreferencesSchema(EduidSchema):
     always_use_security_key = fields.Boolean(required=True, default=True)
 
@@ -50,6 +72,20 @@ class PersonalDataResponsePayload(PersonalDataSchema, CSRFResponseMixin):
     payload = fields.Nested(PersonalDataResponsePayload)
 
 
+class UserNameResponseSchema(FluxStandardAction):
+    class UserNameResponsePayload(UserNameSchema, CSRFResponseMixin):
+        pass
+
+    payload = fields.Nested(UserNameResponsePayload)
+
+
+class UserLanguageResponseSchema(FluxStandardAction):
+    class UserLanguageResponsePayload(UserLanguageSchema, CSRFResponseMixin):
+        pass
+
+    payload = fields.Nested(UserLanguageResponsePayload)
+
+
 class IdentitiesResponseSchema(FluxStandardAction):
     class IdentitiesResponsePayload(EmailSchema, CSRFResponseMixin):
         identities = fields.Nested(IdentitiesSchema)

src/eduid/webapp/personal_data/tests/test_app.py

@@ -93,6 +93,56 @@ def _post_user(
                     data.update(mod_data)
             return client.post("/user", data=json.dumps(data), content_type=self.content_type_json)
 
+    @patch("eduid.common.rpc.am_relay.AmRelay.request_user_sync")
+    def _post_user_name(
+        self, mock_request_user_sync: Any, mod_data: dict[str, Any] | None = None, verified_user: bool = True
+    ):
+        """
+        POST user name for the test user
+        """
+        mock_request_user_sync.side_effect = self.request_user_sync
+        eppn = self.test_user_data["eduPersonPrincipalName"]
+
+        if not verified_user:
+            # Remove verified identities from the users
+            user = self.app.central_userdb.get_user_by_eppn(eppn)
+            for identity in user.identities.verified:
+                user.identities.remove(ElementKey(identity.identity_type.value))
+            self.app.central_userdb.save(user)
+
+        with self.session_cookie(self.browser, eppn) as client:
+            with self.app.test_request_context():
+                with client.session_transaction() as sess:
+                    data = {
+                        "given_name": "Peter",
+                        "surname": "Johnson",
+                        "csrf_token": sess.get_csrf_token(),
+                    }
+                if mod_data:
+                    data.update(mod_data)
+            return client.post("/user/name", data=json.dumps(data), content_type=self.content_type_json)
+
+    @patch("eduid.common.rpc.am_relay.AmRelay.request_user_sync")
+    def _post_user_language(
+        self, mock_request_user_sync: Any, mod_data: dict[str, Any] | None = None, verified_user: bool = True
+    ):
+        """
+        POST user language for the test user
+        """
+        mock_request_user_sync.side_effect = self.request_user_sync
+        eppn = self.test_user_data["eduPersonPrincipalName"]
+
+        with self.session_cookie(self.browser, eppn) as client:
+            with self.app.test_request_context():
+                with client.session_transaction() as sess:
+                    data = {
+                        "language": "en",
+                        "csrf_token": sess.get_csrf_token(),
+                    }
+                if mod_data:
+                    data.update(mod_data)
+            return client.post("/user/language", data=json.dumps(data), content_type=self.content_type_json)
+
     def _get_preferences(self, eppn: str | None = None):
         """
         Send a GET request to get the personal data of a user
@@ -207,6 +257,23 @@ def test_post_user(self):
         }
         self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_SUCCESS", payload=expected_payload)
 
+    def test_post_user_name(self):
+        response = self._post_user_name(verified_user=False)
+        expected_payload = {
+            "surname": "Johnson",
+            "given_name": "Peter",
+        }
+        self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_NAME_SUCCESS", payload=expected_payload)
+
+    def test_post_user_language(self):
+        response = self._post_user_language(verified_user=False)
+        expected_payload = {
+            "language": "en",
+        }
+        self._check_success_response(
+            response, type_="POST_PERSONAL_DATA_USER_LANGUAGE_SUCCESS", payload=expected_payload
+        )
+
     def test_set_chosen_given_name_and_language_verified_user(self):
         expected_payload = {
             "surname": "Smith",
@@ -216,6 +283,23 @@ def test_set_chosen_given_name_and_language_verified_user(self):
         response = self._post_user(mod_data=expected_payload)
         self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_SUCCESS", payload=expected_payload)
 
+    def test_post_user_name_set_chosen_given_name_verified_user(self):
+        expected_payload = {
+            "surname": "Smith",
+            "given_name": "John",
+        }
+        response = self._post_user_name(mod_data=expected_payload)
+        self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_NAME_SUCCESS", payload=expected_payload)
+
+    def test_post_user_language_set_language_verified_user(self):
+        expected_payload = {
+            "language": "sv",
+        }
+        response = self._post_user_language(mod_data=expected_payload)
+        self._check_success_response(
+            response, type_="POST_PERSONAL_DATA_USER_LANGUAGE_SUCCESS", payload=expected_payload
+        )
+
     def test_set_given_name_and_surname_verified_user(self):
         mod_data = {
             "surname": "Johnson",
@@ -230,31 +314,68 @@ def test_set_given_name_and_surname_verified_user(self):
         response = self._post_user(mod_data=mod_data)
         self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_SUCCESS", payload=expected_payload)
 
+    def test_post_user_name_set_given_name_and_surname_verified_user(self):
+        mod_data = {
+            "surname": "Johnson",
+            "given_name": "Peter",
+        }
+        expected_payload = {
+            "surname": "Smith",
+            "given_name": "John",
+        }
+        response = self._post_user_name(mod_data=mod_data)
+        self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_NAME_SUCCESS", payload=expected_payload)
+
     def test_post_user_bad_csrf(self):
         response = self._post_user(mod_data={"csrf_token": "wrong-token"})
         expected_payload = {"error": {"csrf_token": ["CSRF failed to validate"]}}
         self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_FAIL", payload=expected_payload)
 
+    def test_post_user__name_bad_csrf(self):
+        response = self._post_user_name(mod_data={"csrf_token": "wrong-token"})
+        expected_payload = {"error": {"csrf_token": ["CSRF failed to validate"]}}
+        self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_NAME_FAIL", payload=expected_payload)
+
     def test_post_user_no_given_name(self):
         response = self._post_user(mod_data={"given_name": ""})
         expected_payload = {"error": {"given_name": ["pdata.field_required"]}}
         self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_FAIL", payload=expected_payload)
 
+    def test_post_user_name_no_given_name(self):
+        response = self._post_user_name(mod_data={"given_name": ""})
+        expected_payload = {"error": {"given_name": ["pdata.field_required"]}}
+        self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_NAME_FAIL", payload=expected_payload)
+
     def test_post_user_blank_given_name(self):
         response = self._post_user(mod_data={"given_name": " "})
         expected_payload = {"error": {"given_name": ["pdata.field_required"]}}
         self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_FAIL", payload=expected_payload)
 
+    def test_post_user_name_blank_given_name(self):
+        response = self._post_user_name(mod_data={"given_name": " "})
+        expected_payload = {"error": {"given_name": ["pdata.field_required"]}}
+        self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_NAME_FAIL", payload=expected_payload)
+
     def test_post_user_no_surname(self):
         response = self._post_user(mod_data={"surname": ""})
         expected_payload = {"error": {"surname": ["pdata.field_required"]}}
         self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_FAIL", payload=expected_payload)
 
+    def test_post_user_name_no_surname(self):
+        response = self._post_user_name(mod_data={"surname": ""})
+        expected_payload = {"error": {"surname": ["pdata.field_required"]}}
+        self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_NAME_FAIL", payload=expected_payload)
+
     def test_post_user_blank_surname(self):
         response = self._post_user(mod_data={"surname": " "})
         expected_payload = {"error": {"surname": ["pdata.field_required"]}}
         self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_FAIL", payload=expected_payload)
 
+    def test_post_user_name_blank_surname(self):
+        response = self._post_user_name(mod_data={"surname": " "})
+        expected_payload = {"error": {"surname": ["pdata.field_required"]}}
+        self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_NAME_FAIL", payload=expected_payload)
+
     def test_post_user_with_chosen_given_name(self):
         response = self._post_user(mod_data={"chosen_given_name": "Peter"}, verified_user=False)
         expected_payload = {
@@ -265,12 +386,27 @@ def test_post_user_with_chosen_given_name(self):
         }
         self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_SUCCESS", payload=expected_payload)
 
+    def test_post_user_name_with_chosen_given_name(self):
+        response = self._post_user_name(mod_data={"chosen_given_name": "Peter"}, verified_user=False)
+        expected_payload = {
+            "surname": "Johnson",
+            "given_name": "Peter",
+            "chosen_given_name": "Peter",
+        }
+        self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_NAME_SUCCESS", payload=expected_payload)
+
     def test_post_user_with_bad_chosen_given_name(self):
         response = self._post_user(mod_data={"chosen_given_name": "Michael"}, verified_user=False)
         self._check_error_response(
             response, type_="POST_PERSONAL_DATA_USER_FAIL", msg=PDataMsg.chosen_given_name_invalid
         )
 
+    def test_post_user_name_with_bad_chosen_given_name(self):
+        response = self._post_user_name(mod_data={"chosen_given_name": "Michael"}, verified_user=False)
+        self._check_error_response(
+            response, type_="POST_PERSONAL_DATA_USER_NAME_FAIL", msg=PDataMsg.chosen_given_name_invalid
+        )
+
     def test_post_user_to_unset_chosen_given_name(self):
         # set test user chosen given name
         self.test_user.chosen_given_name = "Peter"
@@ -286,16 +422,40 @@ def test_post_user_to_unset_chosen_given_name(self):
         }
         self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_SUCCESS", payload=expected_payload)
 
+    def test_post_user_name_to_unset_chosen_given_name(self):
+        # set test user chosen given name
+        self.test_user.chosen_given_name = "Peter"
+        self.app.central_userdb.save(self.test_user)
+        user = self.app.central_userdb.get_user_by_eppn(eppn=self.test_user.eppn)
+        assert user.chosen_given_name == "Peter"
+
+        response = self._post_user_name(verified_user=False)
+        expected_payload = {
+            "surname": "Johnson",
+            "given_name": "Peter",
+        }
+        self._check_success_response(response, type_="POST_PERSONAL_DATA_USER_NAME_SUCCESS", payload=expected_payload)
+
     def test_post_user_no_language(self):
         response = self._post_user(mod_data={"language": ""})
         expected_payload = {"error": {"language": ["Language '' is not available"]}}
         self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_FAIL", payload=expected_payload)
 
+    def test_post_user_language_no_language(self):
+        response = self._post_user_language(mod_data={"language": ""})
+        expected_payload = {"error": {"language": ["Language '' is not available"]}}
+        self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_LANGUAGE_FAIL", payload=expected_payload)
+
     def test_post_user_unknown_language(self):
         response = self._post_user(mod_data={"language": "es"})
         expected_payload = {"error": {"language": ["Language 'es' is not available"]}}
         self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_FAIL", payload=expected_payload)
 
+    def test_post_user_language_unknown_language(self):
+        response = self._post_user_language(mod_data={"language": "es"})
+        expected_payload = {"error": {"language": ["Language 'es' is not available"]}}
+        self._check_error_response(response, type_="POST_PERSONAL_DATA_USER_LANGUAGE_FAIL", payload=expected_payload)
+
     def test_get_preferences(self):
         response = self._get_preferences()
         expected_payload = {"always_use_security_key": True}

src/eduid/webapp/personal_data/views.py

@@ -1,6 +1,7 @@
 from flask import Blueprint
 
 from eduid.common.config.base import FrontendAction
+from eduid.common.decorators import deprecated
 from eduid.userdb import User
 from eduid.userdb.exceptions import UserOutOfSync
 from eduid.userdb.personal_data import PersonalDataUser
@@ -14,6 +15,10 @@
     IdentitiesResponseSchema,
     PersonalDataRequestSchema,
     PersonalDataResponseSchema,
+    UserLanguageRequestSchema,
+    UserLanguageResponseSchema,
+    UserNameRequestSchema,
+    UserNameResponseSchema,
     UserPreferencesRequestSchema,
     UserPreferencesResponseSchema,
 )
@@ -44,6 +49,7 @@ def get_user(user: User) -> FluxData:
     return success_response(payload=user.to_dict())
 
 
+@deprecated("update_personal_data view is deprecated, use update_user_name or update_user_language view instead")
 @pd_views.route("/user", methods=["POST"])
 @UnmarshalWith(PersonalDataRequestSchema)
 @MarshalWith(PersonalDataResponseSchema)
@@ -83,6 +89,63 @@ def update_personal_data(
     return success_response(payload=personal_data, message=PDataMsg.save_success)
 
 
+@pd_views.route("/user/name", methods=["POST"])
+@UnmarshalWith(UserNameRequestSchema)
+@MarshalWith(UserNameResponseSchema)
+@require_user
+def update_user_name(user: User, given_name: str, surname: str, chosen_given_name: str | None = None) -> FluxData:
+    personal_data_user = PersonalDataUser.from_user(user, current_app.private_userdb)
+    current_app.logger.debug(f"Trying to save user {user}")
+
+    # disallow change of first name, surname if the user is verified
+    if not user.identities.is_verified:
+        personal_data_user.given_name = given_name
+        personal_data_user.surname = surname
+
+    # set chosen given name to either given name or a subset of given name if supplied
+    # also allow to set chosen given name to None
+    if (
+        chosen_given_name is not None
+        and is_valid_chosen_given_name(personal_data_user.given_name, chosen_given_name) is False
+    ):
+        return error_response(message=PDataMsg.chosen_given_name_invalid)
+
+    # mypy borked?
+    # error: Incompatible types in assignment (expression has type "str | None", variable has type "str")
+    personal_data_user.chosen_given_name = chosen_given_name
+
+    try:
+        save_and_sync_user(personal_data_user)
+    except UserOutOfSync:
+        return error_response(message=CommonMsg.out_of_sync)
+    current_app.stats.count(name="user_name_saved", value=1)
+    current_app.logger.info(f"Saved personal data for user {personal_data_user}")
+
+    personal_data = personal_data_user.to_dict()
+    return success_response(payload=personal_data, message=PDataMsg.save_success)
+
+
+@pd_views.route("/user/language", methods=["POST"])
+@UnmarshalWith(UserLanguageRequestSchema)
+@MarshalWith(UserLanguageResponseSchema)
+@require_user
+def update_user_language(user: User, language: str) -> FluxData:
+    personal_data_user = PersonalDataUser.from_user(user, current_app.private_userdb)
+    current_app.logger.debug(f"Trying to save user {user}")
+
+    personal_data_user.language = language
+
+    try:
+        save_and_sync_user(personal_data_user)
+    except UserOutOfSync:
+        return error_response(message=CommonMsg.out_of_sync)
+    current_app.stats.count(name="user_language_saved", value=1)
+    current_app.logger.info(f"Saved personal data for user {personal_data_user}")
+
+    personal_data = personal_data_user.to_dict()
+    return success_response(payload=personal_data, message=PDataMsg.save_success)
+
+
 @pd_views.route("/preferences", methods=["GET"])
 @MarshalWith(UserPreferencesResponseSchema)
 @require_user