Skip to content

Vinter SDK Async

test_async_api

test_filter_by_symbol 

test_filter_by_symbol()

Test that _filter_by_symbol returns a list of dicts

Source code in tests/test_async_api.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def test_filter_by_symbol():
    """
    Test that _filter_by_symbol returns a list of dicts
    """
    api_key = "my_api_key"
    asset_type = "multi_assets"
    api = VinterAPIAsync(api_key=api_key, asset_type=asset_type)
    symbol = "waves-usd-p-d"
    expected_output = [
        {"symbol": "waves-usd-p-d", "contrib": ["waves-usd-p-r"]},
    ]
    mock_data = [
        {"symbol": "waves-usd-p-d", "contrib": ["waves-usd-p-r"]},
        {"symbol": "ton-usdt-p-5-d", "contrib": ["ton-usdt-p-r"]},
    ]
    result = api._filter_by_symbol(mock_data, symbol)
    assert result == expected_output

test_get_all_active_data_async_returns_list async 

test_get_all_active_data_async_returns_list()

Test that get_all_active_data returns a list of dicts

Source code in tests/test_async_api.py 
 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
@pytest.mark.asyncio
async def test_get_all_active_data_async_returns_list():
    """
    Test that get_all_active_data returns a list of dicts
    """
    api_key = "my_api_key"
    asset_type = "multi_assets"
    api = VinterAPIAsync(api_key=api_key, asset_type=asset_type)
    api.httpx_client = httpx.AsyncClient()
    mock_response = {
        "result": "success",
        "message": "Success",
        "data": [
            {"symbol": "waves-usd-p-d", "contrib": ["waves-usd-p-r"]},
            {"symbol": "ton-usdt-p-5-d", "contrib": ["ton-usdt-p-r"]},
        ],
        "params": {},
    }
    expected_output = mock_response["data"]
    with patch.object(
        api.httpx_client, "get", new_callable=AsyncMock
    ) as mock_get:
        mock_get.return_value = AsyncMock(
            json=Mock(return_value=mock_response)
        )
        result = await api.get_all_active_data()
        assert result == expected_output
    await api.httpx_client.aclose()

test_get_all_active_data_async_symbol_only_returns_symbol_list async 

test_get_all_active_data_async_symbol_only_returns_symbol_list()

Test that get_all_active_data_async returns a list of dicts

Source code in tests/test_async_api.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@pytest.mark.asyncio
async def test_get_all_active_data_async_symbol_only_returns_symbol_list():
    """
    Test that get_all_active_data_async returns a list of dicts
    """
    api_key = "my_api_key"
    asset_type = "multi_assets"
    api = VinterAPIAsync(api_key=api_key, asset_type=asset_type)
    api.httpx_client = httpx.AsyncClient()
    mock_response = {
        "result": "success",
        "message": "Success",
        "data": [
            {"symbol": "waves-usd-p-d", "contrib": ["waves-usd-p-r"]},
            {"symbol": "ton-usdt-p-5-d", "contrib": ["ton-usdt-p-r"]},
        ],
        "params": {},
    }
    expected_output = ["waves-usd-p-d", "ton-usdt-p-5-d"]
    with patch.object(
        api.httpx_client, "get", new_callable=AsyncMock
    ) as mock_get:
        mock_get.return_value = AsyncMock(
            json=Mock(return_value=mock_response)
        )
        result = await api.get_all_active_data(symbol_only=True)
        assert result == expected_output
    await api.httpx_client.aclose()

test_get_all_active_data_async_with_frequency_returns_filtered_list async 

test_get_all_active_data_async_with_frequency_returns_filtered_list()

Test that get_all_active_data_async returns a list of dicts

