aboutsummaryrefslogtreecommitdiff
blob: 0c83fe494c0c6b0320a243df31881340506dfa4e (plain) (blame)
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
import pytest
import vcr

from contextlib import contextmanager

try:
    from mock import MagicMock
except ImportError:
    from unittest.mock import MagicMock
import requests_mock

UNLIKELY_HASHTAG = "fgiztsshwiaqqiztpmmjbtvmescsculuvmgjgopwoeidbcrixp"


@contextmanager
def many_statuses(api, n=10, suffix=''):
    statuses = list()
    for i in range(n):
        status = api.status_post(f"Toot number {i}!{suffix}")
        statuses.append(status)
    yield statuses
    for status in statuses:
        api.status_delete(status['id'])


@pytest.mark.vcr()
def test_fetch_next_previous(api):
    account = api.account_verify_credentials()
    with many_statuses(api):
        statuses = api.account_statuses(account['id'], limit=5)
        next_statuses = api.fetch_next(statuses)
        assert next_statuses
        previous_statuses = api.fetch_previous(next_statuses)
        assert previous_statuses


@pytest.mark.vcr()
def test_fetch_next_previous_from_pagination_info(api):
    account = api.account_verify_credentials()
    with many_statuses(api):
        statuses = api.account_statuses(account['id'], limit=5)
        next_statuses = api.fetch_next(statuses._pagination_next)
        assert next_statuses
        previous_statuses = api.fetch_previous(next_statuses._pagination_prev)
        assert previous_statuses

@pytest.mark.vcr()
def test_fetch_next_previous_from_pagination_info_oldstyle(api):
    # Old style compatibility mode. The storage in the list items is not anymore internally used.
    account = api.account_verify_credentials()
    with many_statuses(api):
        statuses = api.account_statuses(account['id'], limit=5)
        next_statuses = api.fetch_next(statuses[-1]._pagination_next)
        assert next_statuses
        previous_statuses = api.fetch_previous(next_statuses[0]._pagination_prev)
        assert previous_statuses

def test_fetch_next_previous_old_pagination(api):
    with vcr.use_cassette('test_fetch_next_previous.yaml', cassette_library_dir='tests/cassettes_old_pagination', record_mode='none'):
        account = api.account_verify_credentials()
        
        with many_statuses(api):
            statuses = api.account_statuses(account['id'], limit=5)
            next_statuses = api.fetch_next(statuses)
            assert next_statuses
            previous_statuses = api.fetch_previous(next_statuses)
            assert previous_statuses

def test_fetch_next_previous_from_pagination_info_old_pagination(api):
    with vcr.use_cassette('test_fetch_next_previous_from_pagination_info.yaml', cassette_library_dir='tests/cassettes_old_pagination', record_mode='none'):
        account = api.account_verify_credentials()
        
        with many_statuses(api):
            statuses = api.account_statuses(account['id'], limit=5)
            next_statuses = api.fetch_next(statuses._pagination_next)
            assert next_statuses
            previous_statuses = api.fetch_previous(next_statuses._pagination_prev)
            assert previous_statuses

    # Old style compatibility mode. The storage in the list items is not anymore internally used.
    with vcr.use_cassette('test_fetch_next_previous_from_pagination_info.yaml', cassette_library_dir='tests/cassettes_old_pagination', record_mode='none'):
        account = api.account_verify_credentials()

        with many_statuses(api):
            statuses = api.account_statuses(account['id'], limit=5)
            next_statuses = api.fetch_next(statuses[-1]._pagination_next)
            assert next_statuses
            previous_statuses = api.fetch_previous(next_statuses[0]._pagination_prev)
            assert previous_statuses

@pytest.mark.vcr()
def test_fetch_remaining(api):
    with many_statuses(api, n=30, suffix=' #'+UNLIKELY_HASHTAG):
        hashtag = api.timeline_hashtag(UNLIKELY_HASHTAG, limit=10)
        hashtag_remaining = api.fetch_remaining(hashtag)
        assert hashtag_remaining
        assert len(hashtag_remaining) >= 30

def test_link_headers(api):
    rmock = requests_mock.Adapter()
    api.session.mount(api.api_base_url, rmock)

    _id='abc1234'

    rmock.register_uri(
        'GET',
        requests_mock.ANY,
        json=[{"foo": "bar"}],
        headers={
            "link": f"<{api.api_base_url}/api/v1/timelines/tag/{UNLIKELY_HASHTAG}?max_id={_id}>; rel=\"next\", "
                    f"<{api.api_base_url}/api/v1/timelines/tag/{UNLIKELY_HASHTAG}?since_id={_id}>; rel=\"prev\""
        }
    )

    resp = api.timeline_hashtag(UNLIKELY_HASHTAG)
    assert resp._pagination_next['max_id'] == _id
    assert resp._pagination_prev['since_id'] == _id

    # Old style compatibility mode. The storage in the list items is not anymore internally used.
    assert resp[0]._pagination_next['max_id'] == _id
    assert resp[0]._pagination_prev['since_id'] == _id
Powered by cgit v1.2.3 (git 2.41.0)