def do_login(self, after_complete_checks=True, user_data_body=None,
                 expected_username=None):
        self.strategy.set_settings({
            'SOCIAL_AUTH_GITHUB_KEY': 'a-key',
            'SOCIAL_AUTH_GITHUB_SECRET': 'a-secret-key',
            'SOCIAL_AUTH_LOGIN_REDIRECT_URL': self.login_redirect_url,
            'SOCIAL_AUTH_AUTHENTICATION_BACKENDS': (
                'social.backends.github.GithubOAuth2',
            )
        })
        start_url = do_auth(self.strategy).url
        target_url = self.strategy.build_absolute_uri(
            '/complete/github/?code=foobar'
        )

        start_query = parse_qs(urlparse(start_url).query)
        location_url = target_url + ('?' in target_url and '&' or '?') + \
                       'state=' + start_query['state']
        location_query = parse_qs(urlparse(location_url).query)

        HTTPretty.register_uri(HTTPretty.GET, start_url, status=301,
                               location=location_url)
        HTTPretty.register_uri(HTTPretty.GET, location_url, status=200,
                               body='foobar')

        response = requests.get(start_url)
        expect(response.url).to.equal(location_url)
        expect(response.text).to.equal('foobar')

        HTTPretty.register_uri(HTTPretty.GET,
                               uri=self.backend.ACCESS_TOKEN_URL,
                               status=200,
                               body=self.access_token_body or '',
                               content_type='text/json')

        if self.user_data_url:
            user_data_body = user_data_body or self.user_data_body or ''
            HTTPretty.register_uri(HTTPretty.GET, self.user_data_url,
                                   body=user_data_body,
                                   content_type='text/json')
        self.strategy.set_request_data(location_query)
        class Request(object):session = {}
        self.strategy.request = Request()

        def _login_lambda(strategy, user, social_user):
            strategy.request = Request()
            return strategy.session_set('username', user.username)

        redirect = do_complete(
            self.strategy,
            user=self.user,
            login=_login_lambda
        )
        if after_complete_checks:
            expect(self.strategy.session_get('username')).to.equal(
                expected_username or self.expected_username
            )
            expect(redirect.url).to.equal(self.login_redirect_url)
        return redirect

def sanitize_redirect(host, redirect_to):
    """
    Given the hostname and an untrusted URL to redirect to,
    this method tests it to make sure it isn't garbage/harmful
    and returns it, else returns None, similar as how's it done
    on django.contrib.auth.views.

    >>> print sanitize_redirect('myapp.com', None)
    None
    >>> print sanitize_redirect('myapp.com', '')
    None
    >>> print sanitize_redirect('myapp.com', {})
    None
    >>> print sanitize_redirect('myapp.com', 'http://notmyapp.com/path/')
    None
    >>> print sanitize_redirect('myapp.com', 'http://myapp.com/path/')
    http://myapp.com/path/
    >>> print sanitize_redirect('myapp.com', '/path/')
    /path/
    """
    # Quick sanity check.
    if not redirect_to:
        return None

    # Heavier security check, don't allow redirection to a different host.
    try:
        netloc = urlparse(redirect_to)[1]
    except TypeError:  # not valid redirect_to value
        return None
    if netloc and netloc != host:
        return None
    return redirect_to

    def run_oauth(self, m, user=None):

        strategy = DjangoStrategy(DjangoStorage)
        backend = self.Backend_Class(strategy, redirect_uri=self.client_complete_url)

        start_url = do_auth(backend).url
        start_query = parse_qs(urlparse(start_url).query)

        # set 'state' in client
        backend.data.update({'state': start_query['state']})

        m.get(backend.USER_DATA_URL,
              json={"username": self.social_username,
                    "email": self.social_email},
              status_code=200)

        m.post(backend.ACCESS_TOKEN_URL,
               json={'access_token': self.access_token,
                     'token_type': self.token_type,
                     'expires_in': self.expires_in,
                     'scope': self.scope,
                     'refresh_token': self.refresh_token},
               status_code=200)

        def _login(backend, user, social_user):
            backend.strategy.session_set('username', user.username)

        do_complete(backend, user=user, login=_login)

        social = backend.strategy.storage.user.get_social_auth(backend.name, self.social_username)

        return strategy.session_get('username'), social, backend

