-
Notifications
You must be signed in to change notification settings - Fork 3.8k
/
test_source.py
47 lines (37 loc) · 2.07 KB
/
test_source.py
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
#
# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
#
from unittest.mock import MagicMock
import pytest
from source_notion.source import SourceNotion
UNAUTHORIZED_ERROR_MESSAGE = "The provided API access token is invalid. Please double-check that you input the correct token and have granted the necessary permissions to your Notion integration."
RESTRICTED_RESOURCE_ERROR_MESSAGE = "The provided API access token does not have the correct permissions configured. Please double-check that you have granted all the necessary permissions to your Notion integration."
GENERIC_ERROR_MESSAGE = "Conflict occured while saving. Please try again."
NO_ERROR_MESSAGE = "An unexpected error occured while connecting to Notion. Please check your credentials and try again."
def test_check_connection(mocker, requests_mock):
source = SourceNotion()
logger_mock, config_mock = MagicMock(), MagicMock()
requests_mock.get("https://api.notion.com/v1/users", json={"results": [{"id": "aaa"}], "next_cursor": None})
assert source.check_connection(logger_mock, config_mock) == (True, None)
@pytest.mark.parametrize(
"status_code,json_response,expected_message",
[
(401, {"code": "unauthorized"}, UNAUTHORIZED_ERROR_MESSAGE),
(403, {"code": "restricted_resource"}, RESTRICTED_RESOURCE_ERROR_MESSAGE),
(409, {"message": GENERIC_ERROR_MESSAGE}, GENERIC_ERROR_MESSAGE),
(400, {}, NO_ERROR_MESSAGE)
]
)
def test_check_connection_errors(mocker, requests_mock, status_code, json_response, expected_message):
source = SourceNotion()
logger_mock, config_mock = MagicMock(), MagicMock()
requests_mock.get("https://api.notion.com/v1/users", status_code=status_code, json=json_response)
result, message = source.check_connection(logger_mock, config_mock)
assert result is False
assert message == expected_message
def test_streams(mocker):
source = SourceNotion()
config_mock = MagicMock()
streams = source.streams(config_mock)
expected_streams_number = 5
assert len(streams) == expected_streams_number