def test_disconnect_with_partial_pipeline(self):
        self.strategy.set_settings({
            'SOCIAL_AUTH_DISCONNECT_PIPELINE': (
                'social.pipeline.partial.save_status_to_session',
                'social.tests.pipeline.ask_for_password',
                'social.tests.pipeline.set_password',
                'social.pipeline.disconnect.allowed_to_disconnect',
                'social.pipeline.disconnect.get_entries',
                'social.pipeline.disconnect.revoke_tokens',
                'social.pipeline.disconnect.disconnect'
            )
        })
        self.do_login()
        user = User.get(self.expected_username)
        redirect = do_disconnect(self.backend, user)

        url = self.strategy.build_absolute_uri('/password')
        expect(redirect.url).to.equal(url)
        HTTPretty.register_uri(HTTPretty.GET, redirect.url, status=200,
                               body='foobar')
        HTTPretty.register_uri(HTTPretty.POST, redirect.url, status=200)

        password = 'foobar'
        requests.get(url)
        requests.post(url, data={'password': password})
        data = parse_qs(HTTPretty.last_request.body)
        expect(data['password']).to.equal(password)
        self.strategy.session_set('password', data['password'])

        redirect = do_disconnect(self.backend, user)
        expect(len(user.social)).to.equal(0)

    def test_oauth_connection_to_user(self, m):
        # tests: 1) connect social to an existing user
        #        2) disconnect social from the user

        # expect for only 1 user: anonymous user
        self.assertEqual(User.objects.all().count(), 1)
        # manually create a new user named self.social_username
        user_sherry = User.objects.create_user(username="sherry",
                                             email="[email protected]",
                                             password='top_secret')
        logger.debug(user_sherry.is_authenticated())
        logger.debug(user_sherry.is_active)

        # expect for 2 users: anonymous and sherry
        self.assertEqual(User.objects.all().count(), 2)

        username_new, social, backend = self.run_oauth(m, user=user_sherry)

        # still expect for 2 users
        self.assertEqual(User.objects.all().count(), 2)
        # check social is connected to user_sherry
        self.assertEqual(social.user, user_sherry)

        # check extra_data
        extra_data_dict = social.extra_data
        self.assertEqual(extra_data_dict["access_token"], self.access_token)
        self.assertEqual(extra_data_dict["refresh_token"], self.refresh_token)
        self.assertEqual(extra_data_dict["expires_in"], self.expires_in)
        self.assertEqual(extra_data_dict["token_type"], self.token_type)
        self.assertEqual(extra_data_dict["scope"], self.scope)

        # test disconnect
        self.assertEqual(UserSocialAuth.objects.count(), 1)
        do_disconnect(backend, user_sherry, association_id=social.id)
        self.assertEqual(UserSocialAuth.objects.count(), 0)

 def test_disconnect_with_association_id(self):
     self.do_login()
     user = User.get(self.expected_username)
     user.password = 'password'
     association_id = user.social[0].id
     second_usa = TestUserSocialAuth(user, user.social[0].provider, "uid2")
     self.assertEqual(len(user.social), 2)
     do_disconnect(self.backend, user, association_id)
     self.assertEqual(len(user.social), 1)
     self.assertEqual(user.social[0], second_usa)

 def test_revoke_token(self):
     self.strategy.set_settings({
         'SOCIAL_AUTH_REVOKE_TOKENS_ON_DISCONNECT': True
     })
     self.do_login()
     user = User.get(self.expected_username)
     user.password = 'password'
     HTTPretty.register_uri(self._method(self.backend.REVOKE_TOKEN_METHOD),
                            self.backend.REVOKE_TOKEN_URL,
                            status=200)
     do_disconnect(self.backend, user)

    def test_full_pipeline_succeeds_for_unlinking_account(self):
        # First, create, the request and strategy that store pipeline state,
        # configure the backend, and mock out wire traffic.
        request, strategy = self.get_request_and_strategy(
            auth_entry=pipeline.AUTH_ENTRY_LOGIN, redirect_uri='social:complete')
        request.backend.auth_complete = mock.MagicMock(return_value=self.fake_auth_complete(strategy))
        user = self.create_user_models_for_existing_account(
            strategy, '[email protected]', 'password', self.get_username())
        self.assert_social_auth_exists_for_user(user, strategy)

        # We're already logged in, so simulate that the cookie is set correctly
        self.set_logged_in_cookies(request)

        # Instrument the pipeline to get to the dashboard with the full
        # expected state.
        self.client.get(
            pipeline.get_login_url(self.provider.provider_id, pipeline.AUTH_ENTRY_LOGIN))
        actions.do_complete(request.backend, social_views._do_login)  # pylint: disable=protected-access

        with self._patch_edxmako_current_request(strategy.request):
            student_views.signin_user(strategy.request)
            student_views.login_user(strategy.request)
            actions.do_complete(request.backend, social_views._do_login, user=user)  # pylint: disable=protected-access

        # First we expect that we're in the linked state, with a backend entry.
        self.assert_account_settings_context_looks_correct(account_settings_context(request), user, linked=True)
        self.assert_social_auth_exists_for_user(request.user, strategy)

        # Fire off the disconnect pipeline to unlink.
        self.assert_redirect_to_dashboard_looks_correct(actions.do_disconnect(
            request.backend, request.user, None, redirect_field_name=auth.REDIRECT_FIELD_NAME))

        # Now we expect to be in the unlinked state, with no backend entry.
        self.assert_account_settings_context_looks_correct(account_settings_context(request), user, linked=False)
        self.assert_social_auth_does_not_exist_for_user(user, strategy)

    def test_full_pipeline_succeeds_for_unlinking_account(self):
        # First, create, the request and strategy that store pipeline state,
        # configure the backend, and mock out wire traffic.
        request, strategy = self.get_request_and_strategy(
            auth_entry=pipeline.AUTH_ENTRY_LOGIN, redirect_uri='social:complete')
        strategy.backend.auth_complete = mock.MagicMock(return_value=self.fake_auth_complete(strategy))
        user = self.create_user_models_for_existing_account(
            strategy, '[email protected]', 'password', self.get_username())
        self.assert_social_auth_exists_for_user(user, strategy)

        # Instrument the pipeline to get to the dashboard with the full
        # expected state.
        self.client.get(
            pipeline.get_login_url(self.PROVIDER_CLASS.NAME, pipeline.AUTH_ENTRY_LOGIN))
        actions.do_complete(strategy, social_views._do_login)  # pylint: disable-msg=protected-access

        mako_middleware_process_request(strategy.request)
        student_views.signin_user(strategy.request)
        student_views.login_user(strategy.request)
        actions.do_complete(strategy, social_views._do_login, user=user)  # pylint: disable-msg=protected-access

        # First we expect that we're in the linked state, with a backend entry.
        self.assert_dashboard_response_looks_correct(student_views.dashboard(request), user, linked=True)
        self.assert_social_auth_exists_for_user(request.user, strategy)

        # Fire off the disconnect pipeline to unlink.
        self.assert_redirect_to_dashboard_looks_correct(actions.do_disconnect(
            request.social_strategy, request.user, None, redirect_field_name=auth.REDIRECT_FIELD_NAME))

        # Now we expect to be in the unlinked state, with no backend entry.
        self.assert_dashboard_response_looks_correct(student_views.dashboard(request), user, linked=False)
        self.assert_social_auth_does_not_exist_for_user(user, strategy)