def url_add_parameters(url, params):
    """Adds parameters to URL, parameter will be repeated if already present"""
    if params:
        fragments = list(urlparse(url))
        fragments[4] = urlencode(parse_qsl(fragments[4]) + params.items())
        url = urlunparse(fragments)
    return url

 def do_start(self):
     start_url = self.strategy.start().url
     target_url = self.auth_handlers(start_url)
     response = requests.get(start_url)
     expect(response.url).to.equal(target_url)
     expect(response.text).to.equal('foobar')
     self.strategy.set_request_data(parse_qs(urlparse(target_url).query))
     return self.strategy.complete()

    def handle_state(self, start_url, target_url):
        start_query = parse_qs(urlparse(start_url).query)
        redirect_uri = start_query.get('redirect_uri')

        if getattr(self.backend, 'STATE_PARAMETER', False):
            if start_query.get('state'):
                target_url = url_add_parameters(target_url, {
                    'state': start_query['state']
                })

        if redirect_uri and getattr(self.backend, 'REDIRECT_STATE', False):
            redirect_query = parse_qs(urlparse(redirect_uri).query)
            if redirect_query.get('redirect_state'):
                target_url = url_add_parameters(target_url, {
                    'redirect_state': redirect_query['redirect_state']
                })
        return target_url

    def do_login(self, after_complete_checks=True, user_data_body=None, expected_username=None):
        self.strategy.set_settings(
            {
                "SOCIAL_AUTH_GITHUB_KEY": "a-key",
                "SOCIAL_AUTH_GITHUB_SECRET": "a-secret-key",
                "SOCIAL_AUTH_LOGIN_REDIRECT_URL": self.login_redirect_url,
                "SOCIAL_AUTH_AUTHENTICATION_BACKENDS": ("social.backends.github.GithubOAuth2",),
            }
        )
        start_url = do_auth(self.backend).url
        target_url = self.strategy.build_absolute_uri("/complete/github/?code=foobar")

        start_query = parse_qs(urlparse(start_url).query)
        location_url = target_url + ("?" in target_url and "&" or "?") + "state=" + start_query["state"]
        location_query = parse_qs(urlparse(location_url).query)

        HTTPretty.register_uri(HTTPretty.GET, start_url, status=301, location=location_url)
        HTTPretty.register_uri(HTTPretty.GET, location_url, status=200, body="foobar")

        response = requests.get(start_url)
        expect(response.url).to.equal(location_url)
        expect(response.text).to.equal("foobar")

        HTTPretty.register_uri(
            HTTPretty.POST,
            uri=self.backend.ACCESS_TOKEN_URL,
            status=200,
            body=self.access_token_body or "",
            content_type="text/json",
        )

        if self.user_data_url:
            user_data_body = user_data_body or self.user_data_body or ""
            HTTPretty.register_uri(HTTPretty.GET, self.user_data_url, body=user_data_body, content_type="text/json")
        self.strategy.set_request_data(location_query, self.backend)

        def _login(backend, user, social_user):
            backend.strategy.session_set("username", user.username)

        redirect = do_complete(self.backend, user=self.user, login=_login)

        if after_complete_checks:
            expect(self.strategy.session_get("username")).to.equal(expected_username or self.expected_username)
            expect(redirect.url).to.equal(self.login_redirect_url)
        return redirect

 def do_start(self):
     start_url = self.backend.start().url
     target_url = self.auth_handlers(start_url)
     response = requests.get(start_url)
     self.assertEqual(response.url, target_url)
     self.assertEqual(response.text, 'foobar')
     self.strategy.set_request_data(parse_qs(urlparse(target_url).query),
                                    self.backend)
     return self.backend.complete()

