fix(integrations): Reparent identity on install (#8625)

When installing an integration, if an identity was already linked to the
identity provider for the user or external ID of the installer, the
integration would fail to install. This corrects that by simply
reparenting the identity to that user + external_id

src/sentry/integrations/pipeline.py

@@ -2,6 +2,8 @@ from __future__ import absolute_import, print_function
 
 __all__ = ['IntegrationPipeline']
 
+from django.db import IntegrityError
+from django.db.models import Q
 from django.http import HttpResponse
 from django.utils import timezone
 
@@ -78,17 +80,33 @@ class IntegrationPipeline(Pipeline):
             if not created:
                 idp.update(config=identity_config)
 
-            Identity.objects.get_or_create(
-                idp=idp,
-                user=self.request.user,
-                external_id=identity['external_id'],
-                defaults={
-                    'status': IdentityStatus.VALID,
-                    'scopes': identity['scopes'],
-                    'data': identity['data'],
-                    'date_verified': timezone.now(),
-                },
-            )
+            identity_data = {
+                'status': IdentityStatus.VALID,
+                'scopes': identity['scopes'],
+                'data': identity['data'],
+                'date_verified': timezone.now(),
+            }
+
+            try:
+                Identity.objects.get_or_create(
+                    idp=idp,
+                    user=self.request.user,
+                    external_id=identity['external_id'],
+                    defaults=identity_data,
+                )
+            except IntegrityError:
+                # If the external_id is already used for a different user or
+                # the user already has a different external_id remove those
+                # identities and recreate it.
+                lookup = Q(external_id=identity['external_id']) | Q(user=self.request.user)
+                Identity.objects.filter(lookup, idp=idp).delete()
+
+                Identity.objects.create(
+                    idp=idp,
+                    user=self.request.user,
+                    external_id=identity['external_id'],
+                    **identity_data
+                )
 
         return self._dialog_response(serialize(org_integration, self.request.user), True)
 

tests/sentry/integrations/slack/test_integration.py

@@ -129,3 +129,13 @@ class SlackIntegrationTest(IntegrationTestCase):
         assert identities.count() == 2
         assert identities[0].external_id != identities[1].external_id
         assert identities[0].idp != identities[1].idp
+
+    @responses.activate
+    def test_reassign_user(self):
+        self.assert_setup_flow()
+        identity = Identity.objects.get()
+        assert identity.external_id == 'UXXXXXXX1'
+
+        self.assert_setup_flow(authorizing_user_id='UXXXXXXX2')
+        identity = Identity.objects.get()
+        assert identity.external_id == 'UXXXXXXX2'