Enable updating nin

src/eduid/webapp/common/api/testing.py

@@ -24,6 +24,7 @@
 from eduid.userdb.db import BaseDB
 from eduid.userdb.element import ElementKey
 from eduid.userdb.fixtures.users import UserFixtures
+from eduid.userdb.identity import IdentityType
 from eduid.userdb.logs.db import ProofingLog
 from eduid.userdb.proofing.state import NinProofingState
 from eduid.userdb.testing import MongoTemporaryInstance, SetupConfig
@@ -271,10 +272,15 @@ def request_user_sync(self, private_user: User, app_name_override: str | None =
 
         central_user = self.app.central_userdb.get_user_by_id(private_user.user_id)
         private_user_dict = private_user.to_dict()
+        replace_lock: IdentityType | None = None
         # fix signup_user data
         if "proofing_reference" in private_user_dict:
             del private_user_dict["proofing_reference"]
 
+        if "replace_lock" in private_user_dict:
+            replace_lock = private_user_dict["replace_lock"]
+            del private_user_dict["replace_lock"]
+
         if central_user is None:
             # This is a new user, create a new user in the central db
             self.app.central_userdb.save(User.from_dict(private_user_dict))
@@ -303,6 +309,11 @@ def request_user_sync(self, private_user: User, app_name_override: str | None =
                     identity.created_by = "test"
                 user.locked_identity.add(identity)
                 continue
+            if replace_lock is locked_identity.identity_type:
+                # replace the locked identity with the new verified identity
+                if identity.created_by is None:
+                    identity.created_by = "test"
+                user.locked_identity.replace(identity)
 
         # Restore metadata that is necessary for the consistency checks in the save() function
         user.modified_ts = central_user.modified_ts

src/eduid/webapp/common/api/tests/test_nin_helpers.py

@@ -12,7 +12,8 @@
 from eduid.userdb import NinIdentity, User
 from eduid.userdb.exceptions import LockedIdentityViolation, UserDoesNotExist
 from eduid.userdb.fixtures.users import UserFixtures
-from eduid.userdb.identity import IdentityList
+from eduid.userdb.identity import IdentityList, IdentityType
+from eduid.userdb.locked_identity import LockedIdentityList
 from eduid.userdb.logs import ProofingLog
 from eduid.userdb.logs.element import (
     ForeignIdProofingLogElement,
@@ -106,10 +107,28 @@ def insert_not_verified_user(self, nin: str | None = None) -> User:
         self.app.central_userdb.save(user)
         return self.app.central_userdb.get_user_by_eppn(user.eppn)
 
+    def insert_not_verified_not_locked_user(self, nin: str | None = None) -> User:
+        user = self.app.central_userdb.get_user_by_eppn(self.test_user.eppn)
+        user.identities = IdentityList()
+        if nin is None:
+            nin = self.test_user_nin
+        nin_element = NinIdentity.from_dict(
+            dict(
+                number=nin,
+                created_by="AlreadyAddedNinHelpersTest",
+                verified=False,
+            )
+        )
+        user.identities.add(nin_element)
+        user.locked_identity = LockedIdentityList()
+        self.app.central_userdb.save(user)
+        return self.app.central_userdb.get_user_by_eppn(user.eppn)
+
     def insert_no_nins_user(self) -> User:
         # Replace user with one without previous proofings
         user = self.app.central_userdb.get_user_by_eppn(self.test_user.eppn)
         user.identities = IdentityList()
+        user.locked_identity = LockedIdentityList()
         self.app.central_userdb.save(user)
         return self.app.central_userdb.get_user_by_eppn(user.eppn)
 
@@ -292,7 +311,7 @@ def test_verify_nin_for_proofing_user_eid(self, mock_reference_nin: MagicMock) -
     @patch("eduid.webapp.common.api.helpers.get_reference_nin_from_navet_data")
     def test_verify_nin_for_user_existing_not_verified(self, mock_reference_nin: MagicMock) -> None:
         mock_reference_nin.return_value = None
-        user = self.insert_not_verified_user()
+        user = self.insert_not_verified_not_locked_user()
         nin_element = NinProofingElement.from_dict(
             dict(number=self.test_user_nin, created_by="NinHelpersTest", verified=False)
         )
@@ -309,6 +328,29 @@ def test_verify_nin_for_user_existing_not_verified(self, mock_reference_nin: Mag
             user=user, proofing_state=proofing_state, number=self.test_user_nin, created_by="AlreadyAddedNinHelpersTest"
         )
 
+    @patch("eduid.webapp.common.api.helpers.get_reference_nin_from_navet_data")
+    def test_verify_nin_for_user_existing_locked_not_verified(self, mock_reference_nin: MagicMock) -> None:
+        mock_reference_nin.return_value = None
+        user = self.insert_not_verified_user()
+        nin_element = NinProofingElement.from_dict(
+            dict(number=self.locked_test_user_nin, created_by="NinHelpersTest", verified=False)
+        )
+        proofing_state = NinProofingState.from_dict({"eduPersonPrincipalName": user.eppn, "nin": nin_element.to_dict()})
+        assert nin_element.created_by is not None
+        proofing_log_entry = self._get_nin_eid_proofing_log_entry(
+            user=user, created_by=nin_element.created_by, nin=nin_element.number
+        )
+        with self.app.app_context():
+            assert verify_nin_for_user(user, proofing_state, proofing_log_entry) is True
+        user = self.app.private_userdb.get_user_by_eppn(user.eppn)
+
+        self._check_nin_verified_ok(
+            user=user,
+            proofing_state=proofing_state,
+            number=self.locked_test_user_nin,
+            created_by="NinHelpersTest",
+        )
+
     @patch("eduid.webapp.common.api.helpers.get_reference_nin_from_navet_data")
     def test_verify_wrong_nin_for_user_existing_not_verified(self, mock_reference_nin: MagicMock) -> None:
         mock_reference_nin.return_value = None
@@ -349,8 +391,8 @@ def test_verify_changed_nin_for_user_existing_not_verified(self, mock_reference_
         # user should be updated with updated nin as it references old locked nin
         assert user.identities.nin is not None
         assert user.identities.nin.number == self.test_user_nin
-        assert user.locked_identity.nin is not None
-        assert user.locked_identity.nin.number == self.test_user_nin
+        # NIN should be updated by am when saving to main DB
+        assert user.replace_lock is IdentityType.NIN
 
     def test_verify_nin_for_user_existing_verified(self) -> None:
         user = self.insert_verified_user()

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

@@ -24,6 +24,7 @@ class LetterProofingTests(EduidAPITestCase[LetterProofingApp]):
     def setUp(self, config: SetupConfig | None = None) -> None:
         self.test_user_eppn = "hubba-baar"
         self.test_user_nin = "200001023456"
+        self.test_old_user_nin = "199909096789"
         self.test_user_wrong_nin = "190001021234"
         if config is None:
             config = SetupConfig()
@@ -371,6 +372,46 @@ def test_proofing_flow_previously_added_wrong_nin(self, mock_reference_nin: Magi
         user = self.app.central_userdb.get_user_by_eppn(self.test_user_eppn)
         self._check_nin_verified_ok(user=user, proofing_state=proofing_state, number=self.test_user_nin)
 
+    @patch("eduid.webapp.common.api.decorators.get_reference_nin_from_navet_data")
+    @patch("eduid.webapp.common.api.helpers.get_reference_nin_from_navet_data")
+    def test_proofing_flow_with_replacement_nin(
+        self, mock_reference_nin: MagicMock, mock_decorator_nin: MagicMock
+    ) -> None:
+        mock_reference_nin.return_value = self.test_old_user_nin
+        mock_decorator_nin.return_value = self.test_old_user_nin
+
+        # Send letter to old nin and verify
+        self.send_letter(self.test_old_user_nin)
+        proofing_state = self.app.proofing_statedb.get_state_by_eppn(self.test_user_eppn)
+        assert proofing_state is not None
+        assert proofing_state.nin is not None
+        assert proofing_state.nin.verification_code is not None
+        self.verify_code(proofing_state.nin.verification_code, None)
+
+        # Check that old nin is locked in
+        user = self.app.central_userdb.get_user_by_eppn(self.test_user_eppn)
+        assert user.locked_identity.nin is not None
+        assert user.locked_identity.nin.number == self.test_old_user_nin
+
+        # Replace old nin with new nin
+        user.identities.remove(ElementKey(IdentityType.NIN))
+        not_verified_nin = NinIdentity(number=self.test_user_nin, created_by="test", is_verified=False)
+        user.identities.add(not_verified_nin)
+        self.app.central_userdb.save(user)
+
+        # Send letter to new nin and verify
+        self.send_letter(self.test_user_nin)
+        new_proofing_state = self.app.proofing_statedb.get_state_by_eppn(self.test_user_eppn)
+        assert new_proofing_state is not None
+        assert new_proofing_state.nin is not None
+        assert new_proofing_state.nin.verification_code is not None
+        self.verify_code(new_proofing_state.nin.verification_code, None)
+
+        # Check that new nin is locked in
+        updated_user = self.app.central_userdb.get_user_by_eppn(self.test_user_eppn)
+        assert updated_user.locked_identity.nin is not None
+        assert updated_user.locked_identity.nin.number == self.test_user_nin
+
     def test_expire_proofing_state(self) -> None:
         self.send_letter(self.test_user_nin)
         json_data = self.get_state()

src/eduid/workers/am/tests/test_tasks.py

@@ -1,6 +1,7 @@
 from bson import ObjectId
 
 import eduid.userdb
+from eduid.common.testing_base import normalised_data
 from eduid.userdb import LockedIdentityList, NinIdentity
 from eduid.userdb.exceptions import EduIDUserDBError, MultipleUsersReturned
 from eduid.userdb.fixtures.users import UserFixtures
@@ -193,14 +194,22 @@ def test_unverify_duplicate_multiple_attribute_values(self) -> None:
 
     def test_create_locked_identity(self) -> None:
         user_id = ObjectId("901234567890123456789012")  # [email protected] / babba-labba
-        attributes = {"$set": {"nins": [{"verified": True, "number": "200102031234", "primary": True}]}}
+        attributes = {
+            "$set": {
+                "identities": [{"identity_type": IdentityType.NIN.value, "number": "200102031234", "verified": True}]
+            }
+        }
         new_attributes = check_locked_identity(self.amdb, user_id, attributes, "test")
 
+        # check_locked_identity should create a locked_identity so add it to the expected attributes
         locked_nin = NinIdentity(number="200102031234", created_by="test", is_verified=True)
         locked_identities = LockedIdentityList(elements=[locked_nin])
         attributes["$set"]["locked_identity"] = locked_identities.to_list_of_dicts()
 
-        self.assertDictEqual(attributes, new_attributes)
+        self.assertDictEqual(
+            normalised_data(attributes, exclude_keys=["created_ts", "modified_ts"]),
+            normalised_data(new_attributes, exclude_keys=["created_ts", "modified_ts"]),
+        )
 
     def test_check_locked_identity(self) -> None:
         user_id = ObjectId("012345678901234567890123")  # [email protected] / hubba-bubba
@@ -212,14 +221,13 @@ def test_check_locked_identity(self) -> None:
         self.amdb.save(user)
         attributes = {
             "$set": {
-                "nins": [{"verified": True, "number": locked_nin.number, "primary": True}],  # hubba-bubba's primary nin
+                "identities": [
+                    {"identity_type": IdentityType.NIN.value, "number": locked_nin.number, "verified": True}
+                ],  # hubba-bubba
             }
         }
         new_attributes = check_locked_identity(self.amdb, user_id, attributes, "test")
-
-        locked_identities = LockedIdentityList(elements=[locked_nin])
-        attributes["$set"]["locked_identity"] = locked_identities.to_list_of_dicts()
-
+        # user has locked_identity that is the same as the verified identity so only identities should be set
         self.assertDictEqual(attributes, new_attributes)
 
     def test_check_locked_identity_wrong_nin(self) -> None:
@@ -236,6 +244,29 @@ def test_check_locked_identity_wrong_nin(self) -> None:
         with self.assertRaises(EduIDUserDBError):
             check_locked_identity(self.amdb, user_id, attributes, "test")
 
+    def test_check_locked_identity_replace_lock(self) -> None:
+        user_id = ObjectId("901234567890123456789012")  # [email protected] / babba-labba
+        user = self.amdb.get_user_by_id(user_id)
+        assert user
+        user.locked_identity.add(NinIdentity(number="200102031234", created_by="test", is_verified=True))
+        self.amdb.save(user)
+        attributes = {
+            "$set": {
+                "identities": [{"identity_type": IdentityType.NIN.value, "verified": True, "number": "200506076789"}]
+            }
+        }
+        new_attributes = check_locked_identity(self.amdb, user_id, attributes, "test", replace_lock=IdentityType.NIN)
+
+        # check_locked_identity should replace locked identity with new identity
+        locked_nin = NinIdentity(number="200506076789", created_by="test", is_verified=True)
+        locked_identities = LockedIdentityList(elements=[locked_nin])
+        attributes["$set"]["locked_identity"] = locked_identities.to_list_of_dicts()
+
+        self.assertDictEqual(
+            normalised_data(attributes, exclude_keys=["created_ts", "modified_ts"]),
+            normalised_data(new_attributes, exclude_keys=["created_ts", "modified_ts"]),
+        )
+
     def test_check_locked_identity_no_verified_nin(self) -> None:
         user_id = ObjectId("012345678901234567890123")  # [email protected] / hubba-bubba
         attributes = {"$set": {"phone": [{"verified": True, "number": "+34609609609", "primary": True}]}}