Skip to content

[py] Move test_output_buffer_size_limit.py to runtime tests #6574

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
ryzhyk merged 1 commit into from
Jul 2, 2026
Merged

[py] Move test_output_buffer_size_limit.py to runtime tests #6574

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
31 changes: 13 additions & 18 deletions ...platform/test_output_buffer_size_limit.py → .../runtime/test_output_buffer_size_limit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,6 @@
from tests import TEST_CLIENT
from tests.utils import DeltaTestLocation, wait_for_condition

from .helper import gen_pipeline_name

# Default value of ``max_output_buffer_size_records``. The buffer must flush
# once it grows past this many records, even without a time limit.
_DEFAULT_MAX_OUTPUT_BUFFER_SIZE_RECORDS = 10_000_000
Expand All @@ -36,7 +34,6 @@
_NUM_RECORDS = 12_000_000


@gen_pipeline_name
def test_output_buffer_flushes_at_default_size_limit(pipeline_name):
"""A buffered Delta sink with no time limit still flushes at 10M records."""

Expand Down Expand Up @@ -98,18 +95,16 @@ def test_output_buffer_flushes_at_default_size_limit(pipeline_name):

pipeline.start()

try:
# The buffer flushes once it crosses the 10M default size cap, pushing
# those records through the Delta sink and advancing the completed
# count past the cap.
wait_for_condition(
"completed-record count advances past the default buffer size cap",
lambda: (
(pipeline.stats().global_metrics.total_completed_records or 0)
>= _DEFAULT_MAX_OUTPUT_BUFFER_SIZE_RECORDS
),
timeout_s=600.0,
poll_interval_s=2.0,
)
finally:
pipeline.stop(force=True)
# The buffer flushes once it crosses the 10M default size cap, pushing
# those records through the Delta sink and advancing the completed count
# past the cap. On timeout the test raises without stopping the pipeline,
# so it is left running on the persistent runtime instance for inspection.
wait_for_condition(
"completed-record count advances past the default buffer size cap",
lambda: (
(pipeline.stats().global_metrics.total_completed_records or 0)
>= _DEFAULT_MAX_OUTPUT_BUFFER_SIZE_RECORDS
),
timeout_s=600.0,
poll_interval_s=2.0,
)