ServerKit/backend/tests/test_source_connection_providers.py

"""What each source provider's OAuth plumbing sends and stores.

GitHub, GitLab and Bitbucket shared one config/authorize/callback shape,
written out three times. These pin the per-provider differences that the
shared implementation has to keep: how the token request authenticates,
which profile fields become the connection, where granted scopes come
back, and which authorize parameters each provider gets.
"""
from types import SimpleNamespace
from unittest.mock import patch
from urllib.parse import parse_qs, urlsplit

import pytest
from flask import session

from app.services.settings_service import SettingsService
from app.services.source_connection_service import SourceConnectionService as SC
from factories import make_user

PROFILES = {
    'github': {'id': 7, 'login': 'octo', 'name': 'Octo Cat', 'avatar_url': 'https://a/gh.png'},
    'gitlab': {'id': 8, 'username': 'tanuki', 'name': 'Tanu Ki', 'avatar_url': 'https://a/gl.png'},
    'bitbucket': {'uuid': '{abc-123}', 'username': 'bucky', 'display_name': 'Bu Cky',
                  'links': {'avatar': {'href': 'https://a/bb.png'}}},
}
TOKENS = {
    'github': {'access_token': 'tok-gh', 'scope': 'repo'},
    'gitlab': {'access_token': 'tok-gl', 'scope': 'read_api'},
    'bitbucket': {'access_token': 'tok-bb', 'scopes': ['repository:read', 'account:read']},
}
EXPECTED = {
    'github': ('7', 'octo', 'Octo Cat', 'https://a/gh.png', 'repo'),
    'gitlab': ('8', 'tanuki', 'Tanu Ki', 'https://a/gl.png', 'read_api'),
    'bitbucket': ('abc-123', 'bucky', 'Bu Cky', 'https://a/bb.png', 'repository:read account:read'),
}
TOKEN_URLS = {'github': SC.GITHUB_TOKEN_URL, 'gitlab': SC.GITLAB_TOKEN_URL,
              'bitbucket': SC.BITBUCKET_TOKEN_URL}
PROVIDERS = ['github', 'gitlab', 'bitbucket']


def configure(provider):
    SettingsService.set(f'source_{provider}_client_id', 'cid')
    SettingsService.set(f'source_{provider}_client_secret', 'sec-1234')
    SettingsService.set('source_github_app_slug', '')


@pytest.mark.parametrize('provider', PROVIDERS)
def test_config_reads_redacts_and_keeps_the_secret_on_a_masked_update(app, provider):
    with app.test_request_context():
        configure(provider)
        config = getattr(SC, f'get_{provider}_config')(redacted=True)
        assert (config['client_id'], config['client_secret'], config['configured']) == \
            ('cid', '****1234', True)
        result = getattr(SC, f'update_{provider}_config')(
            {'client_id': 'cid2', 'client_secret': '****1234'})
        assert result['updated'] == ['client_id']
        assert getattr(SC, f'get_{provider}_config')()['client_secret'] == 'sec-1234'


@pytest.mark.parametrize('provider,scope', [
    ('github', 'repo read:user user:email'),
    ('gitlab', 'read_user read_api read_repository'),
    ('bitbucket', None),  # Bitbucket scopes come from the consumer, not the URL
])
def test_authorize_url_parameters(app, provider, scope):
    with app.test_request_context():
        configure(provider)
        url, state = getattr(SC, f'generate_{provider}_authorize_url')('https://p/cb')
        params = {k: v[0] for k, v in parse_qs(urlsplit(url).query).items()}
        assert params.get('scope') == scope
        assert params['state'] == state == session[f'source_{provider}_state']
        assert ('response_type' in params) == (provider != 'github')
        assert ('allow_signup' in params) == (provider == 'github')


@pytest.mark.parametrize('provider', PROVIDERS)
def test_callback_exchanges_the_code_and_stores_the_profile(app, provider):
    from app import db
    user = make_user(db, f'sc_{provider}')
    posted = {}

    def fake_post(url, **kwargs):
        posted.update(url=url, **kwargs)
        return SimpleNamespace(raise_for_status=lambda: None, json=lambda: TOKENS[provider])

    with app.test_request_context():
        configure(provider)
        session[f'source_{provider}_state'] = 'st'
        with patch('app.services.source_connection_service.requests.post', side_effect=fake_post), \
                patch.object(SC, f'_{provider}_get', return_value=PROFILES[provider]):
            conn = getattr(SC, f'complete_{provider}_callback')(user.id, 'the-code', 'st', 'https://p/cb')

    assert posted['url'] == TOKEN_URLS[provider]
    assert posted['data']['code'] == 'the-code'
    assert posted['data']['redirect_uri'] == 'https://p/cb'
    if provider == 'bitbucket':
        assert posted['auth'] == ('cid', 'sec-1234')
        assert 'client_secret' not in posted['data']
    else:
        assert posted['data']['client_secret'] == 'sec-1234'
        assert 'auth' not in posted
    assert ('grant_type' in posted['data']) == (provider != 'github')
    assert (conn.provider, conn.provider_account_id, conn.provider_username, conn.display_name,
            conn.avatar_url, conn.scope) == (provider, *EXPECTED[provider])
    assert conn.access_token == TOKENS[provider]['access_token']


@pytest.mark.parametrize('provider', PROVIDERS)
def test_callback_rejects_a_state_mismatch(app, provider):
    with app.test_request_context():
        configure(provider)
        session[f'source_{provider}_state'] = 'st'
        with pytest.raises(ValueError, match='OAuth state'):
            getattr(SC, f'complete_{provider}_callback')(1, 'c', 'other', 'https://p/cb')


@pytest.mark.parametrize('provider', PROVIDERS)
def test_token_lookup_needs_a_connection(app, provider):
    with pytest.raises(ValueError, match='is not connected'):
        getattr(SC, f'_get_{provider}_token')(424242)