Resynchronisation des groups à la reconnexion d'un provider

src/pypnusershub/auth/authentication.py

@@ -266,41 +266,56 @@ def insert_or_update_role(
             db.session.add(provider)
             db.session.commit()
 
-        if user_exists:
-            if not provider in user_exists.providers:
-                user_exists.providers.append(provider)
-
-            for attr_key, attr_value in user_dict.items():
-                if fields_to_update and attr_key not in fields_to_update:
-                    continue
-                setattr(user_exists, attr_key, attr_value)
-            db.session.commit()
-            return user_exists
-        else:
-            user_ = models.User(**user_dict)
+        def apply_groups_mapping(user_obj):
             group_id = ""
-            # No group mapping indicated
-            if not (self.group_mapping and source_groups):
-
+            # No group mapping indicated: fallback to default reconciliation group.
+            if not self.group_mapping:
                 if "DEFAULT_RECONCILIATION_GROUP_ID" in current_app.config.get(
                     "AUTHENTICATION", {}
                 ):
-
                     group_id = current_app.config["AUTHENTICATION"][
                         "DEFAULT_RECONCILIATION_GROUP_ID"
                     ]
                     group = db.session.get(models.User, group_id)
-                    if group:
-                        user_.groups.append(group)
-            # Group Mapping indicated
+                    if group and not group in user_obj.groups:
+                        user_obj.groups.append(group)
+            # Group mapping indicated: strict mirror for managed groups.
             else:
+                managed_group_ids = set(self.group_mapping.values())
+                target_group_ids = {
+                    self.group_mapping[group_source_name]
+                    for group_source_name in source_groups
+                    if group_source_name in self.group_mapping
+                }
+
+                for current_group in list(user_obj.groups):
+                    if (
+                        current_group.id_role in managed_group_ids
+                        and current_group.id_role not in target_group_ids
+                    ):
+                        user_obj.groups.remove(current_group)
+
                 for group_source_name in source_groups:
                     group_id = self.group_mapping.get(group_source_name, None)
                     if group_id:
                         group = db.session.get(models.User, group_id)
-                        if group and not group in user_.groups:
-                            user_.groups.append(group)
+                        if group and not group in user_obj.groups:
+                            user_obj.groups.append(group)
 
+        if user_exists:
+            if not provider in user_exists.providers:
+                user_exists.providers.append(provider)
+
+            for attr_key, attr_value in user_dict.items():
+                if fields_to_update and attr_key not in fields_to_update:
+                    continue
+                setattr(user_exists, attr_key, attr_value)
+            apply_groups_mapping(user_exists)
+            db.session.commit()
+            return user_exists
+        else:
+            user_ = models.User(**user_dict)
+            apply_groups_mapping(user_)
             user_.providers.append(provider)
             db.session.add(user_)
             db.session.commit()

src/pypnusershub/tests/test_utilisateurs.py

@@ -131,6 +131,80 @@ def test_insert_or_update_role_grp_reconcialiation(
         user_group_id = map(lambda g: g.id_role, user.groups)
         assert set(user_group_id) == {group_and_users["group1"].id_role}
 
+    def test_insert_or_update_role_grp_reconcialiation_strict_mirror_existing_user(
+        self, provider_instance, group_and_users
+    ):
+        provider_instance.group_mapping = {
+            "group1": group_and_users["group1"].id_role,
+            "group2": group_and_users["group2"].id_role,
+        }
+
+        user_dict = {
+            "id_role": 999997,
+            "identifiant": "mirror.user",
+            "nom_role": "mirror",
+            "prenom_role": "user",
+            "email": "mirror@test.fr",
+        }
+
+        user = provider_instance.insert_or_update_role(
+            user_dict=user_dict,
+            source_groups=["group1", "group2"],
+            reconciliate_attr="identifiant",
+        )
+        assert set(map(lambda g: g.id_role, user.groups)) == {
+            group_and_users["group1"].id_role,
+            group_and_users["group2"].id_role,
+        }
+
+        # Re-login with fewer source groups should remove obsolete mapped groups.
+        user = provider_instance.insert_or_update_role(
+            user_dict=user_dict,
+            source_groups=["group2"],
+            reconciliate_attr="identifiant",
+        )
+        assert set(map(lambda g: g.id_role, user.groups)) == {
+            group_and_users["group2"].id_role
+        }
+
+    def test_insert_or_update_role_grp_reconcialiation_strict_mirror_keeps_unmanaged_groups(
+        self, provider_instance, group_and_users
+    ):
+        provider_instance.group_mapping = {
+            "group1": group_and_users["group1"].id_role,
+            "group2": group_and_users["group2"].id_role,
+        }
+
+        unmanaged_group = User(groupe=True, identifiant="group_unmanaged")
+        db.session.add(unmanaged_group)
+        db.session.flush()
+
+        user_dict = {
+            "id_role": 999996,
+            "identifiant": "mirror.user.unmanaged",
+            "nom_role": "mirror",
+            "prenom_role": "unmanaged",
+            "email": "mirror-unmanaged@test.fr",
+        }
+
+        user = provider_instance.insert_or_update_role(
+            user_dict=user_dict,
+            source_groups=["group1", "group2"],
+            reconciliate_attr="identifiant",
+        )
+        user.groups.append(unmanaged_group)
+        db.session.commit()
+
+        user = provider_instance.insert_or_update_role(
+            user_dict=user_dict,
+            source_groups=["group2"],
+            reconciliate_attr="identifiant",
+        )
+        assert set(map(lambda g: g.id_role, user.groups)) == {
+            group_and_users["group2"].id_role,
+            unmanaged_group.id_role,
+        }
+
     def test_insert_or_update_with_fields_to_update(
         self, provider_instance, group_and_users
     ):