def url_add_parameters(url, params):
    """Adds parameters to URL, parameter will be repeated if already present"""
    if params:
        fragments = list(urlparse(url))
        value = parse_qs(fragments[4])
        value.update(params)
        fragments[4] = urlencode(value)
        url = urlunparse(fragments)
    return url

    def do_start(self):
        name = self.backend.name.upper().replace('-', '_')
        self.strategy.set_settings({
            'SOCIAL_AUTH_' + name + '_KEY': 'a-key',
            'SOCIAL_AUTH_' + name + '_SECRET': 'a-secret-key',
        })
        start_url = self.strategy.start().url
        target_url = self.strategy.build_absolute_uri(self.complete_url)
        start_query = parse_qs(urlparse(start_url).query)

        if self.backend.STATE_PARAMETER:
            location_url = target_url + ('?' in target_url and '&' or '?') + \
                           'state=' + start_query['state']
        elif self.backend.REDIRECT_STATE:
            location_url = target_url + ('?' in target_url and '&' or '?') + \
                           'redirect_state=' + start_query['redirect_state']
        else:
            location_url = target_url
        location_query = parse_qs(urlparse(location_url).query)

        HTTPretty.register_uri(HTTPretty.GET, start_url, status=301,
                               location=location_url)
        HTTPretty.register_uri(HTTPretty.GET, location_url, status=200,
                               body='foobar')

        response = requests.get(start_url)
        expect(response.url).to.equal(location_url)
        expect(response.text).to.equal('foobar')

        method = self.backend.ACCESS_TOKEN_METHOD == 'GET' and HTTPretty.GET \
                                                            or HTTPretty.POST
        HTTPretty.register_uri(method,
                               uri=self.backend.ACCESS_TOKEN_URL,
                               status=200,
                               body=self.access_token_body or '',
                               content_type='text/json')

        if self.user_data_url:
            HTTPretty.register_uri(HTTPretty.GET, self.user_data_url,
                                   body=self.user_data_body or '',
                                   content_type='text/json')
        self.strategy.set_request_data(location_query)

    def build_absolute_uri(self, path=None):
        url = super(DjangoTenantStrategy, self).build_absolute_uri(path)
        auth_domain = settings.TENANT_AUTH_COMPLETE_URL

        if path and path in self.auth_complete_paths:
            parsed_url = urlparse(url)
            url_parts = list(parsed_url)
            url_parts[1] = auth_domain
            return urlunparse(url_parts)

        return url

    def _replace_domain(cls, url, new_domain):
        parsed_url = urlparse(url)
        url_parts = list(parsed_url)
        new_netloc = new_domain

        if ":" not in new_domain and ":" in parsed_url.netloc:
            _, port = parsed_url.netloc.split(":")
            new_netloc = "{}:{}".format(new_domain, port)

        url_parts[1] = new_netloc
        return urlunparse(url_parts)

 def handle_state(self, start_url, target_url):
     try:
         if self.backend.STATE_PARAMETER or self.backend.REDIRECT_STATE:
             query = parse_qs(urlparse(start_url).query)
             target_url = target_url + ('?' in target_url and '&' or '?')
             if 'state' in query or 'redirect_state' in query:
                 name = 'state' in query and 'state' or 'redirect_state'
                 target_url += '{0}={1}'.format(name, query[name])
     except AttributeError:
         pass
     return target_url

def handle_state(backend, start_url, target_url):
    try:
        if backend.STATE_PARAMETER or backend.REDIRECT_STATE:
            query = parse_qs(urlparse(start_url).query)
            target_url = target_url + ("?" in target_url and "&" or "?")
            if "state" in query or "redirect_state" in query:
                name = "state" in query and "state" or "redirect_state"
                target_url += "{0}={1}".format(name, query[name])
    except AttributeError:
        pass
    return bytes(target_url)

def sanitize_redirect(host, redirect_to):
    """
    Given the hostname and an untrusted URL to redirect to,
    this method tests it to make sure it isn't garbage/harmful
    and returns it, else returns None, similar as how's it done
    on django.contrib.auth.views.
    """
    # Quick sanity check.
    if not redirect_to or not isinstance(redirect_to, six.string_types):
        return None

    # Heavier security check, don't allow redirection to a different host.
    netloc = urlparse(redirect_to)[1]
    if netloc and netloc != host:
        return None
    return redirect_to

