Skip to content

test: fix flaky tests by replacing hardcoded sleeps with polls #263

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

Draft
imor wants to merge 4 commits into
base: master
Choose a base branch
from
Draft

test: fix flaky tests by replacing hardcoded sleeps with polls #263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cf43999
test: fix flaky tests by replacing hardcoded sleeps with polls
imor Jul 3, 2026
1163ccc
fix: another race condition causing flakiness
imor Jul 3, 2026
959d9c8
tests: make tests cleanup at exit
imor Jul 7, 2026
8fce899
tests: more tests cleanup at exit
imor Jul 7, 2026
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
23 changes: 23 additions & 0 deletions test/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import time

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
Expand Down Expand Up @@ -41,3 +43,24 @@ def autocommit_sess(engine):
session = Session(ac_engine)

yield session


@pytest.fixture(scope="function")
def wait_until():
"""Poll `fetch()` until `predicate(result)` is true, instead of a fixed sleep.

Returns the last fetched value on success, so callers can use it directly.
Raises AssertionError with the last observed value if `timeout` is exceeded.
"""
def _wait_until(fetch, predicate, timeout=10, interval=0.1, description="condition"):
deadline = time.time() + timeout
result = None
while time.time() < deadline:
result = fetch()
if predicate(result):
return result
time.sleep(interval)
raise AssertionError(
f"Timed out after {timeout}s waiting for {description} (last value: {result!r})"
)
return _wait_until
196 changes: 105 additions & 91 deletions test/test_http_requests_deleted_after_ttl.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,65 +1,73 @@
import time

import pytest
from sqlalchemy import text

def test_http_responses_deleted_after_ttl(sess, autocommit_sess):
def test_http_responses_deleted_after_ttl(sess, autocommit_sess, wait_until):
"""Check that http responses will be deleted when they reach their ttl, not immediately but when the worker wakes again"""

autocommit_sess.execute(text("alter system set pg_net.ttl to '1 second'"))
autocommit_sess.execute(text("select net.worker_restart()"))
autocommit_sess.execute(text("select net.wait_until_running()"))

# Create a request
(request_id,) = sess.execute(text(
"""
select net.http_get(
'http://localhost:8080/anything'
);
"""
)).fetchone()

# Commit so background worker can start
sess.commit()

