Skip to content

test(core): add regression tests for incorrect pagination cursor extraction #17

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
niklashigi merged 14 commits into from
Dec 18, 2025
Merged

test(core): add regression tests for incorrect pagination cursor extraction #17

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
a422ca3
test(core): port general behavior tests from TypeScript SDK
niklashigi Dec 16, 2025
1d92975
Run tests in CI
niklashigi Dec 16, 2025
23abbaf
Fix snapshots
niklashigi Dec 16, 2025
0b81e4a
Add Mise tasks
niklashigi Dec 16, 2025
05f6331
Add Python to Mise config
niklashigi Dec 16, 2025
ac71f83
Unify Python versions
niklashigi Dec 16, 2025
072e8f7
Add `AGENTS.md` to explain testing
niklashigi Dec 16, 2025
959399e
Deslop
niklashigi Dec 16, 2025
45a903b
Add additional dev deps to Speakeasy config
niklashigi Dec 16, 2025
3f1b6ea
Clean up #1
niklashigi Dec 18, 2025
cf19d99
Cleanup #2
niklashigi Dec 18, 2025
38aca2e
Merge branch 'main' into test/core/clean-up-structure
niklashigi Dec 18, 2025
f6bee8a
test(core): add regression tests for incorrect pagination cursor extr…
niklashigi Dec 18, 2025
a9830a9
Merge branch 'main' into test/core/pagination-cursor-regression
niklashigi Dec 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,12 @@ def create_response(request: httpx.Request) -> httpx.Response:
# Read body for non-GET requests
request_body = None
if request.method != "GET" and request.content:
request_body = json.loads(request.content.decode())
try:
# Try to parse as JSON first
request_body = json.loads(request.content.decode())
except (json.JSONDecodeError, UnicodeDecodeError):
# Fall back to raw string if not valid JSON
request_body = request.content.decode()
Copy link

@cursor cursor bot Dec 18, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: Exception handler will re-raise same UnicodeDecodeError

The exception handler catches UnicodeDecodeError from the decode() call on line 95, but then the fallback on line 98 calls request.content.decode() again. If the original error was a UnicodeDecodeError, this second decode call will raise the same exception, which won't be caught. The fallback needs to handle the case where the content cannot be decoded as UTF-8, such as by specifying an error handler (e.g., decode(errors='replace')) or storing raw bytes.

Fix in Cursor Fix in Web


captured = CapturedRequest(
method=request.method,
Expand Down
355 changes: 355 additions & 0 deletions tests/test_pagination.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,355 @@
"""Tests for pagination behavior."""

from datetime import datetime
from tests.conftest import MockContext


class TestPaginationBehavior:
"""Test pagination behavior."""

def test_should_iterate_through_multiple_pages(self):
"""Test that SDK can iterate through multiple pages of results."""
ctx = MockContext()

# Mock 3 pages of results
ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag1",
"remote_id": None,
"name": "Tag 1",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
{
"id": "tag2",
"remote_id": None,
"name": "Tag 2",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": "cursor_page2",
},
},
},
)

ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag3",
"remote_id": None,
"name": "Tag 3",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
{
"id": "tag4",
"remote_id": None,
"name": "Tag 4",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": "cursor_page3",
},
},
},
)

ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag5",
"remote_id": None,
"name": "Tag 5",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": None,
},
},
},
)

page = ctx.kombo.ats.get_tags()
all_results = []

# Iterate through all pages
while page is not None:
all_results.extend(page.result.data.results)
page = page.next()

# Verify all 5 tags were collected
assert len(all_results) == 5
assert [tag.id for tag in all_results] == ["tag1", "tag2", "tag3", "tag4", "tag5"]

# Verify 3 HTTP requests were made
requests = ctx.get_requests()
assert len(requests) == 3

def test_should_pass_cursor_parameter_to_subsequent_requests(self):
"""Test that cursor parameter is passed to subsequent paginated requests."""
ctx = MockContext()

ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag1",
"remote_id": None,
"name": "Tag 1",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": "test_cursor_abc123",
},
},
},
)

ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag2",
"remote_id": None,
"name": "Tag 2",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": None,
},
},
},
)

page = ctx.kombo.ats.get_tags()
# Iterate through all pages
while page is not None:
page = page.next()

requests = ctx.get_requests()
assert len(requests) == 2

# First request should NOT include cursor
assert "cursor=" not in requests[0].path

# Second request SHOULD include cursor
assert "cursor=test_cursor_abc123" in requests[1].path

def test_should_stop_pagination_when_next_is_null(self):
"""Test that pagination stops when next cursor is null."""
ctx = MockContext()

ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag1",
"remote_id": None,
"name": "Tag 1",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
{
"id": "tag2",
"remote_id": None,
"name": "Tag 2",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": None,
},
},
},
)

page = ctx.kombo.ats.get_tags()
page_count = []

while page is not None:
page_count.append(1)
page = page.next()

# Verify only 1 page was returned
assert len(page_count) == 1

# Verify only 1 HTTP request was made
requests = ctx.get_requests()
assert len(requests) == 1

def test_should_preserve_query_parameters_across_paginated_requests(self):
"""Test that original query parameters are preserved across paginated requests."""
ctx = MockContext()

ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag1",
"remote_id": None,
"name": "Tag 1",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": "cursor_for_page2",
},
},
},
)

ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag2",
"remote_id": None,
"name": "Tag 2",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": None,
},
},
},
)

page = ctx.kombo.ats.get_tags(
updated_after=datetime(2024, 1, 1, 0, 0, 0)
)

# Iterate through all pages
while page is not None:
page = page.next()

requests = ctx.get_requests()
assert len(requests) == 2

# Both requests should include the original query parameters
# Check that updated_after parameter is present (URL encoded)
assert "updated_after=2024-01-01T00%3A00%3A00" in requests[0].path
assert "cursor=" not in requests[0].path

assert "updated_after=2024-01-01T00%3A00%3A00" in requests[1].path
assert "cursor=cursor_for_page2" in requests[1].path

def test_should_support_manual_pagination_with_next(self):
"""Test that manual pagination works by calling next() method."""
ctx = MockContext()

ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag1",
"remote_id": None,
"name": "Tag 1",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": "manual_cursor_xyz",
},
},
},
)

ctx.mock_endpoint(
method="GET",
path="/v1/ats/tags",
response={
"body": {
"status": "success",
"data": {
"results": [
{
"id": "tag2",
"remote_id": None,
"name": "Tag 2",
"changed_at": "2024-01-01T00:00:00.000Z",
"remote_deleted_at": None,
},
],
"next": None,
},
},
},
)

page1 = ctx.kombo.ats.get_tags()

# Verify first page was fetched
assert page1.result.data.results is not None
assert len(page1.result.data.results) == 1

# Manually call next()
page2_result = page1.next()

# Verify second page was fetched (should not be null if cursor was read correctly)
# This will fail if cursor extraction bug exists
assert page2_result is not None
if page2_result:
assert len(page2_result.result.data.results) == 1
assert page2_result.result.data.results[0].id == "tag2"

# Verify 2 HTTP requests were made
requests = ctx.get_requests()
assert len(requests) == 2
assert "cursor=manual_cursor_xyz" in requests[1].path

Loading