def sanitize_redirect(host, redirect_to):
    """
    Given the hostname and an untrusted URL to redirect to,
    this method tests it to make sure it isn't garbage/harmful
    and returns it, else returns None, similar as how's it done
    on django.contrib.auth.views.
    """
    if redirect_to:
        try:
            # Don't redirect to a different host
            netloc = urlparse(redirect_to)[1] or host
        except (TypeError, AttributeError):
            pass
        else:
            if netloc == host:
                return redirect_to

 def do_start(self):
     start_url = self.backend.start().url
     # Modify the start URL to make the SAML request consistent
     # from test to test:
     start_url = self.modify_start_url(start_url)
     # If the SAML Identity Provider recognizes the user, we will
     # be redirected back to:
     return_url = self.backend.redirect_uri
     self.install_http_intercepts(start_url, return_url)
     response = requests.get(start_url)
     self.assertTrue(response.url.startswith(return_url))
     self.assertEqual(response.text, 'foobar')
     query_values = dict((k, v[0]) for k, v in
                         parse_qs(urlparse(response.url).query).items())
     self.assertNotIn(' ', query_values['SAMLResponse'])
     self.strategy.set_request_data(query_values, self.backend)
     return self.backend.complete()

 def modify_start_url(self, start_url):
     """
     Given a SAML redirect URL, parse it and change the ID to
     a consistent value, so the request is always identical.
     """
     # Parse the SAML Request URL to get the XML being sent to TestShib
     url_parts = urlparse(start_url)
     query = dict((k, v[0]) for (k, v) in
                  parse_qs(url_parts.query).iteritems())
     xml = OneLogin_Saml2_Utils.decode_base64_and_inflate(
         query['SAMLRequest']
     )
     # Modify the XML:
     xml, changed = re.subn(r'ID="[^"]+"', 'ID="TEST_ID"', xml)
     self.assertEqual(changed, 1)
     # Update the URL to use the modified query string:
     query['SAMLRequest'] = OneLogin_Saml2_Utils.deflate_and_base64_encode(
         xml
     )
     url_parts = list(url_parts)
     url_parts[4] = urlencode(query)
     return urlunparse(url_parts)

    def get_tenant_url(cls, url):
        parsed_url = urlparse(url)
        url_parts = list(parsed_url)
        query = parse_qs(parsed_url.query)
        state_value = (query.get('state', None)
                       or query.get('redirect_state', None))

        if not state_value:
            return None

        if isinstance(state_value, list):
            state_value = state_value[0]

        tenant = cls.tenant_from_state(state_value)
        tenant_domain = tenant.domain_url

        if ":" in settings.TENANT_AUTH_COMPLETE_URL:
            _, port = settings.TENANT_AUTH_COMPLETE_URL.split(":")
            tenant_domain = ":".join((tenant_domain, port))

        url_parts[1] = tenant_domain
        return urlunparse(url_parts)

    def do_start(self):
        name = self.backend.name.upper().replace('-', '_')
        self.strategy.set_settings({
            'SOCIAL_AUTH_' + name + '_KEY': 'a-key',
            'SOCIAL_AUTH_' + name + '_SECRET': 'a-secret-key',
        })

        method = self.backend.REQUEST_TOKEN_METHOD == 'GET' and HTTPretty.GET \
                                                            or HTTPretty.POST
        HTTPretty.register_uri(method, self.backend.REQUEST_TOKEN_URL,
                               status=200, body=self.request_token_body)

        start_url = self.strategy.start().url
        target_url = self.strategy.build_absolute_uri(self.complete_url)
        target_query = parse_qs(urlparse(target_url).query)

        HTTPretty.register_uri(HTTPretty.GET, start_url, status=301,
                               location=target_url)
        HTTPretty.register_uri(HTTPretty.GET, target_url, status=200,
                               body='foobar')

        response = requests.get(start_url)
        expect(response.url).to.equal(target_url)
        expect(response.text).to.equal('foobar')

        method = self.backend.ACCESS_TOKEN_METHOD == 'GET' and HTTPretty.GET \
                                                            or HTTPretty.POST
        HTTPretty.register_uri(method,
                               uri=self.backend.ACCESS_TOKEN_URL,
                               status=200,
                               body=self.access_token_body or '',
                               content_type='text/json')

        if self.user_data_url:
            HTTPretty.register_uri(HTTPretty.GET, self.user_data_url,
                                   body=self.user_data_body or '',
                                   content_type=self.user_data_content_type)
        self.strategy.set_request_data(target_query)