# Confirm that the request was retrievable
response = sess.execute(
text(
try:
# Create a request
(request_id,) = sess.execute(text(
"""
select * from net._http_collect_response(:request_id, async:=false);
"""
),
{"request_id": request_id},
).fetchone()
assert response[0] == "SUCCESS"
select net.http_get(
'http://localhost:8080/anything'
);
"""
)).fetchone()

# Sleep until after request should have been deleted
time.sleep(1.1)
# Commit so background worker can start
sess.commit()

# Wake the worker manually, under normal operation this will happen when new requests are received
sess.execute(text("select net.wake()"))
# Confirm that the request was retrievable
response = sess.execute(
text(
"""
select * from net._http_collect_response(:request_id, async:=false);
"""
),
{"request_id": request_id},
).fetchone()
assert response[0] == "SUCCESS"

# Wait until the request's age exceeds the ttl
wait_until(
fetch=lambda: sess.execute(
text("select extract(epoch from clock_timestamp() - created) from net._http_response where id = :request_id"),
{"request_id": request_id},
).scalar(),
predicate=lambda age: age is None or age >= 1.0,
timeout=5,
description=f"request {request_id} to exceed the ttl",
)

sess.commit() # commit so worker wakes
# Wake the worker manually, under normal operation this will happen when new requests are received
sess.execute(text("select net.wake()"))

time.sleep(0.1) # wait for deletion
sess.commit() # commit so worker wakes

# Ensure the response is now empty
(count,) = sess.execute(
text(
"""
select count(*) from net._http_response where id = :request_id;
"""
),
{"request_id": request_id},
).fetchone()
assert count == 0

autocommit_sess.execute(text("alter system reset pg_net.ttl"))
autocommit_sess.execute(text("select net.worker_restart()"))
autocommit_sess.execute(text("select net.wait_until_running()"))
# Ensure the response is now empty
(count,) = wait_until(
fetch=lambda: sess.execute(
text("select count(*) from net._http_response where id = :request_id"),
{"request_id": request_id},
).fetchone(),
predicate=lambda r: r[0] == 0,
timeout=5,
description=f"deletion of expired response {request_id}",
)
assert count == 0
finally:
# Always restore ttl, even on assertion failure - it's set via ALTER
# SYSTEM so it would otherwise leak into and destabilize later tests.
autocommit_sess.execute(text("alter system reset pg_net.ttl"))
autocommit_sess.execute(text("select net.worker_restart()"))
autocommit_sess.execute(text("select net.wait_until_running()"))


def test_http_responses_will_complete_deletion(sess, autocommit_sess):
def test_http_responses_will_complete_deletion(sess, autocommit_sess, wait_until):
"""Check that http responses will keep being deleted until completion despite no new requests coming"""

(request_id,) = sess.execute(text(
Expand Down Expand Up @@ -95,42 +103,48 @@ def test_http_responses_will_complete_deletion(sess, autocommit_sess):
autocommit_sess.execute(text("alter system set pg_net.batch_size to 2;"))
autocommit_sess.execute(text("select pg_reload_conf();"))

# wait for ttl
time.sleep(1)

# Wake the worker manually, under normal operation this will happen when new requests are received
sess.execute(text("select net.wake()"))
sess.commit() # commit so worker wakes

time.sleep(0.1)

(count,) = sess.execute(
text(
"""
select count(*) from net._http_response
"""
try:
# wait for ttl
wait_until(
fetch=lambda: sess.execute(
text("select extract(epoch from clock_timestamp() - created) from net._http_response where id = :request_id"),
{"request_id": request_id},
).scalar(),
predicate=lambda age: age is None or age >= 1.0,
timeout=5,
description=f"request {request_id} to exceed the ttl",
)
).fetchone()
assert count == 2

# wait for another batch
time.sleep(1.1)
# Wake the worker manually, under normal operation this will happen when new requests are received
sess.execute(text("select net.wake()"))
sess.commit() # commit so worker wakes

(count,) = sess.execute(
text(
"""
select count(*) from net._http_response
"""
# only one batch (batch_size=2) should be deleted by this wake
(count,) = wait_until(
fetch=lambda: sess.execute(text("select count(*) from net._http_response")).fetchone(),
predicate=lambda r: r[0] <= 2,
timeout=2,
description="first batch of expired responses to be deleted",
)
).fetchone()
assert count == 0

autocommit_sess.execute(text("alter system reset pg_net.ttl"))
autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
autocommit_sess.execute(text("select pg_reload_conf();"))
assert count == 2

# wait for another batch
(count,) = wait_until(
fetch=lambda: sess.execute(text("select count(*) from net._http_response")).fetchone(),
predicate=lambda r: r[0] == 0,
timeout=5,
description="remaining batch of expired responses to be deleted",
)
assert count == 0
finally:
# Always restore ttl/batch_size, even on assertion failure - they're set
# via ALTER SYSTEM so they would otherwise leak into later tests.
autocommit_sess.execute(text("alter system reset pg_net.ttl"))
autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
autocommit_sess.execute(text("select pg_reload_conf();"))


def test_http_responses_will_delete_despite_restart(sess, autocommit_sess):
def test_http_responses_will_delete_despite_restart(sess, autocommit_sess, wait_until):
"""Check that http responses will keep being despite no new requests coming" and despite restart"""

(request_id,) = sess.execute(text(
Expand Down Expand Up @@ -168,20 +182,20 @@ def test_http_responses_will_delete_despite_restart(sess, autocommit_sess):
autocommit_sess.execute(text("select net.worker_restart()"))
autocommit_sess.execute(text("select net.wait_until_running()"))

# wait for ttl
time.sleep(1.1)

(count,) = sess.execute(
text(
"""
select count(*) from net._http_response
"""
try:
# wait for ttl to elapse and the worker to clean up all expired responses
# (multiple worker cycles may be needed since batch_size=2 but there are 4 rows)
(count,) = wait_until(
fetch=lambda: sess.execute(text("select count(*) from net._http_response")).fetchone(),
predicate=lambda r: r[0] == 0,
timeout=20,
description="all expired responses to be deleted after restart",
)
).fetchone()
assert count == 0

# reset
autocommit_sess.execute(text("alter system reset pg_net.ttl"))
autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
autocommit_sess.execute(text("select net.worker_restart()"))
autocommit_sess.execute(text("select net.wait_until_running()"))
assert count == 0
finally:
# Always restore ttl/batch_size, even on assertion failure - they're set
# via ALTER SYSTEM so they would otherwise leak into later tests.
autocommit_sess.execute(text("alter system reset pg_net.ttl"))
autocommit_sess.execute(text("alter system reset pg_net.batch_size"))
autocommit_sess.execute(text("select net.worker_restart()"))
autocommit_sess.execute(text("select net.wait_until_running()"))
Loading
Loading