55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
#
|
|
# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
|
|
#
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import responses
|
|
from airbyte_cdk.models import AirbyteConnectionStatus, Status
|
|
from source_github.source import SourceGithub
|
|
|
|
|
|
def check_source(repo_line: str) -> AirbyteConnectionStatus:
|
|
source = SourceGithub()
|
|
config = {"access_token": "test_token", "repository": repo_line}
|
|
logger_mock = MagicMock()
|
|
return source.check(logger_mock, config)
|
|
|
|
|
|
@responses.activate
|
|
def test_check_connection_repos_only():
|
|
responses.add("GET", "https://api.github.com/repos/airbyte", json={})
|
|
|
|
status = check_source("airbyte airbyte airbyte")
|
|
assert not status.message
|
|
assert status.status == Status.SUCCEEDED
|
|
# Only one request since 3 repos have same name
|
|
assert len(responses.calls) == 1
|
|
|
|
|
|
@responses.activate
|
|
def test_check_connection_repos_and_org_repos():
|
|
repos = [{"name": f"name {i}", "full_name": f"full name {i}"} for i in range(1000)]
|
|
responses.add("GET", "https://api.github.com/repos/airbyte/test", json={})
|
|
responses.add("GET", "https://api.github.com/repos/airbyte/test2", json={})
|
|
responses.add("GET", "https://api.github.com/orgs/airbytehq/repos", json=repos)
|
|
responses.add("GET", "https://api.github.com/orgs/org/repos", json=repos)
|
|
|
|
status = check_source("airbyte/test airbyte/test2 airbytehq/* org/*")
|
|
assert not status.message
|
|
assert status.status == Status.SUCCEEDED
|
|
# Two requests for repos and two for organization
|
|
assert len(responses.calls) == 4
|
|
|
|
|
|
@responses.activate
|
|
def test_check_connection_org_only():
|
|
repos = [{"name": f"name {i}", "full_name": f"full name {i}"} for i in range(1000)]
|
|
responses.add("GET", "https://api.github.com/orgs/airbytehq/repos", json=repos)
|
|
|
|
status = check_source("airbytehq/*")
|
|
assert not status.message
|
|
assert status.status == Status.SUCCEEDED
|
|
# One request to check organization
|
|
assert len(responses.calls) == 1
|