zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-20 06:31:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

Feature/generic api connector (#13545)

Browse Source 
# feat: Add Generic REST API Connector

## What problem does this PR solve?

RAGFlow supports many specific data source connectors (MySQL, Slack,
Google Drive, etc.), but there was no way to connect an arbitrary REST
API as a data source. Users with custom or third-party APIs had to write
a new connector class for each one.

This PR adds a **generic, configuration-driven REST API connector** that
lets users connect any REST API as a data source entirely through the UI
— no code changes needed per API.

---

## Features

### Core Connector (`common/data_source/rest_api_connector.py`)

- Implements `LoadConnector` and `PollConnector` interfaces for full and
incremental sync
- **Configurable authentication:** None, API Key (custom header), Bearer
Token, Basic Auth
- **Pluggable pagination:** Page-based, Offset-based, Cursor-based, or
None
- Smart page-size inference from user's query parameters to avoid
duplicate/conflicting params
- Configurable request delay between pages to prevent API rate limiting
- Auto-detection of the items array in JSON responses (`items`,
`results`, `data`, `records`, or first list found)
- **Advanced field mapping** with dot-notation (`country.name`), array
wildcards (`newsType[*].name`), type hints, and default values
- Optional content template rendering (`"Title: {title}\nBody: {body}"`)
- HTML stripping for content fields
- Stable document IDs via `hash128` from a configurable ID field or
auto-generated from item content
- Pydantic configuration schema with automatic coercion of UI string
inputs to dicts/lists

### Backend Registration (`rag/svr/sync_data_source.py`,
`common/constants.py`, `common/data_source/config.py`)

- `REST_API` sync class wired into RAGFlow's `func_factory`
- Full sync (`load_from_state`) and incremental polling (`poll_source`)
support
- Credentials and config passed from task to connector following
existing patterns (MySQL, SeaFile, etc.)

### Test Connection Endpoint (`api/apps/connector_app.py`)

- `POST /v1/connector/<id>/test` validates config schema,
authentication, and API connectivity without triggering a sync
- Clear error messages for auth failures vs. config issues

### Frontend UI (`web/src/pages/user-setting/data-source/constant/`)

- **Postman-style configuration:** Base URL, Query Parameters (key=value
per line), Auth, Content Fields, Metadata Fields, Pagination Type
- Auth-type-aware form: fields for API key header/value, Bearer token,
or Basic username/password appear only when relevant
- **Advanced Settings** toggle for: Custom Headers, Max Pages, Request
Delay, Poll Timestamp Field, Request Body (POST)
- Connector icon (SVG) and i18n strings (English)
- **"Test Connection"** button to validate before syncing

---

## Controls & Safety

- Configurable max pages safety cap (default: 1000, adjustable in UI)
- Configurable request delay between pages (default: 0.5s, adjustable in
UI)
- Auth errors (401/403) fail immediately without retries; transient
errors retry with exponential backoff
- Diagnostic logging: auth setup confirmation, request details on
failure, content field extraction status

---

## Type of change

- [x] New Feature (non-breaking change which adds functionality)


##Visual Screenshots of Features
<img width="482" height="510" alt="Screenshot 2026-03-11 at 5 19 52 PM"
src="https://github.com/user-attachments/assets/dcb7ab4a-1622-44f3-bb02-d6f0527314c4"
/>
(Connector can be configured within the external data sources tab)

Configuration Parameters:
<img width="661" height="682" alt="Screenshot 2026-03-11 at 5 20 46 PM"
src="https://github.com/user-attachments/assets/5e154e71-4ab5-4872-bfb2-04f02b73c18a"
/>
<img width="661" height="682" alt="Screenshot 2026-03-11 at 5 20 54 PM"
src="https://github.com/user-attachments/assets/00cb14b7-0bcf-4b94-9d71-34e93369ecb2"
/>

Connection can be tested before attaching to dataset:
<img width="981" height="681" alt="Screenshot 2026-03-11 at 5 21 40 PM"
src="https://github.com/user-attachments/assets/aaa6eeeb-89a7-4349-bc34-2423bf8be9ee"
/>

Ingestion tested with API connector (works perfectly fine):
<img width="1062" height="705" alt="Screenshot 2026-03-11 at 5 22 30 PM"
src="https://github.com/user-attachments/assets/afcd0d58-cadd-4152-badc-d2f14d96fbec"
/>

