pardner/tests/test_transfer_services/test_base.py at d65c7c9914b1863285b666229608d5115e5d987c · dtinit/pardner · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
from urllib import parse

import pytest

from pardner.services import (
    BaseTransferService,
    InsufficientScopeException,
    UnsupportedVerticalException,
)
from pardner.verticals import SocialPostingVertical
from tests.test_transfer_services.conftest import ExtraScopeVertical, NewVertical

SAMPLE_SCOPE = {'fake', 'scope'}
SAMPLE_BASE_URL = 'https://api.example.com/v1'


class FakeTransferService(BaseTransferService):
    _authorization_url = 'https://auth_url'
    _base_url = SAMPLE_BASE_URL
    _token_url = 'https://token_url'

    def __init__(self, supported_verticals, verticals):
        super().__init__(
            service_name='Fake Transfer Service',
            client_id='fake_client_id',
            client_secret='fake_client_secret',
            redirect_uri='https://redirect_uri',
            supported_verticals=set(supported_verticals),
            state=None,
            verticals=set(verticals),
        )

    def scope_for_verticals(self, verticals):
        if ExtraScopeVertical in verticals:
            return SAMPLE_SCOPE | {'extra_scope'}
        return SAMPLE_SCOPE


@pytest.fixture
def blank_transfer_service(monkeypatch):
    return FakeTransferService([SocialPostingVertical], [])


def test_add_verticals_raises_exception(blank_transfer_service):
    with pytest.raises(InsufficientScopeException):
        blank_transfer_service.add_verticals([ExtraScopeVertical])


def test_set_verticals_raises_exception(blank_transfer_service):
    with pytest.raises(UnsupportedVerticalException):
        blank_transfer_service.verticals = [NewVertical]


@pytest.fixture
def mock_transfer_service():
    mock_transfer_service = FakeTransferService(
        [SocialPostingVertical, NewVertical, ExtraScopeVertical],
        [SocialPostingVertical],
    )
    mock_transfer_service.scope = SAMPLE_SCOPE
    return mock_transfer_service


def test_set_supported_verticals(mock_transfer_service):
    mock_transfer_service.verticals = [NewVertical]
    assert mock_transfer_service.verticals == {NewVertical}


def test_add_supported_verticals(mock_transfer_service):
    assert mock_transfer_service.add_verticals([NewVertical])
    assert mock_transfer_service.verticals == {SocialPostingVertical, NewVertical}


def test_add_unsupported_vertical_new_scope_required(
    monkeypatch, mock_transfer_service
):
    def _mock_scope_for_verticals(verticals):
        if ExtraScopeVertical in verticals:
            return {'new_scope'}
        return SAMPLE_SCOPE

    mock_transfer_service._oAuth2Session.access_token = 'access_token'
    monkeypatch.setattr(
        mock_transfer_service, 'scope_for_verticals', _mock_scope_for_verticals
    )
    assert not mock_transfer_service.add_verticals(
        [ExtraScopeVertical], should_reauth=True
    )
    assert not mock_transfer_service._oAuth2Session.access_token
    assert mock_transfer_service.scope == {'fake', 'scope', 'new_scope'}
    assert mock_transfer_service.verticals == {
        SocialPostingVertical,
        ExtraScopeVertical,
    }


def test_authorization_url(mock_transfer_service):
    auth_url, state = mock_transfer_service.authorization_url()

    auth_url_query = parse.urlsplit(auth_url).query
    auth_url_params = dict(parse.parse_qsl(auth_url_query))

    assert 'client_id' in auth_url_params
    assert auth_url_params['client_id'] == 'fake_client_id'
    assert 'redirect_uri' in auth_url_params
    assert auth_url_params['redirect_uri'] == 'https://redirect_uri'
    assert 'state' in auth_url_params
    assert auth_url_params['state'] == state


def test_fetch_token_raises_error(mock_transfer_service):
    with pytest.raises(ValueError):
        mock_transfer_service.fetch_token()


def test_fetch_token(
    mock_oauth2_session_request,
    mock_oauth2_session_response,
    mock_strava_transfer_service,
):
    mock_strava_transfer_service.fetch_token(code='123code123')
    mock_oauth2_session_request.assert_called_once()
    mock_oauth2_session_response.assert_called_once()


@pytest.mark.parametrize(
    ['path', 'base'],
    [
        ('test/path', SAMPLE_BASE_URL),
        ('test/path', f'{SAMPLE_BASE_URL}/'),
        ('/test/path', SAMPLE_BASE_URL),
        ('test/path', None),
        ('/test/path', None),
    ],
)
def test__build_resource_url(path, base, mock_transfer_service):
    resource_url = mock_transfer_service._build_resource_url(path, base)
    assert resource_url == 'https://api.example.com/v1/test/path'