Source code in tests/test_async_api.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@pytest.mark.asyncio
async def test_get_all_active_data_async_with_frequency_returns_filtered_list():
    """
    Test that get_all_active_data_async returns a list of dicts
    """
    api_key = "my_api_key"
    asset_type = "multi_assets"
    api = VinterAPIAsync(api_key=api_key, asset_type=asset_type)
    api.httpx_client = httpx.AsyncClient()
    mock_response = {
        "result": "success",
        "message": "Success",
        "data": [
            {"symbol": "waves-usd-p-d", "contrib": ["waves-usd-p-r"]},
            {"symbol": "ton-usdt-p-5-r", "contrib": ["ton-usdt-p-r"]},
        ],
        "params": {},
    }
    expected_output = [
        {"symbol": "waves-usd-p-d", "contrib": ["waves-usd-p-r"]}
    ]
    with patch.object(
        api.httpx_client, "get", new_callable=AsyncMock
    ) as mock_get:
        mock_get.return_value = AsyncMock(
            json=Mock(return_value=mock_response)
        )
        result = await api.get_all_active_data(frequency="d")
        assert result == expected_output
    await api.httpx_client.aclose()

test_get_latest_data_raises_exception async 

test_get_latest_data_raises_exception()

Test that get_latest_data raises a ValueError

Source code in tests/test_async_api.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@pytest.mark.asyncio
async def test_get_latest_data_raises_exception():
    """
    Test that get_latest_data raises a ValueError
    """
    api_key = "my_api_key"
    asset_type = "single_assets"
    api = VinterAPIAsync(api_key=api_key, asset_type=asset_type)
    api.httpx_client = httpx.AsyncClient()
    mock_response = {
        "result": "success",
        "message": "Success",
        "data": [],
        "params": {"symbol": "btc-usd-p-r", "limit": 1},
    }
    with patch.object(
        api.httpx_client, "get", new_callable=AsyncMock
    ) as mock_get:
        mock_get.return_value = AsyncMock(
            json=Mock(return_value=mock_response)
        )
        with pytest.raises(ValueError):
            await api.get_latest_data("btc-usd-p-r")
    await api.httpx_client.aclose()

test_get_latest_data_returns_dict async 

test_get_latest_data_returns_dict()

Test that get_latest_data returns a dict

Source code in tests/test_async_api.py 
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
@pytest.mark.asyncio
async def test_get_latest_data_returns_dict():
    """
    Test that get_latest_data returns a dict
    """
    api_key = "my_api_key"
    asset_type = "single_assets"
    api = VinterAPIAsync(api_key=api_key, asset_type=asset_type)
    api.httpx_client = httpx.AsyncClient()
    mock_response = {
        "result": "success",
        "message": "Success",
        "data": [
            {
                "symbol": "btc-usd-p-r",
                "timestamp": 1647724800,
            }
        ],
        "params": {"symbol": "btc-usd-p-r", "limit": 1},
    }
    expected_output = mock_response["data"]
    with patch.object(
        api.httpx_client, "get", new_callable=AsyncMock
    ) as mock_get:
        mock_get.return_value = AsyncMock(
            json=Mock(return_value=mock_response)
        )
        result = await api.get_latest_data("btc-usd-p-r")
        assert result == expected_output
    await api.httpx_client.aclose()

test_get_latest_value async 

test_get_latest_value()

Test that get_latest_value returns a float

Source code in tests/test_async_api.py 
 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
@pytest.mark.asyncio
async def test_get_latest_value():
    """
    Test that get_latest_value returns a float
    """
    api_key = "my_api_key"
    asset_type = "single_assets"
    api = VinterAPIAsync(api_key=api_key, asset_type=asset_type)
    api.httpx_client = httpx.AsyncClient()
    mock_response = {
        "result": "success",
        "message": "Success",
        "data": [
            {"symbol": "btc-usd-p-r", "timestamp": 1647724800, "value": 1000}
        ],
        "params": {"symbol": "btc-usd-p-r", "limit": 1},
    }
    expected_output = mock_response["data"][0]["value"]

    with patch.object(
        api.httpx_client, "get", new_callable=AsyncMock
    ) as mock_get:
        mock_get.return_value = AsyncMock(
            json=Mock(return_value=mock_response)
        )
        result = await api.get_latest_value("btc-usd-p-r")
        assert result == expected_output
    await api.httpx_client.aclose()