Skip to content
Merged
Changes from all commits
Commits
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
24 changes: 22 additions & 2 deletions kernel/tests/test_postgresql_audit_operations.py
Original file line number Diff line number Diff line change
Expand Up @@ -527,9 +527,29 @@ def test_overflow_observation_returns_one_oldest_closeable_bucket(
""",
(newer,),
).fetchone()
assert control.execute(
observed_before = control.execute(
"SELECT pg_catalog.clock_timestamp()"
).fetchone()[0]
remaining = control.execute(
"""
SELECT * FROM
ofarm_security.observe_next_closeable_overflow_bucket()
"""
).fetchall() == []
).fetchall()
observed_after = control.execute(
"SELECT pg_catalog.clock_timestamp()"
).fetchone()[0]
timing = (current_bucket, observed_before, observed_after)
assert observed_before <= observed_after, timing
expires_at = current_bucket + timedelta(minutes=1)
expected_remaining = [
("REQUEST_ROUTER_BOUNDARY_V1", "REQUEST_ROUTER", current_bucket)
]
# The setup bucket can expire while the earlier buckets are closed.
if observed_after < expires_at:
assert remaining == [], timing
elif observed_before >= expires_at:
assert remaining == expected_remaining, timing
else:
# Expiry fell inside the call; either exact result is legitimate.
assert remaining in ([], expected_remaining), timing
Loading