Search & Retrieval works as well with metadata flow:
<img width="1062" height="705" alt="Screenshot 2026-03-11 at 5 23 05 PM"
src="https://github.com/user-attachments/assets/d41ee935-dcf7-4456-b317-22a76ca032c0"
/>

---------

Co-authored-by: Ahmad Intisar <ahmadintisar@Ahmads-MacBook-M4-Pro.local>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Ahmad Intisar
2026-05-13 17:35:01 +05:00
committed by GitHub
parent 30d1c1dc28
commit e994051eb9
34 changed files with 2338 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
test/unit_test/data_source/__init__.py Normal file
View File

36
test/unit_test/data_source/conftest.py Normal file
View File

@@ -0,0 +1,36 @@
#
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Pre-register the ``common.data_source`` package namespace so that
importing individual sub-modules (config, exceptions, rest_api_connector, …)
does **not** trigger ``common/data_source/__init__.py``, which pulls in every
connector and their heavy transitive dependencies (numpy, xgboost, etc.).
This file is executed by pytest before any test module in this directory is
collected, so the lightweight namespace is always in place.
"""
import os
import sys
import types
import common # lightweight top-level package
if "common.data_source" not in sys.modules:
_pkg = types.ModuleType("common.data_source")
_pkg.__path__ = [os.path.join(p, "data_source") for p in common.__path__]
_pkg.__package__ = "common.data_source"
sys.modules["common.data_source"] = _pkg

607
test/unit_test/data_source/test_rest_api_connector.py Normal file
View File

@@ -0,0 +1,607 @@
#
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from contextlib import contextmanager
from unittest.mock import MagicMock, patch
import pytest
import requests
from common.data_source import utils as _ds_utils
from common.data_source.exceptions import (
ConnectorMissingCredentialError,
ConnectorValidationError,
)
from common.data_source.rest_api_connector import (
AuthType,
PaginationType,
RestAPIConnector,
RestAPIConnectorConfig,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
VALID_URL = "https://api.example.com/v1/items"
_MOCK_DNS_ADDRINFO = [(2, 1, 6, "", ("93.184.216.34", 0))]
@contextmanager
def _mocked_rest_api_requests_and_dns():
"""Block real DNS/TCP: mock SSRF getaddrinfo and HTTP at the class layer.
`RestAPIConnector` calls `rl_requests.get` / `.post` on
`utils._RateLimitedRequest`. Replacing only module-level `rl_requests` is not
reliable everywhere (import/rebind quirks), so we patch the class methods
that wrap `requests.get` / `requests.post` and avoid retry backoff delays.
"""
mock_rl = MagicMock()
with patch(
"common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=_MOCK_DNS_ADDRINFO,
), patch.object(_ds_utils._RateLimitedRequest, "get", mock_rl.get), patch.object(
_ds_utils._RateLimitedRequest,
"post",
mock_rl.post,
):
yield mock_rl
def _make_paged_connector(**overrides) -> RestAPIConnector:
defaults = dict(
url=VALID_URL,
content_fields=["title"],
pagination_type=PaginationType.PAGE,
pagination_config={"page_param": "page"},
max_pages=100,
request_delay=0,
)
defaults.update(overrides)
return RestAPIConnector(**defaults)
def _make_connector(**overrides) -> RestAPIConnector:
"""Build a RestAPIConnector with sensible defaults, applying *overrides*."""
defaults = dict(
url=VALID_URL,
content_fields=["title", "body"],
)
defaults.update(overrides)
return RestAPIConnector(**defaults)
def _mock_response(json_data, status_code=200):
"""Return a ``requests.Response``-like mock."""
resp = MagicMock(spec=requests.Response)
resp.status_code = status_code
resp.url = VALID_URL
resp.json.return_value = json_data
if status_code >= 400:
http_error = requests.HTTPError(response=resp)
resp.raise_for_status.side_effect = http_error
resp.status_code = status_code
else:
resp.raise_for_status.return_value = None
return resp
# ===================================================================== #
# 1. Config schema validation #
# ===================================================================== #
class TestRestAPIConfig:
"""Test Pydantic RestAPIConnectorConfig schema validation."""
def test_missing_url_raises_validation_error(self):
"""Missing url should fail Pydantic validation."""
with pytest.raises(Exception):
RestAPIConnectorConfig(content_fields=["title"])
def test_missing_content_fields_detected(self):
"""An empty content_fields list should be caught by ensure_required_fields."""
cfg = RestAPIConnectorConfig(url=VALID_URL, content_fields=[])
with pytest.raises(ConnectorValidationError):
cfg.ensure_required_fields()
def test_valid_minimal_config(self):
"""Minimal valid config: url + content_fields."""
cfg = RestAPIConnectorConfig(url=VALID_URL, content_fields=["title"])
assert str(cfg.url).startswith("https://api.example.com")
assert cfg.content_fields == ["title"]
def test_auth_type_defaults_to_none(self):
"""auth_type should default to 'none'."""
cfg = RestAPIConnectorConfig(url=VALID_URL, content_fields=["t"])
assert cfg.auth_type == AuthType.NONE
def test_pagination_type_defaults_to_none(self):
"""pagination_type should default to 'none'."""
cfg = RestAPIConnectorConfig(url=VALID_URL, content_fields=["t"])
assert cfg.pagination_type == PaginationType.NONE
def test_string_to_dict_coercion_for_headers(self):
"""A key=value string should be coerced to a dict."""
cfg = RestAPIConnectorConfig(
url=VALID_URL, content_fields=["t"], headers="X-Custom=hello"
)
assert cfg.headers == {"X-Custom": "hello"}
def test_string_to_list_coercion_for_content_fields(self):
"""A comma-separated string should be coerced to a list."""
cfg = RestAPIConnectorConfig(url=VALID_URL, content_fields="title,content")
assert cfg.content_fields == ["title", "content"]
# ===================================================================== #
# 2. SSRF URL validation #
# ===================================================================== #
class TestSSRFValidation:
"""Test that unsafe URLs are blocked before any HTTP request is made."""
def test_localhost_blocked(self):
"""localhost should be rejected."""
with pytest.raises(ConnectorValidationError, match="localhost"):
_make_connector(url="http://localhost/api")
@patch("common.data_source.rest_api_connector.socket.getaddrinfo")
def test_loopback_ip_blocked(self, mock_dns):
"""127.0.0.1 should be rejected."""
mock_dns.return_value = [(2, 1, 6, "", ("127.0.0.1", 0))]
with pytest.raises(ConnectorValidationError, match="disallowed"):
_make_connector(url="http://127.0.0.1/api")
@patch("common.data_source.rest_api_connector.socket.getaddrinfo")
def test_cloud_metadata_ip_blocked(self, mock_dns):
"""169.254.169.254 (cloud metadata endpoint) should be rejected."""
mock_dns.return_value = [(2, 1, 6, "", ("169.254.169.254", 0))]
with pytest.raises(ConnectorValidationError, match="disallowed"):
_make_connector(url="http://169.254.169.254/latest/meta-data/")
@patch("common.data_source.rest_api_connector.socket.getaddrinfo")
def test_private_ip_192_blocked(self, mock_dns):
"""192.168.x.x should be rejected."""
mock_dns.return_value = [(2, 1, 6, "", ("192.168.1.1", 0))]
with pytest.raises(ConnectorValidationError, match="disallowed"):
_make_connector(url="http://192.168.1.1/api")
@patch("common.data_source.rest_api_connector.socket.getaddrinfo")
def test_private_ip_10_blocked(self, mock_dns):
"""10.x.x.x should be rejected."""
mock_dns.return_value = [(2, 1, 6, "", ("10.0.0.1", 0))]
with pytest.raises(ConnectorValidationError, match="disallowed"):
_make_connector(url="http://10.0.0.1/api")
@patch("common.data_source.rest_api_connector.socket.getaddrinfo")
def test_public_url_passes(self, mock_dns):
"""A public IP should pass validation."""
mock_dns.return_value = [(2, 1, 6, "", ("93.184.216.34", 0))]
c = _make_connector(url="https://example.com/api")
assert c.url.startswith("https://")
def test_ftp_scheme_blocked(self):
"""ftp:// should be rejected."""
with pytest.raises(ConnectorValidationError, match="scheme"):
_make_connector(url="ftp://example.com/file")
def test_file_scheme_blocked(self):
"""file:// should be rejected."""
with pytest.raises(ConnectorValidationError, match="scheme"):
_make_connector(url="file:///etc/passwd")
# ===================================================================== #
# 3. Authentication setup #
# ===================================================================== #
class TestAuthSetup:
"""Test _build_auth produces the correct headers / auth objects."""
@patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))])
def test_auth_none(self, _dns):
"""auth_type=none should produce no auth headers."""
c = _make_connector(auth_type=AuthType.NONE)
c.load_credentials({})
assert c._auth_headers == {}
assert c._basic_auth is None
@patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))])
def test_api_key_header(self, _dns):
"""api_key_header should set the specified header."""
c = _make_connector(
auth_type=AuthType.API_KEY_HEADER,
auth_config={"header_name": "X-API-Key"},
)
c.load_credentials({"api_key": "secret123"})
assert c._auth_headers == {"X-API-Key": "secret123"}
@patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))])
def test_bearer_token(self, _dns):
"""bearer should set Authorization: Bearer <token>."""
c = _make_connector(auth_type=AuthType.BEARER)
c.load_credentials({"token": "tok_abc"})
assert c._auth_headers == {"Authorization": "Bearer tok_abc"}
@patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))])
def test_basic_auth(self, _dns):
"""basic should produce an HTTPBasicAuth object."""
c = _make_connector(auth_type=AuthType.BASIC)
c.load_credentials({"username": "user", "password": "pass"})
assert c._basic_auth is not None
assert c._basic_auth.username == "user"
assert c._basic_auth.password == "pass"
# ===================================================================== #
# 4. Field extraction #
# ===================================================================== #
class TestFieldExtraction:
"""Test _extract_field / _extract_field_values dot-notation paths."""
@patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))])
def setup_method(self, method, _dns=None):
with patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]):
self.connector = _make_connector()
def test_simple_field(self):
"""Top-level field extraction."""
assert self.connector._extract_field({"title": "Hello"}, "title") == "Hello"
def test_dot_notation_nested(self):
"""Dot-notation nested field."""
item = {"country": {"name": "Kuwait"}}
assert self.connector._extract_field(item, "country.name") == "Kuwait"
def test_array_wildcard(self):
"""Wildcard [*] returns all array elements."""
item = {"tags": [{"name": "A"}, {"name": "B"}]}
result = self.connector._extract_field(item, "tags[*].name")
assert result == ["A", "B"]
def test_missing_field_returns_none(self):
"""Missing field returns None."""
assert self.connector._extract_field({"a": 1}, "nonexistent") is None
def test_missing_field_with_default(self):
"""Missing field returns configured default value."""
with patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]):
c = _make_connector(field_default_values={"missing": "fallback"})
result = c._get_typed_field_value("missing", {"other": 1})
assert result == "fallback"
def test_deeply_nested_path(self):
"""Multi-level dot-notation path."""
item = {"a": {"b": {"c": {"d": 42}}}}
assert self.connector._extract_field(item, "a.b.c.d") == 42
# ===================================================================== #
# 5. Items array detection #
# ===================================================================== #
class TestItemsArrayDetection:
"""Test _extract_items auto-detection of the items array."""
@patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))])
def setup_method(self, method, _dns=None):
with patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]):
self.connector = _make_connector()
def test_items_key(self):
"""Detect 'items' key."""
resp = {"items": [{"id": 1}]}
assert self.connector._extract_items(resp) == [{"id": 1}]
def test_results_key(self):
"""Detect 'results' key."""
resp = {"results": [{"id": 2}]}
assert self.connector._extract_items(resp) == [{"id": 2}]
def test_data_key(self):
"""Detect 'data' key."""
resp = {"data": [{"id": 3}]}
assert self.connector._extract_items(resp) == [{"id": 3}]
def test_records_key(self):
"""Detect 'records' key."""
resp = {"records": [{"id": 4}]}
assert self.connector._extract_items(resp) == [{"id": 4}]
def test_custom_key_fallback(self):
"""Fall back to the first list value in the dict."""
resp = {"totalCount": 5, "stories": [{"id": 5}]}
assert self.connector._extract_items(resp) == [{"id": 5}]
def test_response_is_list(self):
"""Response that is directly a list."""
resp = [{"id": 6}, {"id": 7}]
assert self.connector._extract_items(resp) == [{"id": 6}, {"id": 7}]
def test_empty_response(self):
"""Empty dict returns empty list."""
assert self.connector._extract_items({}) == []
def test_no_list_in_response(self):
"""Dict with no list values returns empty list."""
assert self.connector._extract_items({"count": 0}) == []
# ===================================================================== #
# 6. HTML stripping #
# ===================================================================== #
class TestHTMLStripping:
"""Test the _strip_html static method."""
def test_basic_tag_removal(self):
"""Remove simple HTML tags."""
assert RestAPIConnector._strip_html("<p>Hello</p>") == "Hello"
def test_whitespace_collapsing(self):
"""Multiple whitespace chars collapse to single space."""
assert RestAPIConnector._strip_html("<p>Hello</p> <p>World</p>") == "Hello World"
def test_empty_string(self):
"""Empty input returns empty output."""
assert RestAPIConnector._strip_html("") == ""
def test_plain_text_passthrough(self):
"""Text without HTML passes through unchanged."""
assert RestAPIConnector._strip_html("Hello World") == "Hello World"
def test_nested_tags(self):
"""Nested HTML tags are all stripped."""
result = RestAPIConnector._strip_html("<div><p><b>Bold</b> text</p></div>")
assert result == "Bold text"
def test_html_with_attributes(self):
"""Tags with attributes are stripped."""
result = RestAPIConnector._strip_html('<a href="http://x.com">Link</a>')
assert result == "Link"
# ===================================================================== #
# 7. Document creation #
# ===================================================================== #
class TestDocumentCreation:
"""Test _item_to_document mapping."""
@patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))])
def setup_method(self, method, _dns=None):
with patch("common.data_source.rest_api_connector.socket.getaddrinfo",
return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]):
self.connector = _make_connector(
id_field="id",
content_fields=["title", "body"],
metadata_fields=["author"],
)
def test_document_id_from_configured_field(self):
"""Document ID uses the configured id_field."""
item = {"id": "abc", "title": "T", "body": "B", "author": "A"}
doc = self.connector._item_to_document(item)
assert doc.id is not None and len(doc.id) > 0
def test_semantic_identifier_from_first_content_field(self):
"""semantic_identifier comes from the first content field."""
item = {"id": "1", "title": "My Title", "body": "Body", "author": "A"}
doc = self.connector._item_to_document(item)
assert "My Title" in doc.semantic_identifier
def test_content_blob_contains_all_fields(self):
"""Blob should contain both content fields."""
item = {"id": "1", "title": "Title", "body": "Body text", "author": "A"}
doc = self.connector._item_to_document(item)
content = doc.blob.decode("utf-8")
assert "Title" in content
assert "Body text" in content
def test_metadata_populated(self):
"""Metadata dict is populated from configured metadata_fields."""
item = {"id": "1", "title": "T", "body": "B", "author": "Jane"}
doc = self.connector._item_to_document(item)
assert doc.metadata is not None
assert doc.metadata["author"] == "Jane"
def test_html_stripped_from_content(self):
"""HTML tags are removed from content fields."""
item = {"id": "1", "title": "T", "body": "<p>Clean</p>", "author": "A"}
doc = self.connector._item_to_document(item)
content = doc.blob.decode("utf-8")
assert "<p>" not in content
assert "Clean" in content
def test_extension_is_txt(self):
"""Document extension should be .txt."""
item = {"id": "1", "title": "T", "body": "B", "author": "A"}
doc = self.connector._item_to_document(item)
assert doc.extension == ".txt"
def test_missing_content_fields_graceful(self):
"""Missing content fields produce an empty blob gracefully."""
item = {"id": "1", "author": "A"}
doc = self.connector._item_to_document(item)
assert doc.blob == b""
# ===================================================================== #
# 8. Pagination behaviour #
# ===================================================================== #
class TestPaginationBehavior:
"""Test pagination iteration with mocked HTTP responses."""
def test_page_pagination_increments(self):
"""Page-based pagination should increment the page param."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
page1 = _mock_response({"items": [{"title": "A"}, {"title": "B"}]})
page2 = _mock_response({"items": []})
mock_rl.get.side_effect = [page1, page2]
c = _make_paged_connector()
items = list(c._iter_items())
assert len(items) == 2
assert mock_rl.get.call_count == 2
def test_offset_pagination_increments(self):
"""Offset-based pagination should increment offset by limit."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
page1 = _mock_response({"items": [{"title": "A"}]})
page2 = _mock_response({"items": []})
mock_rl.get.side_effect = [page1, page2]
c = _make_connector(
pagination_type=PaginationType.OFFSET,
pagination_config={
"offset_param": "offset",
"limit_param": "limit",
"limit": 10,
},
request_delay=0,
)
items = list(c._iter_items())
assert len(items) == 1
def test_stops_on_empty_results(self):
"""Pagination stops when empty items are returned."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
mock_rl.get.return_value = _mock_response({"items": []})
c = _make_paged_connector()
items = list(c._iter_items())
assert items == []
assert mock_rl.get.call_count == 1
def test_stops_when_fewer_items_than_page_size(self):
"""Pagination stops when fewer items than page_size are returned."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
page1 = _mock_response({"items": [{"title": "A"}]})
mock_rl.get.return_value = page1
c = _make_paged_connector(
pagination_config={"page_param": "page", "page_size": 10},
)
items = list(c._iter_items())
assert len(items) == 1
assert mock_rl.get.call_count == 1
def test_max_pages_cap(self):
"""Pagination respects the max_pages safety cap."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
mock_rl.get.return_value = _mock_response(
{"items": [{"title": "A"}, {"title": "B"}]}
)
c = _make_paged_connector(
max_pages=3,
pagination_config={"page_param": "page", "page_size": 2},
)
list(c._iter_items())
assert mock_rl.get.call_count == 3
def test_request_delay_applied(self):
"""request_delay should cause a sleep between pages."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
with patch("common.data_source.rest_api_connector.time.sleep") as mock_sleep:
page1 = _mock_response({"items": [{"title": "A"}, {"title": "B"}]})
page2 = _mock_response({"items": []})
mock_rl.get.side_effect = [page1, page2]
c = _make_paged_connector(
pagination_config={"page_param": "page", "page_size": 2},
)
c.request_delay = 1.5
list(c._iter_items())
mock_sleep.assert_called_once_with(1.5)
# ===================================================================== #
# 9. Non-retriable HTTP errors #
# ===================================================================== #
class TestNonRetriableErrors:
"""Test that HTTP errors are classified correctly in _fetch_page."""
def test_401_raises_credential_error(self):
"""401 should raise ConnectorMissingCredentialError immediately."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
mock_rl.get.return_value = _mock_response({}, status_code=401)
c = _make_connector(request_delay=0)
c.load_credentials({})
with pytest.raises(ConnectorMissingCredentialError):
c._fetch_page({})
def test_403_raises_credential_error(self):
"""403 should raise ConnectorMissingCredentialError immediately."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
mock_rl.get.return_value = _mock_response({}, status_code=403)
c = _make_connector(request_delay=0)
c.load_credentials({})
with pytest.raises(ConnectorMissingCredentialError):
c._fetch_page({})
def test_404_raises_validation_error(self):
"""404 should raise ConnectorValidationError (no retry)."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
mock_rl.get.return_value = _mock_response({}, status_code=404)
c = _make_connector(request_delay=0)
c.load_credentials({})
with pytest.raises(ConnectorValidationError, match="non-retriable"):
c._fetch_page({})
def test_400_raises_validation_error(self):
"""400 should raise ConnectorValidationError (no retry)."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
mock_rl.get.return_value = _mock_response({}, status_code=400)
c = _make_connector(request_delay=0)
c.load_credentials({})
with pytest.raises(ConnectorValidationError, match="non-retriable"):
c._fetch_page({})
def test_500_triggers_retry(self):
"""500 should raise HTTPError (which the retry decorator catches)."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
mock_rl.get.return_value = _mock_response({}, status_code=500)
c = _make_connector(request_delay=0)
c.load_credentials({})
with pytest.raises(requests.HTTPError):
c._fetch_page({})
def test_429_triggers_retry(self):
"""429 should raise HTTPError (retriable, not ConnectorValidationError)."""
with _mocked_rest_api_requests_and_dns() as mock_rl:
mock_rl.get.return_value = _mock_response({}, status_code=429)
c = _make_connector(request_delay=0)
c.load_credentials({})
with pytest.raises(requests.HTTPError):
c._fetch_page({})