def disconnect(request, backend, association_id=None):
    """Disconnects given backend from current logged in user."""
    return do_disconnect(request.backend, request.user, association_id,
                         redirect_name=REDIRECT_FIELD_NAME)

 def disconnect(self, backend, association_id=None):
     user = getattr(cherrypy.request, 'user', None)
     return do_disconnect(self.strategy, user, association_id)

 def post(self):
     do_disconnect()

 def test_disconnect(self):
     self.do_login()
     user = User.get(self.expected_username)
     user.password = 'password'
     do_disconnect(self.backend, user)
     expect(len(user.social)).to.equal(0)

 def POST(self, backend, association_id=None):
     return do_disconnect(self.backend, self.get_current_user(),
                          association_id)

def disconnect(backend, association_id=None):
    """Disconnects given backend from current logged in user."""
    return do_disconnect(g.strategy, g.user, association_id)

 def test_disconnect(self):
     self.do_login()
     user = User.get(self.expected_username)
     user.password = "password"
     do_disconnect(self.strategy, user)
     expect(len(user.social)).to.equal(0)

def disconnect(request, backend, association_id=None):
    return do_disconnect(request.strategy, request.user, association_id,
                         redirect_name=REDIRECT_FIELD_NAME)

def disconnect(request):
    return do_disconnect(request.strategy, request.user,
                         request.matchdict.get('association_id'),
                         redirect_name='next')

 def test_not_allowed_to_disconnect(self):
     self.do_login()
     user = User.get(self.expected_username)
     with self.assertRaises(NotAllowedToDisconnect):
         do_disconnect(self.backend, user)